fix(engine,pet): close operation spans on cancel and keep shared frames valid

Review fixes for the engine owner contract: - Started/Completed pairing: an RAII OperationSpanGuard sends Completed{Cancelled} (try_send) when the turn loop drops an in-flight tool future on Esc or a parallel-batch cancel. - Span ids are process-unique (`{call_id}#{seq}`): gateways that elide tool-call ids fall back to `call_{block_index}`, which repeats and was swallowed by the reducer's completed-span dedupe. - Reducer: turn_complete with no turn id (/purge, edit rejection, mid-turn session switch) no longer records a turn outcome, which broke the Rust invariant `turn_outcome requires turn_id` and made every shared view reject frames and reconnect. Presence is idle either way. - live.rs: an owner activity the client cannot parse or that fails validation/binding now reads as none instead of rejecting the frame, so a long-lived `pet serve` owner from an older binary keeps the pet body rendering instead of looping on "Shared pet reconnecting". - Outcomes: only PermissionDenied maps to Denied; invalid input, missing field, path escape, not-available and timeout are Failed. - MCP activity is classified by the resolved server, not the model name. - event_msg: document that the wire OperationActivity pair is reserved (no runtime-thread emitter yet). Evidence (local): - scripts/dev-cargo.sh check -p codewhale-tui -p codewhale-protocol --locked --lib --tests: Finished, 0 warnings (scratch build dir; the shared volume was full). - scripts/dev-cargo.sh test -p codewhale-tui --locked --lib -- dispatch_reports_typed dropped_operation_span tools::activity pet_watch protocol_parity: 35 passed, 0 failed. - scripts/dev-cargo.sh check -p codewhale-protocol --locked --lib: ok. - pet (node 22.20.0): tsc --noEmit exit 0; build-pet --tui --apple ok; node --test tests/*.test.mjs: 70 passed, 0 failed. - cargo fmt --all, git diff --check: clean. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01R9rJEoMuUSRWznU6QjkE7h
2026-11-08 17:10:50 -10:00 · 2026-09-25 06:08:14 -07:00
1 parent 29e85615c1
commit b7cfd5ce9a
@@ -366,6 +366,10 @@ pub enum EventMsg {
    /// Trusted Engine-owned activity for an operation that passed dispatch
    /// and authority checks. No tool name, arguments, command, or result is
    /// included in the pet-facing activity contract.
+    ///
+    /// Reserved on the wire: `protocol_parity` maps the engine event, but no
+    /// runtime thread emits the pair yet, so Runtime API and GPUI clients do
+    /// not receive it. The shared pet (`pet_watch`) is the only consumer today.
    OperationActivityStarted {
        thread_id: ThreadId,
        session_id: SessionId,
@@ -20460,10 +20460,19 @@ async fn dispatch_reports_typed_operation_activity_without_names_or_arguments()
                    activity_kind: OwnerActivityKind::Reading,
                    outcome: OwnerOperationOutcome::Succeeded,
                },
-            ] if started == "call-1" && completed == "call-1"
+            ] if started == completed && started.starts_with("call-1#")
        ),
        "unexpected activity: {events:?}"
    );
+    // A repeated model call id (gateways that elide ids fall back to
+    // `call_{block_index}`) still gets a fresh span, so a consumer that
+    // deduplicates completed spans sees the second call.
+    let again = run("read", Some("call-1"), json!({"path": "private-note.txt"})).await;
+    let span_of = |events: &[Event]| match events.first() {
+        Some(Event::OperationActivityStarted { span_id, .. }) => span_id.clone(),
+        other => panic!("unexpected activity: {other:?}"),
+    };
+    assert_ne!(span_of(&events), span_of(&again));
    let wire = format!("{events:?}");
    assert!(!wire.contains("private-note"), "arguments leaked: {wire}");

@@ -20473,7 +20482,7 @@ async fn dispatch_reports_typed_operation_activity_without_names_or_arguments()
        matches!(
            events.last(),
            Some(Event::OperationActivityCompleted {
-                outcome: OwnerOperationOutcome::Failed | OwnerOperationOutcome::Denied,
+                outcome: OwnerOperationOutcome::Failed,
                ..
            })
        ),
@@ -20501,6 +20510,65 @@ async fn dispatch_reports_typed_operation_activity_without_names_or_arguments()
        .await
        .is_empty()
    );
