adk_graph/
child.rs

1//! Invoking a node from inside another node's body.
2//!
3//! Declared edges express a topology decided before the run. A supervisor often
4//! cannot: how many workers to start, and in what order, depends on what the
5//! first one found. Both adk-python and adk-go answer this the same way, and
6//! neither mutates the graph — a node body calls other nodes directly and awaits
7//! their output.
8//!
9//! # Identity and replay
10//!
11//! Each invocation is recorded under a path, `<parent>/<child>@<run_id>`. When a
12//! parent is re-executed after a resume it runs from the top, so without a record
13//! every child would run again. A path already in the ledger returns its recorded
14//! output instead.
15//!
16//! Only a successful invocation is recorded. A child that failed or interrupted
17//! has to run again, because its work did not finish.
18//!
19//! The default `run_id` counts invocations of that child name within one
20//! activation, so it is stable only while the parent runs once. **A parent that
21//! may be resumed must supply its own `run_id`**, or the counter will hand the
22//! same identity to a different unit of work. adk-go documents the same trap.
23
24use std::collections::HashMap;
25use std::sync::{Arc, Mutex};
26
27use serde_json::Value;
28
29use crate::error::{GraphError, Result};
30use crate::node::{Node, NodeContext};
31
32/// How one child invocation behaves.
33#[derive(Debug, Clone, Default)]
34pub struct RunNodeOptions {
35    /// Stable identity for replay.
36    ///
37    /// Defaults to a counter over invocations of this child name within the
38    /// current activation. Supply one when the parent may be resumed.
39    pub run_id: Option<String>,
40}
41
42impl RunNodeOptions {
43    /// Options with an explicit run id.
44    pub fn with_run_id(run_id: impl Into<String>) -> Self {
45        Self { run_id: Some(run_id.into()) }
46    }
47}
48
49/// The machinery a [`NodeContext`] needs to invoke another node.
50///
51/// Built by the executor for each node it runs, so a node body reaches only the
52/// graph it belongs to.
53pub(crate) struct ChildInvoker {
54    /// Every node in the graph, including any reachable by no edge.
55    nodes: HashMap<String, Arc<dyn Node>>,
56    /// Outputs already recorded, keyed by child path. Shared with the executor so
57    /// the run's checkpoint carries them.
58    ledger: Arc<Mutex<HashMap<String, Value>>>,
59    /// The invoking node's path, which prefixes its children's.
60    parent_path: String,
61    /// Invocations so far per child name, for the default run id.
62    counters: Mutex<HashMap<String, u32>>,
63}
64
65impl ChildInvoker {
66    pub(crate) fn new(
67        nodes: HashMap<String, Arc<dyn Node>>,
68        ledger: Arc<Mutex<HashMap<String, Value>>>,
69        parent_path: String,
70    ) -> Self {
71        Self { nodes, ledger, parent_path, counters: Mutex::new(HashMap::new()) }
72    }
73
74    /// The path this invocation is recorded under.
75    fn path_for(&self, child: &str, options: &RunNodeOptions) -> String {
76        let run_id = match &options.run_id {
77            Some(id) => id.clone(),
78            None => {
79                let mut counters = self.counters.lock().expect("child counters");
80                let count = counters.entry(child.to_string()).or_insert(0);
81                *count += 1;
82                count.to_string()
83            }
84        };
85        format!("{}/{}@{}", self.parent_path, child, run_id)
86    }
87
88    /// Invoke a child and return its updates as one value.
89    pub(crate) async fn run(
90        &self,
91        child: &str,
92        input: Value,
93        options: RunNodeOptions,
94        parent: &NodeContext,
95    ) -> Result<Value> {
96        let path = self.path_for(child, &options);
97
98        // A child that already completed under this identity is not run again.
99        if let Some(recorded) = self.ledger.lock().expect("child ledger").get(&path) {
100            tracing::debug!(path = %path, "child already completed, serving its recorded output");
101            return Ok(recorded.clone());
102        }
103
104        let node = self
105            .nodes
106            .get(child)
107            .ok_or_else(|| GraphError::NodeNotFound(child.to_string()))?
108            .clone();
109
110        // The child sees the parent's state, with its input merged over it.
111        let mut state = parent.state.clone();
112        if let Value::Object(map) = input {
113            for (key, value) in map {
114                state.insert(key, value);
115            }
116        }
117        let child_ctx = NodeContext::new(state, parent.config.clone(), parent.step);
118
119        let output = node.execute(&child_ctx).await?;
120
121        // An interrupt is not a result: the child has not finished, so nothing is
122        // recorded and it runs again on resume.
123        if let Some(interrupt) = output.interrupt {
124            return Err(GraphError::Interrupted(Box::new(
125                crate::error::InterruptedExecution::new(
126                    parent.config.thread_id.clone(),
127                    String::new(),
128                    interrupt,
129                    child_ctx.state.clone(),
130                    parent.step,
131                ),
132            )));
133        }
134
135        let value = Value::Object(output.updates.into_iter().collect());
136        self.ledger.lock().expect("child ledger").insert(path.clone(), value.clone());
137        tracing::debug!(path = %path, "child completed");
138        Ok(value)
139    }
140}
adk_graph/child.rs

adk_graph/
child.rs
