fix(workflow-session): preserve full-history unfinished-call recovery

crates/webcodex-workflow-session/src/continuation.rs

@@ -28,7 +28,7 @@
 use serde::Serialize;
 use serde_json::{json, Value};
 use sha2::{Digest, Sha256};
-use std::collections::{BTreeSet, HashSet};
+use std::collections::{BTreeSet, HashMap, HashSet};
 
 use crate::{
     canonical_tool_call_finished_events, closeout_work_projection, current_attempt_event_view,
@@ -540,6 +540,94 @@ fn empty_attempt() -> AttemptSummary {
     }
 }
 
+/// Whether any mutating tool call in `events` lacks exactly one matching
+/// finish after it. Callers pass the full retained history so an unfinished
+/// mutation older than a bounded summary window is still detected; the scan
+/// is a single reverse pass that allocates only per-identity finish counters.
+///
+/// A successful Job acceptance has its own matching terminal tool event;
+/// its subsequent running/recovering state belongs to the Job lifecycle,
+/// not to this unfinished-call check. Never correlate by wall time or args.
+pub(crate) fn has_unclosed_mutating_call<'a>(
+    events: impl DoubleEndedIterator<Item = &'a SessionEvent>,
+    session_id: &str,
+) -> bool {
+    // Exact call identity. `session_id` is fixed for the whole scan, so the
+    // key carries the remaining match dimensions. Project comes from the
+    // event's own resolved/plain project, never the Session default.
+    type CallIdentity<'a> = (
+        Option<&'a str>,
+        &'a str,
+        &'a str,
+        Option<&'a str>,
+        Option<&'a str>,
+    );
+    fn call_identity<'a>(event: &'a SessionEvent) -> Option<CallIdentity<'a>> {
+        let call_id = event.call_id.as_deref().filter(|id| !id.is_empty())?;
+        Some((
+            event
+                .resolved_project
+                .as_deref()
+                .or(event.project.as_deref()),
+            event.tool_name.as_str(),
+            call_id,
+            event.logical_invocation_id.as_deref(),
+            event.logical_invocation_role.as_deref(),
+        ))
+    }
+    // Suffix finish counts, built back-to-front so each start is closed only
+    // by finishes recorded after it: a duplicated call identity must not let a
+    // later start inherit an earlier invocation's finish. Duplicated or
+    // corrupted finishes are not authoritative confirmation — exactly one
+    // match closes a call.
+    let mut suffix_finishes: HashMap<CallIdentity<'_>, usize> = HashMap::new();
+    for event in events.rev() {
+        if event.kind == "tool_call_finished" {
+            if event.session_id != session_id {
+                continue;
+            }
+            if let Some(identity) = call_identity(event) {
+                *suffix_finishes.entry(identity).or_insert(0) += 1;
+            }
+            continue;
+        }
+        if event.kind != "tool_call_started"
+            || !(event.write_like || event.shell_like)
+            || event.session_id != session_id
+        {
+            continue;
+        }
+        let Some(call_id) = event.call_id.as_deref().filter(|id| !id.is_empty()) else {
+            // Legacy starts cannot be proven to have a matching finish.
+            return true;
+        };
+        // A Session can record implicitly scoped calls and requests for a
+        // different Project (including rejected requests). Neither case is
+        // proof that the recorded start finished. Compare the start's own
+        // Project identity with the finish, not with the Session default.
+        let identity: CallIdentity<'_> = (
+            event
+                .resolved_project
+                .as_deref()
+                .or(event.project.as_deref()),
+            event.tool_name.as_str(),
+            call_id,
+            event.logical_invocation_id.as_deref(),
+            event.logical_invocation_role.as_deref(),
+        );
+        let Some(unmatched_finishes) = suffix_finishes.get_mut(&identity) else {
+            return true;
+        };
+        if *unmatched_finishes != 1 {
+            return true;
+        }
+        // A finish is proof for only one start. Two starts sharing the same
+        // identity followed by one finish must not both be considered closed.
+        *unmatched_finishes = 0;
+    }
+    false
+}
+
 #[allow(clippy::too_many_arguments)]
 fn build_attempt_summary(
     attempt_events: &[SessionEvent],
@@ -588,6 +676,14 @@ fn build_attempt_summary(
     };
 
     // --- activity (meaningful tool calls only) ---
+    // An attempt boundary is not confirmation that a prior mutating request
+    // finished. Only exact retained call identity can resolve such a start.
+    // The store computed the flag over the full retained history (see
+    // `SessionSummary::unclosed_mutating_call`), so an unfinished mutation
+    // older than this bounded summary window still blocks the attempt.
+    let unconfirmed_mutating_call = session_summary.unclosed_mutating_call;
+    let incomplete_call_history =
+        session_summary.events_truncated || session_summary.retention_truncated;
     let canonical_finished = canonical_tool_call_finished_events(semantic_attempt_events);
     let meaningful: Vec<&SessionEvent> = canonical_finished
         .iter()
@@ -665,14 +761,23 @@ fn build_attempt_summary(
     let guidance_block = build_attempt_guidance(discussion);
 
     // --- outcome ---
-    let (outcome_status, reason_codes) = build_attempt_outcome(
+    let (mut outcome_status, mut reason_codes) = build_attempt_outcome(
         &meaningful,
         actionable_failed_tool_calls,
         unresolved_failures,
         &jobs_block,
         &guidance_block,
         &validation_block,
     );
+    if unconfirmed_mutating_call {
+        outcome_status = "blocked".to_string();
+        push_unique(&mut reason_codes, "unconfirmed_tool_execution");
+    }
+    // The bounded tail cannot prove that no older effect remains unresolved.
+    // This is uncertainty about evidence coverage, not permission to replay.
+    if incomplete_call_history {
+        push_unique(&mut reason_codes, "mutating_call_history_incomplete");
+    }
 
     let mut suggested_next_actions = build_suggested_next_actions(
         &outcome_status,
@@ -707,6 +812,17 @@ fn build_attempt_summary(
         suggested_next_actions.insert(0, action.to_string());
         suggested_next_actions.truncate(MAX_SUGGESTED_ACTIONS);
     }
+    if unconfirmed_mutating_call {
+        let action = crate::task_context::RECONCILE_ACTION.to_string();
+        suggested_next_actions.retain(|suggestion| suggestion != &action);
+        suggested_next_actions.insert(0, action);
+        suggested_next_actions.truncate(MAX_SUGGESTED_ACTIONS);
+    } else if incomplete_call_history {
+        let action = "inspect missing Session history before inferring prior tool outcomes";
+        suggested_next_actions.retain(|suggestion| suggestion != action);
+        suggested_next_actions.insert(0, action.to_string());
+        suggested_next_actions.truncate(MAX_SUGGESTED_ACTIONS);
+    }
 
     AttemptSummary {
         boundary: AttemptBoundary {