+
+    // A call refused by the cancel gate never ran, so it reports nothing.
+    let (tx_event, mut rx_event) = mpsc::channel(16);
+    let cancelled = CancellationToken::new();
+    cancelled.cancel();
+    let refused = Engine::execute_tool_with_lock(
+        Arc::new(RwLock::new(())),
+        false,
+        false,
+        tx_event,
+        Some(cancelled),
+        "read".to_string(),
+        Some("call-5".to_string()),
+        json!({"path": "private-note.txt"}),
+        tmp.path().to_path_buf(),
+        Some(&registry),
+        None,
+        Some(context.clone()),
+    )
+    .await;
+    assert!(refused.is_err());
+    while let Ok(event) = rx_event.try_recv() {
+        assert!(
+            !matches!(
+                event,
+                Event::OperationActivityStarted { .. } | Event::OperationActivityCompleted { .. }
+            ),
+            "a refused call reported activity: {event:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn dropped_operation_span_completes_as_cancelled() {
+    use codewhale_protocol::engine_owner::{OwnerActivityKind, OwnerOperationOutcome};
+
+    // The turn loop drops an in-flight tool future on cancel; the span it
+    // opened must still close, or every host leaks an active operation.
+    let (tx_event, mut rx_event) = mpsc::channel(16);
+    let span = super::tool_execution::OperationSpanGuard::start(
+        tx_event,
+        "call-x",
+        OwnerActivityKind::Editing,
+    )
+    .await;
+    drop(span);
+    let started = match rx_event.try_recv() {
+        Ok(Event::OperationActivityStarted { span_id, .. }) => span_id,
+        other => panic!("unexpected event: {other:?}"),
+    };
+    match rx_event.try_recv() {
+        Ok(Event::OperationActivityCompleted {
+            span_id,
+            activity_kind: OwnerActivityKind::Editing,
+            outcome: OwnerOperationOutcome::Cancelled,
+        }) => assert_eq!(span_id, started),
+        other => panic!("unexpected event: {other:?}"),
+    }
+    assert!(rx_event.try_recv().is_err(), "exactly one Completed");
 }

 #[tokio::test]
@@ -24,6 +24,77 @@ fn inherited_interactive_shell_refusal(tool_name: &str, interactive: bool) -> Op
        .map(|message| ToolError::execution_failed(message.to_string()))
 }

+/// Pairs one `OperationActivityStarted` with exactly one
+/// `OperationActivityCompleted`.
+///
+/// The turn loop drops an in-flight tool future when the user cancels
+/// (`tokio::select!` on the cancel token, or `drop(tool_tasks)` for a parallel
+/// batch), so a Completed sent inline after the await would never be sent.
+/// Dropping an unfinished span sends `Completed { Cancelled }` with
+/// `try_send`: best effort, like the other guards here, because `Drop` cannot
+/// await a full channel.
+pub(super) struct OperationSpanGuard {
+    tx: mpsc::Sender<Event>,
+    span: Option<(String, codewhale_protocol::engine_owner::OwnerActivityKind)>,
+}
+
+impl OperationSpanGuard {
+    /// A process-unique span id. The model's tool-call id is not unique:
+    /// gateways that elide ids fall back to `call_{block_index}`, which
+    /// repeats every step, and a consumer deduplicates completed spans.
+    fn span_id(call_id: &str) -> String {
+        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
+        let seq = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        format!("{call_id}#{seq}")
+    }
+
+    pub(super) async fn start(
+        tx: mpsc::Sender<Event>,
+        call_id: &str,
+        activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
+    ) -> Self {
+        let span_id = Self::span_id(call_id);
+        // `Sender::send` is cancel safe: if this future is dropped the event
+        // was either sent or not, and the guard is armed only once it was.
+        let sent = tx
+            .send(Event::OperationActivityStarted {
+                span_id: span_id.clone(),
+                activity_kind,
+            })
+            .await
+            .is_ok();
+        Self {
+            tx,
+            span: sent.then_some((span_id, activity_kind)),
+        }
+    }
+
+    async fn complete(mut self, outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome) {
+        if let Some((span_id, activity_kind)) = self.span.take() {
+            let _ = self
+                .tx
+                .send(Event::OperationActivityCompleted {
+                    span_id,
+                    activity_kind,
+                    outcome,
+                })
+                .await;
+        }
+    }
+}
+
+impl Drop for OperationSpanGuard {
+    fn drop(&mut self) {
+        if let Some((span_id, activity_kind)) = self.span.take() {
+            let _ = self.tx.try_send(Event::OperationActivityCompleted {
+                span_id,
+                activity_kind,
+                outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome::Cancelled,
+            });
+        }
+    }
+}
+
 /// Emits delayed, best-effort liveness pulses for one running tool.
 ///
 /// Keep the ticker in its own task instead of embedding `tokio::time::Interval`
