fix(composio): execute valid first calls + recover Kimi tool-call garble (#5119) (#5132)

src/openhuman/agent/harness/parse.rs

@@ -230,7 +230,38 @@ fn normalize_garbled_tool_call_tags(s: &str) -> Cow<'_, str> {
         let (close_start, close_end) = pair[1];
         out.push_str(&s[cursor..open_start]); // text before the open tag, verbatim
         out.push_str("<tool_call>");
-        out.push_str(strip_call_prefix(&s[open_end..close_start]));
+        // Strip the `call:` prefix, then try to recover a Kimi-family
+        // `NAME{…}` argument-sentinel body into canonical JSON (#5119). When
+        // the body is already canonical JSON / P-Format the recovery is a no-op
+        // and the stripped body flows through unchanged.
+        let stripped = strip_call_prefix(&s[open_end..close_start]);
+        match recover_sentinel_tool_call_body(stripped) {
+            Some(recovered) => {
+                // Recovered a Kimi-family `NAME{…}` sentinel body into canonical
+                // JSON. body_chars only (never the body itself — it may carry
+                // tool arguments with user data); stable `[agent_parse]` prefix
+                // so it aggregates with the other harness log families.
+                tracing::debug!(
+                    body_chars = recovered.chars().count(),
+                    "[agent_parse] recovered Kimi-family sentinel tool-call body into canonical JSON (#5119)"
+                );
+                out.push_str(&recovered)
+            }
+            None => {
+                // A body still carrying the `<|"|>` arg-quote sentinel that
+                // recovery could NOT normalize is a new Kimi garble variant.
+                // Surface it (body_chars only — never the body: it may carry
+                // user data) so operators debugging a future unrecovered variant
+                // get a signal instead of a silently dropped tool call.
+                if stripped.contains(ARG_QUOTE_SENTINEL) {
+                    tracing::warn!(
+                        body_chars = stripped.chars().count(),
+                        "[agent_parse] unrecovered Kimi-family sentinel tool-call body; passing through as text (#5119)"
+                    );
+                }
+                out.push_str(stripped)
+            }
+        }
         out.push_str("</tool_call>");
         cursor = close_end;
     }
@@ -249,6 +280,117 @@ fn strip_call_prefix(body: &str) -> &str {
         .unwrap_or(trimmed)
 }
 
