fix(flows): guarantee a terminal state on every flows_build turn (no silent trail-off)

src/openhuman/flows/ops.rs

@@ -3711,26 +3711,25 @@ pub async fn flows_build(
         }
     };
 
-    // Emit the terminal chat event so a client viewing the copilot thread stops
-    // "processing" and finalizes the assistant bubble (the bridge streams only
-    // intermediate deltas). Success delivers `chat_done`; a run error delivers
-    // `chat_error`. The blocking return below is unchanged.
-    if let Some(target) = &stream {
-        let terminal: Result<String, String> = match &run_error {
-            None => Ok(assistant_text.clone()),
-            Some(err) => Err(err.clone()),
-        };
-        finalize_flow_stream(target, &terminal, &prompt).await;
-    }
-
     // Capture the proposal from the run's tool history (propose/revise/save all
     // emit the same self-describing `{ type: "workflow_proposal", … }` payload).
+    // Extracted BEFORE the stream is finalized below (issue: builder
+    // convergence): the trail-off backstop needs `proposal`/`capped` to decide
+    // whether to override `assistant_text`, and the streamed copilot-pane chat
+    // bubble must render the SAME (possibly-overridden) text as the RPC
+    // response — the frontend renders from the stream, not the return value,
+    // so patching only the latter would still leave an interactive user
+    // staring at the original silent/status-only text.
     let proposal = extract_workflow_proposal(agent.history());
 
     // A run that both errored AND produced no proposal is a hard failure; a run
     // that proposed before erroring still returns the proposal for review.
     if proposal.is_none() {
         if let Some(err) = &run_error {
+            if let Some(target) = &stream {
+                let terminal: Result<String, String> = Err(err.clone());
+                finalize_flow_stream(target, &terminal, &prompt).await;
+            }
             return Err(format!("workflow_builder produced no proposal: {err}"));
         }
     }
@@ -3748,11 +3747,51 @@ pub async fn flows_build(
     let hit_cap = agent.last_turn_hit_cap();
     let capped = hit_cap && proposal.is_none();
 
+    // Terminal-state guarantee (builder convergence fix): a turn can end
+    // "naturally" (no more tool calls, not capped, no run error) yet still
+    // produce neither a proposal nor a real question — the model ran out of
+    // steam mid-build and left a status dump ("Done so far: checked
+    // connections…") as its final reply. `prompt.md` tells the model to
+    // always end a building turn in a proposal or a question, but a prompt
+    // rule can be silently ignored; this is the fail-closed backend backstop
+    // that makes it a hard invariant regardless of model behavior — the user
+    // is NEVER left with silence or an unanswerable status note.
+    let trail_off = !capped && proposal.is_none() && run_error.is_none();
+    let assistant_text = if trail_off && !text_looks_like_question(&assistant_text) {
+        let fallback = build_trail_off_fallback(agent.history());
+        tracing::warn!(
+            target: "flows",
+            flow_id = req.flow_id.as_deref().unwrap_or("<none>"),
+            original_len = assistant_text.len(),
+            fallback_len = fallback.len(),
+            "[flows] flows_build: trail-off detected (no proposal, no cap, no question) — \
+             guaranteeing a fallback question instead of silence"
+        );
+        fallback
+    } else {
+        assistant_text
+    };
+
+    // Emit the terminal chat event so a client viewing the copilot thread stops
+    // "processing" and finalizes the assistant bubble (the bridge streams only
+    // intermediate deltas). Success delivers `chat_done`; a run error delivers
+    // `chat_error`. The blocking return below is unchanged. Uses the
+    // (possibly trail-off-overridden) `assistant_text` above.
+    if let Some(target) = &stream {
+        let terminal: Result<String, String> = match &run_error {
+            None => Ok(assistant_text.clone()),
+            Some(err) => Err(err.clone()),
+        };
+        finalize_flow_stream(target, &terminal, &prompt).await;
+    }
+
     tracing::info!(
         target: "flows",
+        flow_id = req.flow_id.as_deref().unwrap_or("<none>"),
         has_proposal = proposal.is_some(),
         hit_cap,
         capped,
+        trail_off,
         "[flows] flows_build: workflow_builder turn complete"
     );
     Ok(RpcOutcome::single_log(
@@ -3761,11 +3800,155 @@ pub async fn flows_build(
             "assistant_text": assistant_text,
             "error": run_error,
             "capped": capped,
+            "trail_off": trail_off,
         }),
         "workflow builder turn complete",
     ))
 }
 