@@ -570,8 +641,8 @@ impl Engine {
                    .lock()
                    .await
                    .resolved_tool_servers()
-                    .contains_key(&tool_name)
-                    .then(|| crate::tools::activity::mcp_activity_kind(&tool_name)),
+                    .get(&tool_name)
+                    .map(|server| crate::tools::activity::mcp_activity_kind(server)),
                None => None,
            },
            Some(_)
@@ -590,14 +661,12 @@ impl Engine {
                    crate::tools::activity::registry_activity_kind(&tool_name, &tool_input)
                }),
        };
-        if let (Some(span_id), Some(activity_kind)) = (activity_call_id.as_ref(), activity_kind) {
-            let _ = tx_event
-                .send(Event::OperationActivityStarted {
-                    span_id: span_id.clone(),
-                    activity_kind,
-                })
-                .await;
-        }
+        let operation_span = match (activity_call_id.as_deref(), activity_kind) {
+            (Some(call_id), Some(activity_kind)) => {
+                Some(OperationSpanGuard::start(tx_event.clone(), call_id, activity_kind).await)
+            }
+            _ => None,
+        };

        let outcome: Result<RichToolResult, ToolError> = if McpPool::is_mcp_tool(&tool_name) {
            if let Some(pool) = mcp_pool {
@@ -651,17 +720,14 @@ impl Engine {
            )))
        };

-        if let (Some(span_id), Some(activity_kind)) = (activity_call_id, activity_kind) {
+        if let Some(operation_span) = operation_span {
            let cancelled = cancel_token
                .as_ref()
                .is_some_and(CancellationToken::is_cancelled);
-            let operation_outcome = crate::tools::activity::operation_outcome(&outcome, cancelled);
-            let _ = tx_event
-                .send(Event::OperationActivityCompleted {
-                    span_id,
-                    activity_kind,
-                    outcome: operation_outcome,
-                })
+            operation_span
+                .complete(crate::tools::activity::operation_outcome(
+                    &outcome, cancelled,
+                ))
                .await;
        }

@@ -55,11 +55,12 @@ pub(crate) fn registry_activity_kind(tool_name: &str, input: &Value) -> Option<O
    Some(kind_for_operation(canonical_action_alias(tool_name, input)))
 }

-/// Classify an MCP call whose server the pool resolved. Desktop-control
-/// servers report `Computer`; everything else is a generic `Tool`.
+/// Classify an MCP call by the server the pool resolved it to, not by the
+/// model-facing name. Desktop-control servers report `Computer`; everything
+/// else is a generic `Tool`.
 #[must_use]