+/// The Kimi-K2-family argument-quote sentinel that leaks in place of a real `"`
+/// around string values (`[<|"|>INBOX<|"|>]` instead of `["INBOX"]`). It is the
+/// body-level sibling of the tag garble [`normalize_garbled_tool_call_tags`]
+/// already repairs; see [`recover_sentinel_tool_call_body`].
+const ARG_QUOTE_SENTINEL: &str = "<|\"|>";
+
+/// Recover a Kimi-K2-family garbled tool-call **body** into canonical
+/// `{"name":…,"arguments":…}` JSON (#5119).
+///
+/// The managed `burst`/`chat` tiers are Kimi-K2-family models; in text mode they
+/// sometimes render a call as `NAME{…}` — the action name before a JSON-ish
+/// argument object with **unquoted keys** and the `<|"|>` sentinel in place of
+/// string quotes — e.g. `GMAIL_FETCH_EMAILS{label_ids:[<|"|>INBOX<|"|>],max_results:1}`.
+/// After [`normalize_garbled_tool_call_tags`] fixes the surrounding tags this
+/// body still matches neither the JSON nor the P-Format grammar, so the call is
+/// dropped as narrative text and the tool never runs (the turn then loops).
+///
+/// Recovery: replace the `<|"|>` sentinels with real quotes, split the leading
+/// action name off the `{…}` object, quote the object's bare keys, and re-emit
+/// as `{"name":"NAME","arguments":{…}}` for the existing JSON parser. Returns
+/// `None` — leaving the body untouched — whenever the shape does not match: a
+/// canonical JSON body (`{…}`, empty name), a P-Format body (`NAME[…]`, no `{`),
+/// or the already-handled `call:{"name":…}` form all fall through unchanged.
+fn recover_sentinel_tool_call_body(body: &str) -> Option<String> {
+    let repaired = body.replace(ARG_QUOTE_SENTINEL, "\"");
+    let trimmed = repaired.trim();
+
+    // Shape must be `NAME{…}`: a bare action identifier immediately followed by
+    // a brace object. A body already starting with `{` yields an empty name and
+    // is left to the JSON parser; a `NAME[…]` P-Format body has no `{`.
+    let brace = trimmed.find('{')?;
+    let name = trimmed[..brace].trim();
+    if name.is_empty() || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+        return None;
+    }
+    let object = trimmed[brace..].trim_end();
+    if !object.starts_with('{') || !object.ends_with('}') {
+        return None;
+    }
+
+    // Kimi renders object keys unquoted (`{label_ids: …}`); quote them so the
+    // result is strict JSON, then confirm it actually parses as an object before
+    // committing to the rewrite.
+    let quoted = quote_bare_json_object_keys(object);
+    let arguments: serde_json::Value = serde_json::from_str(&quoted).ok()?;
+    if !arguments.is_object() {
+        return None;
+    }
+
+    serde_json::to_string(&serde_json::json!({ "name": name, "arguments": arguments })).ok()
+}
+
+/// Quote every **bare** object key (`{label_ids: …}` → `{"label_ids": …}`) in a
+/// JSON-ish string, tracking string context so a `:` or identifier inside a
+/// value never triggers a spurious rewrite. Bare literal values
+/// (`true`/`false`/`null`/numbers) are left untouched — they are valid JSON —
+/// and already-quoted keys pass through unchanged.
+fn quote_bare_json_object_keys(s: &str) -> String {
+    let mut out = String::with_capacity(s.len() + 16);
+    let mut in_string = false;
+    let mut escaped = false;
+    // True right after a structural `{` or `,` — the only positions where a bare
+    // key may begin.
+    let mut expect_key = false;
+    let mut chars = s.chars().peekable();
+    while let Some(c) = chars.next() {
+        if in_string {
+            out.push(c);
+            if escaped {
+                escaped = false;
+            } else if c == '\\' {
+                escaped = true;
+            } else if c == '"' {
+                in_string = false;
+            }
+            continue;
+        }
+        match c {
+            '"' => {
+                in_string = true;
+                expect_key = false;
+                out.push(c);
+            }
+            '{' | ',' => {
+                expect_key = true;
+                out.push(c);
+            }
+            c if c.is_whitespace() => out.push(c), // keep looking for a key
+            c if expect_key && (c.is_ascii_alphabetic() || c == '_') => {
+                out.push('"');
+                out.push(c);
+                while let Some(&nc) = chars.peek() {
+                    if nc.is_ascii_alphanumeric() || nc == '_' {
+                        out.push(nc);
+                        chars.next();
+                    } else {
+                        break;
+                    }
+                }
+                out.push('"');
+                expect_key = false;
+            }
+            _ => {
+                expect_key = false;
+                out.push(c);
+            }
+        }
+    }
+    out
+}
+
 /// `<invoke` prefix shared by the bare (`<invoke>`) and Claude-native
 /// attribute (`<invoke name="…">`) forms.
 const INVOKE_PREFIX: &str = "<invoke";

src/openhuman/agent/harness/parse_tests.rs

@@ -432,3 +432,58 @@ fn normalize_leaves_pformat_pipe_args_untouched() {
         std::borrow::Cow::Borrowed(_)
     ));
 }
