fix(agent-harness): enforce required structured-output block every turn (#4117) (#4900)

Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/openhuman/agent/harness/mod.rs

@@ -34,6 +34,7 @@ pub(crate) mod memory_context;
 pub(crate) mod memory_context_safety;
 pub(crate) mod memory_protocol;
 pub(crate) mod parse;
+pub(crate) mod required_output;
 pub mod run_queue;
 pub mod sandbox_context;
 pub mod session;

src/openhuman/agent/harness/required_output.rs

@@ -0,0 +1,226 @@
+//! Required structured-output validation & repair (issue #4117).
+//!
+//! Some agent contracts mandate a JSON block — e.g. a `thoughts` block like
+//! `{"thoughts": "…", "next_action": "…"}` — on **every** turn, because
+//! downstream parsing/routing depends on it. Models frequently omit the block
+//! entirely, leaving those consumers with nothing.
+//!
+//! This module supplies the pure primitives the turn engine uses to guarantee a
+//! well-formed block on every accepted turn:
+//!
+//! * [`output_satisfies_contract`] — validate presence + shape of the block,
+//! * [`repair_instruction`] — the corrective re-prompt that asks the model to
+//!   re-emit its reply with the block, and
+//! * [`synthesize_block`] — a minimal, schema-valid block used as a deterministic
+//!   fallback when the re-prompt also omits it.
+//!
+//! The orchestration that ties these together (validate → re-prompt → synthesize)
+//! lives on the session in `session::turn::session_io` so it can drive the extra
+//! provider call and reconcile with streaming; keeping the logic here pure keeps
+//! it unit-testable without a provider.
+
+use crate::openhuman::config::RequiredOutputContract;
+
+/// Whether `text` satisfies `contract`: it contains a JSON object carrying every
+/// required key with a non-null value, in the expected leading position. An
+/// inert contract (no non-blank keys) is treated as always satisfied so
+/// enforcement is a no-op.
+pub(crate) fn output_satisfies_contract(text: &str, contract: &RequiredOutputContract) -> bool {
+    if !contract.is_active() {
+        return true;
+    }
+    find_required_block(text, contract).is_some()
+}
+
+/// The required block when it appears in the expected **leading position** —
+/// the *first* JSON value in `text` must be an object carrying every required
+/// key with a non-null value — else `None`.
+///
+/// Issue #4117 mandates the block in a predictable position so downstream
+/// parsing/routing can rely on it. Prose before the block is fine (prose is not
+/// JSON, so the block is still the first extracted value), but a block buried
+/// after *another* JSON object is rejected and gets repaired rather than
+/// silently accepted. Reuses the harness JSON extractor so fenced, inline, and
+/// whole-object replies are all recognised.
+pub(crate) fn find_required_block(
+    text: &str,
+    contract: &RequiredOutputContract,
+) -> Option<serde_json::Value> {
+    let keys = contract.all_keys();
+    if keys.is_empty() {
+        return None;
+    }
+    let first = super::parse::extract_json_values(text).into_iter().next()?;
+    let obj = first.as_object()?;
+    let has_all = keys
+        .iter()
+        .all(|key| obj.get(key).is_some_and(|v| !v.is_null()));
+    if has_all {
+        Some(first)
+    } else {
+        None
+    }
+}
+
+/// A minimal, schema-valid block synthesised when the model omits the block and
+/// a corrective re-prompt fails to recover it. Every required key maps to an
+/// empty string so downstream parsing always has a well-formed object to
+/// consume. Returns `"{}"` only for an inert contract (which enforcement never
+/// reaches).
+pub(crate) fn synthesize_block(contract: &RequiredOutputContract) -> String {
+    let mut obj = serde_json::Map::new();
+    for key in contract.all_keys() {
+        obj.insert(key, serde_json::Value::String(String::new()));
+    }
+    serde_json::to_string(&serde_json::Value::Object(obj)).unwrap_or_else(|_| "{}".to_string())
+}
+
+/// The corrective instruction that re-prompts the model to re-emit its reply
+/// with the required block. Mirrors the iteration-cap checkpoint re-prompt: a
+/// self-contained user turn appended after the omitting reply.
+///
+/// The model is asked to lead with the JSON object so the repaired reply
+/// satisfies the leading-position rule directly, whether the caller keeps the
+/// re-prompt as the whole reply (the non-streamed *replace* path) or appends it
+/// after prose that was already streamed (the *append* path); see
+/// `Agent::enforce_required_output`.
+pub(crate) fn repair_instruction(contract: &RequiredOutputContract) -> String {
+    let keys = contract.all_keys().join("\", \"");
+    format!(
+        "Your previous reply omitted the required JSON `{}` block that every turn must include. \
+Reply again, leading with a single valid JSON object containing the keys \"{}\" — all present \
+and non-null — then continue with your answer. Do not call any tools.",
+        contract.block_key, keys
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn thoughts_contract() -> RequiredOutputContract {
+        RequiredOutputContract {
+            block_key: "thoughts".into(),
+            required_keys: vec!["next_action".into()],
+        }
+    }
+
+    #[test]
+    fn present_well_formed_block_satisfies_contract() {
+        let contract = thoughts_contract();
+        let text = "Sure! {\"thoughts\": \"planning\", \"next_action\": \"call tool\"}";
+        assert!(output_satisfies_contract(text, &contract));
+        assert!(find_required_block(text, &contract).is_some());
+    }
+
+    #[test]
+    fn prose_only_reply_fails_validation() {
+        let contract = thoughts_contract();
+        assert!(!output_satisfies_contract(
+            "Sure, I'll handle that.",
+            &contract
+        ));
+    }
+
+    #[test]
+    fn block_missing_a_required_sibling_key_fails() {
+        let contract = thoughts_contract();
+        // Has `thoughts` but not `next_action`.
+        let text = "{\"thoughts\": \"planning\"}";
+        assert!(!output_satisfies_contract(text, &contract));
+    }
+
+    #[test]
+    fn null_valued_required_key_fails() {
+        let contract = RequiredOutputContract::new("thoughts");
+        assert!(!output_satisfies_contract(
+            "{\"thoughts\": null}",
+            &contract
+        ));
+    }
+
+    #[test]
+    fn synthesized_block_satisfies_its_own_contract() {
+        let contract = thoughts_contract();
+        let synthesized = synthesize_block(&contract);
+        assert!(
+            output_satisfies_contract(&synthesized, &contract),
+            "synthesized fallback must satisfy the contract it was built from: {synthesized}"
+        );
+    }
+
+    #[test]
+    fn leading_block_after_prose_is_accepted() {
+        let contract = thoughts_contract();
+        // Prose before the block is fine — prose is not JSON, so the block is
+        // still the first extracted value.
+        let text = "Here is my plan.\n{\"thoughts\": \"x\", \"next_action\": \"y\"}";
+        assert!(output_satisfies_contract(text, &contract));
+    }
+
+    #[test]
+    fn synthesized_block_prepended_to_prose_leads_correctly() {
+        // The non-streamed *replace* fallback prepends a synthesized block to the
+        // original prose; the block must be the first JSON value so the reply
+        // validates.
+        let contract = thoughts_contract();
+        let repaired = format!("{}\n\n{}", synthesize_block(&contract), "Working on it.");
+        assert!(output_satisfies_contract(&repaired, &contract));
+    }
+
+    #[test]
+    fn block_buried_after_another_json_object_is_rejected() {
+        let contract = thoughts_contract();
+        // A different JSON object leads; the required block is second → rejected
+        // so it gets repaired rather than silently accepted (issue #4117).
+        let text = "{\"foo\": 1}\n{\"thoughts\": \"x\", \"next_action\": \"y\"}";
+        assert!(!output_satisfies_contract(text, &contract));
+    }
+
+    #[test]
+    fn blank_block_key_makes_contract_inert() {
+        // A blank block key is inert even when sibling keys are listed — the
+        // contract's defining key can never be enforced, so enforcement is
+        // skipped instead of accepting a block missing that key.
+        let contract = RequiredOutputContract {
+            block_key: "   ".into(),
+            required_keys: vec!["next_action".into()],
+        };
+        assert!(!contract.is_active());
+        assert!(output_satisfies_contract(
+            "{\"next_action\": \"y\"}",
+            &contract
+        ));
+    }
+
+    #[test]
+    fn inert_contract_is_always_satisfied() {
+        let contract = RequiredOutputContract::default();
+        assert!(!contract.is_active());
+        assert!(output_satisfies_contract("no block here", &contract));
+        assert!(find_required_block("no block here", &contract).is_none());
+    }
+
+    #[test]
+    fn all_keys_trims_and_dedupes() {
+        let contract = RequiredOutputContract {
+            block_key: "  thoughts  ".into(),
+            required_keys: vec![
+                "thoughts".into(),
+                " next_action ".into(),
+                "next_action".into(),
+            ],
+        };
+        // block_key trimmed; duplicate `thoughts` and repeated `next_action`
+        // collapse to a single occurrence each, order-preserving.
+        assert_eq!(contract.all_keys(), vec!["thoughts", "next_action"]);
+    }
+
+    #[test]
+    fn repair_instruction_names_every_required_key() {
+        let contract = thoughts_contract();
+        let instruction = repair_instruction(&contract);
+        assert!(instruction.contains("thoughts"));
+        assert!(instruction.contains("next_action"));
+    }
+}

src/openhuman/agent/harness/session/turn/core.rs

@@ -332,6 +332,23 @@ fn tool_records_from_conversation(
     records
 }
 
+/// Rewrite the **trailing** assistant `Chat` message in `history` to `text`,
+/// keeping the persisted transcript and the next turn's KV-cache prefix
+/// consistent with a repaired required-output reply (issue #4117). Only the last
+/// row is touched — when the tail is not an assistant `Chat` (defensive; a clean
+/// finish, a cap checkpoint, and the #4093 close all end on one) a fresh
+/// assistant message is appended rather than mutating an older entry.
+fn replace_last_assistant_reply(history: &mut Vec<ConversationMessage>, text: &str) {
+    match history.last_mut() {
+        Some(ConversationMessage::Chat(chat)) if chat.role == "assistant" => {
+            chat.content = text.to_string();
+        }
+        _ => history.push(ConversationMessage::Chat(ChatMessage::assistant(
+            text.to_string(),
+        ))),
+    }
+}
+
 fn render_agent_context_status_note(sources: &[harness::AgentContextPreparedSource]) -> String {
     let sources = if sources.is_empty() {
         "the OpenHuman harness".to_string()
@@ -1373,6 +1390,42 @@ impl Agent {
         } else {
             outcome.text.clone()
         };
+
+        // Enforce the required structured-output contract (issue #4117) on the
+        // accepted reply — for ALL of the branches above (normal finish, cap
+        // checkpoint, #4093 synthesized close), since each delivers a reply
+        // downstream parsing depends on. When this agent must emit a JSON block
+        // every turn and the reply omitted it, validate-and-repair before the
+        // turn is accepted, reconciling with streaming (append-only when a live
+        // stream is attached, replace otherwise — see `enforce_required_output`).
+        // The trailing assistant message is rewritten to match, and the repair
+        // call's usage is folded into the turn accounting. `required_output`
+        // defaults to `None`, so existing agents are entirely unaffected.
+        let reply = if let Some(contract) = self.config.required_output.clone() {
+            match self
+                .enforce_required_output(
+                    &reply,
+                    &contract,
+                    effective_model,
+                    outcome.model_calls as u32 + 1,
+                )
+                .await
+            {
+                Some((repaired, repair_usage)) => {
+                    if let Some(u) = repair_usage {
+                        input_tokens += u.input_tokens;
+                        output_tokens += u.output_tokens;
+                        cached_input_tokens += u.cached_input_tokens;
+                        charged_amount_usd += u.charged_amount_usd;
+                    }
+                    replace_last_assistant_reply(&mut self.history, &repaired);
+                    repaired
+                }
+                None => reply,
+            }
+        } else {
+            reply
+        };
         self.trim_history();
 
         // Fold this turn's sub-agent spend into the cumulative meters and capture

src/openhuman/agent/harness/session/turn/session_io.rs

@@ -236,6 +236,243 @@ impl Agent {
         (checkpoint, usage)
     }
 
+    /// Enforce this agent's required structured-output contract on the turn's
+    /// final `reply` (issue #4117), reconciling with streaming.
+    ///
+    /// When the contract is active and `reply` already carries a well-formed
+    /// leading block, returns `None` (the caller keeps `reply` unchanged). When
+    /// the block is omitted/invalid, the turn is **repaired** so downstream
+    /// parsing/routing always receives one, and the repaired text is returned as
+    /// `Some((repaired_text, usage))` so the caller can fold the extra call's
+    /// usage into the turn accounting and rewrite the trailing assistant message.
+    ///
+    /// ## Streaming reconciliation (the #4387 / sanil-23 blocker)
+    ///
+    /// The original reply is streamed to the client as `TextDelta`s *before* this
+    /// runs (via the harness event bridge, keyed on `on_progress`). #4387
+    /// repaired with a `stream: None` re-prompt whose result then *replaced* the
+    /// already-streamed reply — so the client watched one answer stream in and
+    /// silently got a different one back. This implementation makes the repair
+    /// **append-only**, so the returned/persisted reply is always exactly the
+    /// concatenation of deltas the client saw — the live preview is a *prefix* of
+    /// the final message, never contradicted:
+    ///
+    /// * **Streamed case** (`on_progress` attached — interactive/user-visible):
+    ///   the corrective re-prompt runs *silently* (its raw output is never
+    ///   streamed, so a malformed attempt is never shown), then the chosen
+    ///   correction — the model's block if the concatenation validates, else a
+    ///   deterministic [`synthesize_block`] — is streamed as a continuation and
+    ///   appended after the original prose. Visible text == returned text, and
+    ///   the appended block is the first JSON value (prose isn't JSON), so the
+    ///   leading-position rule holds for the dominant omitted-block case.
+    /// * **Non-streamed case** (`on_progress` absent — background/cron/routing,
+    ///   the "non-user-visible" scope sanil-23 offered as the alternative): no
+    ///   client saw anything, so there is nothing to stay consistent with. The
+    ///   strict #4387 **replace** design applies — recover via re-prompt, else
+    ///   prepend a synthesized block to the prose — guaranteeing strict leading
+    ///   position.
+    ///
+    /// `iteration_for_stream` labels the streamed continuation so the UI groups
+    /// it with the turn's other deltas.
+    ///
+    /// [`synthesize_block`]: harness::required_output::synthesize_block
+    pub(in super::super) async fn enforce_required_output(
+        &self,
+        reply: &str,
+        contract: &crate::openhuman::config::RequiredOutputContract,
+        effective_model: &str,
+        iteration_for_stream: u32,
+    ) -> Option<(String, Option<UsageInfo>)> {
+        use harness::required_output as ro;
+
+        if ro::output_satisfies_contract(reply, contract) {
+            return None;
+        }
+        log::warn!(
+            "[agent_loop] required output block `{}` omitted from turn reply — repairing (streamed={})",
+            contract.block_key,
+            self.on_progress.is_some(),
+        );
+
+        // Corrective re-prompt (native tools disabled), seeded with the current
+        // history — which already carries the omitting assistant reply, so the
+        // model sees exactly what it left out. Run silently: we validate the
+        // result before deciding whether/what to show, so a malformed attempt is
+        // never streamed to the client.
+        let mut base = self.tool_dispatcher.to_provider_messages(&self.history);
+        base.push(ChatMessage::user(ro::repair_instruction(contract)));
+        let (repair_text, usage) = self
+            .reprompt_for_required_block(&base, effective_model)
+            .await;
+        let repair_text = repair_text.trim().to_string();
+
+        // Non-streamed (replace) path: nothing was shown, so the repaired reply
+        // can stand alone with a strictly-leading block.
+        if self.on_progress.is_none() {
+            if !repair_text.is_empty() && ro::output_satisfies_contract(&repair_text, contract) {
+                log::info!(
+                    "[agent_loop] required output block `{}` recovered via re-prompt (replace)",
+                    contract.block_key
+                );
+                return Some((repair_text, usage));
+            }
+            log::warn!(
+                "[agent_loop] required output block `{}` still missing after re-prompt — prepending a synthesized block",
+                contract.block_key
+            );
+            let synthesized = format!("{}\n\n{}", ro::synthesize_block(contract), reply);
+            return Some((synthesized, usage));
+        }
+
+        // Streamed (append) path: the original prose is already on the client, so
+        // never replace it — append a streamed correction and return the exact
+        // concatenation the client sees. Append ONLY the required block, not the
+        // whole re-prompt reply: `repair_instruction` asks the model to re-emit the
+        // block *and continue with its answer*, so appending the full reply would
+        // duplicate the already-streamed answer after the block (#4900). Prefer the
+        // model's own recovered block; fall back to a synthesized one otherwise.
+        let correction = match ro::find_required_block(&repair_text, contract) {
+            Some(block) => {
+                log::info!(
+                    "[agent_loop] required output block `{}` recovered via re-prompt (append)",
+                    contract.block_key
+                );
+                serde_json::to_string(&block).unwrap_or_else(|_| ro::synthesize_block(contract))
+            }
+            None => {
+                log::warn!(
+                    "[agent_loop] required output block `{}` still missing after re-prompt — appending a synthesized block",
+                    contract.block_key
+                );
+                ro::synthesize_block(contract)
+            }
+        };
+        // Stream only the correction as a continuation so the live preview stays
+        // a prefix of the final message (visible == returned).
+        let continuation = format!("\n\n{correction}");
+        self.stream_text_continuation(&continuation, iteration_for_stream)
+            .await;
+        let repaired = format!("{reply}{continuation}");
+        if !ro::output_satisfies_contract(&repaired, contract) {
+            // The only way to reach here is a reply that streamed a *non-conforming
+            // JSON object first*: append-only can't restore strict leading
+            // position without contradicting what the user already saw, so we
+            // accept the trailing valid block and keep stream consistency (the
+            // higher-priority invariant). Downstream still finds a conforming
+            // block via `extract_json_values`; only strict ordering is relaxed,
+            // and only for this pathological already-streamed case.
+            log::warn!(
+                "[agent_loop] required output block `{}` appended but not in leading position (streamed reply led with JSON) — accepting for stream consistency",
+                contract.block_key
+            );
+        }
+        Some((repaired, usage))
+    }
+
+    /// Ask the provider once for a reply that includes the required
+    /// structured-output block, with native tools **disabled** and **without**
+    /// forwarding any delta to the progress sink. Returns the parsed prose paired
+    /// with the call's usage (empty text + `None` usage when the call fails or
+    /// yields only tool-call markup).
+    ///
+    /// Unlike [`summarize_turn_wrapup`](Self::summarize_turn_wrapup) this is
+    /// deliberately silent: `enforce_required_output` validates the result before
+    /// deciding what (if anything) to stream, so a malformed repair attempt is
+    /// never shown to the client.
+    async fn reprompt_for_required_block(
+        &self,
+        base_messages: &[ChatMessage],
+        effective_model: &str,
+    ) -> (String, Option<UsageInfo>) {
+        let chat_model = match self
+            .turn_model_source
+            .build_summarizer(effective_model, self.temperature)
+        {
+            Ok(model) => model,
+            Err(error) => {
+                tracing::error!(
+                    error = %error,
+                    model = effective_model,
+                    "[agent::session] failed to build required-output re-prompt model"
+                );
+                return (String::new(), None);
+            }
+        };
+        let request = ModelRequest::new(
+            base_messages
+                .iter()
+                .map(crate::openhuman::tinyagents::chat_message_to_message)
+                .collect(),
+        )
+        .with_model(effective_model)
+        .with_temperature(self.temperature)
+        .with_max_tokens(AGENT_TURN_MAX_OUTPUT_TOKENS);
+        let mut stream = match chat_model.stream(&(), request).await {
+            Ok(stream) => stream,
+            Err(error) => {
+                tracing::warn!(
+                    error = %error,
+                    model = effective_model,
+                    "[agent::session] required-output re-prompt stream failed to start"
+                );
+                return (String::new(), None);
+            }
+        };
+
+        let mut streamed_text = String::new();
+        let mut completed = None;
+        while let Some(item) = stream.next().await {
+            match item {
+                // Buffer only — deliberately NOT forwarded to `on_progress`.
+                ModelStreamItem::MessageDelta(delta) if !delta.text.is_empty() => {
+                    streamed_text.push_str(&delta.text);
+                }
+                ModelStreamItem::Completed(response) => completed = Some(response),
+                ModelStreamItem::Failed(error) => {
+                    tracing::warn!(%error, "[agent::session] required-output re-prompt stream failed");
+                    return (String::new(), None);
+                }
+                ModelStreamItem::ProviderFailed(error) => {
+                    tracing::warn!(error = %error.message, "[agent::session] required-output re-prompt provider failed");
+                    return (String::new(), None);
+                }
+                _ => {}
+            }
+        }
+        let Some(response) = completed else {
+            tracing::warn!("[agent::session] required-output re-prompt ended without completion");
+            return (String::new(), None);
+        };
+        let usage = crate::openhuman::tinyagents::model::usage_info_from_response(&response);
+        let text = response.text();
+        let out = if !text.trim().is_empty() {
+            text
+        } else if response.tool_calls().is_empty() {
+            streamed_text
+        } else {
+            // Only tool-call markup was present — no usable prose.
+            String::new()
+        };
+        (out, usage)
+    }
+
+    /// Emit `text` to the progress sink as a `TextDelta` continuation so a
+    /// repaired required-output block appears in the UI appended after the
+    /// already-streamed reply (issue #4117). No-op when no sink is attached.
+    async fn stream_text_continuation(&self, text: &str, iteration: u32) {
+        if text.is_empty() {
+            return;
+        }
+        if let Some(sink) = &self.on_progress {
+            let _ = sink
+                .send(AgentProgress::TextDelta {
+                    delta: text.to_string(),
+                    iteration,
+                })
+                .await;
+        }
+    }
+
     /// Persist the exact provider messages as a session transcript.
     ///
     /// Writes JSONL as source of truth and re-renders the companion `.md`

src/openhuman/agent/harness/session/turn_tests.rs

@@ -903,6 +903,528 @@ async fn turn_uses_cached_transcript_prefix_on_first_iteration() {
     );
 }
 
+/// Issue #4117 — a reply that already carries a well-formed leading block is
+/// accepted verbatim: no corrective re-prompt, so exactly one provider call.
+#[tokio::test]
+async fn turn_accepts_valid_required_output_without_repair() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![Ok(ChatResponse {
+            text: Some("{\"thoughts\": \"planning\", \"next_action\": \"answer\"}".into()),
+            tool_calls: vec![],
+            usage: None,
+            reasoning_content: None,
+        })]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract {
+            block_key: "thoughts".into(),
+            required_keys: vec!["next_action".into()],
+        }),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+
+    assert!(
+        response.contains("thoughts") && response.contains("next_action"),
+        "a valid reply must be accepted verbatim, got: {response}"
+    );
+    // No corrective re-prompt fired — a single provider call.
+    assert_eq!(provider_impl.requests.lock().await.len(), 1);
+}
+
+/// Issue #4117 — with no live progress sink (background/routing agent), when the
+/// model emits prose without the mandated JSON block the turn engine re-prompts
+/// and the recovered block-bearing reply *replaces* the omitting one.
+#[tokio::test]
+async fn turn_repairs_missing_required_output_via_reprompt() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            // Turn 1 final reply: prose only, no `thoughts` block.
+            Ok(ChatResponse {
+                text: Some("Sure, I'll handle that.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Corrective re-prompt: the model now emits a valid block.
+            Ok(ChatResponse {
+                text: Some(
+                    "{\"thoughts\": \"planning the work\", \"next_action\": \"answer\"}".into(),
+                ),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract {
+            block_key: "thoughts".into(),
+            required_keys: vec!["next_action".into()],
+        }),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+
+    // The returned reply carries the recovered block.
+    assert!(
+        response.contains("thoughts") && response.contains("next_action"),
+        "repaired reply must contain the required block, got: {response}"
+    );
+    // The omitting prose reply was re-prompted (2 provider calls total).
+    assert_eq!(provider_impl.requests.lock().await.len(), 2);
+    // History's trailing assistant message was rewritten to match.
+    assert!(agent.history.iter().rev().any(|message| matches!(
+        message,
+        ConversationMessage::Chat(chat)
+            if chat.role == "assistant" && chat.content.contains("next_action")
+    )));
+}
+
+/// Issue #4117 — when the corrective re-prompt *also* omits the block (and no
+/// live sink is attached), the turn engine synthesizes a minimal valid block and
+/// prepends it to the original prose so the accepted turn leads with a
+/// well-formed block and never loses the model's answer.
+#[tokio::test]
+async fn turn_synthesizes_required_output_when_reprompt_also_omits() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            // Turn 1 final reply: prose only.
+            Ok(ChatResponse {
+                text: Some("Working on it.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Re-prompt: still no block.
+            Ok(ChatResponse {
+                text: Some("Still just prose, sorry.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract::new(
+            "thoughts",
+        )),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+
+    // The synthesized block is prepended to the ORIGINAL turn reply
+    // ("Working on it."), not the failed corrective re-prompt — so the leading
+    // JSON object carries the block and the original prose is preserved.
+    let first_block = crate::openhuman::agent::harness::parse::extract_json_values(&response)
+        .into_iter()
+        .next();
+    assert!(
+        first_block
+            .as_ref()
+            .is_some_and(|v| v.get("thoughts").is_some()),
+        "synthesized reply must lead with a `thoughts` block, got: {response}"
+    );
+    assert!(response.contains("Working on it."));
+    assert_eq!(provider_impl.requests.lock().await.len(), 2);
+}
+
+/// Issue #4117 (the #4387 / sanil-23 streaming blocker) — when a live progress
+/// sink is attached (the reply was streamed to the client) and the block is
+/// omitted, the repair is **append-only**: the original prose is preserved, the
+/// recovered block is *appended* after it (never replacing what the user saw),
+/// and the appended block is streamed as a `TextDelta` continuation so the
+/// returned reply is exactly the concatenation the client watched.
+#[tokio::test]
+async fn turn_appends_required_output_block_when_streamed_to_preserve_consistency() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            // Turn 1 final reply: prose only, streamed to the client.
+            Ok(ChatResponse {
+                text: Some("Sure, I'll handle that.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Corrective re-prompt: the model now emits a valid block.
+            Ok(ChatResponse {
+                text: Some(
+                    "{\"thoughts\": \"planning the work\", \"next_action\": \"answer\"}".into(),
+                ),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract {
+            block_key: "thoughts".into(),
+            required_keys: vec!["next_action".into()],
+        }),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    // Attach a live progress sink and drain it concurrently so the streamed
+    // deltas are captured without the bounded channel ever back-pressuring the
+    // turn.
+    let (tx, mut rx) = tokio::sync::mpsc::channel(64);
+    let text_deltas = Arc::new(AsyncMutex::new(Vec::<String>::new()));
+    let text_deltas_drain = text_deltas.clone();
+    let drain = tokio::spawn(async move {
+        while let Some(progress) = rx.recv().await {
+            if let crate::openhuman::agent::progress::AgentProgress::TextDelta { delta, .. } =
+                progress
+            {
+                text_deltas_drain.lock().await.push(delta);
+            }
+        }
+    });
+    agent.set_on_progress(Some(tx));
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+
+    // Drop the sink so the drain task observes channel close and finishes.
+    agent.set_on_progress(None);
+    // The continuation delta is sent (awaited) during the turn, so it is already
+    // drained; guard the join with a timeout so a leaked sender clone can never
+    // hang the test.
+    let _ = timeout(Duration::from_secs(5), drain).await;
+
+    // Append-only: the original prose is preserved AND precedes the block — the
+    // streamed preview is a prefix of the final reply, never contradicted.
+    assert!(
+        response.contains("Sure, I'll handle that."),
+        "original streamed prose must be preserved, not replaced: {response}"
+    );
+    assert!(
+        response.contains("next_action"),
+        "repaired reply must contain the required block: {response}"
+    );
+    let prose_at = response
+        .find("Sure, I'll handle that.")
+        .expect("prose present");
+    let block_at = response.find("next_action").expect("block present");
+    assert!(
+        prose_at < block_at,
+        "the block must be appended AFTER the already-streamed prose: {response}"
+    );
+
+    // The appended correction was streamed as a `TextDelta` continuation.
+    let deltas = text_deltas.lock().await;
+    assert!(
+        deltas.iter().any(|d| d.contains("next_action")),
+        "the appended block must be streamed as a TextDelta continuation, got: {deltas:?}"
+    );
+    assert_eq!(provider_impl.requests.lock().await.len(), 2);
+}
+
+/// Issue #4117 / #4900 — streamed path where the corrective re-prompt obeys
+/// `repair_instruction` literally: it leads with the block **and continues with
+/// the answer** (as the prompt asks). Because the original prose is already on
+/// the client, only the block may be appended — appending the whole re-prompt
+/// reply would duplicate the answer. Guards against that regression.
+#[tokio::test]
+async fn turn_appends_only_block_not_restated_answer_when_streamed() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            // Turn 1 final reply: prose only, streamed to the client.
+            Ok(ChatResponse {
+                text: Some("Paris is the capital of France.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Corrective re-prompt: block first, THEN the restated answer — exactly
+            // what `repair_instruction` ("…then continue with your answer") elicits.
+            Ok(ChatResponse {
+                text: Some(
+                    "{\"thoughts\": \"restating\", \"next_action\": \"answer\"}\n\n\
+                     Paris is the capital of France."
+                        .into(),
+                ),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract {
+            block_key: "thoughts".into(),
+            required_keys: vec!["next_action".into()],
+        }),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let (tx, mut rx) = tokio::sync::mpsc::channel(64);
+    let text_deltas = Arc::new(AsyncMutex::new(Vec::<String>::new()));
+    let text_deltas_drain = text_deltas.clone();
+    let drain = tokio::spawn(async move {
+        while let Some(progress) = rx.recv().await {
+            if let crate::openhuman::agent::progress::AgentProgress::TextDelta { delta, .. } =
+                progress
+            {
+                text_deltas_drain.lock().await.push(delta);
+            }
+        }
+    });
+    agent.set_on_progress(Some(tx));
+
+    let response = agent
+        .turn("what is the capital of France?")
+        .await
+        .expect("turn should succeed");
+
+    agent.set_on_progress(None);
+    let _ = timeout(Duration::from_secs(5), drain).await;
+
+    // The block is appended and the original prose preserved …
+    assert!(
+        response.contains("next_action"),
+        "repaired reply must contain the required block: {response}"
+    );
+    // … but the answer must appear exactly ONCE — the restated answer from the
+    // re-prompt reply must not be appended after the block.
+    assert_eq!(
+        response.matches("Paris is the capital of France.").count(),
+        1,
+        "the answer must not be duplicated by appending the re-prompt's restated prose: {response}"
+    );
+    // Streamed continuation carried the block, not a second copy of the answer.
+    let deltas = text_deltas.lock().await;
+    let streamed_continuation: String = deltas.concat();
+    assert!(
+        streamed_continuation.contains("next_action"),
+        "the appended block must be streamed as a continuation, got: {deltas:?}"
+    );
+    assert!(
+        !streamed_continuation.contains("Paris is the capital of France."),
+        "the streamed continuation must not restream the answer, got: {deltas:?}"
+    );
+}
+
+/// Issue #4117 — streamed path, but the corrective re-prompt *also* omits the
+/// block. A synthesized block is appended (never replacing the streamed prose)
+/// and streamed as a continuation, so the accepted turn still leads with the
+/// original prose and carries a well-formed block.
+#[tokio::test]
+async fn turn_appends_synthesized_block_when_streamed_reprompt_also_omits() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            Ok(ChatResponse {
+                text: Some("Working on it.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Re-prompt: still no block.
+            Ok(ChatResponse {
+                text: Some("Still just prose, sorry.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract::new(
+            "thoughts",
+        )),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let (tx, mut rx) = tokio::sync::mpsc::channel(64);
+    let text_deltas = Arc::new(AsyncMutex::new(Vec::<String>::new()));
+    let text_deltas_drain = text_deltas.clone();
+    let drain = tokio::spawn(async move {
+        while let Some(progress) = rx.recv().await {
+            if let crate::openhuman::agent::progress::AgentProgress::TextDelta { delta, .. } =
+                progress
+            {
+                text_deltas_drain.lock().await.push(delta);
+            }
+        }
+    });
+    agent.set_on_progress(Some(tx));
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+    agent.set_on_progress(None);
+    // The continuation delta is sent (awaited) during the turn, so it is already
+    // drained; guard the join with a timeout so a leaked sender clone can never
+    // hang the test.
+    let _ = timeout(Duration::from_secs(5), drain).await;
+
+    // Original prose preserved and leads; a synthesized `thoughts` block follows.
+    assert!(response.contains("Working on it."));
+    let prose_at = response.find("Working on it.").expect("prose present");
+    let block_at = response.find("thoughts").expect("synth block present");
+    assert!(
+        prose_at < block_at,
+        "synthesized block must be appended after the streamed prose: {response}"
+    );
+    let deltas = text_deltas.lock().await;
+    assert!(
+        deltas.iter().any(|d| d.contains("thoughts")),
+        "the synthesized block must be streamed as a continuation, got: {deltas:?}"
+    );
+    assert_eq!(provider_impl.requests.lock().await.len(), 2);
+}
+
+/// Issue #4117 — when the corrective re-prompt call itself fails, enforcement
+/// still guarantees a well-formed block: the deterministic synthesized fallback
+/// is used so the turn is never left without one (no live sink → replace path).
+#[tokio::test]
+async fn turn_synthesizes_required_output_when_reprompt_call_fails() {
+    let provider_impl = Arc::new(SequenceProvider {
+        responses: AsyncMutex::new(vec![
+            Ok(ChatResponse {
+                text: Some("Working on it.".into()),
+                tool_calls: vec![],
+                usage: None,
+                reasoning_content: None,
+            }),
+            // Re-prompt call errors out.
+            Err(anyhow::anyhow!("provider boom")),
+        ]),
+        requests: AsyncMutex::new(Vec::new()),
+    });
+    let provider: Arc<dyn Provider> = provider_impl.clone();
+
+    let config = crate::openhuman::config::AgentConfig {
+        max_tool_iterations: 3,
+        max_history_messages: 10,
+        required_output: Some(crate::openhuman::config::RequiredOutputContract::new(
+            "thoughts",
+        )),
+        ..crate::openhuman::config::AgentConfig::default()
+    };
+
+    let mut agent = make_agent_with_builder(
+        provider,
+        vec![],
+        Box::new(FixedMemoryLoader {
+            context: String::new(),
+        }),
+        vec![],
+        config,
+        crate::openhuman::config::ContextConfig::default(),
+    );
+
+    let response = agent.turn("hello").await.expect("turn should succeed");
+
+    // Deterministic fallback: a synthesized block leads, original prose kept.
+    let first_block = crate::openhuman::agent::harness::parse::extract_json_values(&response)
+        .into_iter()
+        .next();
+    assert!(
+        first_block
+            .as_ref()
+            .is_some_and(|v| v.get("thoughts").is_some()),
+        "failed re-prompt must still yield a synthesized leading block, got: {response}"
+    );
+    assert!(response.contains("Working on it."));
+}
+
 #[tokio::test]
 async fn turn_emits_checkpoint_when_max_tool_iterations_are_exceeded() {
     // First response forces a tool call (consuming the single allowed

src/openhuman/config/mod.rs

@@ -55,6 +55,9 @@ pub use schema::{
     MODEL_SUMMARIZATION_V1, MODEL_VISION_V1, SEARCH_ENGINE_BRAVE, SEARCH_ENGINE_DISABLED,
     SEARCH_ENGINE_MANAGED, SEARCH_ENGINE_PARALLEL, SEARCH_ENGINE_QUERIT,
 };
+// Kept as a separate re-export (issue #4117) so the large alphabetized group
+// above stays byte-identical and rustfmt-stable.
+pub use schema::RequiredOutputContract;
 pub use schemas::{
     all_controller_schemas as all_config_controller_schemas,
     all_registered_controllers as all_config_registered_controllers,

src/openhuman/config/schema/agent.rs

@@ -159,6 +159,71 @@ fn default_max_depth() -> u32 {
     3
 }
 
+/// A per-agent contract requiring a structured JSON block in every turn's final
+/// reply (issue #4117).
+///
+/// Some agents are consumed by downstream parsing/routing that expects a
+/// mandated JSON block — e.g. a `thoughts` block like
+/// `{"thoughts": "…", "next_action": "…"}` — on **every** turn. Models
+/// frequently omit it, leaving those consumers with nothing. When this contract
+/// is set on [`AgentConfig::required_output`], the turn engine validates the
+/// reply and repairs an omitted block before the turn is accepted (see
+/// `crate::openhuman::agent::harness::required_output`), so consumers always get
+/// a well-formed block.
+#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
+#[serde(default)]
+pub struct RequiredOutputContract {
+    /// The JSON key that identifies the required block (e.g. `"thoughts"`). The
+    /// contract is satisfied when the reply contains a JSON object carrying this
+    /// key — plus every key in [`required_keys`](Self::required_keys) — with a
+    /// non-null value. A blank `block_key` makes the contract inert (enforcement
+    /// is skipped), so it is a safe no-op default.
+    pub block_key: String,
+    /// Additional sibling keys that must also be present and non-null in the
+    /// same block (e.g. `["next_action"]`). The `block_key` is always required
+    /// and need not be repeated here.
+    pub required_keys: Vec<String>,
+}
+
+impl RequiredOutputContract {
+    /// Construct a contract from a block key with no extra required siblings.
+    pub fn new(block_key: impl Into<String>) -> Self {
+        Self {
+            block_key: block_key.into(),
+            required_keys: Vec::new(),
+        }
+    }
+
+    /// Every key that must be present — the block key followed by any declared
+    /// siblings — order-preserving, trimmed, and de-duplicated. Empty when the
+    /// contract carries no non-blank keys, in which case it is inert and the
+    /// turn engine skips enforcement.
+    pub fn all_keys(&self) -> Vec<String> {
+        // The block key is the contract's defining key — a blank one makes the
+        // whole contract inert, even if `required_keys` lists siblings, so the
+        // feature never accepts or synthesizes a block missing that key.
+        let block_key = self.block_key.trim();
+        if block_key.is_empty() {
+            return Vec::new();
+        }
+
+        let mut keys: Vec<String> = vec![block_key.to_string()];
+        for key in &self.required_keys {
+            let trimmed = key.trim();
+            if !trimmed.is_empty() && !keys.iter().any(|k| k == trimmed) {
+                keys.push(trimmed.to_string());
+            }
+        }
+        keys
+    }
+
+    /// Whether this contract actually constrains output. A contract with no
+    /// non-blank keys is inert and enforcement is skipped.
+    pub fn is_active(&self) -> bool {
+        !self.all_keys().is_empty()
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
 #[serde(default)]
 pub struct AgentConfig {
@@ -286,6 +351,14 @@ pub struct AgentConfig {
     /// [`crate::openhuman::session_import::live::shadow_reads_enabled`].
     #[serde(default = "default_session_shadow_reads")]
     pub session_shadow_reads: bool,
+
+    /// Optional required structured-output contract. When set to an active
+    /// contract, every turn's final reply must contain the mandated JSON block;
+    /// the turn engine validates and repairs an omitted block before the turn is
+    /// accepted (issue #4117). `None` (the default) disables enforcement so
+    /// existing agents are unaffected.
+    #[serde(default)]
+    pub required_output: Option<RequiredOutputContract>,
 }
 
 fn default_session_dual_write() -> bool {
@@ -427,6 +500,7 @@ impl Default for AgentConfig {
             agent_timeout_secs: default_agent_timeout_secs(),
             session_dual_write: default_session_dual_write(),
             session_shadow_reads: default_session_shadow_reads(),
+            required_output: None,
         }
     }
 }

src/openhuman/config/schema/mod.rs

@@ -50,7 +50,7 @@ mod update;
 pub use accessibility::ScreenIntelligenceConfig;
 pub use agent::{
     AgentConfig, DelegateAgentConfig, MemoryContextWindow, MemoryWindowLimits,
-    OrchestratorModelConfig, TeamModelConfig,
+    OrchestratorModelConfig, RequiredOutputContract, TeamModelConfig,
 };
 pub use autocomplete::AutocompleteConfig;
 pub use autonomy::AutonomyConfig;