-pub(crate) fn mcp_activity_kind(tool_name: &str) -> OwnerActivityKind {
-    if super::subagent::is_machine_control_tool(tool_name) {
+pub(crate) fn mcp_activity_kind(server: &str) -> OwnerActivityKind {
+    if super::subagent::is_machine_control_tool(&format!("mcp_{server}_tool")) {
        OwnerActivityKind::Computer
    } else {
        OwnerActivityKind::Tool
@@ -78,16 +79,17 @@ pub(crate) fn operation_outcome(
        _ if cancelled => OwnerOperationOutcome::Cancelled,
        Ok(_) => OwnerOperationOutcome::Failed,
        Err(ToolError::Cancelled { .. }) => OwnerOperationOutcome::Cancelled,
+        // Only a policy refusal is a denial. Bad input, an escaped path or a
+        // missing tool is the call failing, and the pet should show it.
+        Err(ToolError::PermissionDenied { .. }) => OwnerOperationOutcome::Denied,
        Err(
            ToolError::InvalidInput { .. }
            | ToolError::MissingField { .. }
            | ToolError::PathEscape { .. }
            | ToolError::NotAvailable { .. }
-            | ToolError::PermissionDenied { .. },
-        ) => OwnerOperationOutcome::Denied,
-        Err(ToolError::ExecutionFailed { .. } | ToolError::Timeout { .. }) => {
-            OwnerOperationOutcome::Failed
-        }
+            | ToolError::ExecutionFailed { .. }
+            | ToolError::Timeout { .. },
+        ) => OwnerOperationOutcome::Failed,
    }
 }

@@ -147,15 +149,10 @@ mod tests {

    #[test]
    fn shipped_desktop_control_servers_are_computer_activity() {
-        for name in [
-            "mcp_codewhale-cu_click",
-            "mcp_codewhale--cu_screenshot",
-            "mcp_computer-use_type",
-            "mcp_local-computer_use_key",
-        ] {
-            assert_eq!(mcp_activity_kind(name), Kind::Computer, "{name}");
+        for server in ["codewhale-cu", "computer-use", "local-computer_use"] {
+            assert_eq!(mcp_activity_kind(server), Kind::Computer, "{server}");
        }
-        assert_eq!(mcp_activity_kind("mcp_github_get_issue"), Kind::Tool);
+        assert_eq!(mcp_activity_kind("github"), Kind::Tool);
    }

    #[test]
@@ -180,14 +177,21 @@ mod tests {
            operation_outcome(&denied, false),
            OwnerOperationOutcome::Denied
        );
-        let timeout = Err(ToolError::execution_failed("x"));
        assert_eq!(
            operation_outcome(&Err(ToolError::cancelled("stop")), false),
            OwnerOperationOutcome::Cancelled
        );
-        assert_eq!(
-            operation_outcome(&timeout, false),
-            OwnerOperationOutcome::Failed
-        );
+        for failure in [
+            ToolError::Timeout { seconds: 5 },
+            ToolError::execution_failed("x"),
+            ToolError::invalid_input("bad"),
+            ToolError::missing_field("path"),
+            ToolError::not_available("gone"),
+        ] {
+            assert_eq!(
+                operation_outcome(&Err(failure), false),
+                OwnerOperationOutcome::Failed
+            );
+        }
    }
 }
@@ -51,10 +51,33 @@ pub struct Scene {
    pub still: Pose,
    #[serde(default)]
    pub appearance: super::appearance::Appearance,
-    #[serde(default)]
+    /// Owner activity is an overlay on the body, never a reason to drop the
+    /// frame. A long-lived `pet serve` owner can predate this binary (the
+    /// activity shape changed without a frame version bump), so an activity
+    /// this client cannot parse reads as none and the pet keeps rendering.
+    #[serde(default, deserialize_with = "lenient_activity")]
    pub activity: Option<Activity>,
 }
+fn lenient_activity<'de, D: serde::Deserializer<'de>>(
+    deserializer: D,
+) -> Result<Option<Activity>, D::Error> {
+    Ok(Option::<Value>::deserialize(deserializer)?
+        .and_then(|value| serde_json::from_value(value).ok()))
+}
 impl Scene {
+    /// Drop an activity that is invalid or bound to another cursor or
+    /// session. Rejecting the whole frame instead would clear queued events,
+    /// drop the producer lease and loop every shared view on reconnect.
+    fn settle_activity(&mut self) {
+        let source = (self.source != "unattached").then_some(self.source.as_str());
+        if self.activity.as_ref().is_some_and(|activity| {
+            !activity.is_valid()
+                || activity.cursor != self.cursor
+                || activity.session_id.as_deref() != source
+        }) {
+            self.activity = None;
+        }
+    }
    fn valid(&self) -> bool {
        self.version == 1
            && self.time_ms.is_finite()
@@ -66,12 +89,6 @@ impl Scene {
                .chain(&self.still.points)
                .flatten()
                .all(|p| p.is_finite() && p.abs() <= 8.0)
-            && self.activity.as_ref().is_none_or(|activity| {
-                activity.is_valid()
-                    && activity.cursor == self.cursor
-                    && activity.session_id.as_deref()
-                        == (self.source != "unattached").then_some(self.source.as_str())
-            })
    }
 }
 #[derive(Clone)]
@@ -395,7 +412,10 @@ fn run(
            match client
                .get("/v1/frame")
                .and_then(|value| serde_json::from_value::<Scene>(value).map_err(io::Error::other))
-            {
+                .map(|mut next| {
+                    next.settle_activity();
+                    next
+                }) {
                Ok(next) if next.valid() => {
                    if scene
                        .as_ref()
@@ -1724,7 +1724,7 @@ class PetEngineTelemetry {
                authoritativePresence = 'needs_you';
            else if (terminalFresh && this.turnOutcome === 'completed' && this.turnId)
                authoritativePresence = 'done';
-            else if (terminalFresh && this.turnOutcome)
+            else if (terminalFresh)
                authoritativePresence = 'idle';
            else if (active.length > 0 || this.turnId)
                authoritativePresence = 'working';
@@ -2003,7 +2003,11 @@ class PetEngineTelemetry {
                this.active.clear();
                this.waiting = undefined;
                this.turnId = id;
-                this.turnOutcome = outcome;
+                // An outcome belongs to a turn. `/purge`, an edit rejection or a
+                // session switch mid-turn completes with no turn id; recording the
+                // outcome alone would break the projection invariant the Rust
+                // contract checks (`turn_outcome` requires `turn_id`).
+                this.turnOutcome = id ? outcome : undefined;
                this.terminalAt = at;
                if (outcome === 'completed' && id && this.addOnce(this.completedTurns, id, 256))
                    this.add('turn_completed', 'communication', at).status = 'success';
@@ -1724,7 +1724,7 @@ class PetEngineTelemetry {
                authoritativePresence = 'needs_you';
            else if (terminalFresh && this.turnOutcome === 'completed' && this.turnId)
                authoritativePresence = 'done';
-            else if (terminalFresh && this.turnOutcome)
+            else if (terminalFresh)
                authoritativePresence = 'idle';
            else if (active.length > 0 || this.turnId)
                authoritativePresence = 'working';
@@ -2003,7 +2003,11 @@ class PetEngineTelemetry {
                this.active.clear();
                this.waiting = undefined;
                this.turnId = id;
-                this.turnOutcome = outcome;
+                // An outcome belongs to a turn. `/purge`, an edit rejection or a
+                // session switch mid-turn completes with no turn id; recording the
+                // outcome alone would break the projection invariant the Rust
+                // contract checks (`turn_outcome` requires `turn_id`).
+                this.turnOutcome = id ? outcome : undefined;
                this.terminalAt = at;
                if (outcome === 'completed' && id && this.addOnce(this.completedTurns, id, 256))
                    this.add('turn_completed', 'communication', at).status = 'success';
@@ -106,7 +106,7 @@ export class PetEngineTelemetry {
    if (fresh) {
      if (this.waiting) authoritativePresence = 'needs_you';
      else if (terminalFresh && this.turnOutcome === 'completed' && this.turnId) authoritativePresence = 'done';
-      else if (terminalFresh && this.turnOutcome) authoritativePresence = 'idle';
+      else if (terminalFresh) authoritativePresence = 'idle';
      else if (active.length > 0 || this.turnId) authoritativePresence = 'working';
    }
    const activityKind = fresh && authoritativePresence !== 'needs_you'
@@ -364,7 +364,11 @@ export class PetEngineTelemetry {
        this.active.clear();
        this.waiting = undefined;
        this.turnId = id;
-        this.turnOutcome = outcome as OwnerTurnOutcome;
+        // An outcome belongs to a turn. `/purge`, an edit rejection or a
+        // session switch mid-turn completes with no turn id; recording the
+        // outcome alone would break the projection invariant the Rust
+        // contract checks (`turn_outcome` requires `turn_id`).
+        this.turnOutcome = id ? outcome as OwnerTurnOutcome : undefined;
        this.terminalAt = at;
        if (outcome === 'completed' && id && this.addOnce(this.completedTurns, id, 256))
          this.add('turn_completed', 'communication', at).status = 'success';
@@ -42,6 +42,25 @@ test('only a completed turn is Done; interrupted and failed turns are idle', ()
  untracked.observe({event:'turn_complete',turn_outcome:'completed'},0);
  assert.equal(untracked.activity(10).authoritativePresence,'idle');
  assert.throws(() => untracked.observe({event:'turn_complete',turn_id:null,turn_outcome:'completed'},20));
+  // `/purge`, an edit rejection or a mid-turn session switch complete with no
+  // turn id. An outcome with no turn would fail the Rust projection
+  // invariant (`turn_outcome` requires `turn_id`) and drop the shared frame.
+  for (const outcome of ['completed','interrupted','failed']) {
+    const e = new PetEngineTelemetry();
+    e.observe({event:'turn_started',turn_id:'t'},0);
+    e.observe({event:'turn_complete',turn_outcome:outcome},20);
+    const activity=e.activity(30);
+    assert.equal(activity.turnId,null,outcome); assert.equal(activity.turnOutcome,null,outcome);
+    assert.equal(activity.authoritativePresence,'idle',outcome); assert.equal(activity.doneEffectId,null,outcome);
+  }
+});
+test('a new turn drops spans left open by the previous turn', () => {
+  const e = new PetEngineTelemetry();
+  e.observe({event:'turn_started',turn_id:'t1'},0);
+  e.observe({event:'operation_activity_started',span_id:'orphan',activity_kind:'editing'},10);
+  e.observe({event:'turn_started',turn_id:'t2'},20);
+  const activity=e.activity(30);
+  assert.equal(activity.turnId,'t2'); assert.equal(activity.activityKind,null); assert.deepEqual(activity.activeSpans,[]);
 });
 test('waiting and concurrent activity stay bounded and clear after disconnect', () => {
  const pet=new PetNative(points,'','[]',true);