+/// Heuristic: does `text` already end with a clear, answerable question?
+/// Conservative by design (issue: builder convergence) — a false negative (an
+/// actual question this misses) just wraps it in the trail-off fallback,
+/// which still includes the blocker context, so the safe failure mode is
+/// "over-wrap", never "under-detect and stay silent".
+fn text_looks_like_question(text: &str) -> bool {
+    let trimmed = text
+        .trim()
+        .trim_end_matches(|c: char| matches!(c, '"' | '\'' | ')' | ']' | '*' | '_' | '`' | '.'))
+        .trim_end();
+    if trimmed.is_empty() {
+        return false;
+    }
+    if trimmed.ends_with('?') {
+        return true;
+    }
+    // The question may not be the literal last character (trailing markdown
+    // like a closing code fence or list marker on its own line) — fall back
+    // to the last non-blank line. This does NOT catch a question followed by
+    // a further trailing sentence ("...channel?\n\nLet me know!") — that's
+    // an accepted false negative: the turn still ends in a real (if
+    // over-eagerly replaced) question, never in silence, which is the
+    // invariant this function exists to protect.
+    trimmed
+        .lines()
+        .filter(|line| !line.trim().is_empty())
+        .next_back()
+        .is_some_and(|last_line| last_line.trim_end().ends_with('?'))
+}
+
+/// Builder-authoring tools whose result body can explain a trail-off — the
+/// authoring belt `dry_run_workflow`/`validate_workflow`/`propose_workflow`/
+/// `revise_workflow`/`edit_workflow`/`save_workflow` all report either a hard
+/// gate rejection (`ToolResult::error`) or a self-reported broken-graph
+/// result (`"ok": false` in a successful body), so a plain-text read-only
+/// tool's output is never misattributed as the blocker.
+const TRAIL_OFF_BLOCKER_TOOLS: &[&str] = &[
+    "dry_run_workflow",
+    "validate_workflow",
+    "propose_workflow",
+    "revise_workflow",
+    "edit_workflow",
+    "save_workflow",
+];
+
+/// Synthesizes a guaranteed, user-facing fallback for a trail-off turn (no
+/// proposal, not capped, no run error, and the model's own text isn't a
+/// question). Scans the run's tool history for the last builder-tool result
+/// that looks like a blocker (a hard-gate rejection, or a `dry_run_workflow`/
+/// `validate_workflow` report with `"ok": false`) and asks the user about it;
+/// falls back to a generic "what should I focus on" question when no such
+/// blocker is found (the model may have simply stopped with nothing to point
+/// to).
+fn build_trail_off_fallback(
+    history: &[crate::openhuman::inference::provider::ConversationMessage],
+) -> String {
+    match last_builder_tool_blocker(history) {
+        Some(blocker) => format!(
+            "I wasn't able to finish building this workflow. Here's where I got stuck:\n\n{blocker}\n\n\
+             Could you tell me how you'd like me to resolve that, or share more detail about what's needed here?"
+        ),
+        None => "I wasn't able to finish building this workflow in this turn. Could you describe \
+                  what you'd like in more detail, or tell me which part to focus on?"
+            .to_string(),
+    }
+}
+
+/// Scans `history` in reverse for the last result from a
+/// [`TRAIL_OFF_BLOCKER_TOOLS`] call that reads as a failure — a plain-text
+/// error message (gate rejection), or a JSON body with `"ok": false` — and
+/// returns a truncated, human-readable description of it. Tool names are
+/// resolved by correlating each `ToolResults` entry's `tool_call_id` back to
+/// the `AssistantToolCalls` message that issued it, so this never
+/// misattributes an unrelated read-only tool's plain-text output as a
+/// blocker.
+fn last_builder_tool_blocker(
+    history: &[crate::openhuman::inference::provider::ConversationMessage],
+) -> Option<String> {
+    use crate::openhuman::inference::provider::ConversationMessage;
+
+    let mut call_names: std::collections::HashMap<String, String> =
+        std::collections::HashMap::new();
+    for message in history {
+        if let ConversationMessage::AssistantToolCalls { tool_calls, .. } = message {
+            for call in tool_calls {
+                call_names.insert(call.id.clone(), call.name.clone());
+            }
+        }
+    }
+
+    for message in history.iter().rev() {
+        let ConversationMessage::ToolResults(results) = message else {
+            continue;
+        };
+        for result in results.iter().rev() {
+            let Some(name) = call_names.get(&result.tool_call_id) else {
+                continue;
+            };
+            if !TRAIL_OFF_BLOCKER_TOOLS.contains(&name.as_str()) {
+                continue;
+            }
+            // This is the MOST RECENT authoring-belt tool result in the
+            // turn (results are scanned newest-first). Whatever it reads as
+            // is authoritative: a success/progress result here means any
+            // earlier failure from the same tool was already resolved
+            // within this turn, so we must stop at this result rather than
+            // keep walking backward and surfacing a stale, already-fixed
+            // blocker (see review discussion on this PR).
+            return describe_tool_result_blocker(&result.content)
+                .map(|desc| crate::openhuman::util::truncate_with_ellipsis(&desc, 500));
+        }
+    }
+    None
+}
+
+/// Reads one builder tool result's content as a failure description, or
+/// `None` when it reads as success/progress (a `workflow_proposal` payload,
+/// or an `"ok": true` report). The whole body is the description, never one
+/// hardcoded field, so this stays correct regardless of which fields a given
+/// tool uses to explain its failure.
+fn describe_tool_result_blocker(content: &str) -> Option<String> {
+    let trimmed = content.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+    if let Ok(value) = serde_json::from_str::<Value>(trimmed) {
+        if value.get("type").and_then(Value::as_str) == Some("workflow_proposal") {
+            return None; // Success: a proposal was emitted.
+        }
+        if let Some(ok) = value.get("ok").and_then(Value::as_bool) {
+            return if ok { None } else { Some(value.to_string()) };
+        }
+        // Some other structured payload with no `ok`/`type` marker this
+        // function recognises — not confidently a blocker, skip it.
+        return None;
+    }
+    // Non-JSON content: a hard-gate rejection (`ToolResult::error`) puts the
+    // plain error message straight into the content — since every builder
+    // tool's SUCCESS shape is JSON (a proposal or a `{ ok, ... }` report), a
+    // bare string here is, by elimination, an error message.
+    Some(trimmed.to_string())
+}
+
 /// Scans an agent run's conversation history for the workflow proposal a builder
 /// tool emitted. `propose_workflow` / `revise_workflow` / `save_workflow` all
 /// return a self-describing `{ "type": "workflow_proposal", … }` JSON string as