+
+// ── #5119: recover the Kimi `NAME{…}` argument-sentinel body ─────────────────
+
+#[test]
+fn garbled_kimi_name_brace_body_with_quote_sentinels_parses() {
+    // The EXACT shape observed from `integrations_agent`/`burst-v1` (Kimi-K2) on
+    // the post-contract retry: garbled tags PLUS a `NAME{…}` body with unquoted
+    // keys and the `<|"|>` argument-quote sentinel around string values. Before
+    // the recovery this parsed to zero tool calls (the tag fix alone left an
+    // unparseable body), so GMAIL_FETCH_EMAILS never ran and the turn looped.
+    let garbled = r#"<|tool_call>call:GMAIL_FETCH_EMAILS{label_ids:[<|"|>INBOX<|"|>],max_results:1,verbose:true}<tool_call|>"#;
+    let (_text, calls) = parse_tool_calls(garbled);
+    assert_eq!(calls.len(), 1, "the garbled Kimi call must be recovered");
+    assert_eq!(calls[0].name, "GMAIL_FETCH_EMAILS");
+    assert_eq!(
+        calls[0].arguments["label_ids"],
+        serde_json::json!(["INBOX"])
+    );
+    assert_eq!(calls[0].arguments["max_results"], 1);
+    assert_eq!(calls[0].arguments["verbose"], true);
+}
+
+#[test]
+fn garbled_kimi_name_brace_body_integer_only_parses() {
+    // The integer-only variant (no string values → no `<|"|>` sentinel, but the
+    // body is still the unparseable `NAME{unquoted-keys}` shape). Observed as
+    // `{max_results:5}` on the staging repro.
+    let garbled = r#"<|tool_call>call:GMAIL_FETCH_EMAILS{max_results:5}<tool_call|>"#;
+    let (_text, calls) = parse_tool_calls(garbled);
+    assert_eq!(calls.len(), 1);
+    assert_eq!(calls[0].name, "GMAIL_FETCH_EMAILS");
+    assert_eq!(calls[0].arguments["max_results"], 5);
+}
+
+#[test]
+fn recover_sentinel_body_leaves_canonical_and_pformat_untouched() {
+    // Canonical JSON body → empty leading name → not our shape → None.
+    assert!(recover_sentinel_tool_call_body(r#"{"name":"echo","arguments":{}}"#).is_none());
+    // P-Format body (`NAME[…]`, no brace) → None.
+    assert!(recover_sentinel_tool_call_body("get_weather[London|metric]").is_none());
+    // Trailing garbage after the object → not a clean `NAME{…}` → None.
+    assert!(recover_sentinel_tool_call_body("FOO{a:1} trailing").is_none());
+}
+
+#[test]
+fn quote_bare_json_object_keys_respects_string_values() {
+    // A `,ident:` sequence INSIDE a string value must not be quoted; only
+    // structural keys after `{`/`,` are rewritten.
+    let out = quote_bare_json_object_keys(r#"{query:"from:john,to:x",n:1}"#);
+    assert_eq!(out, r#"{"query":"from:john,to:x","n":1}"#);
+    // Parses as strict JSON with the value preserved verbatim.
+    let v: serde_json::Value = serde_json::from_str(&out).unwrap();
+    assert_eq!(v["query"], "from:john,to:x");
+    assert_eq!(v["n"], 1);
+}

src/openhuman/composio/action_tool.rs

@@ -227,7 +227,9 @@ impl Tool for ComposioActionTool {
         // 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 {
+        match super::contract_gate::consult(&self.gate, &live_config, &self.action_name, &args)
+            .await
+        {
             super::contract_gate::GateDecision::Surface(contract) => {
                 tracing::info!(
                     tool = %self.action_name,

src/openhuman/composio/contract_gate.rs

@@ -83,15 +83,32 @@ pub enum GateDecision {
     Proceed,
 }
 
-/// Consult the gate before executing `action_slug`.
+/// Consult the gate before executing `action_slug` with the model's `args`.
 ///
 /// On the FIRST consult for a slug this turn, if a fuller live contract can be
-/// resolved, returns [`GateDecision::Surface`] with the formatted contract and
-/// marks the slug seen. Every later consult — and any consult where no live
-/// contract is available (unconfigured client, unknown action, network miss) —
-/// returns [`GateDecision::Proceed`], so the gate never blocks an action more
-/// than once and never blocks when it cannot help.
-pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) -> GateDecision {
+/// resolved, the gate compares the model's supplied `args` against it:
+///
+/// - **Args already satisfy the contract** (all required present, every supplied
+///   key a known property, types compatible) → [`GateDecision::Proceed`]. The
+///   model did not need the schema, so bouncing would be pure overhead — and, on
+///   the weak text-mode `integrations_agent` path, forcing a needless retry lets
+///   a Kimi-family model corrupt the re-issued call (`<|"|>` sentinel-token leak)
+///   and loop forever without ever executing (#5119).
+/// - **Args do NOT satisfy the contract** (missing required, unknown key, wrong
+///   type — i.e. the model *guessed*) → [`GateDecision::Surface`] with the
+///   formatted contract, exactly the case the gate exists for (#4853).
+///
+/// The slug is marked seen on this first consult either way, so every later
+/// consult — and any consult where no live contract is available (unconfigured
+/// client, unknown action, network miss) — returns [`GateDecision::Proceed`]:
+/// the gate never blocks an action more than once and never blocks when it
+/// cannot help.
+pub async fn consult(
+    gate: &ContractGate,
+    config: &Config,
+    action_slug: &str,
+    args: &serde_json::Value,
+) -> GateDecision {
     // Mark first (releasing the lock) so the retry — and any concurrent
     // sibling call — proceeds even if the contract lookup below is slow.
     let first_time = gate.mark_seen(action_slug);
@@ -110,6 +127,16 @@ pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) ->
     // just does not get the pre-execute contract nudge).
     #[cfg(feature = "flows")]
     if let Some(contract) = lookup_contract(config, action_slug).await {
+        // Validate-then-pass (#5119): only surface when the model actually needs
+        // the schema. A call whose args already conform is executed directly.
+        if args_satisfy_contract(args, &contract) {
+            tracing::debug!(
+                target: "composio",
+                slug = %action_slug,
+                "[composio][contract-gate] args already satisfy the live contract; proceeding without surfacing"
+            );
+            return GateDecision::Proceed;
+        }
         tracing::debug!(
             target: "composio",
             slug = %action_slug,
@@ -120,9 +147,10 @@ pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) ->
         return GateDecision::Surface(format_contract(action_slug, &contract));
     }
 
-    // `config` is only consulted through the flows-gated lookup above.
+    // `config` and `args` are only consulted through the flows-gated lookup +
+    // validation above.
     #[cfg(not(feature = "flows"))]
-    let _ = config;
+    let _ = (config, args);
 
     tracing::debug!(
         target: "composio",
@@ -132,6 +160,106 @@ pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) ->
     GateDecision::Proceed
 }
 
+/// Whether the model's supplied `args` already conform to `contract` — the test
+/// that lets the gate execute a well-formed first call instead of bouncing it
+/// (#5119). Conservative: an object whose required args are all present, whose
+/// every supplied key is a known schema property, and whose values are
+/// type-compatible with the schema. Anything short of that is treated as a
+/// guess and surfaces the contract (#4853).
+///
+/// Type checks are intentionally lenient about stringified scalars (a model may
+/// send `max_results: "10"`), so only a genuinely wrong shape — a string where
+/// an array is required, an unknown/invented key, a missing required arg — fails.
+/// When the schema publishes no `properties`, only the required-args presence
+/// check applies.
+#[cfg(feature = "flows")]
+fn args_satisfy_contract(args: &serde_json::Value, contract: &ToolContract) -> bool {
+    let obj = match args.as_object() {
+        Some(obj) => obj,
+        // Non-object args satisfy the contract only when nothing is required
+        // (e.g. a no-arg action called with `null`/absent args).
+        None => return contract.required_args.is_empty(),
+    };
+
+    // Every required argument must be present and non-null.
+    for req in &contract.required_args {
+        match obj.get(req) {
+            Some(v) if !v.is_null() => {}
+            _ => return false,
+        }
+    }
+
+    // If the schema publishes its properties, every supplied key must be known
+    // (no invented args) and type-compatible. A hallucinated key or a
+    // wrong-typed value is exactly the guess the gate exists to catch.
+    if let Some(props) = contract
+        .input_schema
+        .as_ref()
+        .and_then(|s| s.get("properties"))
+        .and_then(|p| p.as_object())
+    {
+        for (key, value) in obj {
+            // `connection_id` is an OpenHuman-injected routing parameter
+            // (`ComposioActionTool::parameters_schema` / `ComposioExecuteTool`),
+            // consumed before dispatch and absent from Composio's live catalog
+            // `input_schema`. Skip it so a valid multi-account call isn't bounced
+            // as an "unknown key" into the retry path this gate exists to avoid.
+            if key == "connection_id" {
+                continue;
+            }
+            match props.get(key) {
+                None => return false,
+                Some(prop) => {
+                    if let Some(expected) = prop.get("type").and_then(|t| t.as_str()) {
+                        if !json_value_matches_type(value, expected) {
+                            return false;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    true
+}
+
+/// Loose JSON-Schema scalar/compound `type` check used by
+/// [`args_satisfy_contract`]. Numeric/boolean types also accept a string that
+/// parses to that type, so a model sending `"10"` for an `integer` field is not
+/// treated as a schema violation. An unrecognised or union `type` (the
+/// `and_then(as_str)` returns `None` for a `["string","null"]` array) is never
+/// reached here, so callers simply skip the check — lenient by construction.
+#[cfg(feature = "flows")]
+fn json_value_matches_type(value: &serde_json::Value, expected: &str) -> bool {
+    match expected {
+        "string" => value.is_string(),
+        "integer" => {
+            value.is_i64()
+                || value.is_u64()
+                || value
+                    .as_str()
+                    .is_some_and(|s| s.trim().parse::<i64>().is_ok())
+        }
+        "number" => {
+            value.is_number()
+                || value
+                    .as_str()
+                    .is_some_and(|s| s.trim().parse::<f64>().is_ok())
+        }
+        "boolean" => {
+            value.is_boolean()
+                || value
+                    .as_str()
+                    .is_some_and(|s| matches!(s.trim(), "true" | "false"))
+        }
+        "array" => value.is_array(),
+        "object" => value.is_object(),
+        "null" => value.is_null(),
+        // Unknown/unsupported type keyword → don't reject on type grounds.
+        _ => true,
+    }
+}
+
 /// Resolve the full live contract for `action_slug` from the process-cached
 /// live toolkit catalog. Returns `None` when the toolkit can't be derived, the
 /// catalog can't be fetched (unconfigured / offline — `fetch_live_toolkit_catalog`

src/openhuman/composio/contract_gate_tests.rs

@@ -9,9 +9,9 @@ use super::{consult, ContractGate, GateDecision};
 use crate::openhuman::config::Config;
 use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract};
 
-/// Build a full contract for `slug` in `toolkit` with a `query` input field and
-/// a description that spells out the quoting rule — the exact detail the model
-/// misses when it only sees the thin spawn-time schema.
+/// Build a full contract for `slug` in `toolkit` with a REQUIRED `query` input
+/// field and a description that spells out the quoting rule — the exact detail
+/// the model misses when it only sees the thin spawn-time schema.
 fn full_contract(slug: &str, toolkit: &str) -> ToolContract {
     ToolContract {
         slug: slug.to_string(),
@@ -39,6 +39,38 @@ fn full_contract(slug: &str, toolkit: &str) -> ToolContract {
     }
 }
 
+/// Build a contract with only OPTIONAL, typed args — mirrors the real
+/// `GMAIL_FETCH_EMAILS` shape at the heart of #5119 (no required args; the model
+/// supplies `label_ids`/`max_results`/`verbose`). Used to prove validate-then-pass.
+fn fetch_contract(slug: &str, toolkit: &str) -> ToolContract {
+    ToolContract {
+        slug: slug.to_string(),
+        toolkit: toolkit.to_string(),
+        description: Some("Fetch emails from the inbox.".to_string()),
+        required_args: Vec::new(),
+        input_schema: Some(serde_json::json!({
+            "type": "object",
+            "properties": {
+                "label_ids": { "type": "array", "items": { "type": "string" } },
+                "max_results": { "type": "integer" },
+                "verbose": { "type": "boolean" },
+                "query": { "type": "string" }
+            }
+        })),
+        output_fields: Vec::new(),
+        output_schema: None,
+        primary_array_path: None,
+        is_curated: false,
+    }
+}
+
+/// Args that do NOT satisfy [`full_contract`] (its required `query` is absent),
+/// so the gate surfaces the contract — the "model guessed / needs the schema"
+/// path these legacy tests exercise.
+fn guessing_args() -> serde_json::Value {
+    serde_json::json!({})
+}
+
 #[tokio::test]
 async fn first_call_surfaces_full_contract_then_retry_proceeds() {
     // Toolkit derived from the slug prefix: `GMAILGATE_...` -> `gmailgate`.
@@ -49,10 +81,9 @@ async fn first_call_surfaces_full_contract_then_retry_proceeds() {
     let config = Config::default();
     let gate = ContractGate::new();
 
-    // First call: the gate short-circuits execution and hands back the full
-    // contract (this is the behaviour that is ABSENT before the fix — the thin
-    // per-action tool would execute immediately with a guessed query).
-    match consult(&gate, &config, slug).await {
+    // First call with args that MISS the required `query`: the gate
+    // short-circuits execution and hands back the full contract.
+    match consult(&gate, &config, slug, &guessing_args()).await {
         GateDecision::Surface(message) => {
             assert!(message.contains(slug), "contract names the action slug");
             assert!(
@@ -73,7 +104,10 @@ async fn first_call_surfaces_full_contract_then_retry_proceeds() {
 
     // The retry — now with the contract in context — proceeds to execution.
     assert!(
-        matches!(consult(&gate, &config, slug).await, GateDecision::Proceed),
+        matches!(
+            consult(&gate, &config, slug, &guessing_args()).await,
+            GateDecision::Proceed
+        ),
         "retry must proceed once the contract has been surfaced this turn"
     );
 }
@@ -94,7 +128,7 @@ async fn known_toolkit_but_unknown_action_proceeds_without_blocking() {
 
     assert!(
         matches!(
-            consult(&gate, &config, "PARTIALKIT_FETCH_EMAILS").await,
+            consult(&gate, &config, "PARTIALKIT_FETCH_EMAILS", &guessing_args()).await,
             GateDecision::Proceed
         ),
         "an action missing from the live catalog must not be gated"
@@ -114,21 +148,154 @@ async fn distinct_actions_are_gated_independently() {
     let config = Config::default();
     let gate = ContractGate::new();
 
-    // Each action surfaces its own contract exactly once, independently.
+    // Each action (args miss the required `query`) surfaces its own contract
+    // exactly once, independently.
     assert!(matches!(
-        consult(&gate, &config, fetch).await,
+        consult(&gate, &config, fetch, &guessing_args()).await,
         GateDecision::Surface(_)
     ));
     assert!(matches!(
-        consult(&gate, &config, send).await,
+        consult(&gate, &config, send, &guessing_args()).await,
         GateDecision::Surface(_)
     ));
     assert!(matches!(
-        consult(&gate, &config, fetch).await,
+        consult(&gate, &config, fetch, &guessing_args()).await,
         GateDecision::Proceed
     ));
     assert!(matches!(
-        consult(&gate, &config, send).await,
+        consult(&gate, &config, send, &guessing_args()).await,
         GateDecision::Proceed
     ));
 }
+
+// ── #5119: validate-then-pass — a well-formed first call must NOT be bounced ──
+
+#[tokio::test]
+async fn first_call_with_satisfying_args_proceeds_without_surfacing() {
+    // The exact #5119 scenario: "fetch my latest email" → the model's FIRST
+    // call already carries schema-valid args. Bouncing it forces a needless
+    // retry that a weak text-mode model corrupts, looping forever. The gate must
+    // execute immediately instead.
+    let toolkit = "fetchok";
+    let slug = "FETCHOK_FETCH_EMAILS";
+    seed_live_catalog_cache(toolkit, vec![fetch_contract(slug, toolkit)]);
+
+    let config = Config::default();
+    let gate = ContractGate::new();
+
+    let valid = serde_json::json!({ "label_ids": ["INBOX"], "max_results": 1, "verbose": true });
+    assert!(
+        matches!(
+            consult(&gate, &config, slug, &valid).await,
+            GateDecision::Proceed
+        ),
+        "a first call whose args already satisfy the contract must execute, not surface"
+    );
+}
+
+#[tokio::test]
+async fn satisfied_required_arg_executes_immediately() {
+    // A required arg that IS present (and typed correctly) also passes on the
+    // first call — the gate only surfaces when the model actually guessed.
+    let toolkit = "reqok";
+    let slug = "REQOK_SEARCH";
+    seed_live_catalog_cache(toolkit, vec![full_contract(slug, toolkit)]);
+
+    let config = Config::default();
+    let gate = ContractGate::new();
+
+    let valid = serde_json::json!({ "query": "subject:\"quarterly report\"" });
+    assert!(
+        matches!(
+            consult(&gate, &config, slug, &valid).await,
+            GateDecision::Proceed
+        ),
+        "a satisfied required arg must proceed on the first call"
+    );
+}
+
+#[tokio::test]
+async fn synthetic_connection_id_does_not_bounce_a_valid_call() {
+    // #5119 review: `connection_id` is an OpenHuman-injected routing parameter
+    // (added by `ComposioActionTool::parameters_schema` / `ComposioExecuteTool`
+    // and consumed before dispatch), NOT a field in Composio's live catalog
+    // `input_schema`. A valid multi-account first call carries it, so the
+    // unknown-key check must skip it rather than bounce the call into the retry
+    // path this gate exists to avoid.
+    let toolkit = "connkit";
+    let slug = "CONNKIT_FETCH_EMAILS";
+    seed_live_catalog_cache(toolkit, vec![fetch_contract(slug, toolkit)]);
+
+    let config = Config::default();
+    let gate = ContractGate::new();
+
+    let valid = serde_json::json!({
+        "label_ids": ["INBOX"],
+        "max_results": 1,
+        "connection_id": "conn_abc123"
+    });
+    assert!(
+        matches!(
+            consult(&gate, &config, slug, &valid).await,
+            GateDecision::Proceed
+        ),
+        "a valid call carrying the synthetic connection_id must execute, not surface"
+    );
+}
+
+#[tokio::test]
+async fn missing_required_arg_surfaces() {
+    let toolkit = "missreq";
+    let slug = "MISSREQ_SEARCH";
+    seed_live_catalog_cache(toolkit, vec![full_contract(slug, toolkit)]);
+
+    let config = Config::default();
+    let gate = ContractGate::new();
+
+    // `query` is required but absent → surface.
+    let missing = serde_json::json!({ "verbose": true });
+    assert!(
+        matches!(
+            consult(&gate, &config, slug, &missing).await,
+            GateDecision::Surface(_)
+        ),
+        "a missing required arg must surface the contract"
+    );
+}
+
+#[tokio::test]
+async fn unknown_or_mistyped_args_surface() {
+    let toolkit = "guesskit";
+    let unknown_slug = "GUESSKIT_FETCH_A";
+    let mistyped_slug = "GUESSKIT_FETCH_B";
+    seed_live_catalog_cache(
+        toolkit,
+        vec![
+            fetch_contract(unknown_slug, toolkit),
+            fetch_contract(mistyped_slug, toolkit),
+        ],
+    );
+
+    let config = Config::default();
+    let gate = ContractGate::new();
+
+    // Invented key the schema never declares → the model guessed → surface.
+    let invented = serde_json::json!({ "invented_field": 1 });
+    assert!(
+        matches!(
+            consult(&gate, &config, unknown_slug, &invented).await,
+            GateDecision::Surface(_)
+        ),
+        "an unknown/hallucinated key must surface the contract"
+    );
+
+    // `max_results` is an integer; an array is a genuine type error → surface.
+    let mistyped = serde_json::json!({ "max_results": [1, 2, 3] });
+    assert!(
+        matches!(
+            consult(&gate, &config, mistyped_slug, &mistyped).await,
+            GateDecision::Surface(_)
+        ),
+        "a wrong-typed arg must surface the contract"
+    );
+}