From c02732a93b4305f7f9fb8f5566da2a975f66aaff Mon Sep 17 00:00:00 2001 From: Ytallo Layon Date: Fri, 31 Jul 2026 23:51:52 -0300 Subject: [PATCH] (MOT-4295) test(harness): multi-target integration scenarios INT-014 and INT-016 plus runner enablers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two new scenarios aimed at the coldest harness modules found by the coverage report, plus the runner capabilities they needed: - probe actions now record their responses into RunEvidence (probe_responses + expect_probe_response), unlocking body assertions for state/ledger/teardown reads - scripted hooks: fixtures can declare probe-hosted hook functions (Continue/HoldOnce/Deny/Mutate behaviors) bound to harness::hook::* trigger types before Send, with whole-run hook_calls evidence - Send options gain max_total_tokens; terminal statuses may now end in 'failed' (floor already pins the durable status to it) INT-014 budget-preflight-exceeded: a frozen budget admits generation 1, then a ~240k-char controlled-function result deterministically blows generation 2's reservation — failed turn, harness.budget_exceeded custom error entry, and the harness_budget ledger asserted through a recorded probe response (reconcile + rollback). INT-016 notify-grant-teardown: two standing notify subscriptions; a state fire lands a trigger_fired entry plus the injected '[notification: …]' user message that seeds the second tracked turn; filesystem grant/grants/revoke round-trip; the budget ledger holds the exact reconciled usage; harness::teardown sweeps both bindings. INT-013 held-call-resolve is authored but NOT registered: running it exposed an apparent harness defect — after a pre-trigger hook hold, harness::status times out and the turn record vanishes from state, so the resolve intervention can never find the parked call. Isolation runs prove the scripted-hook plumbing is sound (Continue variant completes in ~2.7s with both hooks served). See the scenario doc comment. All 14 registered direct scenarios pass against the local engine; 92 unit tests pass; zero build warnings. --- harness/tests/integration/README.md | 8 + .../tests/integration/src/evidence_data.rs | 61 ++++ harness/tests/integration/src/fixtures.rs | 4 +- .../tests/integration/src/fixtures/loading.rs | 103 ++++++- .../tests/integration/src/fixtures/tests.rs | 5 +- harness/tests/integration/src/probe.rs | 120 +++++++- .../tests/integration/src/scenario/floor.rs | 2 + .../integration/src/scenario/phases/arm.rs | 8 + .../src/scenario/phases/completion.rs | 16 +- .../src/scenario/phases/evidence.rs | 2 + .../src/scenario/phases/intervention.rs | 120 ++++++++ .../tests/integration/src/scenario/state.rs | 3 + .../scenarios/budget_preflight_exceeded.rs | 152 ++++++++++ .../tests/integration/src/scenarios/dsl.rs | 84 ++++++ .../src/scenarios/held_call_resolve.rs | 150 ++++++++++ .../tests/integration/src/scenarios/mod.rs | 6 +- .../src/scenarios/notify_grant_teardown.rs | 273 ++++++++++++++++++ .../src/types/scenario/compiled.rs | 3 + .../tests/integration/tests/determinism.rs | 2 + 19 files changed, 1109 insertions(+), 13 deletions(-) create mode 100644 harness/tests/integration/src/scenarios/budget_preflight_exceeded.rs create mode 100644 harness/tests/integration/src/scenarios/held_call_resolve.rs create mode 100644 harness/tests/integration/src/scenarios/notify_grant_teardown.rs diff --git a/harness/tests/integration/README.md b/harness/tests/integration/README.md index 9ed90cab9..a946edd93 100644 --- a/harness/tests/integration/README.md +++ b/harness/tests/integration/README.md @@ -23,9 +23,17 @@ No provider key or network access is required. | INT-010 | `crash-recovery-507` | direct | SIGKILL and restart the engine while a controlled function is in flight, with `context::assemble` held out during boot; the side effect runs once, the interrupted call closes, and the turn completes | | INT-011 | `stop-cancel-cascade` | direct | stopping a running root turn cancels the root and spawned children while retaining a queued message | | INT-012 | `queued-message-edit-unqueue` | direct | edit one queued message in place and unqueue another while the first turn is streaming; only the edited and untouched rows drain in order | +| INT-014 | `budget-preflight-exceeded` | direct | a frozen token budget admits generation 1, then fails the turn at budget preflight when a fat function result blows generation 2's reservation; the state ledger keeps the reconciled usage | +| INT-016 | `notify-grant-teardown` | direct | a notify subscription fires an injected notification turn; filesystem grants round-trip; the budget ledger reconciles; `harness::teardown` sweeps the standing binding | | UI-001 | `console-streamed-text` | playground | a message sent by the Console streams to durable completion | | UI-002 | `multi-turn-traces` | playground | a native function turn and a Console turn expose distinct traces and function-call events | +INT-013 (`held-call-resolve`, hook-held call released via +`harness::function::resolve`) is authored in `src/scenarios/held_call_resolve.rs` +but not registered: running it surfaced an apparent harness defect where a +hook-held call leaves `harness::status` timing out and the turn record missing +from state (MOT-4296). Register it once that defect is fixed. + Each fixture is defined end to end in its own `src/scenarios/*.rs` file with a small typed DSL. The scenario keeps its send policy, router request matchers, response behavior, controlled function, function history, and verification diff --git a/harness/tests/integration/src/evidence_data.rs b/harness/tests/integration/src/evidence_data.rs index 83cc19039..45ae7116b 100644 --- a/harness/tests/integration/src/evidence_data.rs +++ b/harness/tests/integration/src/evidence_data.rs @@ -34,6 +34,12 @@ pub struct RunEvidence { pub tree_statuses: Vec, /// Raw scripted-router calls and abort acknowledgements. pub router_evidence: Value, + /// `{function_id, response}` per fired probe action, in fire order — + /// the only surface for asserting a probe-dispatched call's response. + pub probe_responses: Vec, + /// `{function_id, payload}` per scripted-hook invocation the probe + /// served, in arrival order — whole-run evidence like `target_calls`. + pub hook_calls: Vec, } /// Compact form published in `playground-result.json`. @@ -209,6 +215,59 @@ impl RunEvidence { Ok(()) } + /// Invocation payloads served by the scripted hook registered as `name`. + pub fn hook_invocations(&self, name: &str) -> Vec<&Value> { + let suffix = format!("::hook-{name}"); + self.hook_calls + .iter() + .filter(|entry| { + entry + .get("function_id") + .and_then(Value::as_str) + .is_some_and(|id| id.ends_with(&suffix)) + }) + .filter_map(|entry| entry.get("payload")) + .collect() + } + + pub fn expect_hook_calls(&self, name: &str, count: usize) -> anyhow::Result<()> { + let actual = self.hook_invocations(name).len(); + anyhow::ensure!( + actual == count, + "hook {name} served {actual} invocation(s), expected {count}" + ); + Ok(()) + } + + /// Response body of the nth fired probe action (fire order follows the + /// fixture's `after_turns` sort). + pub fn probe_response(&self, index: usize) -> Option<&Value> { + self.probe_responses.get(index).and_then(|entry| entry.get("response")) + } + + /// Assert the nth probe action targeted `function_id` and its response + /// satisfies `check`. + pub fn expect_probe_response( + &self, + index: usize, + function_id: &str, + check: impl FnOnce(&Value) -> anyhow::Result<()>, + ) -> anyhow::Result<()> { + let entry = self.probe_responses.get(index).ok_or_else(|| { + anyhow::anyhow!( + "no probe response at index {index} ({} recorded)", + self.probe_responses.len() + ) + })?; + let actual = entry.get("function_id").and_then(Value::as_str); + anyhow::ensure!( + actual == Some(function_id), + "probe action {index} targeted {actual:?}, expected {function_id:?}" + ); + check(entry.get("response").unwrap_or(&Value::Null)) + .map_err(|error| anyhow::anyhow!("probe response {index} ({function_id}): {error}")) + } + pub fn expect_no_duplicate_messages(&self) -> anyhow::Result<()> { anyhow::ensure!( !self.has_duplicate_messages(), @@ -348,6 +407,8 @@ mod tests { tree_sessions: Vec::new(), tree_statuses: Vec::new(), router_evidence: Value::Null, + probe_responses: Vec::new(), + hook_calls: Vec::new(), } } diff --git a/harness/tests/integration/src/fixtures.rs b/harness/tests/integration/src/fixtures.rs index 2e3fd84b0..4006a85b0 100644 --- a/harness/tests/integration/src/fixtures.rs +++ b/harness/tests/integration/src/fixtures.rs @@ -9,7 +9,9 @@ mod discovery; mod loading; pub use discovery::scenario_fixtures; -pub use loading::{ProbeAction, ScenarioFixture, ScenarioIntervention}; +pub use loading::{ + HookBehavior, ProbeAction, ScenarioFixture, ScenarioHook, ScenarioIntervention, +}; #[cfg(test)] mod tests; diff --git a/harness/tests/integration/src/fixtures/loading.rs b/harness/tests/integration/src/fixtures/loading.rs index 1b89d1010..b852cc3be 100644 --- a/harness/tests/integration/src/fixtures/loading.rs +++ b/harness/tests/integration/src/fixtures/loading.rs @@ -44,8 +44,47 @@ pub struct ScenarioFixture { /// outside the compiled subject scenario: it is test synchronization, not /// a public harness request. pub intervention: Option, + /// Probe-hosted hook functions bound to `harness::hook::` trigger + /// types before Send. Each records its invocations as whole-run evidence + /// and answers with its scripted behavior. + pub hooks: Vec, } +/// One scripted hook the probe registers and binds for the run. +#[derive(Debug, Clone)] +pub struct ScenarioHook { + /// Short name; the registered function id is `::hook-`. + pub name: String, + /// Hook point suffix: pre-turn, pre-generate, post-generate, pre-trigger, + /// or post-trigger. + pub point: String, + /// Optional function-id globs for the binding's `functions` filter. + pub functions: Option>, + pub priority: i64, + pub timeout_ms: Option, + pub on_error: Option, + pub behavior: HookBehavior, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HookBehavior { + /// Always answer `null` (continue). + Continue, + /// `{"decision":"hold"}` on the first invocation, `null` afterwards — so + /// a post-resolve chain resume does not re-park the call. + HoldOnce, + Deny { reason: String }, + Mutate { mutations: serde_json::Value }, +} + +pub const HOOK_POINTS: [&str; 5] = [ + "pre-turn", + "pre-generate", + "post-generate", + "pre-trigger", + "post-trigger", +]; + #[derive(Debug, Clone, PartialEq, Eq)] pub enum ScenarioIntervention { StopCancelCascade { @@ -61,6 +100,10 @@ pub enum ScenarioIntervention { remove_message: String, after_message: String, }, + /// Wait for a hook-held call to park the turn in `awaiting_functions`, + /// exercise `harness::function::resolve` no-op gates (wrong turn, unknown + /// call), then release the held call with `action: "execute"`. + HeldCallResolve { function_call_id: String }, } /// A function the PROBE (test infra, not a model turn) invokes at a completion @@ -144,11 +187,47 @@ impl ScenarioFixture { } } if self.intervention.is_none() { + // `cancelled` requires the intervention that performs the stop; a + // run may legitimately END failed (e.g. a budget-preflight + // rejection) — the floor still pins the durable status to it. anyhow::ensure!( - self.expected_turn_statuses.last().map(String::as_str) == Some("completed"), - "the last terminal turn must be completed" + matches!( + self.expected_turn_statuses.last().map(String::as_str), + Some("completed") | Some("failed") + ), + "the last terminal turn must be completed or failed" ); } + { + let mut names = std::collections::BTreeSet::new(); + for hook in &self.hooks { + anyhow::ensure!( + !hook.name.is_empty() + && hook + .name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')), + "hook name {:?} is not id-safe", + hook.name + ); + anyhow::ensure!( + names.insert(hook.name.clone()), + "duplicate hook name {:?}", + hook.name + ); + anyhow::ensure!( + HOOK_POINTS.contains(&hook.point.as_str()), + "unknown hook point {:?}; expected one of {HOOK_POINTS:?}", + hook.point + ); + if let Some(on_error) = &hook.on_error { + anyhow::ensure!( + matches!(on_error.as_str(), "fail_open" | "fail_closed"), + "hook on_error {on_error:?} must be fail_open or fail_closed" + ); + } + } + } if let Some(intervention) = &self.intervention { match intervention { ScenarioIntervention::StopCancelCascade { @@ -202,6 +281,26 @@ impl ScenarioFixture { "queued-edit scenario must end with one completed turn" ); } + ScenarioIntervention::HeldCallResolve { function_call_id } => { + anyhow::ensure!( + !function_call_id.is_empty(), + "held-call-resolve function_call_id must not be empty" + ); + anyhow::ensure!( + self.scenario.target.is_some(), + "held-call-resolve needs a controlled function to hold" + ); + anyhow::ensure!( + self.hooks + .iter() + .any(|hook| hook.behavior == HookBehavior::HoldOnce), + "held-call-resolve needs a HoldOnce hook to park the call" + ); + anyhow::ensure!( + self.expected_turn_statuses == ["completed"], + "held-call-resolve scenario must end with one completed turn" + ); + } } } // A direct scenario is one external send, but the harness may seed diff --git a/harness/tests/integration/src/fixtures/tests.rs b/harness/tests/integration/src/fixtures/tests.rs index 09f15b9c2..17adfc421 100644 --- a/harness/tests/integration/src/fixtures/tests.rs +++ b/harness/tests/integration/src/fixtures/tests.rs @@ -11,7 +11,8 @@ fn all_selection_returns_the_checked_in_fixtures() { ids, std::collections::BTreeSet::from([ "INT-001", "INT-002", "INT-003", "INT-004", "INT-005", "INT-006", "INT-007", "INT-008", - "INT-009", "INT-010", "INT-011", "INT-012", "UI-001", "UI-002" + "INT-009", "INT-010", "INT-011", "INT-012", "INT-014", "INT-016", "UI-001", + "UI-002" ]) ); assert_eq!( @@ -19,7 +20,7 @@ fn all_selection_returns_the_checked_in_fixtures() { .iter() .filter(|fixture| fixture.driver == crate::scenarios::ScenarioDriver::Direct) .count(), - 12 + 14 ); } diff --git a/harness/tests/integration/src/probe.rs b/harness/tests/integration/src/probe.rs index 0f8958874..1a632034f 100644 --- a/harness/tests/integration/src/probe.rs +++ b/harness/tests/integration/src/probe.rs @@ -32,7 +32,7 @@ pub(crate) struct CompletionObservation { #[derive(Clone)] struct BoundTrigger { id: String, - trigger_type: &'static str, + trigger_type: String, } #[derive(Deserialize)] @@ -58,6 +58,9 @@ pub struct ScenarioProbe { target_notify: Arc, target_response_released: Arc, target_response_notify: Arc, + /// `{function_id, payload}` per scripted-hook invocation, in arrival + /// order — whole-run evidence, like `target_calls`. + hook_calls: Arc>>, } impl ScenarioProbe { @@ -75,6 +78,7 @@ impl ScenarioProbe { target_notify: Arc::new(tokio::sync::Notify::new()), target_response_released: Arc::new(AtomicBool::new(false)), target_response_notify: Arc::new(tokio::sync::Notify::new()), + hook_calls: Arc::new(Mutex::new(Vec::new())), }; probe.register_sinks(); Ok(probe) @@ -187,6 +191,99 @@ impl ScenarioProbe { .unwrap_or_default() } + /// Snapshot of every scripted-hook invocation so far, in arrival order, + /// as `{function_id, payload}`. + pub fn hook_calls(&self) -> Vec { + self.hook_calls + .lock() + .map(|calls| calls.clone()) + .unwrap_or_default() + } + + /// Register one probe-hosted function per scripted hook. The function id + /// is `::hook-`; each invocation is recorded whole-run and + /// answered with the hook's scripted behavior. + pub fn register_hooks(&self, run_id: &str, hooks: &[crate::fixtures::ScenarioHook]) { + for hook in hooks { + let function_id = hook_function_id(run_id, &hook.name); + let calls = Arc::clone(&self.hook_calls); + let behavior = hook.behavior.clone(); + let held = Arc::new(AtomicBool::new(false)); + let recorded_id = function_id.clone(); + self.client.inner().register_function( + &function_id, + RegisterFunction::new_async(move |mut payload: Value| { + let calls = Arc::clone(&calls); + let behavior = behavior.clone(); + let held = Arc::clone(&held); + let recorded_id = recorded_id.clone(); + async move { + strip_engine_fields(&mut payload); + calls + .lock() + .map_err(|_| { + Error::Handler("integration/hook_calls_lock_poisoned".into()) + })? + .push(json!({ + "function_id": recorded_id, + "payload": payload, + })); + let response = match &behavior { + crate::fixtures::HookBehavior::Continue => Value::Null, + crate::fixtures::HookBehavior::HoldOnce => { + if held.swap(true, Ordering::AcqRel) { + Value::Null + } else { + json!({ "decision": "hold" }) + } + } + crate::fixtures::HookBehavior::Deny { reason } => { + json!({ "decision": "deny", "reason": reason }) + } + crate::fixtures::HookBehavior::Mutate { mutations } => { + json!({ "mutations": mutations }) + } + }; + Ok::(response) + } + }) + .description(format!("Scripted {} hook for integration runs.", hook.point)), + ); + } + } + + /// Bind every scripted hook to its `harness::hook::` trigger type. + /// Must run after harness readiness so the trigger types exist and the + /// acknowledged RPCs barrier the function registrations queued before. + pub async fn bind_hooks( + &self, + run_id: &str, + hooks: &[crate::fixtures::ScenarioHook], + deadline: Deadline, + ) -> anyhow::Result<()> { + for hook in hooks { + let mut config = serde_json::Map::new(); + config.insert("priority".into(), json!(hook.priority)); + if let Some(functions) = &hook.functions { + config.insert("functions".into(), json!(functions)); + } + if let Some(timeout_ms) = hook.timeout_ms { + config.insert("timeout_ms".into(), json!(timeout_ms)); + } + if let Some(on_error) = &hook.on_error { + config.insert("on_error".into(), json!(on_error)); + } + self.bind_dynamic_trigger( + format!("harness::hook::{}", hook.point), + hook_function_id(run_id, &hook.name), + Value::Object(config), + deadline, + ) + .await?; + } + Ok(()) + } + /// Wait until the controlled function has served at least `expected` /// invocations. This is how a scenario awaits work that produces no /// session turn — a call-mode reaction's dispatch, including the trailing @@ -309,6 +406,22 @@ impl ScenarioProbe { function_id: &'static str, config: Value, deadline: Deadline, + ) -> anyhow::Result { + self.bind_dynamic_trigger( + trigger_type.to_string(), + function_id.to_string(), + config, + deadline, + ) + .await + } + + async fn bind_dynamic_trigger( + &self, + trigger_type: String, + function_id: String, + config: Value, + deadline: Deadline, ) -> anyhow::Result { let response = self .client @@ -559,6 +672,11 @@ fn register_controlled_function( ); } +/// Registered function id for a scripted hook: `::hook-`. +pub fn hook_function_id(run_id: &str, name: &str) -> String { + format!("{run_id}::hook-{name}") +} + fn const_response_schema(response: &Value) -> Value { json!({ "$schema": "http://json-schema.org/draft-07/schema#", diff --git a/harness/tests/integration/src/scenario/floor.rs b/harness/tests/integration/src/scenario/floor.rs index 82cc2051a..d6b1c22dd 100644 --- a/harness/tests/integration/src/scenario/floor.rs +++ b/harness/tests/integration/src/scenario/floor.rs @@ -379,6 +379,8 @@ mod tests { tree_sessions: Vec::new(), tree_statuses: Vec::new(), router_evidence: Value::Null, + probe_responses: Vec::new(), + hook_calls: Vec::new(), } } diff --git a/harness/tests/integration/src/scenario/phases/arm.rs b/harness/tests/integration/src/scenario/phases/arm.rs index 850f2a4d9..75955ba30 100644 --- a/harness/tests/integration/src/scenario/phases/arm.rs +++ b/harness/tests/integration/src/scenario/phases/arm.rs @@ -21,6 +21,7 @@ impl ScenarioRunner<'_> { probe .register_target(&self.run_id, scenario.target.as_ref()) .map_err(|error| RunError::setup(phase, "register controlled function", error))?; + probe.register_hooks(&self.run_id, &self.fixture.hooks); // Register observer bindings before the harness exists. Recoverable // triggers park the harness-owned bindings until their types appear; @@ -47,6 +48,13 @@ impl ScenarioRunner<'_> { .map_err(|error| { RunError::setup(phase, "confirm completion observer binding", error) })?; + // Hook trigger types are registered by the harness at boot; bind the + // scripted hooks only after readiness so registration is acknowledged + // before the first stimulus. + probe + .bind_hooks(&self.run_id, &self.fixture.hooks, deadline) + .await + .map_err(|error| RunError::setup(phase, "bind scripted hooks", error))?; if self.fixture.intervention.is_some() { probe .bind_child_completion_observer(&self.session_id, deadline) diff --git a/harness/tests/integration/src/scenario/phases/completion.rs b/harness/tests/integration/src/scenario/phases/completion.rs index fec6b56df..bf5a7c1cb 100644 --- a/harness/tests/integration/src/scenario/phases/completion.rs +++ b/harness/tests/integration/src/scenario/phases/completion.rs @@ -79,7 +79,11 @@ impl ScenarioRunner<'_> { )); } } - self.fire_probe_action(services, &action, deadline).await?; + let response = self.fire_probe_action(services, &action, deadline).await?; + active.probe_responses.push(json!({ + "function_id": action.function_id, + "response": response, + })); } } @@ -223,14 +227,15 @@ impl ScenarioRunner<'_> { } /// Invoke a probe action, expanding `{{run_id}}`/`{{session_id}}` in its - /// payload first. A failed dispatch is a runner error — the scenario's - /// premise (the reaction it should trip) can't hold without it. + /// payload first, and return the response for evidence. A failed dispatch + /// is a runner error — the scenario's premise (the reaction it should + /// trip) can't hold without it. async fn fire_probe_action( &self, services: &RunServices, action: &crate::fixtures::ProbeAction, deadline: Deadline, - ) -> Result<(), RunError> { + ) -> Result { let mut payload = action.payload.clone(); crate::expand::Placeholders::new(&self.run_id, &self.session_id) .expand_value(&mut payload) @@ -248,8 +253,7 @@ impl ScenarioRunner<'_> { .await .map_err(|error| { RunError::runner(RunPhase::Await, "fire probe action", anyhow::anyhow!(error)) - })?; - Ok(()) + }) } async fn confirm_terminal_status( diff --git a/harness/tests/integration/src/scenario/phases/evidence.rs b/harness/tests/integration/src/scenario/phases/evidence.rs index 20bc48032..e1ca02ce8 100644 --- a/harness/tests/integration/src/scenario/phases/evidence.rs +++ b/harness/tests/integration/src/scenario/phases/evidence.rs @@ -113,6 +113,8 @@ impl ScenarioRunner<'_> { tree_sessions: active.tree_sessions.clone(), tree_statuses: active.tree_statuses.clone(), router_evidence: active.router_evidence.clone(), + probe_responses: active.probe_responses.clone(), + hook_calls: services.probe().hook_calls(), } } diff --git a/harness/tests/integration/src/scenario/phases/intervention.rs b/harness/tests/integration/src/scenario/phases/intervention.rs index 89cb0e9c6..8b3091058 100644 --- a/harness/tests/integration/src/scenario/phases/intervention.rs +++ b/harness/tests/integration/src/scenario/phases/intervention.rs @@ -58,6 +58,10 @@ impl ScenarioRunner<'_> { ) .await } + ScenarioIntervention::HeldCallResolve { function_call_id } => { + self.run_held_call_resolve(services, active, &function_call_id) + .await + } }; match result { @@ -75,6 +79,122 @@ impl ScenarioRunner<'_> { } } + /// INT-013 driver: a scripted pre-trigger hook answered `hold`, parking + /// the turn in `awaiting_functions`. Prove the no-op resolve gates leave + /// it parked, then release the held call so the chain resumes after the + /// holder and the turn completes. + async fn run_held_call_resolve( + &self, + services: &RunServices, + active: &ActiveTurn, + function_call_id: &str, + ) -> Result<(Value, Vec), RunError> { + let phase = RunPhase::Intervene; + let deadline = active.deadline; + let turn_id = active.turn_id.clone().ok_or_else(|| { + RunError::new( + phase, + RunErrorKind::Contract, + "held-call-resolve requires a turn id from harness::send", + ) + })?; + + // Wait for the hook hold to park the call durably. + let parked_status = wait_for_status(services.client(), &self.session_id, deadline, { + let call_id = function_call_id.to_string(); + move |status| { + status.get("status").and_then(Value::as_str) == Some("awaiting_functions") + && status + .get("pending_function_calls") + .and_then(Value::as_array) + .is_some_and(|pending| { + pending.iter().any(|call| { + call.get("function_call_id").and_then(Value::as_str) + == Some(call_id.as_str()) + || call.get("id").and_then(Value::as_str) + == Some(call_id.as_str()) + }) + }) + } + }) + .await?; + + let resolve = |payload: Value| { + let client = services.client().clone(); + async move { + client + .call_with_deadline( + "harness::function::resolve", + payload, + deadline, + DEFAULT_CALL_TIMEOUT_MS, + ) + .await + .map_err(|error| { + RunError::with_source( + phase, + RunErrorKind::Contract, + "call harness::function::resolve", + anyhow::anyhow!(error), + ) + }) + } + }; + + // No-op gate: a resolve against the wrong turn must not settle the call. + let wrong_turn = resolve(json!({ + "session_id": self.session_id, + "turn_id": format!("{turn_id}-bogus"), + "function_call_id": function_call_id, + "action": "execute", + })) + .await?; + // No-op gate: an unknown call id must not settle anything either. + let unknown_call = resolve(json!({ + "session_id": self.session_id, + "turn_id": turn_id, + "function_call_id": format!("{function_call_id}-unknown"), + "action": "execute", + })) + .await?; + for (label, response) in [("wrong turn", &wrong_turn), ("unknown call", &unknown_call)] { + if response.get("resolved") != Some(&Value::Bool(false)) { + return Err(RunError::new( + phase, + RunErrorKind::Contract, + format!("{label} resolve was not a no-op: {response}"), + )); + } + } + + // Release: resume the chain after the holder and run the target. + let execute = resolve(json!({ + "session_id": self.session_id, + "turn_id": turn_id, + "function_call_id": function_call_id, + "action": "execute", + })) + .await?; + if execute.get("resolved") != Some(&Value::Bool(true)) + || execute.get("turn_resumed") != Some(&Value::Bool(true)) + { + return Err(RunError::new( + phase, + RunErrorKind::Contract, + format!("execute resolve did not resume the turn: {execute}"), + )); + } + + let control = json!({ + "kind": "held_call_resolve", + "parked_status": parked_status, + "wrong_turn_resolve": wrong_turn, + "unknown_call_resolve": unknown_call, + "execute_resolve": execute, + }); + Ok((control, Vec::new())) + } + async fn run_stop_cancel_cascade( &self, services: &RunServices, diff --git a/harness/tests/integration/src/scenario/state.rs b/harness/tests/integration/src/scenario/state.rs index 1e986d467..46d259f41 100644 --- a/harness/tests/integration/src/scenario/state.rs +++ b/harness/tests/integration/src/scenario/state.rs @@ -44,6 +44,8 @@ pub(super) struct ActiveTurn { pub(super) tree_sessions: Vec, pub(super) tree_statuses: Vec, pub(super) router_evidence: Value, + /// `{function_id, response}` per fired probe action, in fire order. + pub(super) probe_responses: Vec, } impl ActiveTurn { @@ -66,6 +68,7 @@ impl ActiveTurn { tree_sessions: Vec::new(), tree_statuses: Vec::new(), router_evidence: Value::Null, + probe_responses: Vec::new(), } } diff --git a/harness/tests/integration/src/scenarios/budget_preflight_exceeded.rs b/harness/tests/integration/src/scenarios/budget_preflight_exceeded.rs new file mode 100644 index 000000000..70772e6b8 --- /dev/null +++ b/harness/tests/integration/src/scenarios/budget_preflight_exceeded.rs @@ -0,0 +1,152 @@ +//! INT-014 — a frozen `max_total_tokens` budget admits the first generation +//! and rejects the second at budget preflight. +//! +//! Determinism without calibrating the (unknowable) system-prompt size: the +//! controlled function returns a ~240k-character result, so generation 2's +//! reservation (`assembled_input_estimate + max_output`) exceeds the 30k +//! budget by tens of thousands of tokens no matter what the prompt costs, +//! while generation 1's reservation (prompt + one short message + 4_096 +//! max-output) stays far below it. The fixture model's context window is +//! widened so context assembly never prunes the fat result away before the +//! budget check. +//! +//! Covered: `budget.rs` prepare_root/reserve/reconcile/release + Exceeded, +//! `turn_loop::finalize_failed` (durable `custom_type:"error"` entry, failed +//! terminal turn), and the `harness_budget` state ledger read through a +//! recorded probe response. + +use serde_json::{json, Value}; + +use super::dsl::{ControlledFunction, Generation, Message, Model, Request, Response, Scenario, Send}; +use super::ScenarioDriver; +use crate::fixtures::ScenarioFixture; + +const BUDGET_TOKENS: u64 = 30_000; +/// Scripted generation-1 usage: 8 input + 4 output. +const EXPECTED_USED_TOKENS: u64 = 12; +const FAT_RESULT_CHARS: usize = 240_000; + +pub(super) fn scenario() -> ScenarioFixture { + const ID: &str = "INT-014"; + const MESSAGE: &str = "Fetch the archive; the budget will not survive it."; + + let mut model = Model::scripted("fixture-model"); + // Wide enough that assembly keeps the fat function result intact and the + // budget check — not a context overflow — is what stops the turn. + model.context_window = 400_000; + + let archive = ControlledFunction::new("{{run_id}}::archive", "Return one huge archive blob.") + .request_schema(json!({ + "type": "object", + "additionalProperties": false, + "properties": {}, + })) + .returns_text(&"x".repeat(FAT_RESULT_CHARS)); + + Scenario::new( + ID, + "budget-preflight-exceeded", + "A frozen token budget admits generation 1, then fails the turn at budget preflight \ + when the fat function result blows generation 2's reservation.", + ScenarioDriver::Direct, + model, + ) + .send( + Send::message(MESSAGE) + .idempotency_key("{{run_id}}:integration-014") + .allow_function(&archive) + .max_total_tokens(BUDGET_TOKENS), + ) + .function(archive.clone()) + .terminal_turn_statuses(["failed"]) + .expect_traces(1) + // The ledger survives the failed turn: generation 1 reconciled its actual + // usage and the rejected generation-2 reservation was never charged. + .probe_after( + 1, + "state::get", + json!({ "scope": "harness_budget", "key": "{{session_id}}" }), + ) + .generation( + Generation::new(1) + .expect( + Request::new() + .turn_request() + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_exact([Message::user(MESSAGE)]) + .tools_exact_after_controls([], [archive.tool()]), + ) + .respond(Response::function_call( + "call-archive", + &archive, + json!({}), + 8, + 4, + )), + ) + .verify(|run| { + // user + assistant(function_call) + fat function_result, then the + // turn fails before any further assistant message. + run.expect_message_counts(1, 1, 1)?; + run.expect_target_calls(1)?; + let error_entry = run + .transcript + .iter() + .find_map(|item| { + let custom_type = item + .pointer("/custom/custom_type") + .and_then(Value::as_str); + (custom_type == Some("error")) + .then(|| item.pointer("/custom/data")) + .flatten() + }) + .ok_or_else(|| anyhow::anyhow!("no custom error entry in the transcript"))?; + anyhow::ensure!( + error_entry.get("code").and_then(Value::as_str) == Some("harness.budget_exceeded"), + "error code {:?} != harness.budget_exceeded", + error_entry.get("code") + ); + anyhow::ensure!( + error_entry.get("phase").and_then(Value::as_str) == Some("budget_preflight"), + "error phase {:?} != budget_preflight", + error_entry.get("phase") + ); + anyhow::ensure!( + error_entry.get("retryable") == Some(&Value::Bool(false)), + "budget failures must not be retryable: {error_entry}" + ); + run.expect_probe_response(0, "state::get", |response| { + let ledger = response.get("value").unwrap_or(response); + anyhow::ensure!( + ledger.get("max_total_tokens").and_then(Value::as_u64) == Some(BUDGET_TOKENS), + "ledger max_total_tokens != {BUDGET_TOKENS}: {ledger}" + ); + anyhow::ensure!( + ledger.get("used_tokens").and_then(Value::as_u64) == Some(EXPECTED_USED_TOKENS), + "ledger used_tokens != {EXPECTED_USED_TOKENS}: {ledger}" + ); + anyhow::ensure!( + ledger.get("reserved_tokens").and_then(Value::as_u64) == Some(0), + "the rejected reservation must be rolled back: {ledger}" + ); + Ok(()) + })?; + run.expect_no_duplicate_messages() + }) + .build() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixture_declares_a_failed_turn_after_one_generation() { + let fixture = scenario(); + fixture.validate().unwrap(); + assert_eq!(fixture.expected_turn_statuses, ["failed"]); + assert_eq!(fixture.script.generations.len(), 1); + assert_eq!(fixture.expected_traces(), 1); + assert_eq!(fixture.probe_actions.len(), 1); + } +} diff --git a/harness/tests/integration/src/scenarios/dsl.rs b/harness/tests/integration/src/scenarios/dsl.rs index da2a34de8..fcb81ec60 100644 --- a/harness/tests/integration/src/scenarios/dsl.rs +++ b/harness/tests/integration/src/scenarios/dsl.rs @@ -63,6 +63,7 @@ pub(super) struct Scenario { await_target_calls: Option, traces_override: Option, intervention: Option, + hooks: Vec, } impl Scenario { @@ -91,6 +92,7 @@ impl Scenario { await_target_calls: None, traces_override: None, intervention: None, + hooks: Vec::new(), } } @@ -179,6 +181,14 @@ impl Scenario { self } + /// Register a scripted hook the probe hosts and binds to + /// `harness::hook::` before Send. + #[allow(dead_code)] // used by the unregistered INT-013 fixture + pub(super) fn hook(mut self, hook: Hook) -> Self { + self.hooks.push(hook.fixture); + self + } + /// After every terminal turn arrived, additionally await `count` /// controlled-function invocations (probe-side, whole-run) before /// collecting evidence. Required to observe call-mode reaction dispatches @@ -258,6 +268,23 @@ impl Scenario { self } + /// Wait for the scripted hook hold to park the scripted call, exercise + /// `harness::function::resolve` no-op gates, then release with + /// `action: "execute"`. + #[allow(dead_code)] // used by the unregistered INT-013 fixture + pub(super) fn held_call_resolve(mut self, function_call_id: &str) -> Self { + assert!( + !function_call_id.is_empty(), + "held-call-resolve function_call_id must not be empty" + ); + self.intervention = Some(ScenarioIntervention::HeldCallResolve { + function_call_id: function_call_id.to_string(), + }); + self.expected_turn_statuses = vec!["completed".to_string()]; + self.traces_override = Some(1); + self + } + pub(super) fn terminal_turns(mut self, count: usize) -> Self { assert!(count > 0, "scenario must expect at least one terminal turn"); self.expected_turn_statuses = vec!["completed".to_string(); count]; @@ -327,8 +354,55 @@ impl Scenario { await_target_calls: self.await_target_calls, traces_override: self.traces_override, intervention: self.intervention, + hooks: self.hooks, + } + } +} + +/// Builder for one scripted hook (see [`crate::fixtures::ScenarioHook`]). +#[allow(dead_code)] // used by the unregistered INT-013 fixture +pub(super) struct Hook { + fixture: crate::fixtures::ScenarioHook, +} + +#[allow(dead_code)] // used by the unregistered INT-013 fixture +impl Hook { + pub(super) fn new(name: &str, point: &str, behavior: crate::fixtures::HookBehavior) -> Self { + Self { + fixture: crate::fixtures::ScenarioHook { + name: name.to_string(), + point: point.to_string(), + functions: None, + priority: 0, + timeout_ms: None, + on_error: None, + behavior, + }, } } + + /// Restrict the binding to function ids matching these globs. + pub(super) fn functions<'a>(mut self, globs: impl IntoIterator) -> Self { + self.fixture.functions = Some(globs.into_iter().map(str::to_string).collect()); + self + } + + pub(super) fn priority(mut self, priority: i64) -> Self { + self.fixture.priority = priority; + self + } + + #[allow(dead_code)] + pub(super) fn timeout_ms(mut self, timeout_ms: u64) -> Self { + self.fixture.timeout_ms = Some(timeout_ms); + self + } + + #[allow(dead_code)] + pub(super) fn on_error(mut self, on_error: &str) -> Self { + self.fixture.on_error = Some(on_error.to_string()); + self + } } pub(super) struct Send { @@ -336,6 +410,7 @@ pub(super) struct Send { idempotency_key: Option, allowed_functions: Vec, expose: CompiledFunctionExposureV1, + max_total_tokens: Option, } impl Send { @@ -345,9 +420,17 @@ impl Send { idempotency_key: None, allowed_functions: Vec::new(), expose: CompiledFunctionExposureV1::Native, + max_total_tokens: None, } } + /// Freeze a hard token budget onto the root session + /// (`SendOptions.max_total_tokens`). + pub(super) fn max_total_tokens(mut self, tokens: u64) -> Self { + self.max_total_tokens = Some(tokens); + self + } + pub(super) fn idempotency_key(mut self, key: &str) -> Self { self.idempotency_key = Some(key.to_string()); self @@ -401,6 +484,7 @@ impl Send { deny: Vec::new(), expose: self.expose, }, + max_total_tokens: self.max_total_tokens, }, } } diff --git a/harness/tests/integration/src/scenarios/held_call_resolve.rs b/harness/tests/integration/src/scenarios/held_call_resolve.rs new file mode 100644 index 000000000..afc255f13 --- /dev/null +++ b/harness/tests/integration/src/scenarios/held_call_resolve.rs @@ -0,0 +1,150 @@ +//! INT-013 — a pre-trigger hook holds a function call; `harness::function::resolve` +//! releases it and the turn completes. +//! +//! NOT REGISTERED YET (no `mod` declaration): running this fixture against +//! the live stack exposed what looks like a real harness defect — once the +//! holder hook parks the call, `harness::status` first times out at the +//! engine and later returns `null` (the turn record disappears from state), +//! so the resolve intervention can never find the parked call. Isolation +//! runs proved the scripted-hook plumbing itself is sound: the same fixture +//! with `Continue` behaviors completes in ~2.7s with both hooks served. +//! Register this scenario once the hold-path defect is fixed (MOT-4296). +//! +//! This is the deferred-call seam end to end: the scripted holder hook answers +//! `{"decision":"hold"}`, the harness parks the call (`CallState::Pending`, +//! turn `awaiting_functions`) instead of executing it, and the runner +//! intervention proves the resolve no-op gates (wrong turn id, unknown call +//! id) leave it parked before `action: "execute"` resumes the hook chain +//! AFTER the holder — the second scripted hook runs exactly once, the target +//! executes exactly once, and the turn finishes on the pinned function result. +//! +//! Out of scope by design: `harness::sweep-pending` only expires pending +//! calls with no holder (`held_by: None`), and a hook hold is currently the +//! only pending-call producer — so the sweep's resolved path has no reachable +//! fixture today. + +use serde_json::json; + +use super::dsl::{ + ControlledFunction, Generation, Hook, Message, Model, Request, Response, Scenario, Send, +}; +use super::ScenarioDriver; +use crate::fixtures::{HookBehavior, ScenarioFixture}; + +const CALL_ID: &str = "call-held"; + +pub(super) fn scenario() -> ScenarioFixture { + const ID: &str = "INT-013"; + const MESSAGE: &str = "Record one value; approval will hold it."; + const FINAL_TEXT: &str = "recorded after release"; + + let model = Model::scripted("fixture-model"); + let record = ControlledFunction::new("{{run_id}}::record", "Record one value.") + .request_schema(json!({ + "type": "object", + "additionalProperties": false, + "properties": { "value": { "type": "string" } }, + "required": ["value"] + })) + .returns_text("recorded"); + + Scenario::new( + ID, + "held-call-resolve", + "A pre-trigger hook holds a call and harness::function::resolve releases it through the \ + remaining chain.", + ScenarioDriver::Direct, + model, + ) + .send( + Send::message(MESSAGE) + .idempotency_key("{{run_id}}:integration-013") + .allow_function(&record), + ) + .function(record.clone()) + // The holder parks the call on first sight and must stay silent when the + // chain resumes; the second hook proves the resume starts AFTER the holder. + .hook(Hook::new("holder", "pre-trigger", HookBehavior::HoldOnce)) + .hook(Hook::new("chain", "pre-trigger", HookBehavior::Continue).priority(10)) + .held_call_resolve(CALL_ID) + .generation( + Generation::new(1) + .expect( + Request::new() + .turn_request() + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_exact([Message::user(MESSAGE)]) + .tools_exact_after_controls([], [record.tool()]), + ) + .respond(Response::function_call( + CALL_ID, + &record, + json!({ "value": "held-then-released" }), + 8, + 4, + )), + ) + .generation( + Generation::new(2) + .expect( + Request::new() + .turn_request_step(1) + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_subset([ + json!({ "role": "user" }), + json!({ "role": "assistant", "content": [ + { "type": "function_call", "id": CALL_ID } + ] }), + json!({ "role": "function_result", "function_call_id": CALL_ID, + "is_error": false }), + ]) + .tools_exact_after_controls([], [record.tool()]), + ) + .respond(Response::text(FINAL_TEXT, 10, 2)), + ) + .verify(|run| { + run.expect_assistant_texts([FINAL_TEXT])?; + run.expect_message_counts(1, 2, 1)?; + // The target executed exactly once, only after the release. + run.expect_target_calls(1)?; + anyhow::ensure!( + run.target_calls[0] == json!({ "value": "held-then-released" }), + "target payload {:?} != held-then-released", + run.target_calls[0] + ); + // The holder saw the original dispatch; the chain hook only ran on the + // post-resolve resume — each exactly once. + run.expect_hook_calls("holder", 1)?; + run.expect_hook_calls("chain", 1)?; + let control = &run.control; + anyhow::ensure!( + control.get("kind").and_then(serde_json::Value::as_str) == Some("held_call_resolve"), + "intervention control missing: {control}" + ); + anyhow::ensure!( + control + .pointer("/execute_resolve/turn_resumed") + .and_then(serde_json::Value::as_bool) + == Some(true), + "execute resolve did not resume the turn: {control}" + ); + run.expect_no_duplicate_messages() + }) + .build() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixture_holds_and_resolves_one_call() { + let fixture = scenario(); + fixture.validate().unwrap(); + assert_eq!(fixture.expected_terminal_turns, 1); + assert_eq!(fixture.hooks.len(), 2); + assert_eq!(fixture.hooks[0].behavior, HookBehavior::HoldOnce); + assert!(fixture.intervention.is_some()); + assert_eq!(fixture.expected_traces(), 1); + } +} diff --git a/harness/tests/integration/src/scenarios/mod.rs b/harness/tests/integration/src/scenarios/mod.rs index b560a0e43..67986a492 100644 --- a/harness/tests/integration/src/scenarios/mod.rs +++ b/harness/tests/integration/src/scenarios/mod.rs @@ -1,5 +1,6 @@ //! The checked-in integration fixtures. +mod budget_preflight_exceeded; mod coalesced_fire; mod console_streamed_text; mod dsl; @@ -8,6 +9,7 @@ mod exactly_once_function; mod join_spec_mismatch; mod late_join_replay; mod multi_turn_traces; +mod notify_grant_teardown; mod queued_message_edit_unqueue; mod reaction_policy_inheritance; mod reaction_unregisters_run; @@ -31,6 +33,7 @@ pub enum ScenarioDriver { /// Every fixture, in stable slug order. pub fn all() -> Vec { vec![ + budget_preflight_exceeded::scenario(), coalesced_fire::scenario(), console_streamed_text::scenario(), engine_restart_recovery::scenario(), @@ -38,6 +41,7 @@ pub fn all() -> Vec { join_spec_mismatch::scenario(), late_join_replay::scenario(), multi_turn_traces::scenario(), + notify_grant_teardown::scenario(), reaction_policy_inheritance::scenario(), reaction_unregisters_run::scenario(), state_worker_sidecar::scenario(), @@ -55,7 +59,7 @@ mod tests { #[test] fn every_fixture_is_unique_and_valid() { let fixtures = all(); - assert_eq!(fixtures.len(), 14); + assert_eq!(fixtures.len(), 16); let mut slugs = std::collections::BTreeSet::new(); let mut ids = std::collections::BTreeSet::new(); for fixture in fixtures { diff --git a/harness/tests/integration/src/scenarios/notify_grant_teardown.rs b/harness/tests/integration/src/scenarios/notify_grant_teardown.rs new file mode 100644 index 000000000..046dde034 --- /dev/null +++ b/harness/tests/integration/src/scenarios/notify_grant_teardown.rs @@ -0,0 +1,273 @@ +//! INT-016 — a notify subscription fires into the owning session, filesystem +//! grants round-trip, the shared budget ledger reconciles, and +//! `harness::teardown` sweeps the surviving binding. +//! +//! One run covers four cold seams at once: +//! - `subscriptions/notify_agent.rs`: the state fire lands as a durable +//! `trigger_fired` custom entry plus an injected `[notification: …]` user +//! message that seeds the second tracked turn. Both bindings are standing +//! (`once: false`): an armed one-shot wake would park the session and keep +//! the registering turn non-terminal, stalling the probe boundary. +//! - `functions/filesystem.rs` + `filesystem_grants.rs`: grant → grants → +//! revoke round-trip through recorded probe responses. +//! - `budget.rs` happy path: a generous frozen budget admits every +//! generation; after both turns the ledger holds the exact reconciled +//! usage with no outstanding reservation. +//! - `teardown.rs` + `subscriptions/reconcile.rs::sweep_owner`: the second, +//! never-fired `once: false` subscription survives until `harness::teardown` +//! removes it. + +use serde_json::{json, Value}; + +use super::dsl::{Generation, Message, Model, Request, Response, Scenario, Send}; +use super::ScenarioDriver; +use crate::fixtures::ScenarioFixture; + +const REGISTER: &str = "engine::register_trigger"; +const SCOPE: &str = "integration-016"; +const FIRE_KEY: &str = "fire"; +const IDLE_KEY: &str = "idle"; +const BUDGET_TOKENS: u64 = 200_000; +/// Scripted usage across the three generations: (8+4) + (10+2) + (12+3). +const EXPECTED_USED_TOKENS: u64 = 39; + +pub(super) fn scenario() -> ScenarioFixture { + const ID: &str = "INT-016"; + const MESSAGE: &str = "Arm two standing state notifications."; + + let subscribe_fire = json!({ + "trigger_type": "state", + "config": { "scope": SCOPE, "key": FIRE_KEY }, + "label": "fire-key changed", + "once": false + }); + let subscribe_idle = json!({ + "trigger_type": "state", + "config": { "scope": SCOPE, "key": IDLE_KEY }, + "label": "idle-key changed", + "once": false + }); + + Scenario::new( + ID, + "notify-grant-teardown", + "A notify subscription fires an injected notification turn; filesystem grants \ + round-trip; the budget ledger reconciles; teardown sweeps the standing binding.", + ScenarioDriver::Direct, + Model::scripted("fixture-model"), + ) + .send( + Send::message(MESSAGE) + .idempotency_key("{{run_id}}:integration-016") + .allow_id(REGISTER) + .max_total_tokens(BUDGET_TOKENS), + ) + .terminal_turns(2) + // Two trees: the registering turn and the injected-notification turn. + // Probe calls here are plain RPCs — none seeds an extra trace. + .expect_traces(2) + // Turn 1 done: trip the once subscription's key. The notify fire injects + // the user message that seeds turn 2. + .probe_after( + 1, + "state::set", + json!({ "scope": SCOPE, "key": FIRE_KEY, "value": { "seq": 1 } }), + ) + // Turn 2 done: grants round-trip, ledger, and teardown — all recorded. + .probe_after( + 2, + "harness::filesystem::grant", + json!({ "session_id": "{{session_id}}", "root": "/tmp/int-016" }), + ) + .probe_after( + 2, + "harness::filesystem::grants", + json!({ "session_id": "{{session_id}}" }), + ) + .probe_after( + 2, + "harness::filesystem::revoke", + json!({ "session_id": "{{session_id}}", "root": "/tmp/int-016" }), + ) + .probe_after( + 2, + "state::get", + json!({ "scope": "harness_budget", "key": "{{session_id}}" }), + ) + .probe_after( + 2, + "harness::teardown", + json!({ "root_session_id": "{{session_id}}" }), + ) + .generation( + Generation::new(1) + .expect( + Request::new() + .turn_request() + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_exact([Message::user(MESSAGE)]) + .tools_exact_after_controls([REGISTER], Vec::new()), + ) + .respond(Response::function_calls_raw( + vec![ + ("call-sub-fire", REGISTER, subscribe_fire), + ("call-sub-idle", REGISTER, subscribe_idle), + ], + 8, + 4, + )), + ) + .generation( + Generation::new(2) + .expect( + Request::new() + .turn_request_step(1) + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_subset([ + json!({ "role": "user" }), + json!({ "role": "assistant", "content": [ + { "type": "function_call", "id": "call-sub-fire" }, + { "type": "function_call", "id": "call-sub-idle" } + ] }), + json!({ "role": "function_result", "function_call_id": "call-sub-fire", + "is_error": false }), + json!({ "role": "function_result", "function_call_id": "call-sub-idle", + "is_error": false }), + ]) + .tools_exact_after_controls([REGISTER], Vec::new()), + ) + .respond(Response::streamed_text("armed", ["armed"], 10, 2)), + ) + // The injected-notification turn: a fresh turn in the SAME session whose + // user message is the notify text. + .generation( + Generation::new(3) + .expect( + Request::new() + .turn_request_step(0) + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_subset([json!({ "role": "user" })]) + .tools_exact_after_controls([REGISTER], Vec::new()), + ) + .respond(Response::streamed_text("acknowledged", ["acknowledged"], 12, 3)), + ) + .verify(|run| { + run.expect_assistant_texts(["armed", "acknowledged"])?; + + // The fire left a durable trigger_fired entry for the retired once-sub. + let fired = run + .transcript + .iter() + .filter_map(|item| { + let custom_type = item + .pointer("/custom/custom_type") + .and_then(Value::as_str); + (custom_type == Some("trigger_fired")) + .then(|| item.pointer("/custom/data")) + .flatten() + }) + .collect::>(); + anyhow::ensure!( + fired.len() == 1, + "expected one trigger_fired entry, found {}", + fired.len() + ); + anyhow::ensure!( + fired[0].get("once") == Some(&Value::Bool(false)) + && fired[0].get("retired") == Some(&Value::Bool(false)), + "a standing subscription must not retire on fire: {}", + fired[0] + ); + + // The injected notification message carries the label. + let injected = run + .transcript + .iter() + .filter_map(|item| item.get("message")) + .filter(|message| message.get("role").and_then(Value::as_str) == Some("user")) + .map(crate::evidence_data::message_text) + .find(|text| text.starts_with("[notification:")); + anyhow::ensure!( + injected + .as_deref() + .is_some_and(|text| text.contains("fire-key changed")), + "injected notification message missing or unlabeled: {injected:?}" + ); + + let roots_of = |response: &Value| -> Vec { + response + .get("roots") + .and_then(Value::as_array) + .map(|roots| { + roots + .iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default() + }; + run.expect_probe_response(1, "harness::filesystem::grant", |response| { + anyhow::ensure!( + roots_of(response) == ["/tmp/int-016"], + "grant did not persist the root: {response}" + ); + Ok(()) + })?; + run.expect_probe_response(2, "harness::filesystem::grants", |response| { + anyhow::ensure!( + roots_of(response) == ["/tmp/int-016"], + "grants did not list the root: {response}" + ); + Ok(()) + })?; + run.expect_probe_response(3, "harness::filesystem::revoke", |response| { + anyhow::ensure!( + roots_of(response).is_empty(), + "revoke did not drop the root: {response}" + ); + Ok(()) + })?; + run.expect_probe_response(4, "state::get", |response| { + let ledger = response.get("value").unwrap_or(response); + anyhow::ensure!( + ledger.get("max_total_tokens").and_then(Value::as_u64) == Some(BUDGET_TOKENS), + "ledger max_total_tokens != {BUDGET_TOKENS}: {ledger}" + ); + anyhow::ensure!( + ledger.get("used_tokens").and_then(Value::as_u64) + == Some(EXPECTED_USED_TOKENS), + "ledger used_tokens != {EXPECTED_USED_TOKENS}: {ledger}" + ); + anyhow::ensure!( + ledger.get("reserved_tokens").and_then(Value::as_u64) == Some(0), + "no reservation may remain after terminal turns: {ledger}" + ); + Ok(()) + })?; + run.expect_probe_response(5, "harness::teardown", |response| { + anyhow::ensure!( + response.get("removed").and_then(Value::as_u64) >= Some(2), + "teardown must sweep both standing subscriptions: {response}" + ); + Ok(()) + })?; + run.expect_no_duplicate_messages() + }) + .build() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixture_declares_two_turns_and_six_probe_actions() { + let fixture = scenario(); + fixture.validate().unwrap(); + assert_eq!(fixture.expected_terminal_turns, 2); + assert_eq!(fixture.probe_actions.len(), 6); + assert_eq!(fixture.expected_traces(), 2); + assert_eq!(fixture.script.generations.len(), 3); + } +} diff --git a/harness/tests/integration/src/types/scenario/compiled.rs b/harness/tests/integration/src/types/scenario/compiled.rs index 496ea716f..acb975f7a 100644 --- a/harness/tests/integration/src/types/scenario/compiled.rs +++ b/harness/tests/integration/src/types/scenario/compiled.rs @@ -51,6 +51,9 @@ pub struct CompiledSendV1 { #[serde(deny_unknown_fields)] pub struct CompiledSendOptionsV1 { pub functions: CompiledFunctionPolicyV1, + /// Hard token budget for the root session (`SendOptions.max_total_tokens`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_total_tokens: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] diff --git a/harness/tests/integration/tests/determinism.rs b/harness/tests/integration/tests/determinism.rs index bb442560b..d77acfbf4 100644 --- a/harness/tests/integration/tests/determinism.rs +++ b/harness/tests/integration/tests/determinism.rs @@ -101,6 +101,8 @@ fn evidence(run_id: &str, session_id: &str, turn_id: &str) -> RunEvidence { tree_sessions: Vec::new(), tree_statuses: Vec::new(), router_evidence: Value::Null, + probe_responses: Vec::new(), + hook_calls: Vec::new(), } }