feat(todos,goals): task runs, dispatch policy, and goal budget enforcement

docs/modules/graph/goals.md

@@ -61,8 +61,32 @@ it spent into `State`, and the caller's `progress` / `run_turn` closure reports
 it. `made_progress == false` is the graph analogue of OpenHuman's "the turn
 produced no tool calls".
 
+## Budget enforcement
+
+`store::account_usage` is the raw write; `graph::goals::budget` is the policy
+around it — the two halves of enforcement a host would otherwise reimplement.
+
+- `account_turn(store, thread_id, input, output, secs, user_initiated)` charges
+  a finished turn against the thread's **active** goal (a paused, complete, or
+  budget-limited goal accrues nothing from incidental conversation) and returns
+  the goal as it stands afterwards, including a flip to `BudgetLimited`. A
+  `user_initiated` turn also clears the one-shot `continuation_suppressed` flag;
+  a continuation must not clear its own, or it would loop.
+- `GoalBudgetGuard::for_goal(goal)` arms only for an active goal that has a
+  budget. `check(store, in_flight_tokens)` adds the turn's spend so far to the
+  accounted total and returns `BudgetVerdict::Stop` once it reaches the ceiling
+  — checked mid-turn, this bounds a run to a small overshoot instead of
+  discovering the overrun after the fact. `verdict_for` is the same decision
+  with no store read. The guard captures the `goal_id` it was armed for, so a
+  goal replaced mid-turn quietly disarms it rather than enforcing a ceiling that
+  no longer describes the work.
+
+Neither aborts anything: `account_turn` reports, the guard returns a verdict,
+and wiring a stop into a turn stays the host's call.
+
 ## Testing
 
-Unit tests in `src/graph/goals/test.rs` (types, store, tools, and the gate loop
-on `InMemoryStore`); an end-to-end self-driving loop in
-`tests/e2e_graph_goals.rs`.
+Unit tests in `src/graph/goals/test.rs` (types, store, tools, the gate loop, and
+budget enforcement on `InMemoryStore`); an end-to-end self-driving loop in
+`tests/e2e_graph_goals.rs`; feature coverage for budget accounting and the
+mid-turn guard in `tests/feature_graph_goal_budget.rs`.

docs/modules/graph/todos.md

@@ -55,7 +55,33 @@ bound to `ToolExecutionContext::thread_id` (never a tool argument). Domain error
 (unknown id, invariant violation) are surfaced to the model as tool errors
 rather than failing the run.
 
