feat(task-chat): add queued message steering

This commit is contained in:
Dotta committed 2026-08-22 13:11:17 -05:00
1 parent edde864925
commit 4fdcf5a141
31 files changed
+3029 -58

No files matched your search

@@ -26,6 +26,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
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<String> = None;
let mut planning_thread = false;
let mut pending_tool_turns = BTreeMap::<String, String>::new();
let stdin = io::stdin();
@@ -122,6 +123,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
} 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<dyn std::error::Error>> {
"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
@@ -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<u32> {
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<Value, LocalRunnerError>;
fn steer_turn(&mut self, _turn_id: &str, _message: &str) -> Result<Value, LocalRunnerError> {
Err(LocalRunnerError::invalid(
"provider does not support same-turn steering",
))
}
fn interrupt_turn(&mut self, turn_id: &str) -> Result<Value, LocalRunnerError>;
fn read(&mut self) -> Result<Value, LocalRunnerError>;
fn poll(&mut self) -> Result<Option<ProviderEvent>, LocalRunnerError>;
@@ -332,6 +340,8 @@ pub struct CodexProvider {
next_request_id: u64,
thread_id: String,
session_id: Option<String>,
active_provider_turn_id: Option<String>,
pending_messages: VecDeque<Value>,
pending_tool_requests: BTreeMap<String, Value>,
collaboration_mode: String,
collaboration_mode_payload: Option<Value>,
@@ -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<Value, LocalRunnerError> {
pub fn interrupt_turn(&mut self, _turn_id: &str) -> Result<Value, LocalRunnerError> {
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<Value, LocalRunnerError> {
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<Option<ProviderEvent>, 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<Value, LocalRunnerError> {
CodexProvider::start_turn(self, message, cwd)
}
fn steer_turn(&mut self, turn_id: &str, message: &str) -> Result<Value, LocalRunnerError> {
CodexProvider::steer_turn(self, turn_id, message)
}
fn interrupt_turn(&mut self, turn_id: &str) -> Result<Value, LocalRunnerError> {
CodexProvider::interrupt_turn(self, turn_id)
}
@@ -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<Mutex<Vec<(String, String)>>>,
fail_next: Arc<Mutex<bool>>,
}
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<Value, LocalRunnerError> {
unreachable!()
}
fn steer_turn(&mut self, turn_id: &str, message: &str) -> Result<Value, LocalRunnerError> {
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<Value, LocalRunnerError> {
unreachable!()
}
fn read(&mut self) -> Result<Value, LocalRunnerError> {
unreachable!()
}
fn poll(&mut self) -> Result<Option<ProviderEvent>, 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<Box<dyn Provider>> = 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<Box<dyn Provider>> = 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");
@@ -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,
@@ -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);
}
@@ -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<void>;
steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise<void>;
interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
pendingRuntimeRequests?(): HarnessRuntimeRequest[];
resolveRuntimeRequest?(input: {
@@ -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<void>;
steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise<void>;
interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
cancel?(input: { reason: string }): Promise<void>;
result(): Promise<{
@@ -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);
@@ -786,6 +786,7 @@ class CodexHarnessSession implements HarnessSession {
#goal: HarnessThreadGoal | null = null;
#interruptQueued = false;
#steerSequence = 0;
readonly #acknowledgedSteeringCorrelations = new Map<string, string>();
#interruptSequence = 0;
constructor(input: {
@@ -986,29 +987,49 @@ class CodexHarnessSession implements HarnessSession {
};
}
async steer(input: { turnId: string; message: NativeUserMessage }): Promise<void> {
async steer(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise<void> {
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);
}
}
@@ -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<never>((_, 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(),
@@ -325,6 +325,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
#threadId = "";
#sessionId: string | null = null;
#turnId = "";
#durableTurnId = "";
#authorizedTools: Record<string, unknown> | 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<string, unknown>): Promise<void> {
async #command(
type: string,
payload: Record<string, unknown>,
correlationId?: string,
): Promise<void> {
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);
}
+4
View File
@@ -1036,6 +1036,10 @@ export type {
IssueExecutionStagePrincipal,
IssueExecutionDecision,
IssueComment,
IssueQueuedCommentEntry,
IssueQueuedCommentProtocol,
IssueQueuedCommentQueue,
IssueQueuedCommentSteeringDisposition,
IssueCommentDerivedAuthorSource,
IssueCommentMetadata,
IssueCommentMetadataSection,
+4
View File
@@ -673,6 +673,10 @@ export type {
IssueReviewRequest,
IssueExecutionDecision,
IssueComment,
IssueQueuedCommentEntry,
IssueQueuedCommentProtocol,
IssueQueuedCommentQueue,
IssueQueuedCommentSteeringDisposition,
IssueCommentDerivedAuthorSource,
IssueCommentMetadata,
IssueCommentMetadataSection,
+27
View File
@@ -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;
@@ -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<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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);
});
});
+489
View File
@@ -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<Parameters<Db["transaction"]>[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<string>();
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<DeferredIssueWake | null> {
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<ReturnType<typeof resolveActiveIssueRun>>;
actor: ReturnType<typeof getActorInfo>;
wake?: DeferredIssueWake | null;
steeringDisposition?: IssueQueuedCommentQueue["steeringDisposition"];
}): Promise<IssueQueuedCommentQueue> {
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<Parameters<Db["transaction"]>[0]>[0];
issue: { id: string; companyId: string; assigneeAgentId: string | null };
actor: ReturnType<typeof getActorInfo>;
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<typeof getActorInfo> }) {
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");
@@ -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<void>((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);
@@ -509,6 +509,101 @@ function record(value: unknown): Record<string, unknown> {
: {};
}
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<NativeSessionSteeringState> {
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<typeof setTimeout> | null = null;
try {
await Promise.race([
active.session.steer({
turnId,
message: { role: "user", text: input.message },
correlationId: input.correlationId,
}),
new Promise<never>((_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<boolean>;
export function cancelNativeSession(
runId: string,
+34
View File
@@ -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<IssueComment[]>(`/issues/${id}/comments${qs ? `?${qs}` : ""}`);
},
getQueuedComments: (id: string) =>
api.get<IssueQueuedCommentQueue>(`/issues/${id}/queued-comments`),
editQueuedComment: (
id: string,
commentId: string,
data: { body: string; targetRunId: string; revision: string },
) =>
api.patch<IssueQueuedCommentQueue>(
`/issues/${id}/queued-comments/${commentId}`,
data,
),
reorderQueuedComments: (
id: string,
data: { orderedCommentIds: string[]; targetRunId: string; revision: string },
) => api.put<IssueQueuedCommentQueue>(`/issues/${id}/queued-comments/order`, data),
steerQueuedComment: (
id: string,
commentId: string,
data: { targetRunId: string; revision: string },
) =>
api.post<IssueQueuedCommentQueue>(
`/issues/${id}/queued-comments/${commentId}/steer`,
data,
),
discardQueuedComment: (
id: string,
commentId: string,
data: { targetRunId: string; revision: string },
) =>
api.delete<IssueQueuedCommentQueue>(
`/issues/${id}/queued-comments/${commentId}`,
data,
),
listInteractions: (id: string) =>
api.get<IssueThreadInteraction[]>(`/issues/${id}/interactions`),
listAcceptedPlanDecompositions: (id: string) =>
+7
View File
@@ -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<void>;
onCancelQueued?: (commentId: string) => void;
/** Authoritative PRP queue. The classic thread intentionally ignores it. */
queuedCommentQueue?: IssueQueuedCommentQueue | null;
onEditQueuedComment?: (commentId: string, body: string, revision: string) => Promise<void>;
onReorderQueuedComments?: (orderedCommentIds: string[], revision: string) => Promise<void>;
onSteerQueuedComment?: (commentId: string, revision: string) => Promise<void>;
onDiscardQueuedComment?: (commentId: string, revision: string) => Promise<void>;
onDeleteComment?: (commentId: string) => Promise<void> | void;
interruptingQueuedRunId?: string | null;
stoppingRunId?: string | null;
+58 -1
View File
@@ -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(<TaskChatThread {...props} />);
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(<TaskChatThread {...props} queuedCommentQueue={{ ...queue, revision: "revision-2", entries: [] }} />);
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;
+160 -18
View File
@@ -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) {
</Button>
);
},
[interruptingQueuedRunId, onInterruptQueued],
[interruptingQueuedRunId, onInterruptQueued, paperclipQueue],
);
const renderInteraction = useCallback(
@@ -1009,22 +1119,54 @@ export function TaskChatThread(props: TaskChatThreadProps) {
)}
>
{composerAccessory}
<TaskChatComposer
onAdd={handleThreadAdd}
workMode={issueWorkMode}
onWorkModeChange={onWorkModeChange}
disabled={Boolean(runtimeComposerDisabledReason)}
disabledReason={runtimeComposerDisabledReason}
onAttachImage={onAttachImage}
onImageUpload={imageUploadHandler}
mentions={mentions}
enableReassign={enableReassign}
reassignOptions={reassignOptions}
currentAssigneeValue={currentAssigneeValue}
issueStatus={issueStatus}
mobile={isMobile}
draftKey={draftKey}
/>
<div
className="relative isolate flex flex-col"
data-testid="task-chat-composer-stack"
>
{paperclipQueue && paperclipQueue.entries.length > 0 ? (
<TaskChatQueuedMessages
queue={paperclipQueue}
onEdit={beginQueuedEdit}
onReorder={async (orderedCommentIds, revision) => {
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}
<div className="relative z-10">
<TaskChatComposer
onAdd={handleThreadAdd}
workMode={issueWorkMode}
onWorkModeChange={onWorkModeChange}
disabled={Boolean(runtimeComposerDisabledReason)}
disabledReason={runtimeComposerDisabledReason}
onAttachImage={onAttachImage}
onImageUpload={imageUploadHandler}
mentions={mentions}
enableReassign={enableReassign}
reassignOptions={reassignOptions}
currentAssigneeValue={currentAssigneeValue}
issueStatus={issueStatus}
mobile={isMobile}
draftKey={draftKey}
queuedEdit={queuedEdit}
onSaveQueuedEdit={saveQueuedEdit}
onCancelQueuedEdit={() => setQueuedEdit(null)}
/>
</div>
</div>
{footer}
</div>
) : null}
@@ -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<void>;
stale?: boolean;
}) {
const [queuedEdit, setQueuedEdit] = useState<{
commentId: string;
body: string;
stale?: boolean;
} | null>({
commentId: "queued-1",
body: "Complete queued markdown",
stale,
});
return (
<TaskChatComposer
onAdd={vi.fn()}
workMode="standard"
draftKey={draftKey}
queuedEdit={queuedEdit}
onSaveQueuedEdit={onSave}
onCancelQueuedEdit={() => setQueuedEdit(null)}
/>
);
}
it("restores the existing composer draft after cancelling an edit", async () => {
localStorage.setItem(draftKey, "Unsent normal draft");
render(<Harness onSave={vi.fn().mockResolvedValue(undefined)} />);
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(<Harness onSave={onSave} />);
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(<Harness onSave={onSave} />);
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(<Harness onSave={onSave} stale />);
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 ",
);
});
});
});
+118 -13
View File
@@ -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<void>;
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<ReturnType<typeof setTimeout> | null>(null);
const queuedEditRef = useRef(queuedEdit);
queuedEditRef.current = queuedEdit;
const queuedEditBackupRef = useRef<QueuedEditBackup | null>(null);
const previousQueuedEditIdRef = useRef<string | null>(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<string> {
@@ -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 ? (
<span className="px-1 text-xs font-medium text-muted-foreground">
{queuedEdit.stale ? "Queued message changed" : "Editing queued message"}
</span>
) : (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<button
@@ -519,10 +602,11 @@ export function TaskChatComposer({
})}
</DropdownMenuContent>
</DropdownMenu>
)}
<div className="flex-1" />
{showAssignee ? (
{showAssignee && !queuedEdit ? (
<InlineEntitySelector
value={assigneeValue}
options={reassignOptions ?? []}
@@ -543,6 +627,17 @@ export function TaskChatComposer({
/>
) : null}
{queuedEdit ? (
<button
type="button"
onClick={onCancelQueuedEdit}
disabled={submitting}
className="h-8 shrink-0 rounded-md px-2.5 text-xs font-medium text-muted-foreground transition-colors hover:bg-accent hover:text-foreground disabled:opacity-50"
>
Cancel
</button>
) : null}
<button
type="button"
onClick={() => void submit()}
@@ -554,13 +649,23 @@ export function TaskChatComposer({
(body.trim().length === 0 && attachedRefs.length === 0)
}
title={
uploadPending
queuedEdit
? queuedEdit.stale
? "Queue as new message"
: "Save queued message"
: uploadPending
? "Waiting for upload to finish"
: uploadFailed
? "Remove the failed attachment to send"
: "Send (⌘+Enter)"
}
aria-label="Send"
aria-label={
queuedEdit
? queuedEdit.stale
? "Queue as new message"
: "Save queued message"
: "Send"
}
className="flex h-8 w-8 shrink-0 items-center justify-center rounded-md bg-primary text-primary-foreground transition-transform hover:scale-105 disabled:scale-100 disabled:bg-muted disabled:text-muted-foreground"
data-testid="task-chat-composer-send"
>
@@ -0,0 +1,101 @@
// @vitest-environment jsdom
import { act, type ComponentProps } 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 type { IssueQueuedCommentQueue } from "@paperclipai/shared";
import { TaskChatQueuedMessages } from "./TaskChatQueuedMessages";
const queue: IssueQueuedCommentQueue = {
issueId: "issue-1",
targetRunId: "run-1",
revision: "rev-1",
protocol: "paperclip_runner_v1",
steeringDisposition: "available",
entries: ["First queued message", "Second queued message"].map((body, position) => ({
comment: {
id: `comment-${position + 1}`,
companyId: "company-1",
issueId: "issue-1",
authorType: "user",
authorAgentId: null,
authorUserId: "user-1",
body,
presentation: null,
metadata: null,
createdAt: new Date(),
updatedAt: new Date(),
},
position,
canEdit: true,
canDiscard: true,
})),
};
describe("TaskChatQueuedMessages", () => {
let container: HTMLDivElement;
let root: Root;
beforeEach(() => {
container = document.createElement("div");
document.body.appendChild(container);
root = createRoot(container);
});
afterEach(() => {
flushSync(() => root.unmount());
container.remove();
});
function render(overrides: Partial<ComponentProps<typeof TaskChatQueuedMessages>> = {}) {
const props = {
queue,
onEdit: vi.fn(),
onReorder: vi.fn().mockResolvedValue(undefined),
onSteer: vi.fn().mockResolvedValue(undefined),
onDiscard: vi.fn().mockResolvedValue(undefined),
...overrides,
};
flushSync(() => root.render(<TaskChatQueuedMessages {...props} />));
return props;
}
it("renders each queued message once as a compact one-line row", () => {
render();
const pane = container.querySelector('[data-testid="task-chat-queued-messages"]');
expect(pane?.classList).toContain("mx-3");
expect(pane?.classList).toContain("rounded-b-none");
expect(pane?.classList).toContain("border-b-0");
expect(pane?.classList).toContain("-mb-px");
expect(container.querySelectorAll('[data-testid^="task-chat-queued-message-"]')).toHaveLength(2);
expect(container.textContent).toContain("First queued message");
expect(container.textContent).toContain("Second queued message");
});
it("removes only the acknowledged steering row", async () => {
const props = render();
await act(async () => {
container.querySelector<HTMLButtonElement>('[data-testid="task-chat-queued-steer-comment-1"]')?.click();
});
expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-1");
expect(container.querySelector('[data-testid="task-chat-queued-message-comment-1"]')).toBeNull();
expect(container.querySelector('[data-testid="task-chat-queued-message-comment-2"]')).not.toBeNull();
});
it("keeps a row queued when steering fails and announces the retryable state", async () => {
render({ onSteer: vi.fn().mockRejectedValue(new Error("steering_timeout")) });
await act(async () => {
container.querySelector<HTMLButtonElement>('[data-testid="task-chat-queued-steer-comment-1"]')?.click();
});
expect(container.querySelector('[data-testid="task-chat-queued-message-comment-1"]')).not.toBeNull();
expect(container.textContent).toContain("Couldn’t steer. Message is still queued.");
});
it("disables steering when the provider does not advertise it", () => {
render({ queue: { ...queue, steeringDisposition: "unsupported" } });
expect(container.querySelector<HTMLButtonElement>('[data-testid="task-chat-queued-steer-comment-1"]')?.disabled)
.toBe(true);
});
});
@@ -0,0 +1,272 @@
import { useEffect, useMemo, useState } from "react";
import {
DndContext,
KeyboardSensor,
PointerSensor,
closestCenter,
useSensor,
useSensors,
type DragEndEvent,
} from "@dnd-kit/core";
import {
SortableContext,
arrayMove,
sortableKeyboardCoordinates,
useSortable,
verticalListSortingStrategy,
} from "@dnd-kit/sortable";
import { CSS } from "@dnd-kit/utilities";
import {
CornerDownRight,
GripVertical,
Loader2,
MoreHorizontal,
Pencil,
Trash2,
} from "lucide-react";
import type {
IssueQueuedCommentEntry,
IssueQueuedCommentQueue,
} from "@paperclipai/shared";
import { cn } from "@/lib/utils";
import {
DropdownMenu,
DropdownMenuContent,
DropdownMenuItem,
DropdownMenuTrigger,
} from "@/components/ui/dropdown-menu";
type QueueAction = "steer" | "discard" | null;
export interface TaskChatQueuedMessagesProps {
queue: IssueQueuedCommentQueue;
onEdit: (commentId: string) => void;
onReorder: (orderedCommentIds: string[], revision: string) => Promise<void>;
onSteer: (commentId: string, revision: string) => Promise<void>;
onDiscard: (commentId: string, revision: string) => Promise<void>;
}
function SortableQueuedMessage({
entry,
queue,
disabled,
action,
onEdit,
onSteer,
onDiscard,
}: {
entry: IssueQueuedCommentEntry;
queue: IssueQueuedCommentQueue;
disabled: boolean;
action: QueueAction;
onEdit: () => void;
onSteer: () => void;
onDiscard: () => void;
}) {
const sortable = useSortable({ id: entry.comment.id, disabled });
const style = {
transform: CSS.Transform.toString(sortable.transform),
transition: sortable.transition,
};
const steerDisabled = disabled || queue.steeringDisposition !== "available";
const steerTitle =
queue.steeringDisposition === "unsupported"
? "This runner does not support steering"
: queue.steeringDisposition === "temporarily_unavailable"
? "Steering is temporarily unavailable"
: "Steer this message into the active turn";
return (
<div
ref={sortable.setNodeRef}
style={style}
className={cn(
"group flex h-11 min-w-0 items-center gap-1 border-b border-border/55 bg-card/95 px-2.5 text-sm last:border-b-0",
sortable.isDragging && "relative z-20 rounded-lg border border-border shadow-lg",
)}
data-testid={`task-chat-queued-message-${entry.comment.id}`}
>
<button
type="button"
ref={sortable.setActivatorNodeRef}
{...sortable.attributes}
{...sortable.listeners}
disabled={disabled}
aria-label={`Reorder queued message: ${entry.comment.body}`}
className="flex h-7 w-7 shrink-0 cursor-grab items-center justify-center rounded-md text-muted-foreground/70 transition-colors hover:bg-accent hover:text-foreground active:cursor-grabbing disabled:cursor-default disabled:opacity-40"
>
<GripVertical className="h-3.5 w-3.5" aria-hidden />
</button>
<CornerDownRight className="h-3.5 w-3.5 shrink-0 text-muted-foreground" aria-hidden />
<span className="min-w-0 flex-1 truncate px-1" title={entry.comment.body}>
{entry.comment.body}
</span>
<button
type="button"
onClick={onSteer}
disabled={steerDisabled}
title={steerTitle}
className="flex h-7 shrink-0 items-center gap-1.5 rounded-md px-2 text-xs font-medium text-muted-foreground transition-colors hover:bg-accent hover:text-foreground disabled:opacity-40"
data-testid={`task-chat-queued-steer-${entry.comment.id}`}
>
{action === "steer" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden />
) : (
<CornerDownRight className="h-3.5 w-3.5" aria-hidden />
)}
Steer
</button>
<button
type="button"
onClick={onDiscard}
disabled={disabled || !entry.canDiscard}
title="Discard queued message"
aria-label={`Discard queued message: ${entry.comment.body}`}
className="flex h-7 w-7 shrink-0 items-center justify-center rounded-md text-muted-foreground transition-colors hover:bg-destructive/10 hover:text-destructive disabled:opacity-40"
data-testid={`task-chat-queued-discard-${entry.comment.id}`}
>
{action === "discard" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden />
) : (
<Trash2 className="h-3.5 w-3.5" aria-hidden />
)}
</button>
<DropdownMenu>
<DropdownMenuTrigger asChild>
<button
type="button"
disabled={disabled}
title="Queued message actions"
aria-label={`Queued message actions: ${entry.comment.body}`}
className="flex h-7 w-7 shrink-0 items-center justify-center rounded-md text-muted-foreground transition-colors hover:bg-accent hover:text-foreground disabled:opacity-40"
>
<MoreHorizontal className="h-4 w-4" aria-hidden />
</button>
</DropdownMenuTrigger>
<DropdownMenuContent align="end" className="w-44">
<DropdownMenuItem disabled={!entry.canEdit} onSelect={onEdit}>
<Pencil className="h-4 w-4" aria-hidden />
Edit message
</DropdownMenuItem>
</DropdownMenuContent>
</DropdownMenu>
</div>
);
}
/** 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<QueueAction, null> } | null>(null);
const [reordering, setReordering] = useState(false);
const [announcement, setAnnouncement] = useState("");
const [visibleError, setVisibleError] = useState<string | null>(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<QueueAction, null>) {
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 (
<div
className="relative z-0 mx-3 -mb-px overflow-hidden rounded-t-xl rounded-b-none border border-b-0 border-border/75 bg-card shadow-sm"
data-testid="task-chat-queued-messages"
aria-label="Queued messages"
>
<DndContext sensors={sensors} collisionDetection={closestCenter} onDragEnd={(event) => void handleDragEnd(event)}>
<SortableContext items={ids} strategy={verticalListSortingStrategy}>
{entries.map((entry) => {
const action = pending?.commentId === entry.comment.id ? pending.action : null;
const disabled = Boolean(pending || reordering);
return (
<SortableQueuedMessage
key={entry.comment.id}
entry={entry}
queue={queue}
disabled={disabled}
action={action}
onEdit={() => onEdit(entry.comment.id)}
onSteer={() => void runRowAction(entry.comment.id, "steer")}
onDiscard={() => void runRowAction(entry.comment.id, "discard")}
/>
);
})}
</SortableContext>
</DndContext>
{visibleError ? (
<div
role="status"
aria-live="polite"
className="border-t border-destructive/20 bg-destructive/5 px-3 py-1.5 text-xs text-destructive"
>
{visibleError}
</div>
) : null}
<div className="sr-only" aria-live="polite" aria-atomic="true">
{announcement}
</div>
</div>
);
}
@@ -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: [],
});
});
});
+60
View File
@@ -0,0 +1,60 @@
import type {
IssueComment,
IssueQueuedCommentQueue,
IssueQueuedCommentSteeringDisposition,
} from "@paperclipai/shared";
function record(value: unknown): Record<string, unknown> | null {
return typeof value === "object" && value !== null && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
const STEERING_DISPOSITIONS = new Set<IssueQueuedCommentSteeringDisposition>([
"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<string>();
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,
};
}
+1
View File
@@ -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) =>
+156 -1
View File
@@ -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<ReadonlySet<string>>(
() => new Set(),
);
const [discardedQueuedCommentIds, setDiscardedQueuedCommentIds] = useState<ReadonlySet<string>>(
() => 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}
@@ -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<IssueComment[]>([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 (
<main className="min-h-screen bg-background p-4 text-foreground">
<section className={mobile ? "mx-auto w-(--sz-390px)" : "mx-auto max-w-(--tc-shell-max-w)"}>
<TaskChatThread
comments={comments}
issueId={ISSUE_ID}
issueStatus="in_progress"
issueAssigneeAgentId="agent-codex"
agentMap={agentMap}
currentUserId={USER_ID}
userLabelMap={new Map([[USER_ID, "Riley Board"]])}
issueBrief={{
description: "Implement Codex-style queueing for the Paperclip Runner.",
author: "human",
authorName: "Riley Board",
createdAt: TS,
onSave: async () => 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<void>(() => 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"}`}
/>
</section>
</main>
);
}
const meta = {
title: "Task Page/Queue & Steering",
component: TaskChatThread,
args: { comments: [], onAdd: async () => undefined },
parameters: { layout: "fullscreen" },
} satisfies Meta<typeof TaskChatThread>;
export default meta;
type Story = StoryObj<typeof meta>;
export const SingleQueuedMessage: Story = {
render: () => <QueueStory initialEntries={[queuedComments[0]]} />,
};
export const MultipleQueuedMessages: Story = {
render: () => <QueueStory />,
};
export const PointerReorder: Story = {
render: () => <QueueStory />,
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: () => <QueueStory />,
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: () => <QueueStory />,
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: () => <QueueStory />,
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: () => <QueueStory failEditStale />,
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: () => <QueueStory />,
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: () => <QueueStory steeringDisposition="unsupported" />,
};
export const SteeringTemporarilyUnavailable: Story = {
render: () => <QueueStory steeringDisposition="temporarily_unavailable" />,
};
export const SteeringPending: Story = {
render: () => <QueueStory holdSteer />,
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: () => <QueueStory failSteer />,
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: () => <QueueStory failSteer />,
parameters: {
docs: { description: { story: "The active provider turn settled before its steering acknowledgement." } },
},
};
export const Trash: Story = {
render: () => <QueueStory />,
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: () => <QueueStory failReorder />,
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: () => <QueueStory failDiscard />,
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: () => <QueueStory initialEntries={[]} />,
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: () => <QueueStory initialEntries={[]} />,
};
export const MalformedQueueData: Story = {
render: () => <QueueStory initialEntries={[]} rawQueue={{
issueId: ISSUE_ID,
targetRunId: RUN_ID,
revision: "malformed-revision",
protocol: "paperclip_runner_v1",
steeringDisposition: "available",
entries: [
null,
{ position: 0, comment: { id: "missing-body" } },
{ position: 1, canEdit: true, canDiscard: true, comment: queuedComments[0] },
{ position: 2, canEdit: true, canDiscard: true, comment: queuedComments[0] },
],
}} />,
};
export const NotMessageOwner: Story = {
render: () => <QueueStory rawQueue={{
issueId: ISSUE_ID,
targetRunId: RUN_ID,
revision: "restricted-revision",
protocol: "paperclip_runner_v1",
steeringDisposition: "available",
entries: [{
position: 0,
canEdit: false,
canDiscard: false,
comment: { ...queuedComments[0], authorUserId: "another-user" },
}],
}} />,
};
export const DesktopKitchenSink: Story = {
render: () => <QueueStory />,
};
export const MobileKitchenSink: Story = {
render: () => <QueueStory mobile />,
parameters: { viewport: { defaultViewport: "mobile1" } },
};