fix(cleanup): fix session-task-cursor-crashes dirty follow-up

n00n-agent/src/agent/tool_dispatch.rs

@@ -264,9 +264,22 @@ fn is_subagent_failure(event: &ToolDoneEvent, ctx: &ToolContext) -> bool {
     let Some(entry) = ctx.registry.get(event.tool.as_ref()) else {
         return false;
     };
-    matches!(
+    let is_subagent = matches!(
         entry.source,
         ToolSource::Lua { plugin } if SUBAGENT_PLUGINS.contains(&plugin.as_ref())
+    );
+    is_subagent
+        && (ctx.cancel.is_cancelled()
+            || !is_cancelled_subagent_output(event.output.as_text().as_str()))
+}
+
+fn is_cancelled_subagent_output(output: &str) -> bool {
+    matches!(
+        output.trim(),
+        "cancelled"
+            | "sub-agent error: cancelled"
+            | "task failed: cancelled"
+            | "task failed: plugin interrupted: task cancelled"
     )
 }
 
@@ -1993,6 +2006,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn child_cancelled_subagent_only_fails_when_parent_is_cancelled() {
+        const CANCELLED: &str = "cancelled";
+        let registry = ToolRegistry::new();
+        let tool: Arc<dyn Tool> = Arc::new(FailingSubagentTool::new("task", CANCELLED));
+        registry
+            .register(
+                &tool,
+                &ToolSource::Lua {
+                    plugin: "task".into(),
+                },
+            )
+            .unwrap();
+        let (parent_cancel, parent_token) = crate::CancelToken::new();
+        let mut ctx = crate::tools::test_support::stub_ctx(&Arc::new(AgentMode::Build));
+        ctx.cancel = parent_token;
+        ctx.registry = Arc::new(registry);
+        let event = ToolDoneEvent {
+            id: "tu1".into(),
+            tool: "task".into(),
+            output: ToolOutput::Plain(CANCELLED.into()),
+            is_error: true,
+            annotation: None,
+            written_path: None,
+        };
+
+        assert!(!is_subagent_failure(&event, &ctx));
+        parent_cancel.cancel();
+        assert!(is_subagent_failure(&event, &ctx));
+    }
+
     #[test]
     fn failed_subagent_tool_aborts_process_tool_calls() {
         smol::block_on(async {

n00n-lua/tests/real_plugins_restore.rs

@@ -29,6 +29,7 @@ const CODEGRAPH_SRC: &str = include_str!("../../plugins/codegraph/init.lua");
 const EXPLORE_SRC: &str = include_str!("../../plugins/explore/init.lua");
 const GREP_SRC: &str = include_str!("../../plugins/grep/init.lua");
 const SEMBLEM_SRC: &str = include_str!("../../plugins/semblem/init.lua");
+const TASK_SRC: &str = include_str!("../../plugins/task/init.lua");
 const WORKFLOW_SRC: &str = include_str!("../../plugins/workflow/init.lua");
 
 /// Only the real `ToolView` emits this when collapsed.
@@ -85,6 +86,7 @@ fn load_host() -> PluginHost {
     host.load_source("explore", EXPLORE_SRC).unwrap();
     host.load_source("grep", GREP_SRC).unwrap();
     host.load_source("semblem", SEMBLEM_SRC).unwrap();
+    host.load_source("task", TASK_SRC).unwrap();
     host
 }
 
@@ -487,6 +489,23 @@ fn restore(
     out
 }
 
+#[test]
+fn task_restore_rebuilds_old_plain_persisted_output() {
+    let host = load_host();
+    let output = "cancelled\nold detail one\nold detail two\nold detail three\nold detail four\nold detail five";
+    let restored = restore(
+        &host,
+        "task",
+        json!({ "description": "restored task", "prompt": "work" }),
+        output,
+        None,
+        vec![],
+    );
+
+    assert!(restored.body.contains("cancelled"));
+    assert!(restored.body.contains(EXPAND_HINT));
+}
+
 #[test_case::test_case(
     "explore",
     json!({ "query": "how does session restore work", "project": "/tmp/project" }),

n00n-providers/src/providers/cursor/run.rs

@@ -723,21 +723,15 @@ fn handle_data_frame(
         status: 502,
         message,
     })?;
-    if let Some(op) = parse_kv_server_message(&payload).map_err(|message| AgentError::Api {
-        status: 502,
-        message,
-    })? {
+    if let Ok(Some(op)) = parse_kv_server_message(&payload) {
         queue_checkpoint_reply(op, checkpoints, outbound)?;
         return Ok(FrameHandleOutcome {
             exec_skipped: false,
             text_deltas: 0,
             kv_op: true,
         });
     }
-    if has_exec_server_message(&payload).map_err(|message| AgentError::Api {
-        status: 502,
-        message,
-    })? {
+    if let Ok(true) = has_exec_server_message(&payload) {
         // Phase 0: n00n owns tools; ignore Cursor-side exec until Phase 1 maps them.
         // Aborting the whole turn drops text deltas that often follow.
         return Ok(FrameHandleOutcome {
@@ -747,18 +741,16 @@ fn handle_data_frame(
         });
     }
     let mut deltas = 0u32;
-    for delta in extract_text_deltas(&payload).map_err(|message| AgentError::Api {
-        status: 502,
-        message,
-    })? {
-        text.push_str(&delta);
-        deltas = deltas.saturating_add(1);
+    if let Ok(text_deltas) = extract_text_deltas(&payload) {
+        for delta in text_deltas {
+            text.push_str(&delta);
+            deltas = deltas.saturating_add(1);
+        }
     }
-    for delta in extract_thinking_deltas(&payload).map_err(|message| AgentError::Api {
-        status: 502,
-        message,
-    })? {
-        thinking.push_str(&delta);
+    if let Ok(thinking_deltas) = extract_thinking_deltas(&payload) {
+        for delta in thinking_deltas {
+            thinking.push_str(&delta);
+        }
     }
     Ok(FrameHandleOutcome {
         exec_skipped: false,
@@ -849,6 +841,48 @@ mod tests {
         assert!(outbound.lock().expect("lock").queue.is_empty());
     }
 
+    #[test]
+    fn handle_data_frame_ignores_unknown_wire_type_three_payload() {
+        let frame = ConnectFrame {
+            end_stream: false,
+            compressed: false,
+            payload: vec![0x0b, 0x0c],
+        };
+        let store = shared_store();
+        let (outbound, _notify) = new_outbound_queue();
+        let mut text = String::new();
+        let mut thinking = String::new();
+
+        let outcome = handle_data_frame(&frame, &mut text, &mut thinking, &store, &outbound)
+            .expect("unknown protobuf variants should be ignored");
+
+        assert!(text.is_empty());
+        assert!(thinking.is_empty());
+        assert!(!outcome.exec_skipped);
+        assert_eq!(outcome.text_deltas, 0);
+        assert!(!outcome.kv_op);
+    }
+
+    #[test]
+    fn handle_data_frame_rejects_corrupt_compressed_payload() {
+        let frame = ConnectFrame {
+            end_stream: false,
+            compressed: true,
+            payload: b"not-gzip".to_vec(),
+        };
+        let store = shared_store();
+        let (outbound, _notify) = new_outbound_queue();
+        let mut text = String::new();
+        let mut thinking = String::new();
+
+        let Err(error) = handle_data_frame(&frame, &mut text, &mut thinking, &store, &outbound)
+        else {
+            panic!("transport decompression errors must remain strict");
+        };
+
+        assert!(error.to_string().contains("gzip"));
+    }
+
     #[test]
     fn handle_data_frame_queues_set_blob_ack() {
         let mut args = field_bytes(1, b"blob-id");

n00n-providers/src/providers/openai/platform.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, OnceLock, Weak};
 use std::time::{Duration, Instant};
@@ -52,6 +53,7 @@ const RESPONSE_CHAIN_LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(25);
 const PROMPT_CACHE_SHARDS: u8 = 16;
 
 static PROCESS_INSTANCE_NONCE: OnceLock<u64> = OnceLock::new();
+static RESPONSE_OPERATIONS: OnceLock<ResponseOperationRegistry> = OnceLock::new();
 
 fn coding_plan_slot_count(slots: u64) -> u8 {
     match u8::try_from(slots.clamp(1, u64::from(CODING_PLAN_MAX_SLOTS))) {
@@ -61,6 +63,8 @@ fn coding_plan_slot_count(slots: u64) -> u8 {
 }
 
 type ResponseOperationSlot = Arc<AsyncMutex<()>>;
+type ResponseOperationKey = (PathBuf, n00nId);
+type ResponseOperationRegistry = Mutex<HashMap<ResponseOperationKey, Weak<AsyncMutex<()>>>>;
 
 #[derive(Debug, Clone, Copy)]
 pub struct OpenAiOptions {
@@ -485,7 +489,6 @@ pub struct OpenAi {
     system_prefix: Option<String>,
     session_state: Arc<Mutex<HashMap<n00nId, OpenAiSessionState>>>,
     response_connections: Arc<Mutex<HashMap<n00nId, ResponseConnectionSlot>>>,
-    response_operations: Arc<Mutex<HashMap<n00nId, Weak<AsyncMutex<()>>>>>,
 }
 
 impl OpenAi {
@@ -522,7 +525,6 @@ impl OpenAi {
             system_prefix: None,
             session_state: Arc::new(Mutex::new(HashMap::new())),
             response_connections: Arc::new(Mutex::new(HashMap::new())),
-            response_operations: Arc::new(Mutex::new(HashMap::new())),
         })
     }
 
@@ -557,7 +559,6 @@ impl OpenAi {
             system_prefix: None,
             session_state: Arc::new(Mutex::new(HashMap::new())),
             response_connections: Arc::new(Mutex::new(HashMap::new())),
-            response_operations: Arc::new(Mutex::new(HashMap::new())),
         })
     }
 
@@ -1797,18 +1798,19 @@ impl OpenAi {
         &self,
         session_id: Option<&SessionRef>,
     ) -> Option<ResponseOperationSlot> {
-        let session_id = session_id?;
-        let session_id = canonical_session_key(session_id);
-        let mut operations = self
-            .response_operations
+        let session_id = canonical_session_key(session_id?);
+        let storage_path = self.response_state_storage.as_ref()?.path().to_path_buf();
+        let mut operations = RESPONSE_OPERATIONS
+            .get_or_init(|| Mutex::new(HashMap::new()))
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let key = (storage_path, session_id);
         operations.retain(|_, operation| operation.strong_count() > 0);
-        if let Some(operation) = operations.get(&session_id).and_then(Weak::upgrade) {
+        if let Some(operation) = operations.get(&key).and_then(Weak::upgrade) {
             return Some(operation);
         }
         let operation = Arc::new(AsyncMutex::new(()));
-        operations.insert(session_id, Arc::downgrade(&operation));
+        operations.insert(key, Arc::downgrade(&operation));
         Some(operation)
     }
 }
@@ -3147,12 +3149,17 @@ mod tests {
     }
 
     #[test]
-    fn response_operation_slot_is_reused_while_request_is_live() {
+    fn response_operation_slot_is_reused_across_provider_instances() {
         let temp_dir = TempDir::new().unwrap();
-        let provider = provider_with_response_storage(temp_dir.path());
+        let first_provider = provider_with_response_storage(temp_dir.path());
+        let second_provider = provider_with_response_storage(temp_dir.path());
         let session_id = SessionRef::generate();
-        let first = provider.response_operation_slot(Some(&session_id)).unwrap();
-        let second = provider.response_operation_slot(Some(&session_id)).unwrap();
+        let first = first_provider
+            .response_operation_slot(Some(&session_id))
+            .unwrap();
+        let second = second_provider
+            .response_operation_slot(Some(&session_id))
+            .unwrap();
 
         assert!(Arc::ptr_eq(&first, &second));
     }

plugins/task/init.lua

@@ -6,6 +6,7 @@
 -- `n00n.async.semaphore`).
 
 local ActivityPreview = require("n00n.activity_preview")
+local ToolView = require("n00n.tool_view")
 local output_limits = require("n00n.output_limits")
 local route_tier = require("n00n.route_tier").route_tier
 local structured_output = require("n00n.structured_output")