+## Runs and dispatch
+
+Two layers sit on top of the board for hosts that run cards autonomously.
+
+`graph::todos::runs` is the claim/heartbeat/reclaim log (`task_run_store`). A
+worker claims a card, opens a `TaskRun`, ticks a heartbeat while it works, and
+closes the run with a `RunOutcome`. `reclaim_stale` sweeps runs whose heartbeat
+or claim aged out under `RunLimits`, closes them `Reclaimed`, and returns their
+card to `Todo` — or parks it at `Blocked` once the card has exceeded
+`max_reclaim_count`, so a card that keeps killing workers stops cycling. The
+staleness policy itself is the pure, clock-injected `staleness_reason`. See
+[`src/graph/todos/runs/README.md`](../../../src/graph/todos/runs/README.md).
+
+`graph::todos::dispatch` is the scheduling policy: `pick_next_card` (urgency,
+then board order, optionally agent-assigned only), `requires_plan_approval`
+(the card's own mode outranks the global gate), `PollCadence` (idle backoff),
+`build_task_prompt` / `build_progress_instruction`, and `ActiveRunRegistry`
+(in-flight runs with race-free removal, so a terminal write-back happens once).
+See [`src/graph/todos/dispatch/README.md`](../../../src/graph/todos/dispatch/README.md).
+
+Executing a card is out of scope for the crate — that needs a host's agent and
+tool belt. `tests/e2e_graph_task_dispatch.rs` is the reference assembly.
+
 ## Testing
 
-Unit tests in `src/graph/todos/test.rs` (types, store invariants, tool); an
-end-to-end model-driven tool run in `tests/e2e_graph_todos.rs`.
+Unit tests in `src/graph/todos/test.rs` (types, store invariants, tool),
+`src/graph/todos/runs/test.rs`, and `src/graph/todos/dispatch/test.rs`; an
+end-to-end model-driven tool run in `tests/e2e_graph_todos.rs`; feature coverage
+for the run lifecycle in `tests/feature_graph_task_runs.rs`; and the full
+dispatch loop in `tests/e2e_graph_task_dispatch.rs`.

src/graph/goals/README.md

@@ -123,4 +123,5 @@ let exec = graph.run_with_thread("thread-1", St::default()).await?;
 | `store.rs` | `Store`-backed CRUD, per-thread RMW lock, budget + CAS guards. |
 | `tool.rs` | `GoalTool` / `GoalToolKind` harness tools. |
 | `continuation.rs` | `goal_gate_node`, `run_continuation_tick`, `note_user_turn`. |
-| `test.rs` | Unit tests (types, store, tools, continuation loop). |
+| `budget.rs` | `account_turn`, `GoalBudgetGuard`, `BudgetVerdict` — charging a finished turn and stopping an overrunning one. |
+| `test.rs` | Unit tests (types, store, tools, continuation loop, budget). |

src/graph/goals/budget.rs

@@ -0,0 +1,178 @@
+//! Charging a turn against a goal, and stopping one that would overrun it.
+//!
+//! [`store::account_usage`](super::store::account_usage) is the raw write.
+//! This module is the policy around it — the two halves of budget enforcement a
+//! host otherwise reimplements:
+//!
+//! - [`account_turn`] — *after* a turn: fold its usage into the thread's active
+//!   goal, and clear the one-shot continuation suppression when the turn was
+//!   user-initiated (a person re-engaging means a later idle period may
+//!   auto-continue again).
+//! - [`GoalBudgetGuard`] — *during* a turn: given the tokens spent so far,
+//!   decide whether to stop now. Checked mid-turn, this bounds an autonomous
+//!   run to a small overshoot past its ceiling instead of discovering the
+//!   overrun only once the turn is over.
+//!
+//! Neither aborts anything itself. `account_turn` reports the goal as it stands
+//! afterwards and the guard returns a [`BudgetVerdict`]; wiring a stop into a
+//! turn is the host's call, because only the host knows whether a graceful
+//! wrap-up or a hard cut is wanted.
+
+use std::sync::Arc;
+
+use super::store;
+use super::types::{ThreadGoal, ThreadGoalStatus};
+use crate::error::Result;
+use crate::harness::store::Store;
+
+/// Total tokens a turn spent — the quantity charged against a goal's budget.
+pub fn turn_tokens(input: u64, output: u64) -> u64 {
+    input.saturating_add(output)
+}
+
+/// Whether an in-flight turn should be stopped to stay inside its goal's budget.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum BudgetVerdict {
+    /// Keep going.
+    Continue,
+    /// The projected spend meets or exceeds the budget.
+    Stop {
+        /// Human-readable reason, suitable for a transcript or a log line.
+        reason: String,
+    },
+}
+
+impl BudgetVerdict {
+    /// Whether this verdict calls for stopping.
+    pub fn is_stop(&self) -> bool {
+        matches!(self, Self::Stop { .. })
+    }
+}
+
+/// Fold a finished turn's usage into the thread's goal.
+///
+/// Only an **active** goal is charged: a paused, complete, or budget-limited
+/// goal does not accrue usage from incidental conversation. Returns the goal as
+/// it stands afterwards (including a flip to
+/// [`BudgetLimited`](ThreadGoalStatus::BudgetLimited)), or `None` when the
+/// thread has no goal or its goal is not active.
+///
+/// `user_initiated` distinguishes a person's turn from an autonomous
+/// continuation. A user turn clears the one-shot `continuation_suppressed`
+/// flag; a continuation must not clear its own suppression, or it would loop.
+pub async fn account_turn(
+    store: &Arc<dyn Store>,
+    thread_id: &str,
+    input_tokens: u64,
+    output_tokens: u64,
+    elapsed_secs: u64,
+    user_initiated: bool,
+) -> Result<Option<ThreadGoal>> {
+    let Some(goal) = store::get(store, thread_id).await? else {
+        return Ok(None);
+    };
+    if !goal.status.is_active() {
+        return Ok(None);
+    }
+
+    let mut current = goal;
+    if current.continuation_suppressed
+        && user_initiated
+        && let Some(updated) =
+            store::set_continuation_suppressed_if(store, thread_id, &current.goal_id, false).await?
+    {
+        current = updated;
+    }
+
+    let delta = turn_tokens(input_tokens, output_tokens);
+    if delta == 0 && elapsed_secs == 0 {
+        return Ok(Some(current));
+    }
+    store::account_usage(store, thread_id, &current.goal_id, delta, elapsed_secs).await
+}
+
+/// Mid-turn budget check, armed for one specific version of a goal.
+///
+/// Built from a goal that is active and has a budget; a goal with neither is
+/// nothing to enforce. The captured `goal_id` is what makes the guard safe to
+/// hold across a long turn: if the objective is replaced while the turn runs,
+/// the new goal has a new id, the guard stops matching, and it quietly stands
+/// down instead of enforcing a budget that no longer applies.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GoalBudgetGuard {
+    thread_id: String,
+    goal_id: String,
+    budget: u64,
+}
+
+impl GoalBudgetGuard {
+    /// A guard for `goal`, or `None` when it is inactive or has no budget.
+    pub fn for_goal(goal: &ThreadGoal) -> Option<Self> {
+        if !goal.status.is_active() {
+            return None;
+        }
+        Some(Self {
+            thread_id: goal.thread_id.clone(),
+            goal_id: goal.goal_id.clone(),
+            budget: goal.token_budget?,
+        })
+    }
+
+    /// The thread this guard watches.
+    pub fn thread_id(&self) -> &str {
+        &self.thread_id
+    }
+
+    /// The goal version this guard was armed for.
+    pub fn goal_id(&self) -> &str {
+        &self.goal_id
+    }
+
+    /// The ceiling being enforced.
+    pub fn budget(&self) -> u64 {
+        self.budget
+    }
+
+    /// Verdict for a turn that has spent `in_flight_tokens` so far.
+    ///
+    /// Reads the goal's already-accounted usage, adds the in-flight spend, and
+    /// stops once the total reaches the budget. Returns
+    /// [`Continue`](BudgetVerdict::Continue) whenever there is nothing left to
+    /// enforce: the goal is gone, was replaced, or is no longer active.
+    pub async fn check(
+        &self,
+        store: &Arc<dyn Store>,
+        in_flight_tokens: u64,
+    ) -> Result<BudgetVerdict> {
+        let Some(goal) = store::get(store, &self.thread_id).await? else {
+            return Ok(BudgetVerdict::Continue);
+        };
+        if goal.goal_id != self.goal_id || !goal.status.is_active() {
+            return Ok(BudgetVerdict::Continue);
+        }
+        Ok(self.verdict_for(goal.tokens_used, in_flight_tokens))
+    }
+
+    /// The pure half of [`check`](Self::check): the verdict for a given
+    /// accounted and in-flight spend, with no store read.
+    pub fn verdict_for(&self, accounted_tokens: u64, in_flight_tokens: u64) -> BudgetVerdict {
+        let projected = accounted_tokens.saturating_add(in_flight_tokens);
+        if projected >= self.budget {
+            BudgetVerdict::Stop {
+                reason: format!(
+                    "thread goal budget reached: {projected} tokens >= {} budget — stopping to \
+                     summarise progress",
+                    self.budget
+                ),
+            }
+        } else {
+            BudgetVerdict::Continue
+        }
+    }
+}
+
+/// Whether `status` still accrues usage. Kept next to the accounting policy so
+/// callers do not re-derive the rule.
+pub fn accrues_usage(status: ThreadGoalStatus) -> bool {
+    status.is_active()
+}

