diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 934a50ee05e..01af2876e94 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -26,6 +26,7 @@ fn run() -> Result<(), Box> { linger_after_turn_start: args.iter().any(|arg| arg == "--linger-after-turn-start"), }; let mut turn_count = 0_u64; + let mut active_turn_id: Option = None; let mut planning_thread = false; let mut pending_tool_turns = BTreeMap::::new(); let stdin = io::stdin(); @@ -122,6 +123,7 @@ fn run() -> Result<(), Box> { } else { format!("fake-turn-{turn_count}") }; + active_turn_id = Some(turn_id.clone()); let call_id = format!("call-{turn_count}"); let request_id = format!("rpc-tool-{turn_count}"); send(json!({"id": id, "result": {"turn": {"id": turn_id}}}))?; @@ -153,6 +155,20 @@ fn run() -> Result<(), Box> { "params": {"threadId": "fake-thread", "turnId": turn_id, "callId": call_id, "tool": "get_task_context", "arguments": {}} }))?; } + "turn/steer" => { + let expected_turn_id = request + .pointer("/params/expectedTurnId") + .and_then(Value::as_str); + if expected_turn_id != active_turn_id.as_deref() { + send(json!({ + "id": id, + "error": {"code": -32000, "message": "stale active turn"} + }))?; + } else { + send(json!({"id": id, "result": {"acknowledged": true}}))?; + } + } + "turn/interrupt" => send(json!({"id": id, "result": {}}))?, _ => {} } let completed_turn = request diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index 17f13b08632..34745025b3b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, VecDeque}; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -250,6 +250,9 @@ pub enum ProviderSessionIdentity { pub trait Provider { fn kind(&self) -> ProviderKind; + fn agent_process_id(&self) -> Option { + None + } fn runtime_identity(&self) -> ProviderRuntimeIdentity; fn session_identity(&self) -> &str; fn provider_session_id(&self) -> Option<&str>; @@ -295,6 +298,11 @@ pub trait Provider { cwd: &str, turn_id: &str, ) -> Result; + fn steer_turn(&mut self, _turn_id: &str, _message: &str) -> Result { + Err(LocalRunnerError::invalid( + "provider does not support same-turn steering", + )) + } fn interrupt_turn(&mut self, turn_id: &str) -> Result; fn read(&mut self) -> Result; fn poll(&mut self) -> Result, LocalRunnerError>; @@ -332,6 +340,8 @@ pub struct CodexProvider { next_request_id: u64, thread_id: String, session_id: Option, + active_provider_turn_id: Option, + pending_messages: VecDeque, pending_tool_requests: BTreeMap, collaboration_mode: String, collaboration_mode_payload: Option, @@ -358,6 +368,8 @@ impl CodexProvider { next_request_id: 1, thread_id: String::new(), session_id: None, + active_provider_turn_id: None, + pending_messages: VecDeque::new(), pending_tool_requests: BTreeMap::new(), collaboration_mode: config.collaboration_mode.clone(), collaboration_mode_payload: None, @@ -504,14 +516,45 @@ impl CodexProvider { .expect("turn parameters are an object") .insert("collaborationMode".to_owned(), collaboration_mode); } - self.request("turn/start", params) + let response = self.request("turn/start", params)?; + self.active_provider_turn_id = response + .pointer("/turn/id") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned); + if self.active_provider_turn_id.is_none() { + return Err(LocalRunnerError::invalid( + "Codex turn/start omitted turn.id", + )); + } + Ok(response) } - pub fn interrupt_turn(&mut self, turn_id: &str) -> Result { + pub fn interrupt_turn(&mut self, _turn_id: &str) -> Result { let thread_id = self.thread_id.clone(); + let provider_turn_id = self + .active_provider_turn_id + .clone() + .ok_or_else(|| LocalRunnerError::invalid("Codex has no active provider turn"))?; self.request( "turn/interrupt", - json!({ "threadId": thread_id, "turnId": turn_id }), + json!({ "threadId": thread_id, "turnId": provider_turn_id }), + ) + } + + pub fn steer_turn(&mut self, _turn_id: &str, message: &str) -> Result { + let thread_id = self.thread_id.clone(); + let provider_turn_id = self + .active_provider_turn_id + .clone() + .ok_or_else(|| LocalRunnerError::invalid("Codex has no active provider turn"))?; + self.request( + "turn/steer", + json!({ + "threadId": thread_id, + "expectedTurnId": provider_turn_id, + "input": [{"type": "text", "text": message, "text_elements": []}], + }), ) } @@ -524,16 +567,20 @@ impl CodexProvider { } pub fn poll(&mut self) -> Result, LocalRunnerError> { - let Some(line) = self.process.receive_stdout_line(Duration::from_millis(1))? else { - return if self.process.try_wait()?.is_some() { - Ok(Some(ProviderEvent::Exited)) - } else { - Ok(None) + let message = if let Some(message) = self.pending_messages.pop_front() { + message + } else { + let Some(line) = self.process.receive_stdout_line(Duration::from_millis(1))? else { + return if self.process.try_wait()?.is_some() { + Ok(Some(ProviderEvent::Exited)) + } else { + Ok(None) + }; }; + serde_json::from_str(&line).map_err(|error| { + LocalRunnerError::invalid(format!("Codex emitted invalid JSON-RPC: {error}")) + })? }; - let message: Value = serde_json::from_str(&line).map_err(|error| { - LocalRunnerError::invalid(format!("Codex emitted invalid JSON-RPC: {error}")) - })?; if message.get("id").is_some() && message.get("method").and_then(Value::as_str) == Some("item/tool/call") { @@ -633,6 +680,7 @@ impl CodexProvider { } return Ok(message.get("result").cloned().unwrap_or(Value::Null)); } + self.pending_messages.push_back(message); } } } @@ -661,6 +709,9 @@ impl Provider for CodexProvider { ) -> Result { CodexProvider::start_turn(self, message, cwd) } + fn steer_turn(&mut self, turn_id: &str, message: &str) -> Result { + CodexProvider::steer_turn(self, turn_id, message) + } fn interrupt_turn(&mut self, turn_id: &str) -> Result { CodexProvider::interrupt_turn(self, turn_id) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 329b205b6ec..ba82d0652c7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -845,6 +845,60 @@ fn process_command_and_provider( )?; store.save(state)?; } + Some("turn.steer") => { + let steering = (|| -> Result<(String, String), DurableRunnerError> { + let text = command + .pointer("/payload/text") + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| { + DurableRunnerError::invalid("turn.steer payload.text is required") + })?; + let turn_id = command + .pointer("/payload/turnId") + .and_then(Value::as_str) + .unwrap_or(state.turn_id.as_str()); + if turn_id != state.turn_id || !state.active_turn { + return Err(DurableRunnerError::invalid( + "turn.steer named a stale turn", + )); + } + provider + .as_mut() + .ok_or_else(|| DurableRunnerError::invalid("turn.steer requires a provider"))? + .steer_turn(turn_id, text) + .map_err(|error| { + DurableRunnerError::invalid(format!("provider turn.steer failed: {error}")) + })?; + Ok((turn_id.to_owned(), text.to_owned())) + })(); + let (turn_id, _text) = match steering { + Ok(value) => value, + Err(error) => { + // process_command persisted its dedupe entry before the + // provider effect. Roll that state back so a rejected or + // timed-out steering command remains genuinely retryable. + *state = before; + store.save(state)?; + return Err(error); + } + }; + enqueue_event( + state, + config, + "item.completed", + 1, + json!({ + "kind": "steering_acknowledgement", + "status": "acknowledged", + "text": "Steering acknowledged for the active turn.", + "commandId": command_id, + "correlationId": command.pointer("/payload/correlationId").cloned().unwrap_or(Value::Null), + }), + Some(&format!("{}:steer:{}", turn_id, command_id)), + )?; + store.save(state)?; + } Some("semantic_tool.result") => { let result: crate::provider_bridge::ToolResult = serde_json::from_value(command["payload"].clone()).map_err(|error| { @@ -1429,12 +1483,63 @@ mod tests { use crate::local_runner::LocalRunnerError; use crate::provider_bridge::ToolResult; use std::net::TcpListener; - use std::sync::{mpsc, Mutex}; + use std::sync::{mpsc, Arc, Mutex}; static ENVIRONMENT_TEST_LOCK: Mutex<()> = Mutex::new(()); struct OversizedFrameProvider; + struct SteeringProvider { + calls: Arc>>, + fail_next: Arc>, + } + + impl Provider for SteeringProvider { + fn kind(&self) -> ProviderKind { + ProviderKind::Codex + } + fn runtime_identity(&self) -> crate::codex_provider::ProviderRuntimeIdentity { + crate::codex_provider::ProviderRuntimeIdentity::LocalProcess { + process_id: 2, + provider_session_id: "steering".to_owned(), + } + } + fn session_identity(&self) -> &str { + "steering-thread" + } + fn provider_session_id(&self) -> Option<&str> { + Some("steering-session") + } + fn start_turn(&mut self, _message: &str, _cwd: &str, _turn_id: &str) -> Result { + unreachable!() + } + fn steer_turn(&mut self, turn_id: &str, message: &str) -> Result { + let mut fail_next = self.fail_next.lock().unwrap(); + if *fail_next { + *fail_next = false; + return Err(LocalRunnerError::invalid("provider rejected steering")); + } + drop(fail_next); + self.calls.lock().unwrap().push((turn_id.to_owned(), message.to_owned())); + Ok(json!({"acknowledged": true})) + } + fn interrupt_turn(&mut self, _turn_id: &str) -> Result { + unreachable!() + } + fn read(&mut self) -> Result { + unreachable!() + } + fn poll(&mut self) -> Result, LocalRunnerError> { + Ok(None) + } + fn deliver_tool_result(&mut self, _result: &ToolResult) -> Result<(), LocalRunnerError> { + unreachable!() + } + fn shutdown(&mut self) -> Result<(), LocalRunnerError> { + Ok(()) + } + } + impl Provider for OversizedFrameProvider { fn kind(&self) -> ProviderKind { ProviderKind::Codex @@ -1873,6 +1978,99 @@ mod tests { let _ = fs::remove_dir_all(root); } + #[test] + fn steering_is_bound_to_the_active_turn_and_duplicate_commands_do_not_redispatch() { + let root = temporary_root("steering-command-dedupe"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + state.active_turn = true; + let calls = Arc::new(Mutex::new(Vec::new())); + let fail_next = Arc::new(Mutex::new(false)); + let mut provider: Option> = Some(Box::new(SteeringProvider { + calls: calls.clone(), + fail_next, + })); + let steering = command( + "command-steer-1", + 1, + "turn.steer", + json!({ "turnId": config.turn_id, "text": "Prioritize mobile overflow." }), + ); + + let first = process_command_and_provider( + &mut state, + &store, + &config, + &mut provider, + &steering, + ).unwrap(); + let event_count = state.outbox.len(); + let duplicate = process_command_and_provider( + &mut state, + &store, + &config, + &mut provider, + &steering, + ).unwrap(); + + assert_eq!(first.status, "completed"); + assert_eq!(duplicate.result, first.result); + assert_eq!(calls.lock().unwrap().as_slice(), &[ + (config.turn_id.clone(), "Prioritize mobile overflow.".to_owned()), + ]); + assert_eq!(state.outbox.len(), event_count); + assert!(state.outbox.iter().any(|event| { + event.envelope["payload"]["eventType"] == "item.completed" + && event.envelope["payload"]["payload"]["kind"] == "steering_acknowledgement" + && event.envelope["payload"]["payload"]["commandId"] == "command-steer-1" + })); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn rejected_steering_rolls_back_the_dedupe_ledger_for_retry() { + let root = temporary_root("steering-command-retry"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + state.active_turn = true; + let calls = Arc::new(Mutex::new(Vec::new())); + let fail_next = Arc::new(Mutex::new(true)); + let mut provider: Option> = Some(Box::new(SteeringProvider { + calls: calls.clone(), + fail_next, + })); + let steering = command( + "command-steer-retry", + 1, + "turn.steer", + json!({ "turnId": config.turn_id, "text": "Retry me." }), + ); + + let first_error = process_command_and_provider( + &mut state, + &store, + &config, + &mut provider, + &steering, + ).unwrap_err(); + assert!(first_error.to_string().contains("provider rejected steering")); + assert!(!state.processed_commands.contains_key("command-steer-retry")); + assert_eq!(state.last_controller_command_seq, 0); + + let retry = process_command_and_provider( + &mut state, + &store, + &config, + &mut provider, + &steering, + ).unwrap(); + assert_eq!(retry.status, "completed"); + assert_eq!(calls.lock().unwrap().len(), 1); + let _ = fs::remove_dir_all(root); + } + #[test] fn provider_bootstrap_failure_returns_a_failed_prepare_result() { let root = temporary_root("provider-bootstrap-failure"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index d99a9de63b7..ba96e9d5273 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -1518,6 +1518,11 @@ fn execute_command_effect( } Ok(("completed".to_owned(), 1, "turn completed".to_owned())) } + "turn.steer" => Ok(( + "completed".to_owned(), + 1, + "provider turn steering requested".to_owned(), + )), "turn.interrupt" => Ok(( "completed".to_owned(), 1, diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index 55dcad1d3f1..b1dc36f6d96 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -153,7 +153,7 @@ class HarnessNativeSession implements NativeSession { return this.#session.startTurn(input); } - steer(input: { turnId: string; message: { role: "user"; text: string } }) { + steer(input: { turnId: string; message: { role: "user"; text: string }; correlationId?: string }) { if (this.#session.steer === undefined) throw new Error("steering is unavailable"); return this.#session.steer(input); } diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 667842177f1..8704ebaf5fd 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -349,7 +349,7 @@ export interface HarnessSession { message: NativeUserMessage; requestedCollaborationMode?: "default" | "plan"; }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan" }>; - steer?(input: { turnId: string; message: NativeUserMessage }): Promise; + steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise; interrupt?(input: { turnId?: string; reason?: string }): Promise; pendingRuntimeRequests?(): HarnessRuntimeRequest[]; resolveRuntimeRequest?(input: { diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 0e94d25b01d..da60d6c0a9d 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -56,7 +56,7 @@ export interface NativeSession { message: NativeUserMessage; requestedCollaborationMode?: "default" | "plan"; }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan" }>; - steer?(input: { turnId: string; message: NativeUserMessage }): Promise; + steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise; interrupt?(input: { turnId?: string; reason?: string }): Promise; cancel?(input: { reason: string }): Promise; result(): Promise<{ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts index f4b2bba3515..68f0ca26520 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts @@ -701,7 +701,32 @@ describe("Codex app-server Codex driver", () => { workingDirectory: "/workspace", }); const { turnId } = await session.startTurn({ message: { role: "user", text: "Start." } }); - await session.steer?.({ turnId, message: { role: "user", text: "Stay concise." } }); + await session.steer?.({ + turnId, + message: { role: "user", text: "Stay concise." }, + correlationId: "queued-comment-1", + }); + await session.steer?.({ + turnId, + message: { role: "user", text: "Stay concise." }, + correlationId: "queued-comment-1", + }); + expect(transport.calls.find((call) => call.method === "turn/steer")?.params) + .toMatchObject({ correlationId: "queued-comment-1" }); + expect(transport.calls.filter((call) => call.method === "turn/steer")).toHaveLength(1); + transport.rejectMethods.set("turn/steer", new Error("provider rejected steering")); + const rejectedSteering = await session.steer?.({ + turnId, + message: { role: "user", text: "Try this again later." }, + correlationId: "queued-comment-2", + }).then( + () => null, + (error: unknown) => error, + ); + expect(rejectedSteering).toBeInstanceOf(Error); + expect(rejectedSteering).not.toBeInstanceOf(HarnessCapabilityUnavailableError); + expect(String(rejectedSteering)).toContain("provider rejected steering"); + transport.rejectMethods.delete("turn/steer"); await expect(session.steer?.({ turnId: fixture.controls.staleTurnSteer.turnId!, message: { role: "user", text: "Stale." }, @@ -732,6 +757,7 @@ describe("Codex app-server Codex driver", () => { expect(observed).toEqual(expect.arrayContaining([ expect.objectContaining({ eventType: "item.completed", + itemId: `${turnId}:steer:queued-comment-1`, payload: expect.objectContaining({ kind: "steering_acknowledgement", status: "acknowledged" }), }), expect.objectContaining({ @@ -1761,7 +1787,10 @@ describe("Codex app-server Codex driver", () => { it("degrades unsupported operations with explicit redacted diagnostics", async () => { const transport = new FakeCodexTransport(); - transport.rejectMethods.set("turn/steer", new Error("Bearer super-secret api_key=also-secret")); + transport.rejectMethods.set( + "turn/steer", + new Error("method not found Bearer super-secret api_key=also-secret"), + ); const session = await makeDriver([transport]).openSession({ runId: "run-degrade", normalizedSessionId: "normalized-degrade", workingDirectory: "/workspace" }); const { turnId } = await session.startTurn({ message: { role: "user", text: "Start." } }); await expect(session.steer?.({ turnId, message: { role: "user", text: "Steer." } })).rejects.toBeInstanceOf(HarnessCapabilityUnavailableError); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts index 2a6bddb9714..d38fd2c2c26 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts @@ -786,6 +786,7 @@ class CodexHarnessSession implements HarnessSession { #goal: HarnessThreadGoal | null = null; #interruptQueued = false; #steerSequence = 0; + readonly #acknowledgedSteeringCorrelations = new Map(); #interruptSequence = 0; constructor(input: { @@ -986,29 +987,49 @@ class CodexHarnessSession implements HarnessSession { }; } - async steer(input: { turnId: string; message: NativeUserMessage }): Promise { + async steer(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise { this.#requireCapability("steering"); this.#requireActiveTurn(input.turnId, "steering"); + if (input.correlationId) { + const acknowledgedTurnId = this.#acknowledgedSteeringCorrelations.get(input.correlationId); + if (acknowledgedTurnId) { + if (acknowledgedTurnId !== input.turnId) throw new HarnessOperationAlreadyTerminalError("steering"); + return; + } + } try { await this.#transport.request("turn/steer", { threadId: this.#opened.threadId, input: [userInput(input.message)], expectedTurnId: input.turnId, + correlationId: input.correlationId, }); if (this.#activeTurnId !== input.turnId) { throw new HarnessOperationAlreadyTerminalError("steering"); } + if (input.correlationId) { + this.#acknowledgedSteeringCorrelations.set(input.correlationId, input.turnId); + } this.#emit("item.completed", { kind: "steering_acknowledgement", text: "Steering acknowledged for the active turn.", status: "acknowledged", }, { turnId: input.turnId, - itemId: `${input.turnId}:steer:${++this.#steerSequence}`, + itemId: input.correlationId + ? `${input.turnId}:steer:${input.correlationId}` + : `${input.turnId}:steer:${++this.#steerSequence}`, }); } catch (error) { if (error instanceof HarnessOperationAlreadyTerminalError) throw error; - throw this.#unsupported("steering", error); + const detail = redactCodexDiagnostic(String(error)); + if (/unsupported|unavailable|capability|method not found/i.test(detail)) { + throw this.#unsupported("steering", detail); + } + // Steering rejection is a retryable operation failure, not evidence + // that the provider lacks the capability. Preserve that distinction for + // the control plane while keeping provider diagnostics redacted. + throw new Error(detail); } } diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 0f906e1be43..2f670eb4ec4 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -125,6 +125,85 @@ it("runs the lab provider boundary through authenticated durable PRP", async () expect(bundle.evidence()).toMatchObject({ runnerExited: true, runnerExitCode: 0 }); }, 30_000); +it("steers the active provider turn through the durable PRP command path", async () => { + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: ["--linger-after-turn-start"], + }); + bundle.transport.setServerRequestHandler(async () => ({ + success: true, + contentItems: [], + })); + try { + await bundle.transport.request("initialize", {}); + await bundle.transport.request("thread/start", { + cwd: tmpdir(), + dynamicTools: [ + { + name: "get_task_context", + description: "Read the active task.", + inputSchema: { + type: "object", + properties: {}, + additionalProperties: false, + }, + }, + ], + }); + await bundle.transport.request("turn/start", { + input: [{ type: "text", text: "Work until I steer you." }], + }); + + await expect( + bundle.transport.request("turn/steer", { + input: [{ type: "text", text: "Prioritize the mobile queue layout." }], + correlationId: "queued-comment-1", + }), + ).resolves.toEqual({}); + await expect( + bundle.transport.request("turn/steer", { + input: [{ type: "text", text: "Prioritize the mobile queue layout." }], + correlationId: "queued-comment-1", + }), + ).resolves.toEqual({}); + await expect( + bundle.transport.request("turn/steer", { + expectedTurnId: "stale-logical-turn", + input: [{ type: "text", text: "This must not dispatch." }], + }), + ).rejects.toThrow("stale turn"); + + const notifications = bundle.transport + .notifications() + [Symbol.asyncIterator](); + const methods: string[] = []; + let acknowledged = false; + const deadline = Date.now() + 5_000; + while (Date.now() < deadline && !acknowledged) { + const next = await Promise.race([ + notifications.next(), + new Promise((_, reject) => + setTimeout( + () => reject(new Error("steering notification timeout")), + 1_000, + ), + ), + ]); + if (!next.value) break; + methods.push(next.value.method); + acknowledged = + next.value.method === "item/completed" && + (next.value.params?.kind === "steering_acknowledgement" || + next.value.params?.item?.kind === "steering_acknowledgement"); + } + expect(methods).toContain("turn/started"); + expect(acknowledged).toBe(true); + } finally { + await bundle.transport.close(); + } +}, 30_000); + it("attaches a second governed run to one warm runner and provider process", async () => { const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: defaultCapabilityRunnerdBinary(), diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 7b0e1a71d7a..719f10f3bc7 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -325,6 +325,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #threadId = ""; #sessionId: string | null = null; #turnId = ""; + #durableTurnId = ""; #authorizedTools: Record | null = null; #closed = false; #failure: Error | null = null; @@ -389,6 +390,34 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : { data: [] }; } if (method === "turn/start") return this.#startTurn(params); + if (method === "turn/steer") { + const input = Array.isArray(params.input) ? params.input.map(record) : []; + const text = input + .map((item) => (typeof item.text === "string" ? item.text : "")) + .join("\n"); + const expectedTurnId = + typeof params.expectedTurnId === "string" + ? params.expectedTurnId + : this.#turnId; + if (!text.trim()) throw new Error("turn/steer requires a message"); + if (expectedTurnId !== this.#turnId) + throw new Error("turn/steer named a stale turn"); + const correlationId = + typeof params.correlationId === "string" + ? params.correlationId + : undefined; + await this.#command( + "turn.steer", + { + text, + turnId: this.#durableTurnId, + providerTurnId: expectedTurnId, + ...(correlationId ? { correlationId } : {}), + }, + correlationId, + ); + return {}; + } if (method === "turn/interrupt") { await this.#command("turn.interrupt", params); return {}; @@ -450,6 +479,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ...(this.#authorizedTools === null ? {} : { authorizedTools: this.#authorizedTools }), }); await this.#waitForAttachment(input.runId); + this.#durableTurnId = input.turnId; this.#turnId = input.turnId; } @@ -520,6 +550,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { turnId: `turn_lab_${token}`, itemId: `item_lab_${token}`, }; + this.#durableTurnId = identity.turnId; const dynamicTools = Array.isArray(params.dynamicTools) ? params.dynamicTools.map(record) : []; const core = new DurablePrpControlPlane({ stateDirectory: resolve(this.#root, "control-plane"), @@ -693,6 +724,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } const controlPlaneDirectory = resolve(this.#root, "control-plane"); const identity = recoveredControlPlaneIdentity(controlPlaneDirectory, desiredIdentity); + this.#durableTurnId = identity.turnId; const attachingNewRun = identity.runId !== desiredIdentity.runId; const core = new DurablePrpControlPlane({ stateDirectory: controlPlaneDirectory, @@ -789,11 +821,35 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return { turn: { id: this.#turnId, status: "inProgress" } }; } - async #command(type: string, payload: Record): Promise { + async #command( + type: string, + payload: Record, + correlationId?: string, + ): Promise { const core = this.#core; if (core === null) throw new Error("PRP provider thread is not started"); - const commandId = `command_lab_${randomUUID().replaceAll("-", "")}`; - core.queueCommand(type, payload, commandId, true); + const commandId = correlationId + ? `command_steer_${createHash("sha256") + .update(`${this.#durableTurnId}:${correlationId}`) + .digest("hex") + .slice(0, 32)}` + : `command_lab_${randomUUID().replaceAll("-", "")}`; + const existing = core.store.state.commands.find( + (command) => command.commandId === commandId, + ); + if (existing) { + if ( + existing.type !== type || + durableRecoveryInternals.canonicalJson(existing.payload) !== + durableRecoveryInternals.canonicalJson(payload) + ) { + throw new Error( + `PRP steering correlation ${correlationId} was reused with different content`, + ); + } + } else { + core.queueCommand(type, payload, commandId, true); + } await this.#waitCommand(type, commandId); } diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index b3c311ab33a..22780fd1146 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -1036,6 +1036,10 @@ export type { IssueExecutionStagePrincipal, IssueExecutionDecision, IssueComment, + IssueQueuedCommentEntry, + IssueQueuedCommentProtocol, + IssueQueuedCommentQueue, + IssueQueuedCommentSteeringDisposition, IssueCommentDerivedAuthorSource, IssueCommentMetadata, IssueCommentMetadataSection, diff --git a/packages/shared/src/types/index.ts b/packages/shared/src/types/index.ts index 8f7a85e5787..f49638af1c2 100644 --- a/packages/shared/src/types/index.ts +++ b/packages/shared/src/types/index.ts @@ -673,6 +673,10 @@ export type { IssueReviewRequest, IssueExecutionDecision, IssueComment, + IssueQueuedCommentEntry, + IssueQueuedCommentProtocol, + IssueQueuedCommentQueue, + IssueQueuedCommentSteeringDisposition, IssueCommentDerivedAuthorSource, IssueCommentMetadata, IssueCommentMetadataSection, diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index e6b7f958dc6..f43bdb29cdc 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -969,6 +969,33 @@ export interface IssueComment { updatedAt: Date; } +export type IssueQueuedCommentProtocol = "paperclip_runner_v1" | "legacy"; +export type IssueQueuedCommentSteeringDisposition = + | "available" + | "unsupported" + | "temporarily_unavailable"; + +export interface IssueQueuedCommentEntry { + comment: IssueComment; + position: number; + canEdit: boolean; + canDiscard: boolean; +} + +/** + * Authoritative projection of the comments waiting behind one active issue + * run. `revision` is opaque and must be echoed by queue mutations so a stale + * browser cannot overwrite a newer ordering. + */ +export interface IssueQueuedCommentQueue { + issueId: string; + targetRunId: string | null; + revision: string; + protocol: IssueQueuedCommentProtocol; + steeringDisposition: IssueQueuedCommentSteeringDisposition; + entries: IssueQueuedCommentEntry[]; +} + interface IssueCommentMetadataRowBase { type: IssueCommentMetadataRowType; label?: string | null; diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts new file mode 100644 index 00000000000..6ea7c57b7f1 --- /dev/null +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -0,0 +1,303 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agentWakeupRequests, + agents, + companies, + companyMemberships, + createDb, + heartbeatRuns, + issueComments, + issues, +} from "@paperclipai/db"; +import { errorHandler } from "../middleware/index.js"; +import { issueRoutes } from "../routes/issues.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe.sequential : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping queued-comment route tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("issue queued-comment routes", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-queued-comments-"); + db = createDb(tempDb.connectionString); + }, 30_000); + + afterEach(async () => { + await db.delete(issueComments).catch(() => undefined); + await db.delete(issues).catch(() => undefined); + await db.delete(heartbeatRuns).catch(() => undefined); + await db.delete(agentWakeupRequests).catch(() => undefined); + await db.delete(companyMemberships).catch(() => undefined); + await db.delete(agents).catch(() => undefined); + await db.delete(companies).catch(() => undefined); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + function app(companyId: string, userId = "queue-owner") { + const testApp = express(); + testApp.use(express.json()); + testApp.use((req, _res, next) => { + (req as any).actor = { + type: "board", + source: "session", + userId, + companyIds: [companyId], + memberships: [{ companyId, status: "active", membershipRole: "operator" }], + isInstanceAdmin: false, + }; + next(); + }); + testApp.use("/api", issueRoutes(db, {} as any, {})); + testApp.use(errorHandler); + return testApp; + } + + async function seedQueue() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + const wakeId = randomUUID(); + const commentIds = [randomUUID(), randomUUID()]; + await db.insert(companies).values({ + id: companyId, + name: "Queue Test Company", + issuePrefix: "QUE", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Paperclip Runner", + role: "engineer", + status: "working", + adapterType: "paperclip_runner", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(companyMemberships).values([ + { + companyId, + principalType: "user", + principalId: "queue-owner", + status: "active", + membershipRole: "operator", + }, + { + companyId, + principalType: "user", + principalId: "other-operator", + status: "active", + membershipRole: "operator", + }, + ]); + await db.insert(agentWakeupRequests).values({ + id: wakeId, + companyId, + agentId, + source: "issue_comment", + reason: "Follow-up comments arrived during the active run", + status: "deferred_issue_execution", + requestedByActorType: "user", + requestedByActorId: "queue-owner", + payload: { + issueId, + commentId: commentIds[1], + _paperclipWakeContext: { + commentId: commentIds[1], + wakeCommentId: commentIds[1], + wakeCommentIds: commentIds, + }, + }, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "queue route test", + status: "running", + runtimeMode: "native", + startedAt: new Date("2026-08-22T15:00:00.000Z"), + contextSnapshot: { issueId }, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + identifier: "QUE-1", + title: "Queued steering", + status: "in_progress", + priority: "medium", + assigneeAgentId: agentId, + executionRunId: runId, + }); + await db.insert(issueComments).values(commentIds.map((id, index) => ({ + id, + companyId, + issueId, + authorType: "user" as const, + authorUserId: "queue-owner", + body: index === 0 ? "First queued message" : "Second queued message", + createdAt: new Date(`2026-08-22T15:0${index + 1}:00.000Z`), + updatedAt: new Date(`2026-08-22T15:0${index + 1}:00.000Z`), + }))); + return { companyId, agentId, issueId, runId, wakeId, commentIds }; + } + + it("returns the authoritative order, preserves full Markdown edits, and rejects stale revisions", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + + expect(initial.status, JSON.stringify(initial.body)).toBe(200); + expect(initial.body).toMatchObject({ + issueId: seeded.issueId, + targetRunId: seeded.runId, + protocol: "paperclip_runner_v1", + entries: [ + { position: 0, canEdit: true, canDiscard: true, comment: { id: seeded.commentIds[0] } }, + { position: 1, canEdit: true, canDiscard: true, comment: { id: seeded.commentIds[1] } }, + ], + }); + + const markdown = " Keep **all** Markdown. \n"; + const edited = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ targetRunId: seeded.runId, revision: initial.body.revision, body: markdown }); + expect(edited.status, JSON.stringify(edited.body)).toBe(200); + expect(edited.body.entries[0].comment.body).toBe(markdown); + expect(edited.body.revision).not.toBe(initial.body.revision); + const stored = await db.select({ body: issueComments.body }) + .from(issueComments) + .where(eq(issueComments.id, seeded.commentIds[0])) + .then((rows) => rows[0]); + expect(stored?.body).toBe(markdown); + + const stale = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ targetRunId: seeded.runId, revision: initial.body.revision, body: "stale" }); + expect(stale.status).toBe(409); + expect(stale.body.details?.code).toBe("queued_comment_revision_conflict"); + }); + + it("rewrites the wake order transactionally and cancels an empty queue after trash", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + const reordered = await request(app(seeded.companyId)) + .put(`/api/issues/${seeded.issueId}/queued-comments/order`) + .send({ + targetRunId: seeded.runId, + revision: initial.body.revision, + orderedCommentIds: [...seeded.commentIds].reverse(), + }); + expect(reordered.status, JSON.stringify(reordered.body)).toBe(200); + expect(reordered.body.entries.map((entry: any) => entry.comment.id)).toEqual([...seeded.commentIds].reverse()); + + const afterFirstTrash = await request(app(seeded.companyId)) + .delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[1]}`) + .send({ targetRunId: seeded.runId, revision: reordered.body.revision }); + expect(afterFirstTrash.status, JSON.stringify(afterFirstTrash.body)).toBe(200); + expect(afterFirstTrash.body.entries.map((entry: any) => entry.comment.id)).toEqual([seeded.commentIds[0]]); + + const emptied = await request(app(seeded.companyId)) + .delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ targetRunId: seeded.runId, revision: afterFirstTrash.body.revision }); + expect(emptied.status, JSON.stringify(emptied.body)).toBe(200); + expect(emptied.body.entries).toEqual([]); + const wake = await db.select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, seeded.wakeId)) + .then((rows) => rows[0]); + expect(wake?.status).toBe("cancelled"); + }); + + it("limits edit and trash to the comment owner", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId, "other-operator")) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(initial.status).toBe(200); + expect(initial.body.entries[0]).toMatchObject({ canEdit: false, canDiscard: false }); + + const edit = await request(app(seeded.companyId, "other-operator")) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ targetRunId: seeded.runId, revision: initial.body.revision, body: "not mine" }); + expect(edit.status).toBe(403); + const discard = await request(app(seeded.companyId, "other-operator")) + .delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ targetRunId: seeded.runId, revision: initial.body.revision }); + expect(discard.status).toBe(403); + }); + + it("leaves the selected row queued when no native steering session is attached", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + const steered = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ targetRunId: seeded.runId, revision: initial.body.revision }); + + expect(steered.status).toBe(409); + expect(steered.body.details).toMatchObject({ + code: "steering_temporarily_unavailable", + retryable: true, + }); + const queueAfterFailure = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id)).toEqual(seeded.commentIds); + }); + + it("does not consume or edit a queued comment when the target run finishes first", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + await db + .update(heartbeatRuns) + .set({ status: "succeeded", finishedAt: new Date("2026-08-22T15:05:00.000Z") }) + .where(eq(heartbeatRuns.id, seeded.runId)); + + const edit = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ + targetRunId: seeded.runId, + revision: initial.body.revision, + body: "must remain deferred", + }); + + expect(edit.status).toBe(409); + expect(edit.body.details?.code).toBe("queued_comment_stale_target"); + const stored = await db + .select({ body: issueComments.body }) + .from(issueComments) + .where(eq(issueComments.id, seeded.commentIds[0])) + .then((rows) => rows[0]); + expect(stored?.body).toBe("First queued message"); + const wake = await db + .select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, seeded.wakeId)) + .then((rows) => rows[0]); + expect(wake?.status).toBe("deferred_issue_execution"); + expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual(seeded.commentIds); + }); +}); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 566a973a7d8..c3ab45a038c 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -6,6 +6,7 @@ import { and, asc, desc, eq, inArray, isNull, notInArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { activityLog, + agentWakeupRequests, agents, approvals, companyMemberships, @@ -92,6 +93,7 @@ import { type IssueReviewPolicy, type IssueThreadInteractionCanonicalResolverPolicy, type IssueCommentPresentation, + type IssueQueuedCommentQueue, type IssueWatchdogDiscoveryKind, type ProjectWorkspace, type SourceTrustMetadata, @@ -248,11 +250,30 @@ import { observeCrossIssueInfluence, type CrossIssueInfluenceKind, } from "../services/cross-issue-influence-limit.js"; +import { + getNativeSessionSteeringState, + NativeSessionSteeringError, + steerNativeSession, +} from "../services/native-runtime/native-session-executor.js"; const MAX_ISSUE_COMMENT_LIMIT = 500; const updateIssueRouteSchema = updateIssueSchema.extend({ interrupt: z.boolean().optional(), }); +const queuedCommentTargetSchema = z.object({ + targetRunId: z.string().min(1), + revision: z.string().min(1), +}); +const editQueuedCommentSchema = queuedCommentTargetSchema.extend({ + body: z + .string() + .min(1) + .max(200_000) + .refine((value) => value.trim().length > 0, "Queued message cannot be empty"), +}); +const reorderQueuedCommentsSchema = queuedCommentTargetSchema.extend({ + orderedCommentIds: z.array(z.string().min(1)).max(MAX_ISSUE_COMMENT_LIMIT), +}); function prefersMinimalIssueUpdateResponse(req: Request) { return (req.get("Prefer") ?? "") @@ -5192,6 +5213,180 @@ export function issueRoutes( return runToInterrupt?.status === "running" ? runToInterrupt : null; } + type IssueQueueDb = Db | Parameters[0]>[0]; + type DeferredIssueWake = typeof agentWakeupRequests.$inferSelect; + const QUEUE_CONTEXT_KEY = "_paperclipWakeContext"; + const QUEUE_IDS_KEY = "wakeCommentIds"; + + function queuedCommentIdsFromWake(wake: DeferredIssueWake | null): string[] { + if (!wake) return []; + const context = readObject(readObject(wake.payload)[QUEUE_CONTEXT_KEY]); + const ids = Array.isArray(context[QUEUE_IDS_KEY]) ? context[QUEUE_IDS_KEY] : []; + const seen = new Set(); + return ids.flatMap((value) => { + if (typeof value !== "string" || !value || seen.has(value)) return []; + seen.add(value); + return [value]; + }); + } + + function queueRevision(input: { + runId: string | null; + wake: DeferredIssueWake | null; + comments: Array<{ id: string; updatedAt: Date }>; + }): string { + return createHash("sha256") + .update(JSON.stringify({ + runId: input.runId, + wakeId: input.wake?.id ?? null, + wakeUpdatedAt: input.wake?.updatedAt?.toISOString() ?? null, + comments: input.comments.map((comment) => [comment.id, comment.updatedAt.toISOString()]), + })) + .digest("hex") + .slice(0, 32); + } + + async function findDeferredIssueWake( + executor: IssueQueueDb, + issue: { id: string; companyId: string; assigneeAgentId: string | null }, + lock = false, + ): Promise { + if (!issue.assigneeAgentId) return null; + const query = executor + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, issue.companyId), + eq(agentWakeupRequests.agentId, issue.assigneeAgentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + )) + .orderBy(asc(agentWakeupRequests.requestedAt)); + const rows = lock ? await query.for("update") : await query; + return rows.find((row) => readObject(row.payload).issueId === issue.id) ?? null; + } + + async function queueCommentsForWake(executor: IssueQueueDb, issueId: string, wake: DeferredIssueWake | null) { + const ids = queuedCommentIdsFromWake(wake); + if (ids.length === 0) return []; + const rows = await executor + .select() + .from(issueComments) + .where(and(eq(issueComments.issueId, issueId), inArray(issueComments.id, ids))); + const byId = new Map(rows.map((row) => [row.id, row])); + return ids.flatMap((id) => { + const row = byId.get(id); + return row && !row.deletedAt ? [row] : []; + }); + } + + async function buildQueuedCommentQueue(input: { + executor: IssueQueueDb; + issue: { id: string; companyId: string; assigneeAgentId: string | null }; + activeRun: Awaited>; + actor: ReturnType; + wake?: DeferredIssueWake | null; + steeringDisposition?: IssueQueuedCommentQueue["steeringDisposition"]; + }): Promise { + const wake = input.wake === undefined + ? await findDeferredIssueWake(input.executor, input.issue) + : input.wake; + const comments = await queueCommentsForWake(input.executor, input.issue.id, wake); + const protocol = input.activeRun?.runtimeMode === "native" + ? "paperclip_runner_v1" as const + : "legacy" as const; + let steeringDisposition = input.steeringDisposition + ?? (protocol === "paperclip_runner_v1" && input.activeRun + ? await getNativeSessionSteeringState(input.activeRun.id) + .then((state) => state.disposition) + .catch(() => "temporarily_unavailable" as const) + : "unsupported" as const); + if (!input.activeRun || comments.length === 0) steeringDisposition = "temporarily_unavailable"; + return { + issueId: input.issue.id, + targetRunId: input.activeRun?.id ?? null, + revision: queueRevision({ runId: input.activeRun?.id ?? null, wake, comments }), + protocol, + steeringDisposition, + entries: comments.map((comment, position) => ({ + comment: comment as IssueQueuedCommentQueue["entries"][number]["comment"], + position, + canEdit: input.actor.actorType === "user" && comment.authorUserId === input.actor.actorId, + canDiscard: input.actor.actorType === "user" && comment.authorUserId === input.actor.actorId, + })), + }; + } + + function updateWakePayloadCommentIds(wake: DeferredIssueWake, ids: string[]) { + const payload = { ...readObject(wake.payload) }; + const context = { ...readObject(payload[QUEUE_CONTEXT_KEY]) }; + if (ids.length > 0) { + context[QUEUE_IDS_KEY] = ids; + context.wakeCommentId = ids[ids.length - 1]; + context.commentId = ids[ids.length - 1]; + payload.commentId = ids[ids.length - 1]; + } else { + delete context[QUEUE_IDS_KEY]; + delete context.wakeCommentId; + delete context.commentId; + delete payload.commentId; + } + payload[QUEUE_CONTEXT_KEY] = context; + return payload; + } + + function assertQueueMutationTarget(input: { + queue: IssueQueuedCommentQueue; + targetRunId: string; + revision: string; + }) { + if (input.queue.targetRunId !== input.targetRunId) { + throw conflict("The queued message targets a stale run", { code: "queued_comment_stale_target" }); + } + if (input.queue.revision !== input.revision) { + throw conflict("The queued messages changed in another session", { code: "queued_comment_revision_conflict" }); + } + } + + async function lockQueuedCommentState(input: { + tx: Parameters[0]>[0]; + issue: { id: string; companyId: string; assigneeAgentId: string | null }; + actor: ReturnType; + targetRunId: string; + }) { + await input.tx + .select({ id: issueRows.id }) + .from(issueRows) + .where(and(eq(issueRows.id, input.issue.id), eq(issueRows.companyId, input.issue.companyId))) + .for("update"); + const activeRun = await input.tx + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.id, input.targetRunId), + eq(heartbeatRuns.companyId, input.issue.companyId), + eq(heartbeatRuns.status, "running"), + )) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + const runContext = readObject(activeRun?.contextSnapshot); + if (!activeRun || (runContext.issueId !== input.issue.id && runContext.taskId !== input.issue.id)) { + throw conflict("The queued message targets a stale run", { code: "queued_comment_stale_target" }); + } + const wake = await findDeferredIssueWake(input.tx, input.issue, true); + const queue = await buildQueuedCommentQueue({ + executor: input.tx, + issue: input.issue, + activeRun, + actor: input.actor, + wake, + steeringDisposition: activeRun.runtimeMode === "native" + ? "temporarily_unavailable" + : "unsupported", + }); + return { activeRun, wake, queue }; + } + function operatorInterruptCancelOptions(input: { issueId: string; actor: ReturnType }) { return { errorCode: "operator_interrupted", @@ -10841,6 +11036,300 @@ export function issueRoutes( res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, comments)); }); + router.get("/issues/:id/queued-comments", async (req, res) => { + const id = req.params.id as string; + const issue = await getAccessibleResource(req, res, getIssueById(req, id), "Issue not found"); + if (!issue) return; + if (!(await assertIssueReadAllowed(req, res, issue))) return; + const queue = await buildQueuedCommentQueue({ + executor: db, + issue, + activeRun: await resolveActiveIssueRun(issue), + actor: getActorInfo(req), + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }); + + router.patch( + "/issues/:id/queued-comments/:commentId", + validate(editQueuedCommentSchema), + async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const id = req.params.id as string; + const commentId = req.params.commentId as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + const actor = getActorInfo(req); + const queue = await db.transaction(async (tx) => { + const locked = await lockQueuedCommentState({ + tx, + issue, + actor, + targetRunId: req.body.targetRunId, + }); + assertQueueMutationTarget({ + queue: locked.queue, + targetRunId: req.body.targetRunId, + revision: req.body.revision, + }); + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === commentId); + if (!entry) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + if (!entry.canEdit) throw forbidden("Only the queued message author can edit it"); + const updatedAt = new Date(); + const updated = await tx + .update(issueComments) + .set({ body: req.body.body, updatedAt }) + .where(and(eq(issueComments.id, commentId), eq(issueComments.issueId, issue.id))) + .returning({ id: issueComments.id }) + .then((rows) => rows[0] ?? null); + if (!updated) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + await tx.update(issueRows).set({ updatedAt }).where(eq(issueRows.id, issue.id)); + await issueReferencesSvc.syncComment(commentId, tx); + await externalObjectsSvc.syncCommentSafely(commentId, tx); + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + wake: locked.wake, + }); + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }, + ); + + router.put( + "/issues/:id/queued-comments/order", + validate(reorderQueuedCommentsSchema), + async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const id = req.params.id as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + const actor = getActorInfo(req); + const queue = await db.transaction(async (tx) => { + const locked = await lockQueuedCommentState({ tx, issue, actor, targetRunId: req.body.targetRunId }); + assertQueueMutationTarget({ + queue: locked.queue, + targetRunId: req.body.targetRunId, + revision: req.body.revision, + }); + if (!locked.wake) throw conflict("The queued messages are no longer pending", { code: "queued_comment_not_pending" }); + const currentIds = locked.queue.entries.map((entry) => entry.comment.id); + const orderedIds = req.body.orderedCommentIds as string[]; + const orderedSet = new Set(orderedIds); + if ( + orderedSet.size !== orderedIds.length + || orderedIds.length !== currentIds.length + || currentIds.some((commentId) => !orderedSet.has(commentId)) + ) { + throw conflict("The queued message order does not match the current queue", { + code: "queued_comment_order_mismatch", + }); + } + const updatedWake = await tx + .update(agentWakeupRequests) + .set({ + payload: updateWakePayloadCommentIds(locked.wake, orderedIds), + updatedAt: new Date(), + }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then((rows) => rows[0] ?? locked.wake); + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + wake: updatedWake, + }); + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }, + ); + + router.post( + "/issues/:id/queued-comments/:commentId/steer", + validate(queuedCommentTargetSchema), + async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const id = req.params.id as string; + const commentId = req.params.commentId as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + const actor = getActorInfo(req); + let acknowledgedTurnId: string | null = null; + let duplicate = false; + let queue: IssueQueuedCommentQueue; + try { + queue = await db.transaction(async (tx) => { + const locked = await lockQueuedCommentState({ tx, issue, actor, targetRunId: req.body.targetRunId }); + const runResult = readObject(locked.activeRun.resultJson); + const acknowledgements = readObject(runResult.queuedSteeringAcknowledgements); + const priorAcknowledgement = readObject(acknowledgements[commentId]); + if (priorAcknowledgement.status === "acknowledged") { + duplicate = true; + acknowledgedTurnId = typeof priorAcknowledgement.turnId === "string" + ? priorAcknowledgement.turnId + : null; + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + wake: locked.wake, + }); + } + assertQueueMutationTarget({ + queue: locked.queue, + targetRunId: req.body.targetRunId, + revision: req.body.revision, + }); + if (locked.queue.protocol !== "paperclip_runner_v1") { + throw conflict("This runner does not support same-turn steering", { code: "steering_unsupported" }); + } + if (!locked.wake) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === commentId); + if (!entry) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + + const acknowledgement = await steerNativeSession({ + runId: locked.activeRun.id, + message: entry.comment.body, + correlationId: commentId, + }); + acknowledgedTurnId = acknowledgement.turnId; + const remainingIds = locked.queue.entries + .map((candidate) => candidate.comment.id) + .filter((candidateId) => candidateId !== commentId); + const now = new Date(); + const nextWake = remainingIds.length === 0 + ? await tx + .update(agentWakeupRequests) + .set({ status: "cancelled", finishedAt: now, updatedAt: now }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then(() => null) + : await tx + .update(agentWakeupRequests) + .set({ payload: updateWakePayloadCommentIds(locked.wake, remainingIds), updatedAt: now }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then((rows) => rows[0] ?? locked.wake); + await tx + .update(heartbeatRuns) + .set({ + resultJson: { + ...runResult, + queuedSteeringAcknowledgements: { + ...acknowledgements, + [commentId]: { + status: "acknowledged", + turnId: acknowledgement.turnId, + acknowledgedAt: now.toISOString(), + }, + }, + }, + updatedAt: now, + }) + .where(eq(heartbeatRuns.id, locked.activeRun.id)); + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: { ...locked.activeRun, resultJson: runResult }, + actor, + wake: nextWake, + }); + }); + } catch (error) { + if (error instanceof NativeSessionSteeringError) { + throw conflict(error.message, { code: error.code, retryable: true }); + } + throw error; + } + await logActivity(db, { + companyId: issue.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + agentApiKeyId: actor.agentApiKeyId, + action: "issue.queued_comment_steered", + entityType: "issue", + entityId: issue.id, + details: { + commentId, + targetRunId: req.body.targetRunId, + turnId: acknowledgedTurnId, + duplicate, + }, + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }, + ); + + router.delete( + "/issues/:id/queued-comments/:commentId", + validate(queuedCommentTargetSchema), + async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const id = req.params.id as string; + const commentId = req.params.commentId as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + const actor = getActorInfo(req); + const queue = await db.transaction(async (tx) => { + const locked = await lockQueuedCommentState({ tx, issue, actor, targetRunId: req.body.targetRunId }); + assertQueueMutationTarget({ + queue: locked.queue, + targetRunId: req.body.targetRunId, + revision: req.body.revision, + }); + if (!locked.wake) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === commentId); + if (!entry) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + if (!entry.canDiscard) throw forbidden("Only the queued message author can discard it"); + const deleted = await tx + .delete(issueComments) + .where(and(eq(issueComments.id, commentId), eq(issueComments.issueId, issue.id))) + .returning({ id: issueComments.id }) + .then((rows) => rows[0] ?? null); + if (!deleted) throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending" }); + await issueReferencesSvc.deleteCommentSource(commentId, tx); + await externalObjectsSvc.syncCommentSafely(commentId, tx); + const remainingIds = locked.queue.entries + .map((candidate) => candidate.comment.id) + .filter((candidateId) => candidateId !== commentId); + const now = new Date(); + await tx.update(issueRows).set({ updatedAt: now }).where(eq(issueRows.id, issue.id)); + const nextWake = remainingIds.length === 0 + ? await tx + .update(agentWakeupRequests) + .set({ status: "cancelled", finishedAt: now, updatedAt: now }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then(() => null) + : await tx + .update(agentWakeupRequests) + .set({ payload: updateWakePayloadCommentIds(locked.wake, remainingIds), updatedAt: now }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then((rows) => rows[0] ?? locked.wake); + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + wake: nextWake, + }); + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }, + ); + router.get("/issues/:id/interactions", async (req, res) => { const id = req.params.id as string; const issue = await getAccessibleResource(req, res, getIssueById(req, id), "Issue not found"); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e1a0b810f6c..cffa43421cd 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -31,10 +31,13 @@ import { buildNativeProviderEnvironment, cancelNativeSession, executePaperclipNativeSession, + getNativeSessionSteeringState, + NativeSessionSteeringError, nativeSessionFailureDisposition, nativeSessionFailureSourceCode, providerPlanMarkdown, semanticProviderPlanMarkdown, + steerNativeSession, } from "./native-session-executor.js"; describe("native provider bootstrap environment", () => { @@ -267,6 +270,116 @@ describe("native session cancellation", () => { }); }); +describe("native session same-turn steering", () => { + const capabilities = vi.fn(); + const snapshot = vi.fn(); + const steer = vi.fn(); + + beforeEach(() => { + state.release = null; + capabilities.mockReset().mockResolvedValue({ steering: true }); + snapshot.mockReset().mockResolvedValue({ activeTurnId: "provider-turn-1" }); + steer.mockReset().mockResolvedValue(undefined); + state.execute.mockReset().mockImplementation(async (options) => { + options.onSession?.({ capabilities, snapshot, steer, cancel: vi.fn() }); + await new Promise((resolve) => { state.release = resolve; }); + options.onSession?.(null); + return { + result: { summary: "completed" }, + terminal: { runTerminalState: "succeeded" }, + turnId: "provider-turn-1", + normalizedSessionId: "session", + providerSessionId: null, + driverKind: "test", + driverVersion: "1", + nativeEventCount: 1, + highestContiguousSourceSeq: 1, + }; + }); + }); + + async function startActiveSession() { + const running = executePaperclipNativeSession({ + db: leaseDb(), + execution, + runnerInstanceId: "runner", + }); + await vi.waitFor(() => expect(state.release).toBeTypeOf("function")); + return { running }; + } + + it("correlates the queued comment with the active provider turn acknowledgement", async () => { + const { running } = await startActiveSession(); + + await expect(getNativeSessionSteeringState(execution.binding.runId)).resolves.toEqual({ + disposition: "available", + activeTurnId: "provider-turn-1", + }); + await expect(steerNativeSession({ + runId: execution.binding.runId, + message: "Check mobile overflow first.", + correlationId: "queued-comment-1", + })).resolves.toEqual({ turnId: "provider-turn-1" }); + expect(steer).toHaveBeenCalledWith({ + turnId: "provider-turn-1", + message: { role: "user", text: "Check mobile overflow first." }, + correlationId: "queued-comment-1", + }); + + state.release?.(); + await running; + }); + + it.each([ + { + label: "unsupported provider", + prepare: () => capabilities.mockResolvedValue({ steering: false }), + code: "steering_unsupported", + }, + { + label: "stale turn", + prepare: () => snapshot.mockResolvedValue({ activeTurnId: null }), + code: "steering_stale_turn", + }, + { + label: "provider rejection", + prepare: () => steer.mockRejectedValue(new Error("request rejected")), + code: "steering_rejected", + }, + ])("keeps $label retryable with a stable code", async ({ prepare, code }) => { + prepare(); + const { running } = await startActiveSession(); + + const error = await steerNativeSession({ + runId: execution.binding.runId, + message: "Retryable steering", + correlationId: "queued-comment-error", + }).catch((value) => value); + expect(error).toBeInstanceOf(NativeSessionSteeringError); + expect(error.code).toBe(code); + + state.release?.(); + await running; + }); + + it("bounds the provider acknowledgement wait", async () => { + steer.mockReturnValue(new Promise(() => undefined)); + const { running } = await startActiveSession(); + + const error = await steerNativeSession({ + runId: execution.binding.runId, + message: "Do not wait forever", + correlationId: "queued-comment-timeout", + timeoutMs: 5, + }).catch((value) => value); + expect(error).toBeInstanceOf(NativeSessionSteeringError); + expect(error.code).toBe("steering_timeout"); + + state.release?.(); + await running; + }); +}); + describe("native warm session supervision", () => { it("reuses one session across distinct governed runs and closes it after idle expiry", async () => { const close = vi.fn(async () => undefined); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 73d45962f82..8266cecb978 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -509,6 +509,101 @@ function record(value: unknown): Record { : {}; } +export type NativeSessionSteeringState = { + disposition: "available" | "unsupported" | "temporarily_unavailable"; + activeTurnId: string | null; +}; + +export class NativeSessionSteeringError extends Error { + constructor( + readonly code: + | "steering_unsupported" + | "steering_temporarily_unavailable" + | "steering_stale_turn" + | "steering_timeout" + | "steering_rejected", + message: string, + ) { + super(message); + this.name = "NativeSessionSteeringError"; + } +} + +export async function getNativeSessionSteeringState(runId: string): Promise { + const active = activeNativeSessions.get(runId); + if (!active) return { disposition: "temporarily_unavailable", activeTurnId: null }; + const capabilities = await active.session.capabilities(); + if (!capabilities.steering || !active.session.steer) { + return { disposition: "unsupported", activeTurnId: null }; + } + const snapshot = await active.session.snapshot(); + return { + disposition: snapshot.activeTurnId ? "available" : "temporarily_unavailable", + activeTurnId: snapshot.activeTurnId ?? null, + }; +} + +/** Dispatches a true same-turn steering message and resolves only after ack. */ +export async function steerNativeSession(input: { + runId: string; + message: string; + correlationId: string; + timeoutMs?: number; +}): Promise<{ turnId: string }> { + const active = activeNativeSessions.get(input.runId); + if (!active) { + throw new NativeSessionSteeringError( + "steering_temporarily_unavailable", + "The active native session is not attached.", + ); + } + const capabilities = await active.session.capabilities(); + if (!capabilities.steering || !active.session.steer) { + throw new NativeSessionSteeringError( + "steering_unsupported", + "This provider does not support same-turn steering.", + ); + } + const snapshot = await active.session.snapshot(); + const turnId = snapshot.activeTurnId ?? null; + if (!turnId) { + throw new NativeSessionSteeringError( + "steering_stale_turn", + "The target turn is no longer active.", + ); + } + + let timeout: ReturnType | null = null; + try { + await Promise.race([ + active.session.steer({ + turnId, + message: { role: "user", text: input.message }, + correlationId: input.correlationId, + }), + new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new NativeSessionSteeringError( + "steering_timeout", + "The provider did not acknowledge steering in time.", + )), input.timeoutMs ?? 10_000); + }), + ]); + return { turnId }; + } catch (error) { + if (error instanceof NativeSessionSteeringError) throw error; + const message = error instanceof Error ? error.message : String(error); + if (/stale|terminal|active turn/i.test(message)) { + throw new NativeSessionSteeringError("steering_stale_turn", "The target turn is no longer active."); + } + if (/unsupported|unavailable|capability/i.test(message)) { + throw new NativeSessionSteeringError("steering_unsupported", "This provider does not support same-turn steering."); + } + throw new NativeSessionSteeringError("steering_rejected", "The provider rejected the steering message."); + } finally { + if (timeout) clearTimeout(timeout); + } +} + export function cancelNativeSession(runId: string, reason: string): Promise; export function cancelNativeSession( runId: string, diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index 5eae6f0685e..07a832fca6d 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -13,6 +13,7 @@ import type { IssueAttachment, IssueCostSummary, IssueComment, + IssueQueuedCommentQueue, IssueDocument, IssueLabel, IssueRecoveryAction, @@ -237,6 +238,39 @@ export const issuesApi = { const qs = params.toString(); return api.get(`/issues/${id}/comments${qs ? `?${qs}` : ""}`); }, + getQueuedComments: (id: string) => + api.get(`/issues/${id}/queued-comments`), + editQueuedComment: ( + id: string, + commentId: string, + data: { body: string; targetRunId: string; revision: string }, + ) => + api.patch( + `/issues/${id}/queued-comments/${commentId}`, + data, + ), + reorderQueuedComments: ( + id: string, + data: { orderedCommentIds: string[]; targetRunId: string; revision: string }, + ) => api.put(`/issues/${id}/queued-comments/order`, data), + steerQueuedComment: ( + id: string, + commentId: string, + data: { targetRunId: string; revision: string }, + ) => + api.post( + `/issues/${id}/queued-comments/${commentId}/steer`, + data, + ), + discardQueuedComment: ( + id: string, + commentId: string, + data: { targetRunId: string; revision: string }, + ) => + api.delete( + `/issues/${id}/queued-comments/${commentId}`, + data, + ), listInteractions: (id: string) => api.get(`/issues/${id}/interactions`), listAcceptedPlanDecompositions: (id: string) => diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 60ce07e3962..1312664d7a4 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -39,6 +39,7 @@ import type { IssueDocumentSummary, IssueBlockerAttention, IssueRecoveryAction, + IssueQueuedCommentQueue, IssueRelationIssueSummary, IssueScheduledRetry, SuccessfulRunHandoffState, @@ -537,6 +538,12 @@ interface IssueChatThreadProps { includeSucceededRunsWithoutOutput?: boolean; onInterruptQueued?: (runId: string) => Promise; onCancelQueued?: (commentId: string) => void; + /** Authoritative PRP queue. The classic thread intentionally ignores it. */ + queuedCommentQueue?: IssueQueuedCommentQueue | null; + onEditQueuedComment?: (commentId: string, body: string, revision: string) => Promise; + onReorderQueuedComments?: (orderedCommentIds: string[], revision: string) => Promise; + onSteerQueuedComment?: (commentId: string, revision: string) => Promise; + onDiscardQueuedComment?: (commentId: string, revision: string) => Promise; onDeleteComment?: (commentId: string) => Promise | void; interruptingQueuedRunId?: string | null; stoppingRunId?: string | null; diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index e4909b6af9a..fb621c1bf8a 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -1,11 +1,12 @@ // @vitest-environment jsdom import type { ReactElement } from "react"; -import { forwardRef, useImperativeHandle, type ForwardedRef } from "react"; +import { act, forwardRef, useImperativeHandle, type ForwardedRef } from "react"; import { flushSync } from "react-dom"; import { createRoot, type Root } from "react-dom/client"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ThemeProvider } from "@/context/ThemeContext"; +import type { IssueQueuedCommentQueue } from "@paperclipai/shared"; import { TaskChatThread } from "./TaskChatThread"; const transcriptState = vi.hoisted(() => ({ transcriptByRun: new Map() })); @@ -560,6 +561,62 @@ describe("TaskChatThread queued message actions", () => { }); }); +describe("TaskChatThread Paperclip Runner queue", () => { + const queuedComment = { + id: "queued-prp-1", + companyId: "company-1", + issueId: "issue-1", + authorType: "user" as const, + authorAgentId: null, + authorUserId: "user-1", + body: "Render this queued message exactly once.", + presentation: null, + metadata: null, + createdAt: new Date("2026-08-22T15:00:00.000Z"), + updatedAt: new Date("2026-08-22T15:00:00.000Z"), + }; + const queue: IssueQueuedCommentQueue = { + issueId: "issue-1", + targetRunId: "run-1", + revision: "revision-1", + protocol: "paperclip_runner_v1", + steeringDisposition: "available", + entries: [{ comment: queuedComment, position: 0, canEdit: true, canDiscard: true }], + }; + + function occurrenceCount(text: string) { + return container.textContent?.split(text).length! - 1; + } + + it("suppresses the transcript echo until the queued entry is consumed", async () => { + const props = { + comments: [queuedComment], + onAdd: async () => {}, + queuedCommentQueue: queue, + onEditQueuedComment: async () => {}, + onReorderQueuedComments: async () => {}, + onSteerQueuedComment: async () => {}, + onDiscardQueuedComment: async () => {}, + }; + render(); + + const stack = container.querySelector('[data-testid="task-chat-composer-stack"]'); + const queuePane = container.querySelector('[data-testid="task-chat-queued-messages"]'); + expect(queuePane?.parentElement).toBe(stack); + expect(stack?.classList).not.toContain("gap-2"); + expect(container.querySelector('[data-testid="task-chat-queued-message-queued-prp-1"]')).not.toBeNull(); + expect(occurrenceCount(queuedComment.body)).toBe(1); + + await act(async () => { + render(); + await Promise.resolve(); + }); + + expect(container.querySelector('[data-testid="task-chat-queued-message-queued-prp-1"]')).toBeNull(); + expect(occurrenceCount(queuedComment.body)).toBe(1); + }); +}); + describe("TaskChatThread mobile composer dock (PAP-495)", () => { it("pins the composer to the nav-aware bottom offset so its action row clears the auto-hiding bottom nav", () => { sidebarState.isMobile = true; diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index a56d44bb7d7..3e50447593e 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -42,6 +42,7 @@ import { TaskChatBubbleActions } from "@/components/task-chat/TaskChatBubbleActi import type { FeedbackVoteValue } from "@paperclipai/shared"; import { TaskChatThreadView, taskChatContentKey } from "@/components/task-chat/TaskChatThreadView"; import { TaskChatComposer } from "@/components/task-chat/TaskChatComposer"; +import { TaskChatQueuedMessages } from "@/components/task-chat/TaskChatQueuedMessages"; import { useWindowAutoFollow } from "@/components/task-chat/useWindowAutoFollow"; import { useSidebar } from "@/context/SidebarContext"; import { cn } from "@/lib/utils"; @@ -183,11 +184,118 @@ export function TaskChatThread(props: TaskChatThreadProps) { draftKey, onInterruptQueued, interruptingQueuedRunId, + queuedCommentQueue, + onEditQueuedComment, + onReorderQueuedComments, + onSteerQueuedComment, + onDiscardQueuedComment, blockedBy = [], blockerAttention, liveIssueIds, } = props; + const paperclipQueue = + queuedCommentQueue?.protocol === "paperclip_runner_v1" + ? queuedCommentQueue + : null; + const queuedCommentIds = useMemo( + () => + new Set(paperclipQueue?.entries.map((entry) => entry.comment.id) ?? []), + [paperclipQueue], + ); + const [queuedEdit, setQueuedEdit] = useState<{ + commentId: string; + body: string; + revision: string; + stale?: boolean; + } | null>(null); + + const beginQueuedEdit = useCallback( + (commentId: string) => { + const entry = paperclipQueue?.entries.find( + (candidate) => candidate.comment.id === commentId, + ); + if (!entry?.canEdit || !paperclipQueue) return; + setQueuedEdit({ + commentId, + body: entry.comment.body, + revision: paperclipQueue.revision, + }); + }, + [paperclipQueue], + ); + + const saveQueuedEdit = useCallback( + async (commentId: string, body: string) => { + if (!queuedEdit || queuedEdit.commentId !== commentId) { + throw new Error("This queued message is no longer editable."); + } + if (queuedEdit.stale) { + await onAdd(body); + return; + } + if (!onEditQueuedComment) + throw new Error("This queued message is no longer editable."); + try { + await onEditQueuedComment(commentId, body, queuedEdit.revision); + } catch (error) { + const payload = + typeof error === "object" && error !== null + ? (error as { status?: unknown; body?: unknown; message?: unknown }) + : null; + const responseBody = + typeof payload?.body === "object" && payload.body !== null + ? (payload.body as { code?: unknown; details?: unknown }) + : null; + const details = + typeof responseBody?.details === "object" && + responseBody.details !== null + ? (responseBody.details as { code?: unknown }) + : null; + const code = + typeof details?.code === "string" + ? details.code + : typeof responseBody?.code === "string" + ? responseBody.code + : null; + const stale = + payload?.status === 409 && + (code === "queued_comment_stale_target" || + code === "queued_comment_not_pending"); + if (stale) { + setQueuedEdit((current) => + current?.commentId === commentId + ? { ...current, stale: true } + : current, + ); + } + throw error; + } + }, + [onAdd, onEditQueuedComment, queuedEdit], + ); + + useEffect(() => { + if (!queuedEdit || queuedEdit.stale) return; + const targetStillQueued = paperclipQueue?.entries.some( + (entry) => entry.comment.id === queuedEdit.commentId, + ); + if (!targetStillQueued) { + setQueuedEdit((current) => + current ? { ...current, stale: true } : current, + ); + } else if ( + paperclipQueue && + queuedEdit.revision !== paperclipQueue.revision + ) { + // A concurrent reorder/edit refreshes the optimistic-lock token without + // replacing the Markdown currently in the editor. + setQueuedEdit((current) => + current ? { ...current, revision: paperclipQueue.revision } : current, + ); + } + }, [paperclipQueue, queuedEdit]); + const liveWorkLinks = useMemo( () => issueStatus === "blocked" && blockerAttention?.state === "covered" ? resolveTaskChatLiveWork(blockedBy, liveIssueIds ?? EMPTY_LIVE_ISSUE_IDS, blockerAttention.terminalBlocker) @@ -372,6 +480,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { // two lists stay index-aligned. const visibleComments = comments.filter((comment) => !comment.deletedAt); visibleComments.forEach((comment, index) => { + if (queuedCommentIds.has(comment.id)) return; // The live/settling runner lane already renders this run's final answer. // Keep a just-persisted duplicate out of the backbone until the complete // canonical bubble + Worked header can replace that lane atomically. @@ -499,7 +608,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { return entries.sort( (a, b) => a.ms - b.ms || a.order - b.order || a.id.localeCompare(b.id), ); - }, [comments, commentItems, interactions, documents, workProducts, attachments, timelineEvents, linkedRuns, liveRuns, planDocument, heldPaperclipRunnerRunId, heldPaperclipRunnerStartedAtMs, heldPaperclipRunnerFinalText]); + }, [comments, commentItems, interactions, documents, workProducts, attachments, timelineEvents, linkedRuns, liveRuns, planDocument, heldPaperclipRunnerRunId, heldPaperclipRunnerStartedAtMs, heldPaperclipRunnerFinalText, queuedCommentIds]); // Boolean gate (stable across the host's per-render brief objects) so the // heavy assembly memo doesn't recompute on every parent render. @@ -851,6 +960,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { const renderQueuedAction = useCallback( (item: TaskChatMessageItem) => { + if (paperclipQueue) return null; const runId = item.queueTargetRunId; if (item.optimistic !== "queued" || !runId || !onInterruptQueued) return null; @@ -867,7 +977,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { ); }, - [interruptingQueuedRunId, onInterruptQueued], + [interruptingQueuedRunId, onInterruptQueued, paperclipQueue], ); const renderInteraction = useCallback( @@ -1009,22 +1119,54 @@ export function TaskChatThread(props: TaskChatThreadProps) { )} > {composerAccessory} - +
+ {paperclipQueue && paperclipQueue.entries.length > 0 ? ( + { + if (!onReorderQueuedComments) + throw new Error("Queue reordering is unavailable."); + await onReorderQueuedComments(orderedCommentIds, revision); + }} + onSteer={async (commentId, revision) => { + if (!onSteerQueuedComment) + throw new Error("Steering is unavailable."); + await onSteerQueuedComment(commentId, revision); + }} + onDiscard={async (commentId, revision) => { + if (!onDiscardQueuedComment) + throw new Error("Discard is unavailable."); + await onDiscardQueuedComment(commentId, revision); + if (queuedEdit?.commentId === commentId) setQueuedEdit(null); + }} + /> + ) : null} +
+ setQueuedEdit(null)} + /> +
+
{footer} ) : null} diff --git a/ui/src/components/task-chat/TaskChatComposer.test.tsx b/ui/src/components/task-chat/TaskChatComposer.test.tsx index dd4b0a04487..3cfb471b65b 100644 --- a/ui/src/components/task-chat/TaskChatComposer.test.tsx +++ b/ui/src/components/task-chat/TaskChatComposer.test.tsx @@ -1,6 +1,6 @@ // @vitest-environment jsdom -import { StrictMode, type ReactElement } from "react"; +import { StrictMode, useState, type ReactElement } from "react"; import { flushSync } from "react-dom"; import { createRoot, type Root } from "react-dom/client"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -756,4 +756,98 @@ describe("TaskChatComposer", () => { } }); }); + + describe("queued message editing", () => { + const draftKey = "task-chat-draft:queued-edit"; + + function Harness({ + onSave, + stale = false, + }: { + onSave: (commentId: string, body: string) => Promise; + stale?: boolean; + }) { + const [queuedEdit, setQueuedEdit] = useState<{ + commentId: string; + body: string; + stale?: boolean; + } | null>({ + commentId: "queued-1", + body: "Complete queued markdown", + stale, + }); + return ( + setQueuedEdit(null)} + /> + ); + } + + it("restores the existing composer draft after cancelling an edit", async () => { + localStorage.setItem(draftKey, "Unsent normal draft"); + render(); + await flushAsync(); + expect(editable().textContent).toBe("Complete queued markdown"); + expect(localStorage.getItem(draftKey)).toBe("Unsent normal draft"); + + const cancel = Array.from(container.querySelectorAll("button")) + .find((button) => button.textContent?.trim() === "Cancel"); + flushSync(() => cancel?.click()); + await flushAsync(); + + expect(editable().textContent).toBe("Unsent normal draft"); + expect(localStorage.getItem(draftKey)).toBe("Unsent normal draft"); + }); + + it("saves through the queue callback without posting a new comment", async () => { + localStorage.setItem(draftKey, "Normal draft stays here"); + const onSave = vi.fn().mockResolvedValue(undefined); + render(); + await flushAsync(); + typeText("Edited queued markdown"); + + pressKey("Enter", { metaKey: true }); + await flushAsync(); + await flushAsync(); + + expect(onSave).toHaveBeenCalledWith("queued-1", "Edited queued markdown"); + expect(editable().textContent).toBe("Normal draft stays here"); + }); + + it("retains edited text when the queue target rejects the save", async () => { + const onSave = vi.fn().mockRejectedValue(new Error("queued_comment_stale_target")); + render(); + await flushAsync(); + typeText("Keep this replacement text"); + + pressKey("Enter", { metaKey: true }); + await flushAsync(); + + expect(editable().textContent).toBe("Keep this replacement text"); + expect(container.textContent).toContain("Editing queued message"); + }); + + it("offers a stale edit as a new queued message and preserves its Markdown source", async () => { + const onSave = vi.fn().mockResolvedValue(undefined); + render(); + await flushAsync(); + + typeText(" replacement with trailing Markdown "); + expect(container.textContent).toContain("Queued message changed"); + expect(sendButton().getAttribute("aria-label")).toBe("Queue as new message"); + pressKey("Enter", { metaKey: true }); + await flushAsync(); + await flushAsync(); + + expect(onSave).toHaveBeenCalledWith( + "queued-1", + " replacement with trailing Markdown ", + ); + }); + }); }); diff --git a/ui/src/components/task-chat/TaskChatComposer.tsx b/ui/src/components/task-chat/TaskChatComposer.tsx index 4062b1ced6b..548e29bb616 100644 --- a/ui/src/components/task-chat/TaskChatComposer.tsx +++ b/ui/src/components/task-chat/TaskChatComposer.tsx @@ -59,6 +59,10 @@ interface TaskChatComposerProps { mobile?: boolean; /** Storage key used to restore, persist, and clear this task's text draft. */ draftKey?: string; + /** When set, the main composer temporarily edits this queued message. */ + queuedEdit?: { commentId: string; body: string; stale?: boolean } | null; + onSaveQueuedEdit?: (commentId: string, body: string) => Promise; + onCancelQueuedEdit?: () => void; } /** Per-mode hue token (see ui/src/index.css `--tc-mode-*`). */ @@ -100,6 +104,13 @@ type ComposerAttachment = { contentPath?: string; }; +type QueuedEditBackup = { + body: string; + attachments: ComposerAttachment[]; + pendingMode: IssueWorkMode; + pendingAssignee: string | null; +}; + /** Local duplicate of IssueChatThread's module-private helper (same rule). */ function shouldImplicitlyReopenComment(issueStatus: string | undefined, assigneeValue: string) { const resumesToTodo = issueStatus === "done" || issueStatus === "cancelled" || issueStatus === "blocked"; @@ -151,6 +162,9 @@ export function TaskChatComposer({ issueStatus, mobile = false, draftKey, + queuedEdit = null, + onSaveQueuedEdit, + onCancelQueuedEdit, }: TaskChatComposerProps) { const [body, setBody] = useState(() => (draftKey ? loadDraft(draftKey) : "")); const [submitting, setSubmitting] = useState(false); @@ -166,42 +180,96 @@ export function TaskChatComposer({ const bodyRef = useRef(body); bodyRef.current = body; const draftTimer = useRef | null>(null); + const queuedEditRef = useRef(queuedEdit); + queuedEditRef.current = queuedEdit; + const queuedEditBackupRef = useRef(null); + const previousQueuedEditIdRef = useRef(null); useEffect(() => { - if (!draftKey) return; + const previousId = previousQueuedEditIdRef.current; + const nextEdit = queuedEdit; + const nextId = nextEdit?.commentId ?? null; + + if (nextEdit && previousId !== nextId) { + if (!queuedEditBackupRef.current) { + queuedEditBackupRef.current = { + body: bodyRef.current, + attachments: attachmentsRef.current, + pendingMode, + pendingAssignee: pendingAssigneeRef.current, + }; + if (draftKey) saveDraft(draftKey, bodyRef.current); + } + if (draftTimer.current) { + clearTimeout(draftTimer.current); + draftTimer.current = null; + } + bodyRef.current = nextEdit.body; + setBody(nextEdit.body); + setAttachments([]); + requestAnimationFrame(() => editorRef.current?.focus()); + } else if (!nextId && previousId) { + const backup = queuedEditBackupRef.current; + if (backup) { + bodyRef.current = backup.body; + setBody(backup.body); + setAttachments(backup.attachments); + setPendingMode(backup.pendingMode); + setPendingAssignee(backup.pendingAssignee); + if (draftKey) saveDraft(draftKey, backup.body); + } + queuedEditBackupRef.current = null; + requestAnimationFrame(() => editorRef.current?.focus()); + } + + previousQueuedEditIdRef.current = nextId; + }, [draftKey, pendingMode, queuedEdit]); + + useEffect(() => { + if (!draftKey || queuedEdit) return; setBody(loadDraft(draftKey)); - }, [draftKey]); + }, [draftKey, queuedEdit]); useEffect(() => { - if (!draftKey) return; + if (!draftKey || queuedEdit) { + if (draftTimer.current) { + clearTimeout(draftTimer.current); + draftTimer.current = null; + } + return; + } if (draftTimer.current) clearTimeout(draftTimer.current); draftTimer.current = setTimeout(() => { saveDraft(draftKey, body); }, DRAFT_DEBOUNCE_MS); - }, [body, draftKey]); + }, [body, draftKey, queuedEdit]); useEffect(() => { return () => { if (draftTimer.current) clearTimeout(draftTimer.current); - if (draftKey) saveDraft(draftKey, bodyRef.current); + if (draftKey && !queuedEditRef.current) saveDraft(draftKey, bodyRef.current); }; }, [draftKey]); useEffect(() => { if (!draftKey) return; - const flushDraft = () => saveDraft(draftKey, bodyRef.current); + const flushDraft = () => { + if (!queuedEditRef.current) saveDraft(draftKey, bodyRef.current); + }; window.addEventListener("beforeunload", flushDraft); return () => window.removeEventListener("beforeunload", flushDraft); }, [draftKey]); const modeMeta = workModeMetaFor(pendingMode); - const canAcceptFiles = Boolean(onAttachImage || onImageUpload); + const canAcceptFiles = !queuedEdit && Boolean(onAttachImage || onImageUpload); const showAssignee = Boolean(enableReassign && reassignOptions && reassignOptions.length > 0); const assigneeValue = pendingAssignee ?? currentAssigneeValue; const assigneeLabel = reassignOptions?.find((o) => o.id === assigneeValue)?.label ?? "Unassigned"; const assigneeName = assigneeLabel === "Unassigned" ? "the agent" : assigneeLabel; - const effectivePlaceholder = placeholder ?? modePlaceholder(pendingMode, assigneeName); + const effectivePlaceholder = queuedEdit + ? "Edit queued message…" + : (placeholder ?? modePlaceholder(pendingMode, assigneeName)); /** Upload an image and return its URL for inline `![](src)` markdown. */ async function uploadInlineImage(file: File): Promise { @@ -331,13 +399,23 @@ export function TaskChatComposer({ const refLines = attachedRefs .map((item) => `[${escapeMarkdownLabel(item.name)}](${item.contentPath})`) .join("\n"); - const fullBody = [trimmed, refLines].filter(Boolean).join("\n\n"); + // A queued edit must preserve the complete Markdown source. Normal sends + // keep the longstanding trimmed-body behavior. + const fullBody = queuedEdit + ? submittedBody + : [trimmed, refLines].filter(Boolean).join("\n\n"); const hasReassignment = showAssignee && assigneeValue !== currentAssigneeValue; const reassignment = hasReassignment ? parseAssigneeValue(assigneeValue) : undefined; const reopen = shouldImplicitlyReopenComment(issueStatus, assigneeValue) ? true : undefined; setSubmitting(true); try { + if (queuedEdit) { + if (!onSaveQueuedEdit) return; + await onSaveQueuedEdit(queuedEdit.commentId, fullBody); + onCancelQueuedEdit?.(); + return; + } if (pendingMode !== workMode && onWorkModeChange) { await onWorkModeChange(pendingMode); } @@ -376,7 +454,7 @@ export function TaskChatComposer({ onKeyDownCapture={(e) => { // Shift+Tab cycles the pending mode; captured on the wrapper so it // wins over Lexical's list-outdent binding inside the editor. - if (disabled) return; + if (disabled || queuedEdit) return; if (e.key === "Tab" && e.shiftKey) { e.preventDefault(); e.stopPropagation(); @@ -480,6 +558,11 @@ export function TaskChatComposer({ ) : null} + {queuedEdit ? ( + + {queuedEdit.stale ? "Queued message changed" : "Editing queued message"} + + ) : ( + ) : null} + + + + + {entry.comment.body} + + + + + + + + + + + + + + Edit message + + + + + ); +} + +/** Compact production queue shown immediately above the default task composer. */ +export function TaskChatQueuedMessages({ + queue, + onEdit, + onReorder, + onSteer, + onDiscard, +}: TaskChatQueuedMessagesProps) { + const [entries, setEntries] = useState(queue.entries); + const [pending, setPending] = useState<{ commentId: string; action: Exclude } | null>(null); + const [reordering, setReordering] = useState(false); + const [announcement, setAnnouncement] = useState(""); + const [visibleError, setVisibleError] = useState(null); + const sensors = useSensors( + useSensor(PointerSensor, { activationConstraint: { distance: 4 } }), + useSensor(KeyboardSensor, { coordinateGetter: sortableKeyboardCoordinates }), + ); + + useEffect(() => { + setEntries(queue.entries); + }, [queue.entries, queue.revision]); + + const ids = useMemo(() => entries.map((entry) => entry.comment.id), [entries]); + + async function handleDragEnd(event: DragEndEvent) { + const { active, over } = event; + if (!over || active.id === over.id || reordering || pending) return; + const from = ids.indexOf(String(active.id)); + const to = ids.indexOf(String(over.id)); + if (from < 0 || to < 0) return; + + const previous = entries; + const next = arrayMove(entries, from, to).map((entry, position) => ({ ...entry, position })); + const orderedIds = next.map((entry) => entry.comment.id); + setEntries(next); + setReordering(true); + setVisibleError(null); + setAnnouncement(`Moved queued message to position ${to + 1} of ${next.length}.`); + try { + await onReorder(orderedIds, queue.revision); + } catch { + setEntries(previous); + setAnnouncement(""); + setVisibleError("Couldn’t reorder. Previous order restored."); + } finally { + setReordering(false); + } + } + + async function runRowAction(commentId: string, action: Exclude) { + if (pending || reordering) return; + setPending({ commentId, action }); + setVisibleError(null); + setAnnouncement(action === "steer" ? "Steering queued message." : "Discarding queued message."); + try { + if (action === "steer") await onSteer(commentId, queue.revision); + else await onDiscard(commentId, queue.revision); + setEntries((current) => current.filter((entry) => entry.comment.id !== commentId)); + setAnnouncement(action === "steer" ? "Message steered into the active turn." : "Queued message discarded."); + } catch { + setAnnouncement(""); + setVisibleError(action === "steer" + ? "Couldn’t steer. Message is still queued." + : "Couldn’t discard. Message is still queued."); + } finally { + setPending(null); + } + } + + if (entries.length === 0) return null; + + return ( +
+ void handleDragEnd(event)}> + + {entries.map((entry) => { + const action = pending?.commentId === entry.comment.id ? pending.action : null; + const disabled = Boolean(pending || reordering); + return ( + onEdit(entry.comment.id)} + onSteer={() => void runRowAction(entry.comment.id, "steer")} + onDiscard={() => void runRowAction(entry.comment.id, "discard")} + /> + ); + })} + + + {visibleError ? ( +
+ {visibleError} +
+ ) : null} +
+ {announcement} +
+
+ ); +} diff --git a/ui/src/lib/issue-queued-comment-queue.test.ts b/ui/src/lib/issue-queued-comment-queue.test.ts new file mode 100644 index 00000000000..42a8fa9a06d --- /dev/null +++ b/ui/src/lib/issue-queued-comment-queue.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import { normalizeIssueQueuedCommentQueue } from "./issue-queued-comment-queue"; + +describe("normalizeIssueQueuedCommentQueue", () => { + it("sorts, deduplicates, and drops malformed queue entries", () => { + const queue = normalizeIssueQueuedCommentQueue({ + issueId: "issue-1", + targetRunId: "run-1", + revision: "rev-1", + protocol: "paperclip_runner_v1", + steeringDisposition: "available", + entries: [ + { comment: { id: "second", body: "Second" }, position: 2, canEdit: true, canDiscard: true }, + { comment: { id: "first", body: "First" }, position: 0, canEdit: false, canDiscard: true }, + { comment: { id: "first", body: "Duplicate" }, position: 1, canEdit: true, canDiscard: true }, + { comment: { body: "Missing id" }, position: 3 }, + ], + }, "fallback"); + + expect(queue.entries.map((entry) => entry.comment.id)).toEqual(["first", "second"]); + expect(queue.entries.map((entry) => entry.position)).toEqual([0, 1]); + expect(queue.protocol).toBe("paperclip_runner_v1"); + expect(queue.steeringDisposition).toBe("available"); + }); + + it("fails closed for malformed protocol and steering data", () => { + const queue = normalizeIssueQueuedCommentQueue({ entries: "nope" }, "issue-fallback"); + expect(queue).toMatchObject({ + issueId: "issue-fallback", + targetRunId: null, + revision: "unavailable", + protocol: "legacy", + steeringDisposition: "unsupported", + entries: [], + }); + }); +}); diff --git a/ui/src/lib/issue-queued-comment-queue.ts b/ui/src/lib/issue-queued-comment-queue.ts new file mode 100644 index 00000000000..a352061ed38 --- /dev/null +++ b/ui/src/lib/issue-queued-comment-queue.ts @@ -0,0 +1,60 @@ +import type { + IssueComment, + IssueQueuedCommentQueue, + IssueQueuedCommentSteeringDisposition, +} from "@paperclipai/shared"; + +function record(value: unknown): Record | null { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : null; +} + +const STEERING_DISPOSITIONS = new Set([ + "available", + "unsupported", + "temporarily_unavailable", +]); + +/** Defensive boundary shared by IssueDetail and Storybook queue fixtures. */ +export function normalizeIssueQueuedCommentQueue( + value: unknown, + fallbackIssueId: string, +): IssueQueuedCommentQueue { + const source = record(value); + const seen = new Set(); + const entries = (Array.isArray(source?.entries) ? source.entries : []) + .flatMap((candidate, sourcePosition) => { + const entry = record(candidate); + const comment = record(entry?.comment); + const id = typeof comment?.id === "string" ? comment.id : ""; + const body = typeof comment?.body === "string" ? comment.body : null; + if (!id || body === null || seen.has(id)) return []; + seen.add(id); + const position = typeof entry?.position === "number" && Number.isFinite(entry.position) + ? entry.position + : sourcePosition; + return [{ + comment: comment as unknown as IssueComment, + position, + canEdit: entry?.canEdit === true, + canDiscard: entry?.canDiscard === true, + }]; + }) + .sort((left, right) => left.position - right.position) + .map((entry, position) => ({ ...entry, position })); + const disposition = source?.steeringDisposition; + + return { + issueId: typeof source?.issueId === "string" ? source.issueId : fallbackIssueId, + targetRunId: typeof source?.targetRunId === "string" ? source.targetRunId : null, + revision: typeof source?.revision === "string" ? source.revision : "unavailable", + protocol: source?.protocol === "paperclip_runner_v1" ? "paperclip_runner_v1" : "legacy", + steeringDisposition: + typeof disposition === "string" + && STEERING_DISPOSITIONS.has(disposition as IssueQueuedCommentSteeringDisposition) + ? disposition as IssueQueuedCommentSteeringDisposition + : "unsupported", + entries, + }; +} diff --git a/ui/src/lib/queryKeys.ts b/ui/src/lib/queryKeys.ts index b4680ed1899..c21dba53c58 100644 --- a/ui/src/lib/queryKeys.ts +++ b/ui/src/lib/queryKeys.ts @@ -195,6 +195,7 @@ export const queryKeys = { ["issues", companyId, "execution-workspace", executionWorkspaceId] as const, detail: (id: string) => ["issues", "detail", id] as const, comments: (issueId: string) => ["issues", "comments", issueId] as const, + queuedComments: (issueId: string) => ["issues", "queued-comments", issueId] as const, commentsList: (issueId: string) => ["issues", "comments", issueId, "list"] as const, interactions: (issueId: string) => ["issues", "interactions", issueId] as const, acceptedPlanDecompositions: (issueId: string) => diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index 7180bab5941..ceb992271a8 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -29,6 +29,7 @@ import { buildCompanyUserInlineOptions, buildCompanyUserLabelMap, buildCompanyUs import { extractIssueTimelineEvents, extractIssueWorkModeChanges } from "../lib/issue-timeline-events"; import { queryKeys } from "../lib/queryKeys"; import { keepPreviousDataForSameQueryTail } from "../lib/query-placeholder-data"; +import { normalizeIssueQueuedCommentQueue } from "../lib/issue-queued-comment-queue"; import { collectLiveIssueIds } from "../lib/liveIssueIds"; import { hasLegacyIssueDetailQuery, @@ -1142,6 +1143,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ // prop type, so no cast is needed. const { enabled: classicTaskInterfaceEnabled } = useClassicTaskInterfaceEnabled(); const ThreadComponent = classicTaskInterfaceEnabled ? IssueChatThread : TaskChatThread; + const queryClient = useQueryClient(); const { data: activity } = useQuery({ queryKey: queryKeys.issues.activity(issueId), queryFn: () => activityApi.forIssue(issueId), @@ -1166,6 +1168,27 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ () => resolveIssueActiveRun({ status: issueStatus, executionRunId }, activeRun), [activeRun, executionRunId, issueStatus], ); + const paperclipQueueEnabled = !classicTaskInterfaceEnabled + && resolvedActiveRun?.adapterType === "paperclip_runner"; + const { data: authoritativeQueuedCommentQueue } = useQuery({ + queryKey: queryKeys.issues.queuedComments(issueId), + queryFn: async () => normalizeIssueQueuedCommentQueue( + await issuesApi.getQueuedComments(issueId), + issueId, + ), + enabled: paperclipQueueEnabled, + refetchInterval: paperclipQueueEnabled ? 1000 : false, + }); + const [consumedQueuedCommentIds, setConsumedQueuedCommentIds] = useState>( + () => new Set(), + ); + const [discardedQueuedCommentIds, setDiscardedQueuedCommentIds] = useState>( + () => new Set(), + ); + useEffect(() => { + setConsumedQueuedCommentIds(new Set()); + setDiscardedQueuedCommentIds(new Set()); + }, [resolvedActiveRun?.id]); const hasLiveRuns = liveRunCount > 0 || !!resolvedActiveRun; const { data: linkedRuns } = useQuery({ queryKey: queryKeys.issues.runs(issueId), @@ -1279,6 +1302,133 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ resolvedLinkedRuns, interruptibleIssueRun, ]); + const effectiveQueuedCommentQueue = useMemo(() => { + if (!paperclipQueueEnabled || !resolvedActiveRun) return null; + const base = authoritativeQueuedCommentQueue ?? { + issueId, + targetRunId: resolvedActiveRun.id, + revision: "pending-canonical-queue", + protocol: "paperclip_runner_v1" as const, + steeringDisposition: "temporarily_unavailable" as const, + entries: [], + }; + const existingIds = new Set(base.entries.map((entry) => entry.comment.id)); + const optimisticEntries = commentsWithRunMeta.flatMap((comment) => { + if ( + existingIds.has(comment.id) + || consumedQueuedCommentIds.has(comment.id) + || discardedQueuedCommentIds.has(comment.id) + || comment.queueState !== "queued" + || comment.queueTargetRunId !== resolvedActiveRun.id + ) return []; + return [{ + comment, + position: base.entries.length, + canEdit: comment.authorUserId === currentUserId, + canDiscard: comment.authorUserId === currentUserId, + }]; + }); + return { + ...base, + entries: [...base.entries, ...optimisticEntries] + .map((entry, position) => ({ ...entry, position })), + }; + }, [ + authoritativeQueuedCommentQueue, + commentsWithRunMeta, + consumedQueuedCommentIds, + currentUserId, + discardedQueuedCommentIds, + issueId, + paperclipQueueEnabled, + resolvedActiveRun, + ]); + const commentsForThread = useMemo(() => commentsWithRunMeta.flatMap((comment) => { + if (discardedQueuedCommentIds.has(comment.id)) return []; + if (!consumedQueuedCommentIds.has(comment.id)) return [comment]; + return [{ + ...comment, + queueState: undefined, + queueTargetRunId: null, + queueReason: undefined, + }]; + }), [commentsWithRunMeta, consumedQueuedCommentIds, discardedQueuedCommentIds]); + + const storeQueuedCommentQueue = useCallback((value: unknown) => { + const queue = normalizeIssueQueuedCommentQueue(value, issueId); + queryClient.setQueryData(queryKeys.issues.queuedComments(issueId), queue); + return queue; + }, [issueId, queryClient]); + + const refreshQueueAfterConflict = useCallback(async (error: unknown) => { + if (error instanceof ApiError && error.status === 409) { + await queryClient.invalidateQueries({ queryKey: queryKeys.issues.queuedComments(issueId) }); + } + throw error; + }, [issueId, queryClient]); + + const editQueuedComment = useCallback(async (commentId: string, body: string, revision: string) => { + const targetRunId = effectiveQueuedCommentQueue?.targetRunId; + if (!targetRunId) throw new Error("The queued message no longer has an active run target."); + try { + storeQueuedCommentQueue(await issuesApi.editQueuedComment(issueId, commentId, { + body, + targetRunId, + revision, + })); + await queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(issueId) }); + } catch (error) { + await refreshQueueAfterConflict(error); + } + }, [effectiveQueuedCommentQueue?.targetRunId, issueId, queryClient, refreshQueueAfterConflict, storeQueuedCommentQueue]); + + const reorderQueuedComments = useCallback(async (orderedCommentIds: string[], revision: string) => { + const targetRunId = effectiveQueuedCommentQueue?.targetRunId; + if (!targetRunId) throw new Error("The queued messages no longer have an active run target."); + try { + storeQueuedCommentQueue(await issuesApi.reorderQueuedComments(issueId, { + orderedCommentIds, + targetRunId, + revision, + })); + } catch (error) { + await refreshQueueAfterConflict(error); + } + }, [effectiveQueuedCommentQueue?.targetRunId, issueId, refreshQueueAfterConflict, storeQueuedCommentQueue]); + + const steerQueuedComment = useCallback(async (commentId: string, revision: string) => { + const targetRunId = effectiveQueuedCommentQueue?.targetRunId; + if (!targetRunId) throw new Error("The queued message no longer has an active run target."); + try { + storeQueuedCommentQueue(await issuesApi.steerQueuedComment(issueId, commentId, { + targetRunId, + revision, + })); + setConsumedQueuedCommentIds((current) => new Set(current).add(commentId)); + await Promise.all([ + queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(issueId) }), + queryClient.invalidateQueries({ queryKey: queryKeys.issues.activity(issueId) }), + ]); + } catch (error) { + await refreshQueueAfterConflict(error); + } + }, [effectiveQueuedCommentQueue?.targetRunId, issueId, queryClient, refreshQueueAfterConflict, storeQueuedCommentQueue]); + + const discardQueuedComment = useCallback(async (commentId: string, revision: string) => { + const targetRunId = effectiveQueuedCommentQueue?.targetRunId; + if (!targetRunId) throw new Error("The queued message no longer has an active run target."); + try { + storeQueuedCommentQueue(await issuesApi.discardQueuedComment(issueId, commentId, { + targetRunId, + revision, + })); + setDiscardedQueuedCommentIds((current) => new Set(current).add(commentId)); + await queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(issueId) }); + } catch (error) { + await refreshQueueAfterConflict(error); + } + }, [effectiveQueuedCommentQueue?.targetRunId, issueId, queryClient, refreshQueueAfterConflict, storeQueuedCommentQueue]); + const timelineEvents = useMemo( () => extractIssueTimelineEvents(resolvedActivity), [resolvedActivity], @@ -1331,7 +1481,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ ) : undefined } issueBrief={issueBrief} - comments={commentsWithRunMeta} + comments={commentsForThread} interactions={interactions} documents={documents} workProducts={workProducts} @@ -1383,6 +1533,11 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ imageUploadHandler={onImageUpload} onAttachImage={onAttachImage} onInterruptQueued={onInterruptQueued} + queuedCommentQueue={effectiveQueuedCommentQueue} + onEditQueuedComment={editQueuedComment} + onReorderQueuedComments={reorderQueuedComments} + onSteerQueuedComment={steerQueuedComment} + onDiscardQueuedComment={discardQueuedComment} onDeleteComment={onDeleteComment} onCancelQueued={onCancelQueued} interruptingQueuedRunId={interruptingQueuedRunId} diff --git a/ui/storybook/stories/task-queue-steering.stories.tsx b/ui/storybook/stories/task-queue-steering.stories.tsx new file mode 100644 index 00000000000..6a17d0b6c95 --- /dev/null +++ b/ui/storybook/stories/task-queue-steering.stories.tsx @@ -0,0 +1,416 @@ +import { useMemo, useState } from "react"; +import type { Meta, StoryObj } from "@storybook/react-vite"; +import { expect, userEvent, waitFor, within } from "storybook/test"; +import type { + IssueComment, + IssueQueuedCommentQueue, + IssueQueuedCommentSteeringDisposition, +} from "@paperclipai/shared"; +import { TaskChatThread } from "@/components/TaskChatThread"; +import { normalizeIssueQueuedCommentQueue } from "@/lib/issue-queued-comment-queue"; +import { storybookAgentMap } from "../fixtures/paperclipData"; + +const ISSUE_ID = "issue-queue-story"; +const RUN_ID = "run-queue-story"; +const USER_ID = "user-board"; +const TS = new Date("2026-08-22T15:00:00.000Z"); + +function comment(id: string, body: string, minutes: number): IssueComment { + return { + id, + companyId: "company-storybook", + issueId: ISSUE_ID, + authorType: "user", + authorAgentId: null, + authorUserId: USER_ID, + body, + presentation: null, + metadata: null, + createdAt: new Date(TS.getTime() + minutes * 60_000), + updatedAt: new Date(TS.getTime() + minutes * 60_000), + }; +} + +const priorComment = comment("comment-prior", "Build queueing and true same-turn steering.", -2); +const queuedComments = [ + comment("queued-one", "Keep the compact strip aligned with the task composer.", 1), + comment("queued-two", "Add keyboard reorder and announce the new position.", 2), + comment("queued-three", "Verify the mobile layout before shipping.", 3), +]; + +function queueFixture( + entries = queuedComments, + steeringDisposition: IssueQueuedCommentSteeringDisposition = "available", +): IssueQueuedCommentQueue { + return normalizeIssueQueuedCommentQueue({ + issueId: ISSUE_ID, + targetRunId: RUN_ID, + revision: "queue-revision-1", + protocol: "paperclip_runner_v1", + steeringDisposition, + entries: entries.map((queuedComment, position) => ({ + comment: queuedComment, + position, + canEdit: true, + canDiscard: true, + })), + }, ISSUE_ID); +} + +function QueueStory({ + initialEntries = queuedComments, + steeringDisposition = "available", + failReorder = false, + failSteer = false, + holdSteer = false, + failEditStale = false, + failDiscard = false, + rawQueue, + mobile = false, +}: { + initialEntries?: IssueComment[]; + steeringDisposition?: IssueQueuedCommentSteeringDisposition; + failReorder?: boolean; + failSteer?: boolean; + holdSteer?: boolean; + failEditStale?: boolean; + failDiscard?: boolean; + rawQueue?: unknown; + mobile?: boolean; +}) { + const [queue, setQueue] = useState(() => rawQueue === undefined + ? queueFixture(initialEntries, steeringDisposition) + : normalizeIssueQueuedCommentQueue(rawQueue, ISSUE_ID)); + const [comments, setComments] = useState([priorComment, ...initialEntries]); + const agentMap = useMemo(() => { + const next = new Map(storybookAgentMap); + const codex = next.get("agent-codex"); + if (codex) next.set("agent-codex", { ...codex, adapterType: "paperclip_runner" }); + return next; + }, []); + + const updateQueue = (nextEntries: IssueComment[]) => { + setQueue((current) => normalizeIssueQueuedCommentQueue({ + ...current, + revision: `${current.revision}:next`, + entries: nextEntries.map((queuedComment, position) => ({ + comment: queuedComment, + position, + canEdit: true, + canDiscard: true, + })), + }, ISSUE_ID)); + }; + + return ( +
+
+ undefined, + }} + activeRun={{ + id: RUN_ID, + status: "running", + invocationSource: "issue_comment", + triggerDetail: "queued message story", + startedAt: TS, + finishedAt: null, + createdAt: TS, + agentId: "agent-codex", + agentName: "Codex", + adapterType: "paperclip_runner", + issueId: ISSUE_ID, + currentStatusMessage: "Working", + }} + queuedCommentQueue={queue} + onAdd={async (body) => { + const next = comment(`queued-${Date.now()}`, body, 4); + setComments((current) => [...current, next]); + updateQueue([...queue.entries.map((entry) => entry.comment), next]); + }} + onEditQueuedComment={async (commentId, body) => { + if (failEditStale) { + setComments((current) => current.filter((entry) => entry.id !== commentId)); + updateQueue(queue.entries + .filter((entry) => entry.comment.id !== commentId) + .map((entry) => entry.comment)); + throw { + status: 409, + body: { details: { code: "queued_comment_stale_target" } }, + }; + } + const nextEntries = queue.entries.map((entry) => ( + entry.comment.id === commentId + ? { ...entry.comment, body, updatedAt: new Date() } + : entry.comment + )); + setComments((current) => current.map((entry) => ( + entry.id === commentId ? { ...entry, body, updatedAt: new Date() } : entry + ))); + updateQueue(nextEntries); + }} + onReorderQueuedComments={async (orderedCommentIds) => { + if (failReorder) throw new Error("revision_conflict"); + const byId = new Map(queue.entries.map((entry) => [entry.comment.id, entry.comment])); + updateQueue(orderedCommentIds.flatMap((id) => byId.get(id) ?? [])); + }} + onSteerQueuedComment={async (commentId) => { + if (holdSteer) await new Promise(() => undefined); + if (failSteer) throw new Error("steering_timeout"); + updateQueue(queue.entries.filter((entry) => entry.comment.id !== commentId).map((entry) => entry.comment)); + }} + onDiscardQueuedComment={async (commentId) => { + if (failDiscard) throw new Error("queued_comment_revision_conflict"); + setComments((current) => current.filter((entry) => entry.id !== commentId)); + updateQueue(queue.entries.filter((entry) => entry.comment.id !== commentId).map((entry) => entry.comment)); + }} + draftKey={`storybook:queue-steering:${mobile ? "mobile" : "desktop"}`} + /> +
+
+ ); +} + +const meta = { + title: "Task Page/Queue & Steering", + component: TaskChatThread, + args: { comments: [], onAdd: async () => undefined }, + parameters: { layout: "fullscreen" }, +} satisfies Meta; + +export default meta; +type Story = StoryObj; + +export const SingleQueuedMessage: Story = { + render: () => , +}; + +export const MultipleQueuedMessages: Story = { + render: () => , +}; + +export const PointerReorder: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const source = canvas.getByRole("button", { name: /Reorder queued message: Keep the compact strip/ }); + const target = canvas.getByRole("button", { name: /Reorder queued message: Add keyboard reorder/ }); + const sourceRect = source.getBoundingClientRect(); + const targetRect = target.getBoundingClientRect(); + await userEvent.pointer([ + { keys: "[MouseLeft>]", target: source, coords: { x: sourceRect.x + 8, y: sourceRect.y + 8 } }, + { target, coords: { x: targetRect.x + 8, y: targetRect.y + targetRect.height - 2 } }, + { keys: "[/MouseLeft]", target, coords: { x: targetRect.x + 8, y: targetRect.y + targetRect.height - 2 } }, + ]); + await waitFor(() => { + const rows = [...canvasElement.querySelectorAll('[data-testid^="task-chat-queued-message-"]')]; + expect(rows[0]).toHaveTextContent("Add keyboard reorder"); + }); + }, +}; + +export const KeyboardReorder: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const handle = canvas.getByRole("button", { name: /Reorder queued message: Keep the compact strip/ }); + await userEvent.click(handle); + await userEvent.keyboard("{Space}{ArrowDown}{Space}"); + await waitFor(() => { + const rows = [...canvasElement.querySelectorAll('[data-testid^="task-chat-queued-message-"]')]; + expect(rows[0]).toHaveTextContent("Add keyboard reorder"); + }); + }, +}; + +export const EditAndRestoreDraft: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const editor = canvas.getByRole("textbox"); + await userEvent.type(editor, "Unsent composer draft"); + await userEvent.click(canvas.getByRole("button", { name: /Queued message actions: Keep the compact strip/ })); + await userEvent.click(within(canvasElement.ownerDocument.body).getByRole("menuitem", { name: "Edit message" })); + await expect(canvas.getByText("Editing queued message")).toBeInTheDocument(); + await userEvent.click(canvas.getByRole("button", { name: "Cancel" })); + await waitFor(() => expect(canvas.getByRole("textbox")).toHaveTextContent("Unsent composer draft")); + await expect(canvas.getByRole("textbox")).toHaveFocus(); + }, +}; + +export const EditAndSave: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByRole("button", { name: /Queued message actions: Keep the compact strip/ })); + await userEvent.click(within(canvasElement.ownerDocument.body).getByRole("menuitem", { name: "Edit message" })); + const editor = canvas.getByRole("textbox"); + await userEvent.clear(editor); + await userEvent.type(editor, "Keep the compact strip aligned and keyboard accessible."); + await userEvent.click(canvas.getByRole("button", { name: "Save queued message" })); + await waitFor(() => expect(canvas.getByTestId("task-chat-queued-message-queued-one")) + .toHaveTextContent("aligned and keyboard accessible")); + }, +}; + +export const StaleEditRecovery: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByRole("button", { name: /Queued message actions: Keep the compact strip/ })); + await userEvent.click(within(canvasElement.ownerDocument.body).getByRole("menuitem", { name: "Edit message" })); + const editor = canvas.getByRole("textbox"); + await userEvent.clear(editor); + await userEvent.type(editor, "Queue this replacement without losing it."); + await userEvent.click(canvas.getByRole("button", { name: "Save queued message" })); + await expect(canvas.getByText("Queued message changed")).toBeInTheDocument(); + await userEvent.click(canvas.getByRole("button", { name: "Queue as new message" })); + await waitFor(() => expect(canvas.getByText("Queue this replacement without losing it.")).toBeInTheDocument()); + }, +}; + +export const SteerAcknowledged: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByTestId("task-chat-queued-steer-queued-one")); + await waitFor(() => expect(canvas.queryByTestId("task-chat-queued-message-queued-one")).not.toBeInTheDocument()); + }, +}; + +export const SteeringUnsupported: Story = { + render: () => , +}; + +export const SteeringTemporarilyUnavailable: Story = { + render: () => , +}; + +export const SteeringPending: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByTestId("task-chat-queued-steer-queued-one")); + await expect(canvas.getByTestId("task-chat-queued-steer-queued-one")).toBeDisabled(); + await expect(canvas.getByText("Steering queued message.")).toBeInTheDocument(); + }, +}; + +export const SteeringFailed: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByTestId("task-chat-queued-steer-queued-one")); + await waitFor(() => expect(canvas.getByTestId("task-chat-queued-message-queued-one")).toBeInTheDocument()); + await expect(canvas.getByText("Couldn’t steer. Message is still queued.")).toBeInTheDocument(); + }, +}; + +export const SteeringStaleTurn: Story = { + render: () => , + parameters: { + docs: { description: { story: "The active provider turn settled before its steering acknowledgement." } }, + }, +}; + +export const Trash: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByTestId("task-chat-queued-discard-queued-one")); + await waitFor(() => expect(canvas.queryByTestId("task-chat-queued-message-queued-one")).not.toBeInTheDocument()); + }, +}; + +export const ConcurrentRevisionFailure: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const handle = canvas.getByRole("button", { name: /Reorder queued message: Keep the compact strip/ }); + await userEvent.click(handle); + await userEvent.keyboard("{Space}{ArrowDown}{Space}"); + await waitFor(() => expect(canvas.getByText("Couldn’t reorder. Previous order restored.")) + .toBeInTheDocument()); + const rows = [...canvasElement.querySelectorAll('[data-testid^="task-chat-queued-message-"]')]; + expect(rows[0]).toHaveTextContent("Keep the compact strip"); + }, +}; + +export const TrashFailed: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await userEvent.click(canvas.getByTestId("task-chat-queued-discard-queued-one")); + await expect(canvas.getByTestId("task-chat-queued-message-queued-one")).toBeInTheDocument(); + await expect(canvas.getByText("Couldn’t discard. Message is still queued.")).toBeInTheDocument(); + }, +}; + +export const OptimisticToCanonicalReplacement: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + const editor = canvas.getByRole("textbox"); + await userEvent.type(editor, "Queue this while the run is active."); + await userEvent.click(canvas.getByRole("button", { name: "Send" })); + await waitFor(() => expect(canvas.getByText("Queue this while the run is active.")).toBeInTheDocument()); + expect(canvasElement.textContent?.match(/Queue this while the run is active\./g)).toHaveLength(1); + }, +}; + +export const EmptyQueue: Story = { + render: () => , +}; + +export const MalformedQueueData: Story = { + render: () => , +}; + +export const NotMessageOwner: Story = { + render: () => , +}; + +export const DesktopKitchenSink: Story = { + render: () => , +}; + +export const MobileKitchenSink: Story = { + render: () => , + parameters: { viewport: { defaultViewport: "mobile1" } }, +};