diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index a5162b892c..4fa74553cf 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -452,6 +452,7 @@ jobs: openhuman/agent/harness/subagent_runner/tool_prep.rs openhuman/agent_registry/agents/loader.rs openhuman/inference/local/mod.rs + openhuman/tinyagents/contract_gate_tests.rs openhuman/tool_registry/ops_tests.rs openhuman/tool_registry/schemas.rs openhuman/tools/ops_tests.rs diff --git a/Cargo.lock b/Cargo.lock index d426c29ebb..7890b54e56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4507,6 +4507,7 @@ dependencies = [ "windows-sys 0.61.2", "wiremock", "x25519-dalek", + "xxhash-rust", "xz2", "zeroize", "zip 2.4.2", @@ -9121,6 +9122,12 @@ version = "0.8.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ae8337f8a065cfc972643663ea4279e04e7256de865aa66fe25cec5fb912d3f" +[[package]] +name = "xxhash-rust" +version = "0.8.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985eec839aaf2a1270af8f4ebcf63cf9401cfd90f0902f97c28d9f104ffbde72" + [[package]] name = "xz2" version = "0.1.7" diff --git a/Cargo.toml b/Cargo.toml index 14b8e35b92..4a6773dcd3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -134,6 +134,10 @@ argon2 = "0.5" rand = "0.10" dirs = "5" sha2 = "0.10" +# Fast, stable (cross-process reproducible) non-cryptographic hash. Used by the +# contract gate to fingerprint a delivered contract body so a later rescan can +# detect that summarization / a size cap / rewriting mutated it. +xxhash-rust = { version = "0.8", features = ["xxh3"] } # Line-level text diffs for the memory_diff module (modified-item unified diffs). similar = "2" # Git-backed change ledger for the memory_diff module: snapshots are commits, diff --git a/src/openhuman/agent/harness/subagent_runner/handoff.rs b/src/openhuman/agent/harness/subagent_runner/handoff.rs index 26d30b8110..04fede3d29 100644 --- a/src/openhuman/agent/harness/subagent_runner/handoff.rs +++ b/src/openhuman/agent/harness/subagent_runner/handoff.rs @@ -285,3 +285,30 @@ pub(super) fn chunk_content(content: &str, budget: usize) -> Vec { chunks } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn clean_tool_output_collapses_indentation_whitespace() { + // Documents the hazard the contract gate's `trusted_verbatim` guard exists + // for: `clean_tool_output` collapses every run of spaces/tabs to ONE space, + // flattening a pretty-printed JSON schema's 2/4/6-space nesting to 1 space. + // A contract-gate delivery run through this would change the bytes after its + // `[contract-gate:…]` marker, so the rescan digest would miss and re-deliver + // forever — hence `HandoffMiddleware::after_tool` must skip trusted results. + // If this assertion ever changes, revisit that guard. + let pretty = + "{\n \"properties\": {\n \"max_results\": {\n \"default\": 10\n }\n }\n}"; + let cleaned = clean_tool_output(pretty); + assert!( + cleaned.contains("\n \"properties\": {"), + "nested indentation must collapse to a single space, got:\n{cleaned}" + ); + assert!( + !cleaned.contains(" \"properties\""), + "no run of 2+ spaces should survive, got:\n{cleaned}" + ); + } +} diff --git a/src/openhuman/agent/harness/subagent_runner/ops/provider.rs b/src/openhuman/agent/harness/subagent_runner/ops/provider.rs index e8acc67533..1cc453fd34 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/provider.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/provider.rs @@ -247,14 +247,11 @@ pub(crate) struct LazyToolkitResolver { const TIER4_MIN_SLUG_LEN: usize = 8; impl LazyToolkitResolver { - /// NOTE (contract gate, #4853): this builds a *fresh* `ComposioActionTool` - /// — and therefore a fresh, empty `ContractGate` — on every call. The gate's - /// surface-once state lives in the tool instance, so if a future wiring - /// resolves a new tool per invocation the gate would surface the full - /// contract on every call and the retry would never see `first_time == false` - /// to proceed. When this path is wired to actually dispatch, cache the - /// resolved tool (and its gate) per turn so a given action can proceed after - /// its contract has been surfaced once. + /// NOTE (contract gate, #4853): this may build a *fresh* `ComposioActionTool` + /// per call, which is fine — per-action contract gating now lives in the + /// `tinyagents::contract_gate` middleware and derives "already surfaced" from + /// the transcript, not from any per-tool-instance state. So a new tool per + /// invocation no longer risks re-surfacing the contract every call. pub(super) fn resolve(&self, name: &str) -> Option> { let action = self.find_action(name)?; Some(Box::new( diff --git a/src/openhuman/agent/tools/run_workflow.rs b/src/openhuman/agent/tools/run_workflow.rs index f148854f0c..10ad85f14a 100644 --- a/src/openhuman/agent/tools/run_workflow.rs +++ b/src/openhuman/agent/tools/run_workflow.rs @@ -242,6 +242,13 @@ impl Tool for RunWorkflowTool { RUN_WORKFLOW_TOOL_NAME } + /// Expose this profile's runnable-workflow allowlist to the contract gate so + /// it never renders the input contract of a workflow this profile scopes out + /// (issue #4853; matches the `execute` allowlist check below). + fn workflow_gate_allowlist(&self) -> Option<&std::collections::HashSet> { + self.skill_allowlist.as_ref() + } + fn description(&self) -> &str { "Run another workflow as a subagent and wait for its result, the way a \ function call waits on its callee. Spawns the target workflow as a \ diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index bf2139eb1f..8ec1126306 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -67,13 +67,6 @@ pub struct ComposioActionTool { /// Composio connection. Used when the sub-agent is spawned for a /// particular account (e.g. "send from my work Gmail"). connection_id: Option, - /// Per-turn contract gate (#4853). On the first call to this action the - /// gate surfaces the action's FULL live input schema/description so the - /// model composes well-formed arguments (e.g. correctly-quoted Gmail - /// queries) instead of guessing from the thin spawn-time schema; the retry - /// executes normally. Held per tool instance, which lives for one - /// `integrations_agent` spawn, so "seen" is scoped to that turn. - gate: super::contract_gate::ContractGate, } impl ComposioActionTool { @@ -100,7 +93,6 @@ impl ComposioActionTool { description, parameters, connection_id, - gate: super::contract_gate::ContractGate::new(), } } } @@ -123,6 +115,10 @@ impl Tool for ComposioActionTool { &self.action_name } + fn composio_action_slug(&self) -> Option<&str> { + Some(&self.action_name) + } + fn description(&self) -> &str { &self.description } @@ -192,15 +188,13 @@ impl Tool for ComposioActionTool { } } - // [#1710 Wave 4 / #4853] Reload the live config snapshot ONCE, up front, - // and use it for BOTH the contract-gate lookup and dispatch. A mid-session - // `composio.mode` / credential / workspace change must route the gate's - // live-catalog fetch and the actual execution through the SAME config; - // consulting the captured spawn-time `self.config` here would let the gate - // resolve (or skip) a contract against stale routing while dispatch used - // fresh routing. Anchored to this tool's original config path rather than - // re-resolving process-global `OPENHUMAN_WORKSPACE` (the tool is scoped to - // the user/workspace it was created for). + // [#1710 Wave 4] Reload the live config snapshot ONCE, up front, and use it + // for dispatch. A mid-session `composio.mode` / credential / workspace + // change must route execution through fresh config; consulting the captured + // spawn-time `self.config` here would dispatch against stale routing. + // Anchored to this tool's original config path rather than re-resolving + // process-global `OPENHUMAN_WORKSPACE` (the tool is scoped to the + // user/workspace it was created for). let live_config = match config_rpc::reload_config_snapshot_with_timeout(self.config.as_ref()).await { Ok(c) => c, @@ -217,26 +211,14 @@ impl Tool for ComposioActionTool { } }; - // Contract gate (#4853): the per-action tool is built from the thin - // spawn-time `list_tools` schema (often `{"type":"object"}` with no - // field descriptions), so the model guesses argument formats — most - // visibly sending unquoted Gmail `query` strings that return zero - // results. On the first call this turn, surface the action's FULL live - // contract (input schema + description) as a recoverable tool error and - // let the retry — now with the schema in context — execute. Degrades to - // a normal execute whenever the contract can't be resolved (see - // `contract_gate::consult`), so an unconfigured/offline client never - // blocks the action. Uses `live_config` so gate routing matches dispatch. - match super::contract_gate::consult(&self.gate, &live_config, &self.action_name).await { - super::contract_gate::GateDecision::Surface(contract) => { - tracing::info!( - tool = %self.action_name, - "[composio][contract-gate] returning full contract before first execute" - ); - return Ok(ToolResult::error(contract)); - } - super::contract_gate::GateDecision::Proceed => {} - } + // NOTE: per-action contract gating now lives in the unified + // `tinyagents::contract_gate` middleware (issue #4853), whose + // transcript-derived presence + payload-digest integrity re-delivers the + // contract if it is summarised or compacted out of context — closing the + // gap the in-memory `composio::contract_gate` (#4995) documented. That + // module is left in place but no longer consulted here (removal tracked as + // follow-up); the middleware short-circuits the first call before execute, + // so this path only runs once the contract is already in context. // Inject `timeZone` / `singleEvents` defaults for Google // Calendar list slugs (issue #1714). The per-action surface is @@ -258,9 +240,9 @@ impl Tool for ComposioActionTool { // Resolve the client through the mode-aware factory on every call so a // direct-mode toggle takes effect immediately (#1710), reusing the - // `live_config` snapshot reloaded above so the gate lookup and this - // dispatch share identical routing. The pre-baked-client variant routed - // all executions through the backend tinyhumans tenant regardless of mode. + // `live_config` snapshot reloaded above so a mid-session mode/credential + // change routes this dispatch through fresh config. The pre-baked-client + // variant routed all executions through the backend tenant regardless of mode. let kind = match create_composio_client(&live_config) { Ok(kind) => kind, Err(e) => { @@ -523,77 +505,6 @@ mod tests { ); } - // Seeds the flows/tinyflows live-catalog cache, so it only builds with the - // `flows` feature on (the gate degrades to a no-op when flows is off). - #[cfg(feature = "flows")] - #[tokio::test] - async fn contract_gate_surfaces_full_contract_then_proceeds_on_retry() { - // Regression for #4853: the FIRST per-action execute this turn must - // return the action's FULL live contract (so the model composes a - // well-formed query) instead of running with the thin spawn-time - // schema; the retry then proceeds to real dispatch. A unique toolkit is - // seeded so this is deterministic and never touches the network. - use crate::openhuman::config::TEST_ENV_LOCK; - use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; - let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - - let toolkit = "cgateexec"; - let slug = "CGATEEXEC_FETCH_ITEMS"; - seed_live_catalog_cache( - toolkit, - vec![ToolContract { - slug: slug.to_string(), - toolkit: toolkit.to_string(), - description: Some("Search items. Quote multi-word phrases.".to_string()), - required_args: vec!["query".to_string()], - input_schema: Some(serde_json::json!({ - "type": "object", - "properties": { "query": { "type": "string" } }, - "required": ["query"] - })), - output_fields: Vec::new(), - output_schema: None, - primary_array_path: None, - is_curated: false, - }], - ); - - let tmp = tempfile::tempdir().expect("tempdir"); - let _workspace_guard = WorkspaceEnvGuard::set(tmp.path()); - let mut config = Config::default(); - config.config_path = tmp.path().join("config.toml"); - config.workspace_dir = tmp.path().join("workspace"); - config.save().await.expect("save fake config to disk"); - - let t = ComposioActionTool::new( - Arc::new(config), - slug.to_string(), - "search items".to_string(), - None, - ); - - // First call: gate surfaces the contract (recoverable tool error). - let first = t.execute(serde_json::json!({})).await.unwrap(); - assert!( - first.is_error, - "first call must surface a recoverable error" - ); - let first_msg = error_text(&first); - assert!( - first_msg.contains("Input JSON schema"), - "first call must carry the full contract, got: {first_msg}" - ); - - // Retry: gate proceeds; dispatch fails downstream (no session token) but - // crucially NOT with the contract text — proving the gate did not block. - let second = t.execute(serde_json::json!({})).await.unwrap(); - let second_msg = error_text(&second); - assert!( - !second_msg.contains("Input JSON schema"), - "retry must proceed past the gate to real dispatch, got: {second_msg}" - ); - } - // ── Factory routing (#1710) ────────────────────────────────────── // // Regression coverage for the bug fix: `ComposioActionTool` now diff --git a/src/openhuman/composio/tools.rs b/src/openhuman/composio/tools.rs index 89ad36f3bd..b2298c1eb9 100644 --- a/src/openhuman/composio/tools.rs +++ b/src/openhuman/composio/tools.rs @@ -35,6 +35,7 @@ use crate::openhuman::agent::harness::current_task_recency_window; use crate::openhuman::agent::harness::definition::SandboxMode; use crate::openhuman::config::rpc as config_rpc; use crate::openhuman::config::Config; +use crate::openhuman::tinyagents::contract_gate; use crate::openhuman::tools::traits::{ PermissionLevel, Tool, ToolCallOptions, ToolCategory, ToolResult, }; @@ -1229,9 +1230,26 @@ impl Tool for ComposioListToolsTool { } } - let mut result = ToolResult::success( - serde_json::to_string(&resp).unwrap_or_else(|_| "{}".into()), - ); + let json = serde_json::to_string(&resp).unwrap_or_else(|_| "{}".into()); + // Pre-credit the contract gate ONLY when the model will actually + // read the full JSON schemas (i.e. NOT the thin markdown path): + // then a `composio_execute` for any listed slug skips a redundant + // contract re-delivery. Under `prefer_markdown` the model reads + // the compact rendering (schemas dropped) via `output_for_llm`, + // which is exactly the thin listing the gate guards against — so + // leave the marker off there. Marker leads for the prefix scan + // (issue #4853). + let content = if options.prefer_markdown { + json + } else { + contract_gate::prefix_with_present_marker( + resp.tools + .iter() + .map(|t| contract_gate::composio_key(&t.function.name)), + &json, + ) + }; + let mut result = ToolResult::success(content); if options.prefer_markdown { result.markdown_formatted = Some(render_tools_markdown(&resp)); } diff --git a/src/openhuman/mcp_registry/tools.rs b/src/openhuman/mcp_registry/tools.rs index a2698a14cc..35f7317416 100644 --- a/src/openhuman/mcp_registry/tools.rs +++ b/src/openhuman/mcp_registry/tools.rs @@ -21,10 +21,33 @@ use async_trait::async_trait; use serde_json::{json, Value}; use crate::openhuman::config::Config; +use crate::openhuman::tinyagents::contract_gate; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; use super::ops; +/// Contract-gate presence keys for a `mcp_clients_list_tools` result: one key +/// per `(server_id, tool.name)` the listing fully described (each entry carries +/// the tool's full input schema). Pre-crediting these lets a +/// `mcp_registry_tool_call` right after skip a redundant contract re-delivery +/// (issue #4853). Empty when the value has no server/tools. +fn mcp_list_tools_keys(value: &Value) -> Vec { + let Some(server) = value.get("server_id").and_then(Value::as_str) else { + return Vec::new(); + }; + value + .get("tools") + .and_then(Value::as_array) + .map(|tools| { + tools + .iter() + .filter_map(|t| t.get("name").and_then(Value::as_str)) + .map(|name| contract_gate::mcp_key(server, name)) + .collect() + }) + .unwrap_or_default() +} + macro_rules! emit { ($outcome:expr, $name:literal) => {{ let outcome = $outcome.map_err(|e| anyhow::anyhow!(concat!($name, ": {}"), e))?; @@ -216,10 +239,15 @@ impl Tool for McpRegistryListToolsTool { } async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let sid = req_str(&args, "server_id")?; - emit!( - ops::mcp_clients_list_tools(sid).await, - "mcp_registry_list_tools" - ) + let outcome = ops::mcp_clients_list_tools(sid) + .await + .map_err(|e| anyhow::anyhow!("mcp_registry_list_tools: {}", e))?; + let body = serde_json::to_string(&outcome.value)?; + // Full input schemas are in `body`, so pre-credit every (server, tool) + // it described; the marker leads the message for the gate's prefix scan. + let body = + contract_gate::prefix_with_present_marker(mcp_list_tools_keys(&outcome.value), &body); + Ok(ToolResult::success(body)) } fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { true @@ -494,6 +522,29 @@ mod tests { assert_eq!(McpRegistrySearchTool::new(cfg()).scope(), ToolScope::All); } + #[test] + fn list_tools_keys_credits_every_tool_for_the_server() { + // The pure key builder over a `mcp_clients_list_tools` value: one + // contract-gate key per (server_id, tool.name), so a + // `mcp_registry_tool_call` right after skips a redundant re-delivery + // (issue #4853). + let value = json!({ + "server_id": "srv", + "tools": [ + { "name": "do_thing", "input_schema": { "type": "object" } }, + { "name": "other", "input_schema": { "type": "object" } }, + ] + }); + assert_eq!( + mcp_list_tools_keys(&value), + vec!["mcp:srv:do_thing".to_string(), "mcp:srv:other".to_string()] + ); + + // No tools / no server → no keys (caller pre-credits nothing). + assert!(mcp_list_tools_keys(&json!({ "server_id": "srv", "tools": [] })).is_empty()); + assert!(mcp_list_tools_keys(&json!({ "tools": [] })).is_empty()); + } + #[tokio::test] async fn get_requires_qualified_name() { let err = McpRegistryGetTool::new(cfg()) diff --git a/src/openhuman/skills/tools.rs b/src/openhuman/skills/tools.rs index 144a0b3bbd..b4fed99de6 100644 --- a/src/openhuman/skills/tools.rs +++ b/src/openhuman/skills/tools.rs @@ -21,6 +21,7 @@ use async_trait::async_trait; use serde_json::json; use crate::openhuman::config::Config; +use crate::openhuman::tinyagents::contract_gate::{prefix_with_present_marker, workflow_key}; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; use super::ops_create::{create_workflow, CreateWorkflowParams}; @@ -177,11 +178,17 @@ impl Tool for WorkflowDescribeTool { } let def = get_workflow(&self.workspace_dir, &skill_id) .ok_or_else(|| anyhow::anyhow!("describe_workflow: workflow `{skill_id}` not found"))?; - Ok(ToolResult::success(serde_json::to_string(&json!({ + let body = serde_json::to_string(&json!({ "definition": def.definition, "inputs": def.inputs, "github_gated": def.github.is_some(), - }))?)) + }))?; + // This is the workflow's FULL input contract (same source the contract + // gate delivers), so pre-credit it: a `run_workflow` right after won't + // pay a redundant re-delivery. The marker leads the message so the + // gate's fixed-length prefix scan picks it up (issue #4853). + let body = prefix_with_present_marker([workflow_key(&skill_id)], &body); + Ok(ToolResult::success(body)) } fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { @@ -582,6 +589,54 @@ mod tests { ); } + #[tokio::test] + async fn describe_workflow_prepends_the_contract_gate_marker() { + // describe_workflow returns the workflow's FULL input contract, so its + // output must lead with the gate marker for that id — a `run_workflow` + // right after then skips a redundant contract re-delivery (issue #4853). + let ws = tempfile::tempdir().unwrap(); + // Seed a trusted project-scope workflow on disk (hermetic; matches the + // discovery roots `get_workflow` reads). + let skill_dir = ws + .path() + .join(".openhuman") + .join("skills") + .join("triage-inbox"); + std::fs::create_dir_all(&skill_dir).unwrap(); + std::fs::write(ws.path().join(".openhuman").join("trust"), "").unwrap(); + std::fs::write( + skill_dir.join("SKILL.md"), + "---\nname: triage-inbox\ndescription: Summarise the inbox.\n---\n\nSummarise.\n", + ) + .unwrap(); + + let mut config = Config::default(); + config.workspace_dir = ws.path().to_path_buf(); + let res = WorkflowDescribeTool::new(Arc::new(config)) + .execute(json!({ "workflow_id": "triage-inbox" })) + .await + .expect("execute"); + + assert!( + !res.is_error, + "describe should succeed for a seeded workflow" + ); + let text = res.text(); + // Leads with the contract-gate marker (now `[contract-gate::]`) + // carrying the workflow's gate key, so a `run_workflow` right after is + // pre-credited. + assert!( + text.starts_with("[contract-gate:"), + "output must lead with the contract-gate marker, got: {text}" + ); + let marker = &text[..text.find(']').expect("marker is closed")]; + assert!( + marker.ends_with(":workflow:triage-inbox"), + "marker must carry the workflow gate key, got: {marker}" + ); + assert!(text.contains("inputs"), "full contract still follows"); + } + #[test] fn names_and_levels() { let c = cfg(); diff --git a/src/openhuman/tinyagents/contract_gate.rs b/src/openhuman/tinyagents/contract_gate.rs new file mode 100644 index 0000000000..c20fde4267 --- /dev/null +++ b/src/openhuman/tinyagents/contract_gate.rs @@ -0,0 +1,901 @@ +//! Contract gate middleware — require the full tool contract before a +//! late-bound tool call executes (issue #4853). +//! +//! Several tool surfaces expose only a **thin** representation to the model — a +//! generic dispatcher with an opaque `arguments`/`inputs` object, or a +//! per-action tool built from a one-line `list_tools` description whose +//! parameter schema may be empty. The **full** contract (input JSON schema + +//! description) lives behind a separate discovery/describe meta-tool. So the +//! model composes a call *before* the real schema is in its context, and guesses +//! argument formats and slugs — e.g. a Composio Gmail search sent without the +//! quotes Gmail's query syntax needs, or an invented `GMAIL_LIST_MESSAGES` in +//! place of the real `GMAIL_FETCH_EMAILS`. +//! +//! This middleware closes that gap. The **first** time a gated target (a +//! Composio action slug, an MCP-registry remote tool, or a Workflow) is invoked +//! while its contract is **not present in the transcript**, execution is +//! short-circuited and the model is handed that target's full contract as a tool +//! error; the retry — now with the contract in context — runs for real. An +//! unknown target returns a "no such action" error plus the valid list so the +//! model can correct itself. +//! +//! **Presence is derived from the transcript, not tracked as turn state.** A +//! delivered contract leads its tool message with a `[contract-gate:]` +//! marker carrying only the slug(s) — **no digest** — immediately followed by the +//! contract payload. The payload's hash is recorded in the process-global +//! [`DELIVERED`] map, keyed by that exact slug list. Before every model call the +//! gate rescans the tool-role messages of `request.messages`: for each message +//! that leads with the marker (a fixed-length prefix compare, one marker per +//! message) it re-hashes the marker's trailing payload and credits the slug(s) +//! **only if that hash still matches** the value `DELIVERED` recorded at delivery. +//! So a contract summarized, truncated by a size cap, or whitespace-collapsed by +//! the sub-agent handoff cleaner no longer matches and is re-delivered — the +//! transcript marker stays short and whitespace-free while the integrity check +//! rides on the payload hash, not on the marker text. +//! +//! Two things emit that marker: a **gate delivery** (one slug), and a +//! **full-schema discovery tool** that pre-credits every target it fully +//! described, packing the slugs into one marker (comma-joined, sorted) so the +//! same prefix compare still recognises it. So `describe_workflow` → +//! `run_workflow`, or `mcp_registry_list_tools` → `mcp_registry_tool_call`, pays +//! no redundant re-delivery once the model has already read the full contract. A +//! **thin** listing (arg names only, schema dropped) never emits a marker — +//! crediting it would mark a not-yet-seen contract present and defeat the gate; +//! presence is scanned from the model-facing text +//! ([`ToolResult::output_for_llm`](crate::openhuman::tools::ToolResult), the same +//! text the model reads), so a tool that shows thin markdown under +//! `prefer_markdown` simply leaves the marker out of that rendering. +//! +//! This is correct across every context-management path *by construction*: a +//! contract that survives byte-for-byte in a resumed sub-agent's +//! `initial_history` still hash-matches and is seen (no needless re-delivery); one +//! folded away by the summarizer, blanked by microcompact, or dropped by the +//! hard-trim is no longer seen (re-delivered). Only tool-role messages are +//! scanned, so a model echoing the marker in its own text cannot spoof presence. +//! The in-context `present` set is per-run — rebuilt every `before_model` from the +//! transcript (the harness rebuilds every middleware in `assemble_turn_harness`), +//! so nothing leaks across turns or between concurrent background agents. The +//! `DELIVERED` hash map is process-global (a contract delivered in one run stays +//! creditable in a later run or resume), but it only ever credits a marker whose +//! payload is present **and** byte-identical in that run's own transcript — so a +//! stale entry can never mark an absent contract present. +//! +//! Gated surfaces (each pairs an opaque dispatcher with a `*_list`/`describe` +//! meta-tool): +//! - **Composio** — `composio_execute` (slug in `tool`) and the registered +//! per-action Composio tools (name is the slug; identified authoritatively via +//! [`Tool::composio_action_slug`](crate::openhuman::tools::Tool::composio_action_slug), +//! never by name shape); contract via [`fetch_live_toolkit_catalog`]. +//! - **MCP registry bridge** — `mcp_registry_tool_call`; contract via +//! [`all_connected_tools`]. +//! - **Workflows** — `run_workflow`; contract via [`get_workflow`]. +//! +//! The MCP *config/legacy* bridge (`mcp_call_tool`) is intentionally **not** +//! gated here: its registry handle is held privately inside the tool with no +//! ambient accessor, so wiring it needs extra plumbing — tracked as follow-up. +//! +//! Kill switch: set `OPENHUMAN_CONTRACT_GATE=0` to disable the gate entirely. + +use std::collections::{HashMap, HashSet}; +// Only the domain-gated contract renderers use `write!`/`writeln!`; with every +// domain off they compile out, so gate the trait import to match. +#[cfg(any(feature = "flows", feature = "skills", feature = "mcp"))] +use std::fmt::Write as _; +use std::sync::{LazyLock, Mutex}; + +use async_trait::async_trait; +use serde_json::Value; + +use tinyagents::error::Result as TaResult; +use tinyagents::harness::context::RunContext; +use tinyagents::harness::message::Message as TaMessage; +use tinyagents::harness::middleware::{ + Middleware, MiddlewareToolOutcome, ToolHandler, ToolMiddleware, +}; +use tinyagents::harness::model::ModelRequest; +use tinyagents::harness::tool::{ToolCall as TaToolCall, ToolResult as TaToolResult}; + +// The Composio and workflow/MCP contract sources live in domain modules that are +// compiled out by their feature gates (`flows` → `tinyflows`, `skills` → +// `skills::registry`, `mcp` → `mcp_registry`). The gate references them only +// behind the matching `#[cfg]`; with a domain off there is nothing to resolve a +// contract against, so the fetch degrades to `Unavailable` (pass-through) below. +#[cfg(feature = "flows")] +use crate::openhuman::memory_sync::composio::providers::tool_scope::toolkit_from_slug; +#[cfg(feature = "flows")] +use crate::openhuman::tinyflows::caps::{fetch_live_toolkit_catalog, ToolContract}; + +/// Env var that disables the gate when set to `0`. +const KILL_SWITCH_ENV: &str = "OPENHUMAN_CONTRACT_GATE"; + +/// Fixed opening of the transcript marker a contract-carrying tool message +/// leads with, placed at the START so a later model call recognises it with a +/// fixed-length prefix compare (no full-message scan). Closed by `]`, enclosing +/// only the marker's **slug list** — `[,…]`. There is **no** digest in +/// the marker itself; the payload hash lives in [`DELIVERED`], keyed by this exact +/// slug list. A gate delivery carries one slug; a full-schema discovery tool packs +/// several. +const MARKER_OPEN: &str = "[contract-gate:"; + +/// Separator between the slugs packed into a single marker's slug list. A gate +/// slug never contains it — slugs are `composio:` / `mcp::` / +/// `workflow:`, drawn from validated slugs, registry names, and directory ids +/// — and [`normalize_slug_list`] drops any slug that would (defence in depth) so +/// one pathological name can't corrupt the parse of the others. +const MARKER_SEP: char = ','; + +/// Process-global map from a marker's exact slug list (as built by +/// [`normalize_slug_list`]) to the XXH3-64 hash of the **payload** that followed +/// the marker at delivery (everything after the marker's `]`). +/// +/// Presence is decided in [`ContractGateMiddleware::refresh_present`] by +/// re-hashing a transcript marker's payload and matching it here — so the +/// transcript marker carries only slugs (short, whitespace-free), and a payload +/// later reformatted downstream (e.g. the sub-agent handoff's whitespace-collapse) +/// no longer matches → the contract is re-delivered (fail-safe). Global, not +/// per-run: a contract delivered in one turn stays credited across later turns / +/// resumes as long as its verbatim payload survives in the transcript. Keyed by +/// the **exact** slug list, so a discovery tool that credited `(s1,s2)` is matched +/// only by that same `(s1,s2)` marker. +static DELIVERED: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// A tool call resolved to a gated late-bound target. +#[derive(Debug, Clone, PartialEq, Eq)] +enum GateTarget { + /// A Composio action slug (from `composio_execute`'s `tool` arg, or a + /// per-action tool whose name IS the slug). + Composio(String), + /// An MCP-registry remote tool addressed by `(server_id, tool_name)`. + McpRegistry { server: String, tool: String }, + /// A Workflow addressed by `workflow_id`. + Workflow(String), +} + +impl GateTarget { + /// Stable key identifying this target in the transcript marker. + fn key(&self) -> String { + match self { + GateTarget::Composio(slug) => format!("composio:{}", slug.to_ascii_uppercase()), + GateTarget::McpRegistry { server, tool } => format!("mcp:{server}:{tool}"), + GateTarget::Workflow(id) => format!("workflow:{id}"), + } + } +} + +/// Outcome of resolving a [`GateTarget`] to its contract. +enum ContractResult { + /// Full contract text to hand the model; the target is real. + Found(String), + /// The target does not exist → hand the model a "no such target" message with + /// the valid list. The repeated-tool-failure breaker bounds any loop where + /// the model keeps re-issuing the same bad id. + NotFound(String), + /// The contract source was unreachable (no client / transient) — let the call + /// run so the real tool surfaces its own error rather than blocking on a + /// transient lookup failure. + Unavailable, +} + +/// The gate's pre-fetch decision for a call — pure (no network / no harness), so +/// the deliver → retry transition is unit-testable in isolation. +enum GateOutcome { + /// Not a gated tool, or its contract is already in the transcript → run it. + PassThrough, + /// The contract is not in context → fetch and deliver it before executing. + Deliver(GateTarget), +} + +/// Requires the model to see a late-bound tool's full contract before it runs. +/// +/// Implements both [`Middleware`] (`before_model`, to refresh the in-context set +/// from the transcript) and [`ToolMiddleware`] (`wrap_tool`, the gate) over one +/// shared state — the dual-trait shape [`SchemaGuardMiddleware`](super::middleware) +/// uses. Same `Arc`, two registration lists. +pub(super) struct ContractGateMiddleware { + /// Names of the registered per-action Composio tools, each self-reported via + /// [`Tool::composio_action_slug`](crate::openhuman::tools::Tool::composio_action_slug). + /// A bare tool-name call is gated as a Composio action only when its name is + /// in this set — by identity, never by name shape — so a non-Composio tool + /// can't be mistaken for a Composio action and gated into never executing. + composio_actions: HashSet, + /// The active profile's runnable-workflow `workflow_id` allowlist, sourced + /// from the registered [`RunWorkflowTool`] via + /// [`Tool::workflow_gate_allowlist`](crate::openhuman::tools::Tool::workflow_gate_allowlist). + /// `Some(set)` restricts which workflows the gate will resolve a contract + /// for; `None` leaves all installed workflows visible (default). Without it + /// the gate would render the input contract of a workflow the profile scopes + /// out — a metadata leak the executing tool otherwise prevents (issue #4853). + workflow_allowlist: Option>, + /// Gate keys whose contract marker is currently present in the transcript. + /// Rebuilt from `request.messages` on every `before_model` and mutated + /// nowhere else — presence is derived solely from the transcript. So several + /// same-batch calls to a not-yet-seen target are each gated until the model + /// has actually seen the delivered contract (a within-call "already handed + /// it out" shortcut would let a same-batch guess execute before the retry). + present: Mutex>, + /// Kill switch (`OPENHUMAN_CONTRACT_GATE=0` → false). + enabled: bool, +} + +impl ContractGateMiddleware { + pub(super) fn new( + composio_actions: HashSet, + workflow_allowlist: Option>, + ) -> Self { + let enabled = std::env::var(KILL_SWITCH_ENV) + .map(|v| v.trim() != "0") + .unwrap_or(true); + Self { + composio_actions, + workflow_allowlist, + present: Mutex::new(HashSet::new()), + enabled, + } + } + + /// True once the target's contract marker is in the transcript. + fn is_present(&self, key: &str) -> bool { + self.present + .lock() + .map(|set| set.contains(key)) + .unwrap_or(false) + } + + /// Rebuild the in-context slug set by scanning the **tool-role** messages of + /// the current request for contract markers, crediting a marker's slugs only + /// when its payload still hashes to the value recorded in [`DELIVERED`] for that + /// exact slug list. Authoritative: a marker no longer in the transcript + /// (compaction / microcompact / trim / a fresh resume), or one whose payload was + /// reformatted so the hash no longer matches, drops out of the set and the + /// contract is re-delivered. + fn refresh_present(&self, messages: &[TaMessage]) { + let mut present = HashSet::new(); + if let Ok(delivered) = DELIVERED.lock() { + for text in messages + .iter() + .filter(|m| matches!(m, TaMessage::Tool(_))) + .map(|m| m.text()) + { + credit_marker(&text, &delivered, &mut present); + } + } + if let Ok(mut set) = self.present.lock() { + *set = present; + } + } + + /// Whether `id` is runnable under the active profile's workflow allowlist. + /// A `None` allowlist admits every installed workflow (the default). + fn workflow_allowed(&self, id: &str) -> bool { + self.workflow_allowlist + .as_ref() + .map(|allow| allow.contains(id)) + .unwrap_or(true) + } + + /// Pure pre-fetch decision: gate a call only when it targets a gated + /// late-bound tool whose contract is not already in the transcript. + fn decide(&self, call: &TaToolCall) -> GateOutcome { + match gate_target(call, &self.composio_actions) { + None => GateOutcome::PassThrough, + Some(target) => { + // A workflow the active profile scopes out is never gated: + // rendering its contract would leak metadata the profile hides. + // Defer to the real `run_workflow` tool, which owns the scoped + // rejection — the gate never reveals such a workflow exists. + if let GateTarget::Workflow(id) = &target { + if !self.workflow_allowed(id) { + return GateOutcome::PassThrough; + } + } + if self.is_present(&target.key()) { + GateOutcome::PassThrough + } else { + GateOutcome::Deliver(target) + } + } + } + } + + /// Resolve a gated target to its full contract (network on cache-miss). + async fn fetch_contract(&self, target: &GateTarget) -> ContractResult { + // Test-only seam (never compiled into production): an integration test that + // drives the gate through the real `assemble_turn_harness` middleware + // ordering has no handle to this internally-built middleware and cannot + // reach a live Composio / workflow / MCP source, so it injects the contract + // keyed by `GateTarget::key()`. See [`test_support`]. + #[cfg(test)] + if let Some(text) = test_support::injected_found(&target.key()) { + return ContractResult::Found(text); + } + match target { + GateTarget::Composio(slug) => self.fetch_composio(slug).await, + GateTarget::McpRegistry { server, tool } => fetch_mcp_registry(server, tool).await, + GateTarget::Workflow(id) => self.fetch_workflow(id).await, + } + } + + #[cfg(feature = "flows")] + async fn fetch_composio(&self, slug: &str) -> ContractResult { + let Some(toolkit) = toolkit_from_slug(slug) else { + return ContractResult::NotFound(format!( + "`{slug}` is not a valid Composio action slug (expected `_`, \ + e.g. `GMAIL_FETCH_EMAILS`). Call `composio_list_tools` to see the real slugs." + )); + }; + let config = match crate::openhuman::config::ops::load_config_with_timeout().await { + Ok(cfg) => cfg, + Err(_) => return ContractResult::Unavailable, + }; + let Some(catalog) = fetch_live_toolkit_catalog(&config, &toolkit).await else { + return ContractResult::Unavailable; + }; + resolve_composio(slug, &toolkit, &catalog) + } + + /// `flows` off ⇒ no live Composio catalog to resolve against ⇒ pass through. + #[cfg(not(feature = "flows"))] + async fn fetch_composio(&self, _slug: &str) -> ContractResult { + ContractResult::Unavailable + } + + #[cfg(feature = "skills")] + async fn fetch_workflow(&self, id: &str) -> ContractResult { + // Invariant: `decide` has already passed through any workflow outside the + // active profile's allowlist (see `workflow_allowed`), so `id` here is + // always profile-visible — this resolver never renders a scoped-out + // workflow's contract. + let config = match crate::openhuman::config::ops::load_config_with_timeout().await { + Ok(cfg) => cfg, + Err(_) => return ContractResult::Unavailable, + }; + match crate::openhuman::skills::registry::get_workflow(&config.workspace_dir, id) { + Some(def) => ContractResult::Found(render_workflow_contract(id, &def)), + None => ContractResult::NotFound(format!( + "`{id}` is not an installed workflow. Call `list_workflows` to see valid \ + workflow ids, then re-issue `run_workflow` with a real id." + )), + } + } + + /// `skills` off ⇒ no workflow registry to read ⇒ pass through. + #[cfg(not(feature = "skills"))] + async fn fetch_workflow(&self, _id: &str) -> ContractResult { + ContractResult::Unavailable + } +} + +/// Extract the gated target (if any) from a tool call — pure string work, no +/// network. Returns `None` for non-gated tools or when a required addressing arg +/// is missing (the tool's own required-arg error handles that). A bare tool-name +/// call is treated as a Composio action only if its name is in `composio_actions` +/// (the authoritative registered per-action set) — by identity, never by name +/// shape. +fn gate_target(call: &TaToolCall, composio_actions: &HashSet) -> Option { + let arg_str = |k: &str| -> Option { + call.arguments + .get(k) + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + }; + match call.name.as_str() { + "composio_execute" | "composio" => arg_str("tool").map(GateTarget::Composio), + "mcp_registry_tool_call" => { + let server = arg_str("server_id")?; + let tool = arg_str("tool_name")?; + Some(GateTarget::McpRegistry { server, tool }) + } + "run_workflow" => arg_str("workflow_id").map(GateTarget::Workflow), + name if composio_actions.contains(name) => Some(GateTarget::Composio(name.to_string())), + _ => None, + } +} + +#[cfg(feature = "mcp")] +async fn fetch_mcp_registry(server: &str, tool: &str) -> ContractResult { + let all = crate::openhuman::mcp_registry::connections::all_connected_tools().await; + resolve_mcp(server, tool, &all) +} + +/// `mcp` off ⇒ no MCP-registry connections to read ⇒ pass through. +#[cfg(not(feature = "mcp"))] +async fn fetch_mcp_registry(_server: &str, _tool: &str) -> ContractResult { + ContractResult::Unavailable +} + +/// Pure Composio resolver: find the slug in the fetched catalog → the rendered +/// contract, or a not-found + valid list. Split from the network fetch so it is +/// unit-testable against a synthetic catalog. +#[cfg(feature = "flows")] +fn resolve_composio(slug: &str, toolkit: &str, catalog: &[ToolContract]) -> ContractResult { + match catalog.iter().find(|c| c.slug.eq_ignore_ascii_case(slug)) { + Some(contract) => ContractResult::Found(render_composio_contract(contract)), + None => { + let valid = join_capped(catalog.iter().map(|c| c.slug.as_str()), 40); + ContractResult::NotFound(format!( + "`{slug}` is not a real action in the `{toolkit}` toolkit. \ + Valid actions: [{valid}]. Re-issue with a real slug." + )) + } + } +} + +/// Pure MCP-registry resolver: find `(server, tool)` in the connected-tools +/// snapshot → the rendered contract, or a not-found + valid list. Split from the +/// network fetch so it is unit-testable. +#[cfg(feature = "mcp")] +fn resolve_mcp( + server: &str, + tool: &str, + all: &[( + String, + String, + crate::openhuman::mcp_registry::types::McpTool, + )], +) -> ContractResult { + match all + .iter() + .find(|(sid, _, t)| sid == server && t.name == tool) + { + Some((_, _, t)) => ContractResult::Found(render_mcp_contract(server, t)), + None => { + let valid = join_capped( + all.iter() + .filter(|(sid, _, _)| sid == server) + .map(|(_, _, t)| t.name.as_str()), + 40, + ); + if valid.is_empty() { + ContractResult::NotFound(format!( + "No connected MCP server `{server}` (or it advertises no tools). \ + Call `mcp_registry_list_tools` first." + )) + } else { + ContractResult::NotFound(format!( + "`{tool}` is not a tool on MCP server `{server}`. Valid tools: [{valid}]. \ + Re-issue with a real tool_name." + )) + } + } + } +} + +// ── transcript marker ─────────────────────────────────────────────────────── + +/// XXH3-64 hash of a marker's **payload** (the bytes after the marker's `]`). +/// Recorded in [`DELIVERED`] at delivery and recomputed on rescan: a fast, stable +/// (cross-process reproducible) fingerprint that detects any downstream reformat +/// (summarizer / size cap / the sub-agent handoff whitespace-collapse) of the +/// delivered contract. Non-cryptographic — guards accidental mutation, not a +/// forged collision. +fn payload_hash(payload: &str) -> u64 { + xxhash_rust::xxh3::xxh3_64(payload.as_bytes()) +} + +/// Canonical slug-list string used **both** as a transcript marker's body and as +/// the [`DELIVERED`] key, so a marker and its recorded hash always agree. Slugs are +/// de-duplicated and **sorted** (order-independent), and any slug carrying the +/// [`MARKER_SEP`] or `]` is dropped so one pathological name can't corrupt the +/// parse. Returns empty when nothing is creditable. +fn normalize_slug_list(slugs: impl IntoIterator) -> String { + let mut v: Vec = slugs + .into_iter() + .filter(|s| !s.is_empty() && !s.contains(MARKER_SEP) && !s.contains(']')) + .collect(); + v.sort(); + v.dedup(); + v.join(",") +} + +/// The transcript marker for a `normalize_slug_list` string: +/// `[contract-gate:]` — slugs only, no digest. +fn contract_marker(slug_list: &str) -> String { + format!("{MARKER_OPEN}{slug_list}]") +} + +/// Banner inserted right after the marker in every delivered contract. A weak +/// model (observed in the integrations sub-agent) reads the contract — the +/// tool's *input schema* — as if it were the tool's *output*, concludes "the +/// call ran and returned nothing", and gives up instead of retrying. State +/// plainly that the tool did NOT run and a retry is mandatory. The marker still +/// leads the message, so `before_model`'s fixed-length prefix scan is +/// unaffected. +const RETRY_BANNER: &str = "This tool was NOT executed and returned NO result. You are seeing its \ + input contract because the tool must be read before it can run. This is not a failure, an \ + error, or an empty result — nothing has been searched, fetched, or run yet. To actually run \ + it, re-issue the SAME tool call now with arguments matching the schema below. Do NOT report \ + \"no results\" or stop: the call has not happened."; + +/// The tool-error body a delivered contract carries, **plus the payload hash** the +/// caller records in [`DELIVERED`] under `slug_list`. Layout: the marker FIRST (so +/// `before_model` recognises it with a fixed-length prefix compare), then the +/// payload — [`RETRY_BANNER`] (making clear the tool has NOT run and must be +/// retried) then the contract text. The hash is of that exact payload, so a later +/// rescan credits the slug only while the payload survives byte-for-byte. +fn deliver_body(slug_list: &str, contract: &str) -> (String, u64) { + let payload = format!("\n\n{RETRY_BANNER}\n\n{contract}"); + let hash = payload_hash(&payload); + (format!("{}{payload}", contract_marker(slug_list)), hash) +} + +/// True when a tool-result body is a contract-gate **Found** delivery — it +/// leads with the [`MARKER_OPEN`] marker. A NotFound "no such action" message +/// does NOT, so a model looping on a bogus slug still trips the failure breaker. +/// +/// [`RepeatedToolFailureMiddleware`](super::middleware) calls this to exempt +/// Found deliveries from the repeated-tool-failure ladder: they are not +/// failures (they hand the model the contract and expect a retry), and they are +/// bounded to at most once per tool while the contract stays in the transcript. +pub(super) fn is_contract_delivery(content: &str) -> bool { + content.starts_with(MARKER_OPEN) +} + +/// If `text` leads with a `[contract-gate:]` marker whose payload +/// (everything after the marker's `]`) still hashes to the value recorded in +/// `delivered` for that **exact** slug list, credit every slug in the list into +/// `present`. Recognition is a fixed-length prefix compare (marker at the START), +/// so a model echoing the marker mid-text can't spoof presence, and there is one +/// marker per message. +/// +/// The hash match is the integrity gate: a delivered contract summarized, truncated +/// by a result-size cap, or whitespace-collapsed by the sub-agent handoff cleaner +/// no longer hashes to the recorded value, so its slugs are dropped and the gate +/// re-delivers rather than treating a mutated (possibly partial) body as present. +fn credit_marker(text: &str, delivered: &HashMap, present: &mut HashSet) { + let Some(rest) = text.strip_prefix(MARKER_OPEN) else { + return; + }; + let Some(j) = rest.find(']') else { + return; + }; + let slug_list = &rest[..j]; + let payload = &rest[j + 1..]; + if delivered.get(slug_list) != Some(&payload_hash(payload)) { + return; + } + for slug in slug_list.split(MARKER_SEP) { + let slug = slug.trim(); + if !slug.is_empty() { + present.insert(slug.to_string()); + } + } +} + +/// Prepend a **full-schema** discovery tool's presence marker to its output +/// `body`, returning the marker-led message the gate later credits — or `body` +/// unchanged when nothing is creditable — and **record the payload hash** in +/// [`DELIVERED`] under the marker's slug list. So a later `refresh_present` credits +/// every slug once the model has read this full listing: `describe_workflow` / +/// `mcp_registry_list_tools` / full `composio_list_tools` before the real call pays +/// no redundant re-delivery. +/// +/// This function OWNS the marker↔body concatenation so the recorded hash covers the +/// exact bytes that follow the marker: the payload is `"\n\n" + body`, and the +/// returned string is `marker + payload`. Slugs are de-duplicated + sorted (see +/// [`normalize_slug_list`]). +/// +/// Only a rendering that puts the **full** contract in the model's context may call +/// this; a thin listing must not (see the module docs). The slugs come from +/// [`composio_key`] / [`mcp_key`] / [`workflow_key`] so they match exactly what the +/// gate later checks presence against. +/// +/// NOTE: the discovery output is NOT flagged `trusted_verbatim`, so under a +/// sub-agent's handoff cleaning its payload can be whitespace-collapsed and stop +/// matching — the pre-credit is then simply skipped and the first real call +/// re-delivers once (best-effort, never a loop). +pub(crate) fn prefix_with_present_marker( + slugs: impl IntoIterator, + body: &str, +) -> String { + let slug_list = normalize_slug_list(slugs); + if slug_list.is_empty() { + return body.to_string(); + } + let payload = format!("\n\n{body}"); + if let Ok(mut delivered) = DELIVERED.lock() { + delivered.insert(slug_list.clone(), payload_hash(&payload)); + } + format!("{}{payload}", contract_marker(&slug_list)) +} + +/// Presence key for a Composio action slug — matches [`GateTarget::Composio`]'s +/// key so a `composio_list_tools` that carries the full schema credits the slug +/// the later `composio_execute` (or per-action tool) gates on. +pub(crate) fn composio_key(slug: &str) -> String { + GateTarget::Composio(slug.to_string()).key() +} + +/// Presence key for an MCP-registry `(server, tool)` — matches +/// [`GateTarget::McpRegistry`]'s key so a `mcp_registry_list_tools` credits the +/// tool the later `mcp_registry_tool_call` gates on. +pub(crate) fn mcp_key(server: &str, tool: &str) -> String { + GateTarget::McpRegistry { + server: server.to_string(), + tool: tool.to_string(), + } + .key() +} + +/// Presence key for a workflow id — matches [`GateTarget::Workflow`]'s key so +/// `describe_workflow` credits the id the later `run_workflow` gates on. +pub(crate) fn workflow_key(id: &str) -> String { + GateTarget::Workflow(id.to_string()).key() +} + +// ── contract rendering ────────────────────────────────────────────────────── + +#[cfg(feature = "flows")] +fn render_composio_contract(c: &ToolContract) -> String { + let mut out = String::new(); + let _ = write!( + out, + "Read the FULL contract for `{}` before calling it — you had only a short listing, \ + which is why argument formats must not be guessed.\n\nToolkit: {}\nDescription: {}\n\ + Required arguments: {}\n", + c.slug, + c.toolkit, + c.description.as_deref().unwrap_or("(none provided)"), + if c.required_args.is_empty() { + "(none)".to_string() + } else { + c.required_args.join(", ") + }, + ); + match &c.input_schema { + Some(schema) => { + let _ = write!(out, "\nFull input JSON Schema:\n{}\n", pretty(schema)); + } + None => { + let _ = write!(out, "\n(No input schema published for this action.)\n"); + } + } + let _ = write!( + out, + "\nNow re-issue the SAME call with arguments that conform to this schema (mind any \ + quoting/format the field descriptions call for)." + ); + out +} + +#[cfg(feature = "mcp")] +fn render_mcp_contract(server: &str, t: &crate::openhuman::mcp_registry::types::McpTool) -> String { + format!( + "Read the FULL contract for MCP tool `{}` on server `{}` before calling it.\n\n\ + Description: {}\n\nFull input JSON Schema:\n{}\n\nNow re-issue the SAME \ + `mcp_registry_tool_call` with `arguments` that conform to this schema.", + t.name, + server, + t.description.as_deref().unwrap_or("(none provided)"), + pretty(&t.input_schema), + ) +} + +#[cfg(feature = "skills")] +fn render_workflow_contract( + id: &str, + def: &crate::openhuman::skills::registry::WorkflowDefinition, +) -> String { + let mut out = String::new(); + let _ = write!( + out, + "Read the FULL input contract for workflow `{id}` before running it.\n\nInputs:\n" + ); + if def.inputs.is_empty() { + let _ = writeln!(out, " (this workflow declares no inputs)"); + } else { + for input in &def.inputs { + let _ = writeln!( + out, + " - {}{}{}: {}", + input.name, + input + .kind + .as_deref() + .map(|k| format!(" ({k})")) + .unwrap_or_default(), + if input.required { " [required]" } else { "" }, + if input.description.is_empty() { + "(no description)" + } else { + input.description.as_str() + }, + ); + } + } + let _ = write!( + out, + "\nNow re-issue the SAME `run_workflow` call with `inputs` covering every required field." + ); + out +} + +#[cfg(any(feature = "flows", feature = "mcp"))] +fn pretty(v: &Value) -> String { + serde_json::to_string_pretty(v).unwrap_or_else(|_| v.to_string()) +} + +/// Join an iterator of names, comma-separated, capped at `max` items with a +/// `+N more` tail so a huge toolkit doesn't flood the message. +#[cfg(any(feature = "flows", feature = "mcp"))] +fn join_capped<'a>(items: impl Iterator, max: usize) -> String { + let all: Vec<&str> = items.collect(); + if all.len() <= max { + return all.join(", "); + } + let shown = all[..max].join(", "); + format!("{shown}, +{} more", all.len() - max) +} + +#[async_trait] +impl Middleware<()> for ContractGateMiddleware { + fn name(&self) -> &str { + "contract_gate" + } + + async fn before_model( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + request: &mut ModelRequest, + ) -> TaResult<()> { + if self.enabled { + self.refresh_present(&request.messages); + } + Ok(()) + } +} + +#[async_trait] +impl ToolMiddleware<()> for ContractGateMiddleware { + fn name(&self) -> &str { + "contract_gate" + } + + async fn wrap_tool( + &self, + ctx: &mut RunContext<()>, + state: &(), + call: TaToolCall, + next: ToolHandler<'_, (), ()>, + ) -> TaResult { + if !self.enabled { + return next.run(ctx, state, call).await; + } + let target = match self.decide(&call) { + GateOutcome::PassThrough => return next.run(ctx, state, call).await, + GateOutcome::Deliver(target) => target, + }; + let key = target.key(); + // NB: no lock is held across this `.await` (std Mutex guards are !Send). + match self.fetch_contract(&target).await { + ContractResult::Unavailable => { + tracing::debug!( + tool = call.name.as_str(), + "[contract_gate] contract unavailable (transient); passing through to real tool" + ); + next.run(ctx, state, call).await + } + ContractResult::Found(text) => { + // Do NOT credit the slug present here: presence must come only from + // the transcript (via `before_model`), or a second same-batch call to + // this target would slip through before the model has seen the + // contract. Both same-batch calls are gated; the retry passes once the + // delivered tool message is rescanned. + tracing::debug!( + tool = call.name.as_str(), + "[contract_gate] contract not in context; returning it before executing" + ); + // A gate delivery's slug list is the single target slug. Record the + // payload hash in DELIVERED under it, then hand back the marker-led + // body; `refresh_present` credits the slug on the retry while the + // transcript still carries the byte-identical payload. + let slug_list = normalize_slug_list([key]); + let (body, hash) = deliver_body(&slug_list, &text); + if let Ok(mut delivered) = DELIVERED.lock() { + delivered.insert(slug_list, hash); + } + // Keep the delivered contract byte-for-byte: the payload's hash is what + // `refresh_present` re-checks, so any after-tool rewrite — notably the + // sub-agent handoff's `clean_tool_output` whitespace-collapse — would + // break the match and re-deliver every turn. `NotFound` is NOT flagged + // (no hash recorded; a looping bad slug should still trip the + // repeated-tool-failure breaker). + let mut result = TaToolResult::error(call.id, call.name, body); + super::middleware::mark_trusted_verbatim(&mut result); + Ok(MiddlewareToolOutcome::Result(result)) + } + ContractResult::NotFound(text) => { + tracing::debug!( + tool = call.name.as_str(), + "[contract_gate] unknown target; returning not-found + valid list" + ); + Ok(MiddlewareToolOutcome::Result(TaToolResult::error( + call.id, call.name, text, + ))) + } + } + } +} + +/// Test-only contract-source injection seam. +/// +/// Production [`ContractGateMiddleware::fetch_contract`] reaches live Composio / +/// the workflow registry / the MCP registry. An integration test that exercises +/// the deliver → retry through the real [`assemble_turn_harness`] middleware +/// ordering builds the middleware *internally* (so the test never holds the +/// instance) and runs with no live sources — this global, serial-locked seam is +/// the injection point. It is consulted ONLY under `cfg(test)`, at the top of +/// `fetch_contract`, and the whole module is absent from a production build. +/// +/// The value type is `String` (delivered as [`ContractResult::Found`]) so the +/// private `ContractResult` need not widen its visibility for tests. +#[cfg(test)] +pub(crate) mod test_support { + use std::collections::HashMap; + use std::sync::{Mutex, MutexGuard}; + + /// Serializes seam-using tests: the injection is process-global, so two of + /// them running in parallel would clobber each other. Held by the guard. + static SERIAL: Mutex<()> = Mutex::new(()); + /// `GateTarget::key()` → contract text handed back as `Found`. + static INJECTED: Mutex>> = Mutex::new(None); + + /// Installs a keyed contract set for the lifetime of the returned guard, + /// holding the serial lock; both the injection and the lock clear on drop. + /// Hold the guard across the turn under test. + #[must_use = "the injection is cleared when this guard is dropped"] + pub(crate) struct InjectedContracts { + _serial: MutexGuard<'static, ()>, + } + + impl InjectedContracts { + /// `contracts`: `GateTarget::key()` → the full contract text to deliver. + /// Also clears the process-global `DELIVERED` map so a prior test's recorded + /// deliveries can't leak into this one. + pub(crate) fn found(contracts: HashMap) -> Self { + let serial = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + clear_delivered(); + *INJECTED.lock().unwrap_or_else(|e| e.into_inner()) = Some(contracts); + Self { _serial: serial } + } + } + + impl Drop for InjectedContracts { + fn drop(&mut self) { + *INJECTED.lock().unwrap_or_else(|e| e.into_inner()) = None; + clear_delivered(); + } + } + + /// Clear the process-global [`DELIVERED`](super::DELIVERED) map. Exposed for + /// tests that record deliveries directly, so they isolate from one another. + pub(crate) fn clear_delivered() { + if let Ok(mut d) = super::DELIVERED.lock() { + d.clear(); + } + } + + /// Serialize `DELIVERED`-touching tests (the map is process-global) and reset + /// it. Hold the returned guard for the test's duration. + pub(crate) fn isolate_delivered() -> MutexGuard<'static, ()> { + let g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + clear_delivered(); + g + } + + /// The injected contract for `key`, if a test installed one. + pub(super) fn injected_found(key: &str) -> Option { + INJECTED + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + .and_then(|m| m.get(key).cloned()) + } +} + +#[cfg(test)] +#[path = "contract_gate_tests.rs"] +mod tests; diff --git a/src/openhuman/tinyagents/contract_gate_tests.rs b/src/openhuman/tinyagents/contract_gate_tests.rs new file mode 100644 index 0000000000..16f1c3200d --- /dev/null +++ b/src/openhuman/tinyagents/contract_gate_tests.rs @@ -0,0 +1,611 @@ +//! Unit tests for the contract gate (issue #4853). Pure logic only — target +//! extraction, key stability, slug detection, transcript-marker presence, and +//! contract rendering. The network resolvers (`fetch_*`) are exercised by the +//! Composio / MCP / workflow integration suites, not here. +//! +//! Presence tests touch the process-global [`DELIVERED`] map, so they hold +//! `test_support::isolate_delivered()` for their duration — it serializes them +//! and resets the map, so a recorded delivery from one test can't leak into +//! another. + +use super::*; +use serde_json::json; +use std::collections::HashSet; + +fn call(name: &str, args: serde_json::Value) -> TaToolCall { + TaToolCall { + id: "call-1".to_string(), + name: name.to_string(), + arguments: args, + invalid: None, + } +} + +/// The authoritative registered per-action Composio slug set the gate is built +/// with (in production, sourced from `Tool::composio_action_slug`). +fn actions(names: &[&str]) -> HashSet { + names.iter().map(|s| s.to_string()).collect() +} + +/// A tool-role transcript message carrying `content` — what `refresh_present` +/// scans (it filters `TaMessage::Tool` and reads `.text()`). +fn tool_msg(content: impl Into) -> TaMessage { + TaMessage::tool("call-1", content) +} + +/// Deliver a single-slug contract exactly as `wrap_tool`'s Found arm does: +/// record the payload hash in `DELIVERED` under the (normalized) slug list, and +/// return the marker-led transcript body. +fn delivered(slug_list: &str, contract: &str) -> String { + let (body, hash) = deliver_body(slug_list, contract); + DELIVERED + .lock() + .unwrap() + .insert(slug_list.to_string(), hash); + body +} + +/// The slugs a set of transcript texts credit, given the current `DELIVERED` +/// map — i.e. what `refresh_present` would build into the in-context set. +fn credited(texts: &[&str]) -> HashSet { + let delivered = DELIVERED.lock().unwrap(); + let mut present = HashSet::new(); + for t in texts { + credit_marker(t, &delivered, &mut present); + } + present +} + +// ── target extraction (pure; no DELIVERED) ──────────────────────────────────── + +#[test] +fn composio_execute_reads_the_tool_arg_as_the_slug() { + assert_eq!( + gate_target( + &call("composio_execute", json!({ "tool": "GMAIL_FETCH_EMAILS" })), + &actions(&[]), + ), + Some(GateTarget::Composio("GMAIL_FETCH_EMAILS".to_string())) + ); +} + +#[test] +fn ordinary_lowercase_tools_are_never_gated() { + let registered = actions(&["GMAIL_FETCH_EMAILS"]); + assert_eq!( + gate_target(&call("web_search", json!({ "query": "x" })), ®istered), + None + ); + assert_eq!( + gate_target(&call("shell", json!({ "command": "ls" })), ®istered), + None + ); + assert_eq!( + gate_target(&call("memory_recall", json!({ "query": "x" })), ®istered), + None + ); +} + +#[test] +fn mcp_registry_tool_call_needs_both_ids() { + assert_eq!( + gate_target( + &call( + "mcp_registry_tool_call", + json!({ "server_id": "srv", "tool_name": "do_thing" }) + ), + &actions(&[]), + ), + Some(GateTarget::McpRegistry { + server: "srv".to_string(), + tool: "do_thing".to_string() + }) + ); + // Missing tool_name → not gated (the tool's own required-arg error handles it). + assert_eq!( + gate_target( + &call("mcp_registry_tool_call", json!({ "server_id": "srv" })), + &actions(&[]), + ), + None + ); +} + +#[test] +fn run_workflow_reads_workflow_id() { + assert_eq!( + gate_target( + &call( + "run_workflow", + json!({ "workflow_id": "pr-review-shepherd" }) + ), + &actions(&[]), + ), + Some(GateTarget::Workflow("pr-review-shepherd".to_string())) + ); +} + +#[test] +fn composio_execute_without_tool_arg_passes_through() { + assert_eq!( + gate_target(&call("composio_execute", json!({})), &actions(&[])), + None + ); +} + +#[test] +fn composio_key_is_case_insensitive() { + assert_eq!( + GateTarget::Composio("gmail_fetch_emails".to_string()).key(), + GateTarget::Composio("GMAIL_FETCH_EMAILS".to_string()).key() + ); +} + +#[test] +fn bare_call_is_gated_only_when_the_name_is_a_registered_composio_action() { + // Identity, not name shape: only a name in the registered per-action set is + // gated as a Composio action. + let registered = actions(&["GMAIL_FETCH_EMAILS"]); + assert_eq!( + gate_target(&call("GMAIL_FETCH_EMAILS", json!({})), ®istered), + Some(GateTarget::Composio("GMAIL_FETCH_EMAILS".to_string())) + ); + // A non-Composio tool that happens to be UPPER_SNAKE → NOT gated, so it can + // never be short-circuited into never executing. + assert_eq!( + gate_target(&call("EXPORT_REPORT_PDF", json!({})), ®istered), + None + ); + // A Composio-slug-shaped name that isn't registered → NOT gated either. + assert_eq!( + gate_target(&call("GMAIL_SEND_EMAIL", json!({})), ®istered), + None + ); +} + +// ── presence: the global DELIVERED dict + slug-list marker + payload hash ────── + +#[test] +fn is_present_reflects_the_in_context_set() { + let mw = ContractGateMiddleware::new(actions(&[]), None); + assert!(!mw.is_present("composio:GMAIL_FETCH_EMAILS")); + mw.present + .lock() + .unwrap() + .insert("composio:GMAIL_FETCH_EMAILS".to_string()); + assert!(mw.is_present("composio:GMAIL_FETCH_EMAILS")); +} + +#[test] +fn gated_call_delivers_the_contract_then_the_retry_passes() { + let _g = test_support::isolate_delivered(); + let mw = ContractGateMiddleware::new(actions(&[]), None); + let c = call("composio_execute", json!({ "tool": "GMAIL_FETCH_EMAILS" })); + + // 1) Contract not in context → the call is gated: wrap_tool would short- + // circuit with the contract as a tool error, WITHOUT executing the tool. + let target = match mw.decide(&c) { + GateOutcome::Deliver(t) => t, + GateOutcome::PassThrough => panic!("first call must be gated"), + }; + let key = target.key(); + assert_eq!(key, "composio:GMAIL_FETCH_EMAILS"); + + // The body wrap_tool returns for a found contract: the marker FIRST (so the + // rescan recognises it) then the contract, which names the slug. Delivery + // records the payload hash in DELIVERED under the slug list. + let body = delivered(&key, ""); + assert!(is_contract_delivery(&body)); + assert!(body.contains("GMAIL_FETCH_EMAILS")); + + // 2) That tool message re-enters the transcript; the next before_model + // rescans it into the in-context set (payload still hashes to DELIVERED). + mw.refresh_present(&[tool_msg(body)]); + + // 3) The identical call now passes the gate → the real tool runs. + assert!(matches!(mw.decide(&c), GateOutcome::PassThrough)); +} + +#[test] +fn same_batch_second_call_is_still_gated() { + // Two calls to the same target in one assistant message: the second must NOT + // slip through before the model has actually seen the delivered contract. + // Presence comes only from the transcript (refreshed by before_model between + // model turns), so both same-batch calls are gated. + let mw = ContractGateMiddleware::new(actions(&[]), None); + let c = call("composio_execute", json!({ "tool": "GMAIL_FETCH_EMAILS" })); + assert!(matches!(mw.decide(&c), GateOutcome::Deliver(_))); + assert!(matches!(mw.decide(&c), GateOutcome::Deliver(_))); +} + +#[test] +fn a_recorded_delivery_reads_as_present() { + let _g = test_support::isolate_delivered(); + let key = GateTarget::Composio("GMAIL_FETCH_EMAILS".to_string()).key(); + let body = delivered(&key, "...full contract..."); + assert!(credited(&[body.as_str()]).contains(&key)); +} + +#[test] +fn resumed_history_with_the_delivery_reads_as_present() { + // A durable sub-agent resumed with initial_history that still carries the + // delivered contract byte-for-byte (so its payload still hashes to the + // DELIVERED value) → the gate sees it and must NOT re-deliver. + let _g = test_support::isolate_delivered(); + let key = "composio:GMAIL_FETCH_EMAILS"; + let body = delivered(key, "Read the FULL contract ..."); + assert!(credited(&[body.as_str()]).contains(key)); +} + +#[test] +fn marker_not_at_the_start_is_ignored() { + // Only a marker at the very start of a tool message counts — a model echoing + // the marker mid-text cannot spoof presence. + let _g = test_support::isolate_delivered(); + let body = delivered("composio:GMAIL_FETCH_EMAILS", "contract"); + let text = format!("some preamble {body}"); + assert!(credited(&[text.as_str()]).is_empty()); +} + +#[test] +fn compacted_away_contract_is_not_present() { + // Transcript no longer carries the marker (summarizer / microcompact / trim + // dropped it) → the contract is re-delivered on the next call. + let _g = test_support::isolate_delivered(); + assert!(credited(&["some tool output", "a summary without the marker"]).is_empty()); +} + +#[test] +fn credits_one_marker_per_message_across_messages() { + // Each delivered contract is its own tool message with the marker at its + // start; the gate credits keys across all such messages. + let _g = test_support::isolate_delivered(); + let a = delivered("composio:GMAIL_FETCH_EMAILS", "Gmail contract"); + let b = delivered("mcp:srv:do_thing", "Mcp contract"); + let present = credited(&[a.as_str(), b.as_str()]); + assert!(present.contains("composio:GMAIL_FETCH_EMAILS")); + assert!(present.contains("mcp:srv:do_thing")); + assert_eq!(present.len(), 2); +} + +#[test] +fn deliver_body_puts_the_marker_first_then_a_retry_banner() { + // Pure layout check — deliver_body computes the body + hash, it does not + // touch DELIVERED, so no isolation guard is needed. + let (body, _hash) = deliver_body("composio:GMAIL_FETCH_EMAILS", ""); + // Marker leads (before_model's prefix scan depends on it) and carries ONLY + // the slug list — no digest lives in the marker anymore. + assert!(body.starts_with(MARKER_OPEN)); + let marker = &body[..=body.find(']').unwrap()]; + assert_eq!(marker, "[contract-gate:composio:GMAIL_FETCH_EMAILS]"); + // The retry banner sits between the marker and the contract so a weak model + // can't misread the delivered contract as an empty/failed result and give up + // instead of retrying (observed in the integrations sub-agent). + assert!(body.contains("NOT executed")); + assert!(body.contains("re-issue the SAME tool call")); + let marker_end = body.find(']').unwrap(); + let banner_at = body.find("NOT executed").unwrap(); + let contract_at = body.find("").unwrap(); + assert!( + marker_end < banner_at && banner_at < contract_at, + "order must be: marker, then banner, then contract" + ); +} + +#[test] +fn is_contract_delivery_matches_found_but_not_notfound() { + // A Found delivery leads with the marker → exempt from the failure breaker. + let (found, _) = deliver_body("composio:GMAIL_FETCH_EMAILS", ""); + assert!(is_contract_delivery(&found)); + // A NotFound "no such action" message has no leading marker → still counts. + assert!(!is_contract_delivery( + "`GMAIL_NOPE` is not a real action in the `gmail` toolkit." + )); + // An ordinary tool output is not a delivery. + assert!(!is_contract_delivery("some real tool output")); +} + +#[test] +fn a_rewritten_or_truncated_payload_is_not_credited() { + // The delivered payload's hash is recorded in DELIVERED. If a downstream stage + // (summarizer / size cap / whitespace-collapse) mutates the payload, its hash + // no longer matches the recorded value and the slug is NOT credited — so the + // gate re-delivers the full contract rather than treating a partial/rewritten + // body as present. + let _g = test_support::isolate_delivered(); + let key = "composio:GMAIL_FETCH_EMAILS"; + let body = delivered(key, ""); + assert!(credited(&[body.as_str()]).contains(key)); + + // A size cap truncating the tail of the delivered message → hash mismatch. + let truncated = &body[..body.len() - 10]; + assert!( + credited(&[truncated]).is_empty(), + "a truncated payload must not read as present" + ); + + // A summarizer keeping the marker but replacing the payload → hash mismatch. + let marker_end = body.find(']').unwrap(); + let rewritten = format!("{}\n\n(summarized: 1 field)", &body[..=marker_end]); + assert!( + credited(&[rewritten.as_str()]).is_empty(), + "a rewritten payload must not read as present" + ); +} + +#[test] +fn a_marker_whose_slug_list_was_never_delivered_is_ignored() { + // A hand-written / spoofed marker whose slug list has no entry in DELIVERED + // (nothing was ever delivered under it) cannot be credited — crediting + // requires a recorded payload hash that the marker's payload matches, so a + // fabricated marker fails closed. + let _g = test_support::isolate_delivered(); + let text = "[contract-gate:composio:GMAIL_FETCH_EMAILS]\n\nbody"; + assert!(credited(&[text]).is_empty()); +} + +// `join_capped` / the Composio + MCP resolvers and renderers live behind the +// `flows`/`mcp` domain gates (their contract types are compiled out otherwise), +// so the tests that name them are gated to match — else the gates-off test binary +// fails to compile (the feature-gate-smoke lane builds and runs it). +#[cfg(any(feature = "flows", feature = "mcp"))] +#[test] +fn join_capped_adds_a_more_tail() { + let many: Vec = (0..50).map(|i| format!("T{i}")).collect(); + let out = join_capped(many.iter().map(String::as_str), 40); + assert!(out.contains("+10 more"), "got: {out}"); + let few = ["A", "B", "C"]; + assert_eq!(join_capped(few.into_iter(), 40), "A, B, C"); +} + +#[cfg(feature = "flows")] +#[test] +fn composio_contract_render_surfaces_schema_and_required_args() { + let c = ToolContract { + slug: "GMAIL_FETCH_EMAILS".to_string(), + toolkit: "gmail".to_string(), + description: Some("Fetch emails.".to_string()), + required_args: vec!["query".to_string()], + input_schema: Some(json!({ + "type": "object", + "properties": { + "query": { "type": "string", "description": "Gmail search query; quote phrases" } + } + })), + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: false, + }; + let text = render_composio_contract(&c); + assert!(text.contains("GMAIL_FETCH_EMAILS")); + assert!(text.contains("query")); + assert!(text.contains("Gmail search query")); // the field description the model was missing + assert!(text.contains("Required arguments: query")); +} + +#[cfg(feature = "mcp")] +#[test] +fn mcp_contract_render_surfaces_schema() { + let t = crate::openhuman::mcp_registry::types::McpTool { + name: "do_thing".to_string(), + description: Some("Does a thing.".to_string()), + input_schema: json!({ "type": "object", "properties": { "x": { "type": "number" } } }), + }; + let text = render_mcp_contract("srv", &t); + assert!(text.contains("do_thing")); + assert!(text.contains("srv")); + assert!(text.contains("\"x\"")); +} + +#[cfg(feature = "flows")] +#[test] +fn resolve_composio_found_and_not_found() { + let catalog = vec![ToolContract { + slug: "GMAIL_FETCH_EMAILS".to_string(), + toolkit: "gmail".to_string(), + description: Some("Fetch emails.".to_string()), + required_args: vec!["query".to_string()], + input_schema: Some(json!({ "type": "object" })), + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: false, + }]; + // Found (case-insensitive) → rendered contract naming the slug. + match resolve_composio("gmail_fetch_emails", "gmail", &catalog) { + ContractResult::Found(text) => assert!(text.contains("GMAIL_FETCH_EMAILS")), + _ => panic!("expected Found"), + } + // Not found → message carrying the valid list. + match resolve_composio("GMAIL_NOPE", "gmail", &catalog) { + ContractResult::NotFound(text) => { + assert!(text.contains("GMAIL_NOPE")); + assert!(text.contains("GMAIL_FETCH_EMAILS")); // valid list + } + _ => panic!("expected NotFound"), + } +} + +#[cfg(feature = "mcp")] +#[test] +fn resolve_mcp_found_not_found_and_empty_server() { + let all = vec![( + "srv".to_string(), + "srv".to_string(), + crate::openhuman::mcp_registry::types::McpTool { + name: "do_thing".to_string(), + description: Some("Does a thing.".to_string()), + input_schema: json!({ "type": "object" }), + }, + )]; + match resolve_mcp("srv", "do_thing", &all) { + ContractResult::Found(text) => assert!(text.contains("do_thing")), + _ => panic!("expected Found"), + } + match resolve_mcp("srv", "missing", &all) { + ContractResult::NotFound(text) => assert!(text.contains("do_thing")), // valid list + _ => panic!("expected NotFound"), + } + match resolve_mcp("other", "x", &all) { + ContractResult::NotFound(text) => assert!(text.contains("No connected MCP server")), + _ => panic!("expected NotFound for an unknown server"), + } +} + +// ── multi-slug discovery markers (full-schema discovery tools pre-crediting) ─── + +#[test] +fn key_helpers_match_gate_target_keys() { + // The helpers a discovery tool builds markers from must produce EXACTLY the + // key the gate later checks presence against, or crediting would silently + // miss. Composio uppercases; MCP/workflow are verbatim. + assert_eq!( + composio_key("gmail_fetch_emails"), + GateTarget::Composio("gmail_fetch_emails".to_string()).key() + ); + assert_eq!( + composio_key("gmail_fetch_emails"), + "composio:GMAIL_FETCH_EMAILS" + ); + assert_eq!( + mcp_key("srv", "do_thing"), + GateTarget::McpRegistry { + server: "srv".to_string(), + tool: "do_thing".to_string() + } + .key() + ); + assert_eq!(mcp_key("srv", "do_thing"), "mcp:srv:do_thing"); + assert_eq!( + workflow_key("pr-review"), + GateTarget::Workflow("pr-review".to_string()).key() + ); + assert_eq!(workflow_key("pr-review"), "workflow:pr-review"); +} + +#[test] +fn discovery_marker_credits_multiple_slugs() { + let _g = test_support::isolate_delivered(); + let msg = prefix_with_present_marker( + [ + "composio:GMAIL_FETCH_EMAILS".to_string(), + "mcp:srv:do_thing".to_string(), + ], + "...full schemas...", + ); + // One marker at the very start; both slugs recovered from it on rescan. + assert!(msg.starts_with(MARKER_OPEN)); + let present = credited(&[msg.as_str()]); + assert!(present.contains("composio:GMAIL_FETCH_EMAILS")); + assert!(present.contains("mcp:srv:do_thing")); + assert_eq!(present.len(), 2); +} + +#[test] +fn discovery_marker_dedupes_and_sorts_slugs() { + let _g = test_support::isolate_delivered(); + let msg = prefix_with_present_marker( + [ + "workflow:b".to_string(), + "workflow:a".to_string(), + "workflow:a".to_string(), + ], + "body", + ); + // Slugs are de-duped and sorted (order-independent) inside the marker. + let marker = &msg[..=msg.find(']').unwrap()]; + assert_eq!(marker, "[contract-gate:workflow:a,workflow:b]"); + assert_eq!(credited(&[msg.as_str()]).len(), 2); +} + +#[test] +fn discovery_marker_leaves_body_unchanged_when_nothing_creditable() { + let _g = test_support::isolate_delivered(); + // Empty input → body returned as-is (no marker prepended, nothing recorded). + assert_eq!( + prefix_with_present_marker(std::iter::empty(), "raw body"), + "raw body" + ); + // A slug that would corrupt the parse (contains the separator or `]`) is + // dropped rather than emitted — so a pathological name can't break the + // others, and an all-bad set yields the body unchanged (gate re-delivers). + assert_eq!( + prefix_with_present_marker(["bad,key".to_string(), "also]bad".to_string()], "raw body"), + "raw body" + ); + // Good slugs survive alongside a dropped bad one. + let msg = prefix_with_present_marker(["ok:key".to_string(), "bad,key".to_string()], "b"); + assert!(msg.starts_with(MARKER_OPEN)); + assert_eq!( + credited(&[msg.as_str()]), + ["ok:key".to_string()].into_iter().collect() + ); +} + +#[test] +fn discovery_marker_round_trips_all_slugs() { + let _g = test_support::isolate_delivered(); + let keys = [ + composio_key("gmail_fetch_emails"), + mcp_key("srv", "do_thing"), + workflow_key("wf"), + ]; + let msg = prefix_with_present_marker(keys.clone(), "...full schemas..."); + let present = credited(&[msg.as_str()]); + for key in keys { + assert!(present.contains(&key), "missing {key}"); + } + assert_eq!(present.len(), 3); +} + +#[test] +fn a_discovery_marker_pre_credits_the_gate_end_to_end() { + // Model calls `describe_workflow` (full inputs) → its output carries the + // marker → before_model rescans it → the later `run_workflow` passes the + // gate WITHOUT a redundant re-delivery. This is the whole point of C. + let _g = test_support::isolate_delivered(); + let mw = ContractGateMiddleware::new(actions(&[]), None); + let run = call("run_workflow", json!({ "workflow_id": "wf" })); + + // Before the describe read: the run is gated. + assert!(matches!(mw.decide(&run), GateOutcome::Deliver(_))); + + // describe_workflow's output: marker at the start (built exactly as the + // skills tool builds it) then the full contract; it records the payload hash. + let describe_output = prefix_with_present_marker([workflow_key("wf")], "{\"inputs\":[...]}"); + mw.refresh_present(&[tool_msg(describe_output)]); + + // Now the run passes — the contract is already in context. + assert!(matches!(mw.decide(&run), GateOutcome::PassThrough)); +} + +// ── workflow allowlist scoping (pure decide; no DELIVERED) ───────────────────── + +#[test] +fn workflow_outside_the_profile_allowlist_is_not_gated() { + // A profile that scopes workflows to `allowed-wf` must not let the gate + // render `hidden-wf`'s contract — the gate defers a scoped-out workflow to + // the real `run_workflow` tool's rejection (PassThrough), never revealing it + // exists or leaking its input contract (issue #4853 review). + let allow: HashSet = ["allowed-wf".to_string()].into_iter().collect(); + let mw = ContractGateMiddleware::new(actions(&[]), Some(allow)); + + let hidden = call("run_workflow", json!({ "workflow_id": "hidden-wf" })); + assert!( + matches!(mw.decide(&hidden), GateOutcome::PassThrough), + "a scoped-out workflow must not be gated (its contract must not be rendered)" + ); + + // An allowed workflow is still gated normally on its first use. + let allowed = call("run_workflow", json!({ "workflow_id": "allowed-wf" })); + assert!(matches!(mw.decide(&allowed), GateOutcome::Deliver(_))); +} + +#[test] +fn no_workflow_allowlist_gates_every_workflow() { + // The default (unrestricted profile) gates any workflow on first use. + let mw = ContractGateMiddleware::new(actions(&[]), None); + let any = call("run_workflow", json!({ "workflow_id": "whatever" })); + assert!(matches!(mw.decide(&any), GateOutcome::Deliver(_))); +} diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index 6ca9ef1f75..fb0de743c9 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -469,6 +469,46 @@ impl Middleware<()> for OpenHumanToolExposureShadowMiddleware { } } +/// Key under a tool result's `raw` JSON marking its `content` as trusted, +/// system-authored, and **verbatim** — it must reach the transcript BYTE-FOR-BYTE. +/// Every at-creation `after_tool` mutator (the progressive-disclosure handoff's +/// `clean_tool_output`, semantic summarization, TokenJuice compaction, per-tool / +/// byte caps) skips a flagged result, so its exact bytes survive. +/// +/// The contract gate sets it on a delivered contract: presence is decided by +/// re-hashing the bytes after the `[contract-gate:…]` marker and matching the digest +/// baked into it, so ANY rewrite (notably `clean_tool_output`, which collapses runs +/// of whitespace and would flatten the contract's JSON-schema indentation) makes the +/// rescan digest miss — re-delivering the contract on every turn, forever. Keeping +/// the delivery verbatim is what makes the digest match on the retry. Carried in +/// `raw` rather than a typed field to avoid widening the vendored `ToolResult`; +/// `scrub_json_credentials` only rewrites string leaves, so the boolean survives. +pub(crate) const TRUSTED_VERBATIM_RAW_KEY: &str = "trusted_verbatim"; + +/// Mark `result` trusted-verbatim so the `after_tool` mutators leave its `content` +/// byte-for-byte. See [`TRUSTED_VERBATIM_RAW_KEY`]. +pub(crate) fn mark_trusted_verbatim(result: &mut TaToolResult) { + let raw = result + .raw + .get_or_insert_with(|| serde_json::Value::Object(Default::default())); + if let Some(map) = raw.as_object_mut() { + map.insert( + TRUSTED_VERBATIM_RAW_KEY.to_string(), + serde_json::Value::Bool(true), + ); + } +} + +/// Whether `result` is flagged [`TRUSTED_VERBATIM_RAW_KEY`]. +pub(crate) fn is_trusted_verbatim(result: &TaToolResult) -> bool { + result + .raw + .as_ref() + .and_then(|v| v.get(TRUSTED_VERBATIM_RAW_KEY)) + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) +} + /// `after_tool`: progressive-disclosure handoff (issue #4249 1b). An oversized /// sub-agent tool result is stashed in the shared [`ResultHandoffCache`] and its /// content replaced with a short placeholder naming a `result_id` the model can @@ -494,6 +534,17 @@ impl Middleware<()> for HandoffMiddleware { _state: &(), result: &mut TaToolResult, ) -> TaResult<()> { + // A `trusted_verbatim` result (a contract-gate delivery) must reach the + // transcript byte-for-byte. `apply_handoff` runs `clean_tool_output`, which + // collapses runs of whitespace — flattening the delivered JSON schema's + // indentation — so the gate's rescan digest would miss and re-deliver the + // contract forever. Stashing is moot too: the delivery is small and must stay + // inline for the model to read. Runs FIRST in the reverse-order `after_tool` + // chain, so this guard (not the tool-output one below) is what preserves the + // bytes. + if is_trusted_verbatim(result) { + return Ok(()); + } result.content = crate::openhuman::agent::harness::subagent_runner::apply_handoff( &self.cache, &result.name, @@ -817,6 +868,13 @@ impl Middleware<()> for ToolOutputMiddleware { _state: &(), result: &mut TaToolResult, ) -> TaResult<()> { + // A `trusted_verbatim` result (a contract-gate delivery) must reach the + // transcript byte-for-byte: summarization / TokenJuice compaction / a size + // cap would rewrite it, breaking the gate's rescan-digest presence check. + if is_trusted_verbatim(result) { + return Ok(()); + } + // Proposal-/persistence-emitting workflow tools return a self-describing // `{ "type": "workflow_proposal", … }` JSON payload that `flows::ops`' // `extract_workflow_proposal` (and the frontend's content-based @@ -2456,6 +2514,20 @@ impl Middleware<()> for RepeatedToolFailureMiddleware { _state: &(), result: &mut TaToolResult, ) -> TaResult<()> { + // Contract-gate Found deliveries (issue #4853) are NOT failures — they + // hand the model the full contract and expect a retry, so they must not + // feed the repeated-tool-failure ladder. They lead with the contract + // marker; a "no such action" NotFound does not, so a model looping on a + // bogus slug still trips the breaker. Bounded to ~once per tool (once the + // contract is in context the gate passes the retry through). Drop the + // fingerprint `before_tool` staged and exit before any counting/reset. + if result.error.is_some() && super::contract_gate::is_contract_delivery(&result.content) { + if let Ok(mut sigs) = self.arg_sigs.lock() { + sigs.remove(&result.call_id); + } + return Ok(()); + } + let arg_fp = self .arg_sigs .lock() @@ -3328,6 +3400,27 @@ mod tests { assert_eq!(estimate_text_tokens(""), 0); } + #[test] + fn trusted_verbatim_flag_round_trips_and_survives_credential_scrubbing() { + let mut result = + TaToolResult::error("call-1", "GMAIL_LIST_THREADS", "[contract-gate:…]\nbody"); + assert!(!is_trusted_verbatim(&result), "unflagged by default"); + mark_trusted_verbatim(&mut result); + assert!( + is_trusted_verbatim(&result), + "flag reads back after marking" + ); + // `scrub_json_credentials` rewrites only string leaves, so the boolean flag + // in `raw` survives — the `after_tool` guards still see it downstream. + if let Some(raw) = result.raw.take() { + result.raw = Some(scrub_json_credentials(raw)); + } + assert!( + is_trusted_verbatim(&result), + "flag must survive credential scrubbing" + ); + } + #[test] fn estimate_text_tokens_prices_image_marker_flat_not_by_length() { let huge = "x".repeat(40_000); @@ -4179,6 +4272,45 @@ mod tests { ); } + #[tokio::test] + async fn contract_gate_found_deliveries_are_exempt_but_not_found_still_trips() { + let handle = SteeringHandle::allow_all(); + let mw = RepeatedToolFailureMiddleware::new( + handle.clone(), + 3, + std::sync::Arc::new(std::sync::Mutex::new(None)), + ); + // Three identical contract-gate Found deliveries (issue #4853): they lead + // with the `[contract-gate:…]` marker, so they are not failures and must + // NOT trip the breaker even at the identical-repeat threshold. + for _ in 0..3 { + let mut r = failing_result( + "composio_execute", + "[contract-gate:composio:GMAIL_FETCH_EMAILS]\n\nRead the FULL contract before calling it.", + ); + mw.after_tool(&mut ctx(), &(), &mut r).await.unwrap(); + } + assert_eq!( + drain_pause_count(&handle), + 0, + "contract-gate Found deliveries must be exempt from the failure breaker" + ); + // Contrast: a NotFound "no such action" carries no marker → it is a real + // error, so a model looping on a bogus slug still trips the breaker. + for _ in 0..3 { + let mut r = failing_result( + "composio_execute", + "`GMAIL_NOPE` is not a real action in the `gmail` toolkit.", + ); + mw.after_tool(&mut ctx(), &(), &mut r).await.unwrap(); + } + assert_eq!( + drain_pause_count(&handle), + 1, + "a repeated NotFound (bogus slug) must still trip the breaker" + ); + } + #[tokio::test] async fn repeated_tool_failure_resets_on_a_success() { let handle = SteeringHandle::allow_all(); diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 4412ae314f..2d2578b382 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -20,6 +20,7 @@ //! tinyagents harness. mod abort_guard; +pub(crate) mod contract_gate; mod convert; pub(crate) mod delegation; mod embeddings; @@ -2254,6 +2255,38 @@ fn assemble_turn_harness( )); } + // Contract gate (issue #4853): require the model to see a late-bound tool's + // FULL contract before it executes. The `before_model` hook is registered + // here — AFTER the compaction/trim middlewares above — so it rebuilds the + // in-context contract set from the post-compaction transcript (presence is + // derived from `[contract-gate:]` markers, not tracked as turn state). + // The `wrap_tool` gate itself is registered further down, among the + // tool-wrap middlewares. Same `Arc`, two lists (the dual-trait pattern + // `SchemaGuardMiddleware` uses). + // + // The gate is handed the authoritative set of registered Composio per-action + // slugs (each tool self-reports via `Tool::composio_action_slug`), so a bare + // per-action call is gated by identity, never by name shape — a non-Composio + // tool can't be mistaken for a Composio action and gated into never running. + let composio_actions: HashSet = tool_sets + .iter() + .flat_map(|set| set.iter()) + .filter_map(|tool| tool.composio_action_slug().map(str::to_string)) + .collect(); + // Same identity-from-the-tool pattern for workflows: the registered + // `run_workflow` tool self-reports the active profile's runnable-workflow + // allowlist, so the gate never renders the contract of a workflow the profile + // scopes out (it defers to the tool's own scoped rejection instead). + let workflow_allowlist: Option> = tool_sets + .iter() + .flat_map(|set| set.iter()) + .find_map(|tool| tool.workflow_gate_allowlist().cloned()); + let contract_gate = Arc::new(contract_gate::ContractGateMiddleware::new( + composio_actions, + workflow_allowlist, + )); + harness.push_middleware(contract_gate.clone()); + // SDK-owned tool-policy projection (issue #4249 / tinyagents-full-migration // 01.1). Keep this narrow for now: enforce sandbox requirements declared by // adapter policies without enabling classification/approval/result-byte @@ -2279,6 +2312,17 @@ fn assemble_turn_harness( let schema_guard = Arc::new(middleware::SchemaGuardMiddleware::new(tool_sets.clone())); harness.push_tool_middleware(schema_guard.clone()); + // Contract gate `wrap_tool` (issue #4853; see the `before_model` + // registration above for the shared `Arc`). The first call this turn to a + // gated late-bound tool (a Composio action, an MCP-registry remote tool, or + // a Workflow) is short-circuited with that target's full contract as a tool + // error; the retry — now with the contract in context — executes normally. + // Installed right after schema_guard and BEFORE approval/policy so a + // first-call contract delivery returns without triggering an approval prompt + // or policy evaluation (nothing is executed — the model just gets the + // contract). Disable with `OPENHUMAN_CONTRACT_GATE=0`. + harness.push_tool_middleware(contract_gate.clone()); + // Human-in-the-loop approval as a named tool middleware (issue #4249, // Phase 1): an external-effect tool intercepts through the global // `ApprovalGate`, a denial short-circuits with a model-consumable result, and diff --git a/src/openhuman/tinyagents/tests.rs b/src/openhuman/tinyagents/tests.rs index bce571eeeb..a9e8a14d18 100644 --- a/src/openhuman/tinyagents/tests.rs +++ b/src/openhuman/tinyagents/tests.rs @@ -549,20 +549,20 @@ fn adapter_inventory_registers_model_tools_and_middleware() { // shadow tool-exposure, prompt-cache segment + guard, cache-align + tool-output // (TurnContextMiddleware::defaults), observe-only crate BudgetMiddleware // (W2-budget-dedupe), cost budget (local enforcement + budget_shadow), - // context compression + message trim (window known + autocompact on), SDK - // tool-policy projection, tool-outcome capture, arg recovery, schema guard - // (#4451 before_tool). + // context compression + message trim (window known + autocompact on), + // contract gate (#4853 before_model), SDK tool-policy projection, tool-outcome + // capture, arg recovery, schema guard (#4451 before_tool). let mw = assembled.harness.middleware(); - // NOTE(parity merge): these inventory counts are the upstream base (13 / 2) - // plus the lifecycle + around-tool middlewares this parity branch adds. They - // are NOT compiled by `cargo check --lib` (this is a #[cfg(test)] block) and - // are pending the deferred test pass — verify/adjust the exact numbers when - // the test suite actually compiles. - assert_eq!(mw.len(), 15, "lifecycle middleware inventory"); - // Around-tool wraps: schema guard (#4451, outermost) + approval/security + - // CLI/RPC-only scope gate + credential scrub (#4453, innermost). No builder - // tool policy on this call. - assert_eq!(mw.tool_middleware_len(), 4, "tool middleware inventory"); + // These counts pin the exact lifecycle + around-tool middleware set + // `assemble_turn_harness` installs. This is a LIVE test — it runs in CI's + // Rust Core Coverage job (`cargo test -- openhuman::tinyagents`), so adding + // or removing a middleware fails this assertion until the count is updated. + // Keep both numbers in sync when you change the harness assembly. + assert_eq!(mw.len(), 16, "lifecycle middleware inventory"); + // Around-tool wraps: schema guard (#4451, outermost) + contract gate (#4853) + // + approval/security + CLI/RPC-only scope gate + credential scrub (#4453, + // innermost). No builder tool policy on this call. + assert_eq!(mw.tool_middleware_len(), 5, "tool middleware inventory"); // One around-model wrap: the cost `UsageCarryMiddleware` (always installed). // RequiredCapabilities/FallbackObserver are not installed on this call // (no required caps; `mock-model` is not a tier, so no fallback chain). @@ -633,8 +633,8 @@ fn adapter_inventory_gates_context_middleware_on_window() { let mw = assembled.harness.middleware(); assert_eq!( mw.len(), - 13, - "compression + trim must not install without a window" + 14, + "compression + trim must not install without a window (contract gate still installs)" ); assert!(assembled.early_exit_hook.is_none()); } @@ -797,3 +797,165 @@ fn agent_turn_wall_clock_ms_parses_env_override() { Some(DEFAULT_AGENT_TURN_TIMEOUT_SECS * 1_000) ); } + +/// Mock provider for the contract-gate integration test: calls 0 and 1 both issue +/// the SAME gated Composio action (call 0 is the gated first attempt, call 1 the +/// retry once the contract is in context), then it answers. +struct GatedComposioThenRetry { + calls: AtomicUsize, +} + +#[async_trait] +impl Provider for GatedComposioThenRetry { + async fn chat_with_system( + &self, + _s: Option<&str>, + _m: &str, + _model: &str, + _t: f64, + ) -> anyhow::Result { + Ok(String::new()) + } + async fn chat( + &self, + _r: ChatRequest<'_>, + _model: &str, + _t: f64, + ) -> anyhow::Result { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n <= 1 { + Ok(ChatResponse { + tool_calls: vec![ToolCall { + id: format!("call-{n}"), + name: "composio_execute".to_string(), + arguments: r#"{"tool":"GMAIL_FETCH_EMAILS"}"#.to_string(), + extra_content: None, + }], + ..Default::default() + }) + } else { + Ok(ChatResponse { + text: Some("done".to_string()), + ..Default::default() + }) + } + } + fn supports_native_tools(&self) -> bool { + true + } +} + +/// Stub for the gated tool: records how many times it actually executed, so the +/// test can prove the gated first attempt never reached it. +struct RecordingComposioExecute { + executed: Arc, +} + +#[async_trait] +impl Tool for RecordingComposioExecute { + fn name(&self) -> &str { + "composio_execute" + } + fn description(&self) -> &str { + "stub Composio dispatcher for the contract-gate integration test" + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { "tool": { "type": "string" } }, + "required": ["tool"] + }) + } + async fn execute(&self, _args: serde_json::Value) -> anyhow::Result { + self.executed.fetch_add(1, Ordering::SeqCst); + Ok(ToolResult::success("FAKE_EXECUTED")) + } +} + +/// Integration test (issue #4853, reviewer request): drive a gated Composio call +/// through the REAL `assemble_turn_harness` middleware ordering and prove the +/// deliver → retry contract: +/// - the gated first attempt is short-circuited with the contract (the tool +/// does NOT run), +/// - the delivered contract re-enters the transcript, `before_model` rescans it, +/// - the retry passes the gate and the tool executes exactly once. +/// +/// A `cfg(test)` contract-source seam supplies the contract so the gate resolves +/// without a live Composio lookup. +#[tokio::test] +async fn contract_gate_delivers_the_contract_then_the_retry_executes() { + let mut contracts = std::collections::HashMap::new(); + contracts.insert( + "composio:GMAIL_FETCH_EMAILS".to_string(), + "GMAIL_FETCH_EMAILS(query: string, max_results: integer) — full contract".to_string(), + ); + // Held for the whole turn: clears the injection (and the serial lock) on drop. + let _injected = contract_gate::test_support::InjectedContracts::found(contracts); + + let executed = Arc::new(AtomicUsize::new(0)); + let provider: Arc = Arc::new(GatedComposioThenRetry { + calls: AtomicUsize::new(0), + }); + let provider_id = provider.telemetry_provider_id(); + let tool: Box = Box::new(RecordingComposioExecute { + executed: executed.clone(), + }); + let registry: Arc>> = Arc::new(vec![tool]); + let history = vec![ChatMessage::user("fetch my gmail")]; + + let turn_models = build_turn_models(provider, "mock-model", 0.0, None); + let outcome = run_turn_via_tinyagents_shared( + turn_models, + provider_id, + "mock-model", + history, + vec![registry], + None, + 6, + None, + None, + None, + None, + &[], + false, + None, + TurnContextMiddleware::defaults(), + None, + None, + false, + false, + ) + .await + .expect("gated turn runs"); + + // The single strongest assertion: exactly one execution. + // - 2 would mean the gate never blocked the first attempt. + // - 0 would mean the delivered contract was not recognised on the retry, so + // the gate kept blocking (the marker never round-tripped through + // `before_model`). + // 1 proves both halves of deliver → retry end to end. + assert_eq!( + executed.load(Ordering::SeqCst), + 1, + "the tool must run exactly once — on the retry, never on the gated first call" + ); + assert_eq!(outcome.text, "done"); + + // The delivery (marker + contract naming the slug) precedes the execution. + let delivery_idx = outcome + .history + .iter() + .position(|m| { + m.content.contains("contract-gate") && m.content.contains("GMAIL_FETCH_EMAILS") + }) + .expect("a contract-delivery tool message naming the slug"); + let exec_idx = outcome + .history + .iter() + .position(|m| m.content.contains("FAKE_EXECUTED")) + .expect("a real-execution tool message"); + assert!( + delivery_idx < exec_idx, + "the contract must be delivered before the retry executes (delivery {delivery_idx}, exec {exec_idx})" + ); +} diff --git a/src/openhuman/tools/traits.rs b/src/openhuman/tools/traits.rs index 22bb92da89..c863fe9dc8 100644 --- a/src/openhuman/tools/traits.rs +++ b/src/openhuman/tools/traits.rs @@ -351,6 +351,31 @@ pub trait Tool: Send + Sync { ToolCategory::System } + /// The Composio action slug this tool executes, if it is a per-action + /// Composio tool (`ComposioActionTool`). `None` for every other tool. + /// + /// The contract gate (`tinyagents::contract_gate`, issue #4853) uses this to + /// decide whether a bare tool-name call is a real registered Composio action + /// — authoritatively, from the tool itself, never by name shape — so a + /// non-Composio tool can never be mistaken for one and gated into never + /// executing. + fn composio_action_slug(&self) -> Option<&str> { + None + } + + /// The per-profile workflow-id allowlist this tool enforces when it runs + /// late-bound workflows (`RunWorkflowTool`). `Some(set)` restricts to those + /// `workflow_id`s; `None` for every other tool and for an unrestricted + /// profile (all workflows visible). + /// + /// The contract gate (`tinyagents::contract_gate`, issue #4853) reads this + /// so it never renders the input contract of a workflow the active profile + /// scopes out — it defers such a call to the tool's own scoped rejection, + /// mirroring the `run_workflow` / `describe_workflow` allowlist checks. + fn workflow_gate_allowlist(&self) -> Option<&std::collections::HashSet> { + None + } + /// Whether two concurrent invocations of this tool are safe to /// run in parallel inside a single LLM iteration. ///