Skip to content

Commit 6bf67ac

Browse files
authored
Add optional whole-run wall-clock deadline to graph execution (#29)
CompiledGraph::with_run_deadline(d) bounds the whole run by a wall clock, checked at every super-step boundary. When the elapsed run time first reaches the deadline, the run stops *between* super-steps with TinyAgentsError::Timeout, leaving the last committed boundary checkpoint intact and resumable. This is the durable alternative to wrapping run() in an external tokio::time::timeout, which aborts mid-super-step and cannot leave a clean checkpoint. The deadline bounds scheduling, not a single in-flight node — pair it with with_node_timeout to also bound individual handlers. Tests: deadline trips between super-steps (Timeout); a run that finishes in time is unaffected; a deadline trip on a checkpointed thread leaves the last boundary checkpoint intact and the run resumes to completion. fmt + clippy -D warnings clean; full suite green. Docs: builder.md.
1 parent 3e81e49 commit 6bf67ac

5 files changed

Lines changed: 151 additions & 0 deletions

File tree

‎docs/modules/graph/builder.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,15 @@ parallel superstep (the active set runs in chunks of at most `n`).
4343
`with_node_timeout(d)` fails the run with `TinyAgentsError::Timeout` if any node
4444
handler does not resolve within `d`.
4545

46+
`CompiledGraph::with_run_deadline(d)` bounds the *whole run* by a wall-clock
47+
`d`, checked at every super-step boundary: when the elapsed run time first
48+
reaches `d` the run stops *between* super-steps with `TinyAgentsError::Timeout`,
49+
leaving the last committed boundary checkpoint intact and resumable. Prefer this
50+
over wrapping `run` in an external `tokio::time::timeout`, which aborts
51+
mid-super-step and cannot leave a clean checkpoint. It bounds scheduling, not a
52+
single in-flight node — pair it with `with_node_timeout` to also bound
53+
individual handlers.
54+
4655
Compile-time options:
4756

4857
```rust

‎src/graph/compiled/executor.rs‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -418,6 +418,23 @@ where
418418
.await;
419419
return Err(err);
420420
}
421+
// Whole-run wall-clock deadline: stop *between* super-steps once the
422+
// elapsed run time reaches it, leaving the last committed boundary
423+
// checkpoint intact (unlike an external `tokio::time::timeout`, which
424+
// aborts mid-super-step and cannot). The already-completed super-steps
425+
// and their checkpoints are preserved; the run fails with `Timeout`.
426+
if let Some(deadline) = self.run_deadline {
427+
let elapsed = started_at.elapsed().unwrap_or_default();
428+
if elapsed >= deadline {
429+
let err = TinyAgentsError::Timeout(format!(
430+
"graph run exceeded its {deadline:?} deadline after {steps} super-step(s) \
431+
({elapsed:?} elapsed)"
432+
));
433+
self.fail_run(&run_id, &thread_id, started_at, steps, &err, None)
434+
.await;
435+
return Err(err);
436+
}
437+
}
421438
// Node-loop recursion: enforce `max_visits_per_node` per activation.
422439
for activation in &active {
423440
if let Err(err) = recursion.record_node_visit(&mut node_visits, &activation.node) {

‎src/graph/compiled/mod.rs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -277,6 +277,7 @@ impl<State, Update> CompiledGraph<State, Update> {
277277
parallel,
278278
max_concurrency,
279279
node_timeout,
280+
run_deadline: None,
280281
durability: crate::graph::checkpoint::DurabilityMode::default(),
281282
node_retry: None,
282283
}
@@ -352,6 +353,26 @@ impl<State, Update> CompiledGraph<State, Update> {
352353
self
353354
}
354355

356+
/// Bounds the whole run by a wall-clock `deadline`, checked at every
357+
/// super-step boundary.
358+
///
359+
/// Unlike wrapping [`run`](Self::run) in an external
360+
/// [`tokio::time::timeout`] — which aborts mid-super-step and cannot leave a
361+
/// clean checkpoint — this stops the run *between* super-steps: when the
362+
/// elapsed run time first reaches `deadline`, the run fails with
363+
/// [`TinyAgentsError::Timeout`] and (on a checkpointed thread) the last
364+
/// committed boundary checkpoint stays intact, so the run can be resumed or
365+
/// inspected rather than lost.
366+
///
367+
/// The deadline bounds *scheduling*, not a single in-flight node: a
368+
/// long-running node still runs to completion within its super-step (bound
369+
/// it independently with a per-node timeout). `None` (the default) imposes
370+
/// no deadline.
371+
pub fn with_run_deadline(mut self, deadline: std::time::Duration) -> Self {
372+
self.run_deadline = Some(deadline);
373+
self
374+
}
375+
355376
/// Sets the checkpoint namespace (used by subgraph wrappers).
356377
pub fn with_namespace(mut self, namespace: Vec<String>) -> Self {
357378
self.namespace = namespace;

‎src/graph/compiled/test.rs‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1861,6 +1861,103 @@ async fn node_timeout_fails_slow_handler() {
18611861
assert!(matches!(err, TinyAgentsError::Timeout(_)));
18621862
}
18631863

1864+
// ── Whole-run wall-clock deadline ────────────────────────────────────────────
1865+
1866+
/// A per-run deadline stops the run *between* super-steps once the elapsed run
1867+
/// time reaches it, surfacing [`TinyAgentsError::Timeout`].
1868+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1869+
async fn run_deadline_stops_between_supersteps() {
1870+
let graph = GraphBuilder::<i32, i32>::overwrite()
1871+
.add_node("a", |s: i32, _c: NodeContext| async move {
1872+
tokio::time::sleep(Duration::from_millis(40)).await;
1873+
Ok(NodeResult::Update(s + 1))
1874+
})
1875+
.add_node("b", |s: i32, _c: NodeContext| async move {
1876+
Ok(NodeResult::Update(s + 1))
1877+
})
1878+
.set_entry("a")
1879+
.add_edge("a", "b")
1880+
.set_finish("b")
1881+
.compile()
1882+
.unwrap()
1883+
.with_run_deadline(Duration::from_millis(20));
1884+
1885+
// The first boundary (elapsed ~0) admits node `a`; the next boundary
1886+
// (elapsed ~40ms ≥ 20ms) trips the deadline before `b` ever runs.
1887+
let err = graph.run(0).await.unwrap_err();
1888+
assert!(matches!(err, TinyAgentsError::Timeout(_)), "got {err:?}");
1889+
}
1890+
1891+
/// A run that finishes within its deadline is unaffected — no false trip.
1892+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1893+
async fn run_deadline_allows_a_run_that_finishes_in_time() {
1894+
let graph = GraphBuilder::<i32, i32>::overwrite()
1895+
.add_node("a", |s: i32, _c: NodeContext| async move {
1896+
Ok(NodeResult::Update(s + 1))
1897+
})
1898+
.add_node("b", |s: i32, _c: NodeContext| async move {
1899+
Ok(NodeResult::Update(s + 1))
1900+
})
1901+
.set_entry("a")
1902+
.add_edge("a", "b")
1903+
.set_finish("b")
1904+
.compile()
1905+
.unwrap()
1906+
.with_run_deadline(Duration::from_secs(30));
1907+
1908+
let run = graph.run(0).await.unwrap();
1909+
assert_eq!(run.state, 2);
1910+
}
1911+
1912+
/// On a checkpointed thread, a deadline trip leaves the last committed boundary
1913+
/// checkpoint intact — so the run can be resumed to completion rather than lost
1914+
/// (the durability win over an external `tokio::time::timeout`).
1915+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1916+
async fn run_deadline_leaves_last_checkpoint_resumable() {
1917+
let cp = Arc::new(InMemoryCheckpointer::<i32>::new());
1918+
let topology = || {
1919+
GraphBuilder::<i32, i32>::overwrite()
1920+
.add_node("a", |s: i32, _c: NodeContext| async move {
1921+
tokio::time::sleep(Duration::from_millis(40)).await;
1922+
Ok(NodeResult::Update(s + 1))
1923+
})
1924+
.add_node("b", |s: i32, _c: NodeContext| async move {
1925+
Ok(NodeResult::Update(s + 1))
1926+
})
1927+
.add_node("c", |s: i32, _c: NodeContext| async move {
1928+
Ok(NodeResult::Update(s + 1))
1929+
})
1930+
.set_entry("a")
1931+
.add_edge("a", "b")
1932+
.add_edge("b", "c")
1933+
.set_finish("c")
1934+
.compile()
1935+
.unwrap()
1936+
};
1937+
1938+
// Trips after `a`'s boundary (state=1, next=[b]) but before `b` runs.
1939+
let deadlined = topology()
1940+
.with_checkpointer(cp.clone())
1941+
.with_run_deadline(Duration::from_millis(20));
1942+
let err = deadlined.run_with_thread("t", 0).await.unwrap_err();
1943+
assert!(matches!(err, TinyAgentsError::Timeout(_)), "got {err:?}");
1944+
1945+
// The boundary checkpoint from the completed super-step survived intact.
1946+
let list = cp.list("t").await.unwrap();
1947+
assert!(
1948+
!list.is_empty(),
1949+
"the pre-deadline boundary checkpoint is intact"
1950+
);
1951+
1952+
// Resuming (no deadline) continues from that checkpoint to completion.
1953+
let resumed = topology().with_checkpointer(cp.clone());
1954+
let run = resumed
1955+
.resume("t", Command::resume(json!(null)))
1956+
.await
1957+
.unwrap();
1958+
assert_eq!(run.state, 3, "resume ran the remaining super-steps b and c");
1959+
}
1960+
18641961
// ── Network resilience: node retry + resumable failures ──────────────────────
18651962

18661963
/// A single-node graph whose handler fails (with a retryable model error) the

‎src/graph/compiled/types.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,12 @@ pub struct CompiledGraph<State, Update> {
6464
pub(crate) max_concurrency: Option<usize>,
6565
/// Default per-node handler timeout (`None` = no timeout).
6666
pub(crate) node_timeout: Option<std::time::Duration>,
67+
/// Optional whole-run wall-clock deadline (`None` = no deadline). Checked at
68+
/// every super-step boundary: when the elapsed run time reaches it the run
69+
/// stops *between* super-steps with a [`TinyAgentsError::Timeout`], leaving
70+
/// the last committed boundary checkpoint intact and resumable. Configured
71+
/// via [`CompiledGraph::with_run_deadline`](crate::graph::CompiledGraph::with_run_deadline).
72+
pub(crate) run_deadline: Option<std::time::Duration>,
6773
/// When checkpoints are persisted relative to execution (default
6874
/// [`DurabilityMode::Sync`]).
6975
pub(crate) durability: DurabilityMode,
@@ -115,6 +121,7 @@ impl<State, Update> Clone for CompiledGraph<State, Update> {
115121
parallel: self.parallel,
116122
max_concurrency: self.max_concurrency,
117123
node_timeout: self.node_timeout,
124+
run_deadline: self.run_deadline,
118125
durability: self.durability,
119126
node_retry: self.node_retry.clone(),
120127
}

0 commit comments

Comments
 (0)