src/graph/goals/mod.rs

@@ -6,18 +6,22 @@
 //! is exhausted, or a host pauses it. This module owns the data model
 //! ([`types`]), harness-[`Store`](crate::harness::store::Store)-backed
 //! persistence ([`store`]), the model-facing controls exposed as harness tools
-//! ([`tool`]), and the graph-native continuation surface ([`continuation`]).
+//! ([`tool`]), the graph-native continuation surface ([`continuation`]), and
+//! budget enforcement ([`budget`]) — charging a finished turn against the goal
+//! and stopping an in-flight one that would overrun its ceiling.
 //!
 //! It is the graph analogue of OpenHuman's `thread_goals`, minus the
 //! app-specific coupling (event bus, RPC envelopes, heartbeat scheduler): the
 //! primitive is provider-neutral and drives off the graph runtime.
 
+pub mod budget;
 mod continuation;
 mod prompt;
 pub mod store;
 mod tool;
 mod types;
 
+pub use budget::{BudgetVerdict, GoalBudgetGuard, account_turn, accrues_usage, turn_tokens};
 pub use continuation::{goal_gate_node, note_user_turn, run_continuation_tick};
 pub use prompt::active_goal_context_block;
 pub use tool::{GoalTool, GoalToolKind, goal_tools, register_goal_tools};