diff --git a/tests/restart-chaos/ttft-baseline-stats.ts b/tests/restart-chaos/ttft-baseline-stats.ts index 2034038e4a1..a0b490bcb22 100644 --- a/tests/restart-chaos/ttft-baseline-stats.ts +++ b/tests/restart-chaos/ttft-baseline-stats.ts @@ -16,7 +16,7 @@ export interface BaselineRun { adapter: string; environment: string; /** Driver intent: a new task, a follow-up comment on the same task, or a run crossed by a restart. */ - phase: "cold" | "warm" | "restart"; + phase: "cold" | "warm" | "restart" | "midturn"; events: BaselineEvent[]; } diff --git a/tests/restart-chaos/ttft-baseline.ts b/tests/restart-chaos/ttft-baseline.ts index d527afda5b9..974226ddc8f 100644 --- a/tests/restart-chaos/ttft-baseline.ts +++ b/tests/restart-chaos/ttft-baseline.ts @@ -50,6 +50,43 @@ const restartTask = (nonce: string) => [ "Do not create files, documents or other tasks.", ].join("\n"); +const MIDTURN_SLEEP_SECONDS = 40; +const midturnTask = (nonce: string) => [ + `Baseline mid-turn ${nonce}. First run the shell command \`python3 -c "import time; time.sleep(${MIDTURN_SLEEP_SECONDS})"\` and wait for it to finish.`, + `Then reply with a one-line comment that says \`DONE ${nonce}\` and mark this task done.`, + "Do not create files, documents or other tasks.", +].join("\n"); +const midturnFollowUp = (nonce: string) => + `Mid-turn follow-up ${nonce}: also reply with a one-line comment that says \`HELLO ${nonce}\`.`; + +/** The agent's replies naming `marker`, oldest first. A failed read throws: it is not a lost reply. */ +async function agentReplies(api: ChaosApi, issueId: string, marker: string): Promise { + const comments = await api.get(`/api/issues/${issueId}/comments`, 10_000); + return comments.filter(comment => Boolean(comment.authorAgentId) && String(comment.body ?? "").includes(marker)) + .sort((a, b) => Date.parse(String(a.createdAt)) - Date.parse(String(b.createdAt))); +} + +/** + * Steer a queued comment into the live turn once the runner reports steering + * available. Returns the acknowledgement time, or the disposition that blocked it. + */ +async function steerQueuedComment(api: ChaosApi, issueId: string, commentId: string): Promise<{ acknowledgedAt: string | null; disposition: string }> { + const deadline = Date.now() + 60_000; + let disposition = "none"; + while (Date.now() < deadline) { + const queue = await api.get(`/api/issues/${issueId}/queued-comments`, 10_000).catch(() => null); + disposition = String(queue?.steeringDisposition ?? "none"); + if (queue && disposition === "available" && queue.queueId && queue.targetRunId) { + await api.post(`/api/issues/${issueId}/queued-comments/${commentId}/steer`, + { queueId: queue.queueId, revision: queue.revision, targetRunId: queue.targetRunId }); + return { acknowledgedAt: new Date().toISOString(), disposition }; + } + if (disposition === "unsupported") break; + await delay(500); + } + return { acknowledgedAt: null, disposition }; +} + async function createTask(api: ChaosApi, lab: ChaosLab, adapter: AdapterId, title: string, description: string) { const issue = await api.post(`/api/companies/${lab.companyId}/issues`, { title, description, status: "todo", priority: "medium", assigneeAgentId: lab.agents[adapter].id, @@ -105,21 +142,21 @@ async function allEvents(api: ChaosApi, runId: string): Promise * command text (OpenCode's `bash`) counts as inside an open shell call, and the * sample is marked `commandUnverified`. */ -async function runInShellCall(api: ChaosApi, issueId: string): Promise<{ runId: string; commandUnverified: boolean } | null> { +async function runInShellCall(api: ChaosApi, issueId: string, marker = RESTART_SLEEP_MARKER): Promise<{ runId: string; commandUnverified: boolean } | null> { const active = (await runsFor(api, issueId)).find(run => String(run.status) === "running"); if (!active) return null; const events = await allEvents(api, runIdOf(active)).catch(() => []); - const open = openShellExecutions(events, RESTART_SLEEP_MARKER); + const open = openShellExecutions(events, marker); if (open.some(call => call.matched)) return { runId: runIdOf(active), commandUnverified: false }; if (open.length && open.every(call => call.unlabelled)) return { runId: runIdOf(active), commandUnverified: true }; return null; } /** Wait until the run is inside its shell tool call (started, not ended). */ -async function waitForShellCall(api: ChaosApi, issueId: string): Promise<{ runId: string; commandUnverified: boolean } | null> { +async function waitForShellCall(api: ChaosApi, issueId: string, marker = RESTART_SLEEP_MARKER): Promise<{ runId: string; commandUnverified: boolean } | null> { const deadline = Date.now() + 4 * 60_000; while (Date.now() < deadline) { - const inside = await runInShellCall(api, issueId); + const inside = await runInShellCall(api, issueId, marker); if (inside) return inside; await delay(1_500); } @@ -185,6 +222,10 @@ async function main() { // lifecycle, so the server resolves the product default). Absent keeps the // catalog profile's setting. lifecycle: { type: "string" }, + // Mid-turn follow-up delivery: queue (default, wait for the turn) or steer (into the live turn). + "midturn-mode": { type: "string", default: "queue" }, + // Fault the control plane right after a steered comment is acknowledged (K1, K2 or K6). + "midturn-fault": { type: "string" }, "idle-ms": { type: "string", default: "600000" }, } }); if (values.report) { @@ -198,6 +239,11 @@ async function main() { const restarts = Math.max(0, Number(values.restarts)); const missing = requested.map(adapter => CREDENTIAL_FOR_ADAPTER[adapter]).filter(key => !process.env[key]); if (missing.length) throw new Error(`Missing ${missing.join(", ")}; run: set -a; source ~/.secrets; set +a`); + const midturnMode = values["midturn-mode"]; + if (midturnMode !== "queue" && midturnMode !== "steer") throw new Error("--midturn-mode must be queue or steer"); + const midturnFault = values["midturn-fault"] as FaultKind | undefined; + if (midturnFault !== undefined && !["K1", "K2", "K6"].includes(midturnFault)) throw new Error("--midturn-fault must be K1, K2 or K6"); + if (midturnFault && midturnMode !== "steer") throw new Error("--midturn-fault applies to --midturn-mode steer"); const lifecycle = values.lifecycle; if (lifecycle !== undefined && lifecycle !== "per_turn" && lifecycle !== "warm" && lifecycle !== "default") { throw new Error("--lifecycle must be per_turn, warm or default"); @@ -216,6 +262,7 @@ async function main() { const faults: BaselineFault[] = []; const runPhase = new Map(); const agentLifecycles: Record = {}; + const midturn: Json[] = []; const timedOutIssues = new Set(); // One shutdown path for normal completion and for SIGINT/SIGTERM: the server // runs in a detached process group and the embedded Postgres has no exit hook. @@ -272,6 +319,51 @@ async function main() { } })); + // Mid-turn follow-up: a comment posted while the agent is inside a long + // tool call. Records when the comment landed and every run of the task, + // so the wait splits into "rest of the turn" and "hand-off". + const runMidturn = async (adapter: (typeof adapters)[number]) => { + for (let index = 0; index < samples; index += 1) { + const nonce = `${adapter}-m${index}-${Date.now().toString(36)}`; + const issueId = await createTask(api, lab, adapter, `TTFT mid-turn ${nonce}`, midturnTask(nonce)); + tasks.push({ issueId, adapter, phase: "midturn" }); + const inside = await waitForShellCall(api, issueId, `time.sleep(${MIDTURN_SLEEP_SECONDS})`); + if (!inside) { + timeline.add("baseline_midturn_sample_skipped", { adapter, issueId }); + continue; + } + const comment = await api.post(`/api/issues/${issueId}/comments`, { body: midturnFollowUp(nonce) }); + const steer = midturnMode === "steer" ? await steerQueuedComment(api, issueId, String(comment.id)) : null; + // Restart right after the steered comment was acknowledged: it must be answered exactly once. + const fault = midturnFault && steer?.acknowledgedAt + ? await supervisor.fault(midturnFault, 0, `midturn-${index}`) : null; + if (fault) faults.push({ kind: midturnFault!, readyAt: fault.readyAt ?? Date.now(), runIds: [inside.runId] }); + // A steered comment is answered inside the same run; a queued one needs a second run. + const settled = await waitForRest(api, supervisor, issueId, steer?.acknowledgedAt ? 1 : 2); + if (settled.timedOut) { + timedOutIssues.add(issueId); + timeline.add("baseline_sample_timeout", { adapter, issueId, phase: "midturn" }); + continue; + } + // Read the replies once; a failed read is recorded, not counted as a lost reply. + const replies = await agentReplies(api, issueId, `HELLO ${nonce}`) + .catch((error: unknown) => ({ error: error instanceof Error ? error.message : String(error) })); + midturn.push({ adapter, issueId, mode: midturnMode, interruptedRunId: inside.runId, commentId: String(comment.id), + commentCreatedAt: String(comment.createdAt), steer, fault: fault ? midturnFault : null, + replyCount: Array.isArray(replies) ? replies.length : null, + replyReadError: Array.isArray(replies) ? null : replies.error, + replyCreatedAt: Array.isArray(replies) && replies[0] ? String(replies[0].createdAt) : null, + runs: settled.runs }); + timeline.add("baseline_midturn_done", { adapter, issueId, runs: settled.runs.length }); + } + }; + // Every fault restarts the one shared server, so faulted samples run one + // at a time: a restart must never land in another sample's comment or steer. + if (phases.has("midturn")) { + if (midturnFault) for (const adapter of adapters) await runMidturn(adapter); + else await Promise.all(adapters.map(runMidturn)); + } + // Restart recovery: one in-flight tool call per adapter, then one fault crosses all of them. const kinds: FaultKind[] = ["K1", "K2"]; for (let index = 0; phases.has("restart") && index < restarts * kinds.length; index += 1) { @@ -321,11 +413,18 @@ async function main() { } } const data = { waveId, lifecycle: lifecycle ?? "catalog", agentLifecycles, adapters, notReady: requested.filter(adapter => !adapters.includes(adapter)), - timedOutTasks: timedOutIssues.size, samples, restarts, gitSha: process.env.GIT_SHA ?? null, runs, faults }; + timedOutTasks: timedOutIssues.size, samples, restarts, midturn, gitSha: process.env.GIT_SHA ?? null, runs, faults }; await writeFile(path.join(outDir, "baseline-data.json"), `${JSON.stringify(data)}\n`, { mode: 0o600 }); await writeFile(path.join(outDir, "baseline-report.md"), report(runs, faults), { mode: 0o600 }); process.stdout.write(report(runs, faults)); if (timedOutIssues.size) process.stdout.write(`\n${timedOutIssues.size} sample(s) timed out and were left out.\n`); + // Every answered mid-turn comment must be answered exactly once. + const wrongReplies = midturn.filter(sample => sample.replyCount !== 1); + for (const sample of wrongReplies) { + process.stdout.write(`\nMid-turn sample ${String(sample.adapter)} ${String(sample.issueId)}: ${sample.replyReadError + ? `reply read failed (${String(sample.replyReadError)})` : `${String(sample.replyCount)} replies, expected 1`}\n`); + } + if (wrongReplies.length) process.exitCode = 1; process.stdout.write(`\nWrote ${outDir}\n`); } finally { // Always stop the wrapper-owned embedded Postgres (SysV shared memory is scarce). diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index bfcd2e18ba3..274e7c3fea0 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -696,6 +696,8 @@ interface IssueChatThreadProps { revision: string, ) => Promise; onSteerQueuedComment?: (commentId: string, revision: string) => Promise; + /** Delay before an eligible queued message is steered automatically; null keeps every message queued. */ + queuedMessageAutoSteerDelayMs?: number | null; onDiscardQueuedComment?: ( commentId: string, revision: string, diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index ae9a99cd87c..0b0e9934f11 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -5036,8 +5036,9 @@ describe("TaskChatThread Paperclip Runner queue", () => { let rejectDelivery!: (error: Error) => void; const delivery = new Promise((_, reject) => { rejectDelivery = reject; }); const actionQueue = { ...queue, protocol: action === "interrupt" ? "legacy" as const : queue.protocol }; + // Exercises the explicit Steer button, so automatic steering is off here. const props = { comments: [queuedComment], onAdd: async () => {}, queuedCommentQueue: actionQueue, - onSteerQueuedComment: () => delivery, onInterruptQueued: () => delivery }; + onSteerQueuedComment: () => delivery, onInterruptQueued: () => delivery, queuedMessageAutoSteerDelayMs: null }; render(); await act(async () => { container.querySelector(`[data-testid="task-chat-queued-${action}-queued-prp-1"]`)!.click(); diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index bcf46e8bdc2..c802274f646 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -123,6 +123,7 @@ import { useRunnerGoalControl, } from "@/components/task-chat/RunnerGoalWidget"; import { TaskChatQueuedMessages } from "@/components/task-chat/TaskChatQueuedMessages"; +import { QUEUED_MESSAGE_AUTO_STEER_DELAY_MS } from "@/components/task-chat/queued-message-auto-steer"; import { useSidebar } from "@/context/SidebarContext"; import { useStreamlinedUiEnabled } from "@/hooks/useStreamlinedUiEnabled"; import { useDismissedTaskQuestions } from "@/hooks/useDismissedTaskQuestions"; @@ -586,6 +587,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { onEditQueuedComment, onReorderQueuedComments, onSteerQueuedComment, + queuedMessageAutoSteerDelayMs = QUEUED_MESSAGE_AUTO_STEER_DELAY_MS, onDiscardQueuedComment, blockedBy = [], blockerAttention, @@ -3542,6 +3544,8 @@ export function TaskChatThread(props: TaskChatThreadProps) { key={issueId} queue={queuedMessageQueue ?? emptyQueuedCommentQueue} onEdit={beginQueuedEdit} + editingCommentId={queuedEdit?.commentId ?? null} + autoSteerDelayMs={queuedMessageAutoSteerDelayMs} onReorder={async (orderedCommentIds, revision) => { if (!onReorderQueuedComments) throw new Error("Queue reordering is unavailable."); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index f44f60b77cd..fa9b4d6bd07 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -477,4 +477,123 @@ describe("TaskChatQueuedMessages", () => { expect(container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')).toBeNull(); }); + + describe("automatic steering", () => { + const delay = 5_000; + const fresh = (overrides: Partial = {}): IssueQueuedCommentQueue => { + const at = new Date(Date.now()); + return { ...queue, ...overrides, entries: queue.entries.slice(0, 1).map(entry => ({ + ...entry, comment: { ...entry.comment, createdAt: at, updatedAt: at } })) }; + }; + const advance = async (ms: number) => { await act(async () => { vi.advanceTimersByTime(ms); await Promise.resolve(); }); }; + const countdown = () => container.querySelector('[data-testid="task-chat-queued-auto-steer-comment-1"]'); + + beforeEach(() => { vi.useFakeTimers({ now: new Date("2026-11-07T12:00:00Z") }); }); + afterEach(() => { vi.useRealTimers(); }); + + it("shows a countdown with Edit and Cancel, then steers into the live turn", async () => { + const props = render({ queue: fresh(), autoSteerDelayMs: delay }); + expect(countdown()?.textContent).toContain("Sending in 5s"); + expect(container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')).toBeNull(); + await advance(2_100); + expect(countdown()?.textContent).toContain("Sending in 3s"); + expect(props.onSteer).not.toHaveBeenCalled(); + await advance(3_000); + expect(props.onSteer).toHaveBeenCalledOnce(); + expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-1"); + }); + + it("gives a saved edit a full window while another row's countdown keeps the clock running", async () => { + const start = Date.now(); + const twoRows: IssueQueuedCommentQueue = { ...queue, entries: queue.entries.slice(0, 2).map((entry, index) => { + const at = new Date(start + index * 4_000); + return { ...entry, comment: { ...entry.comment, createdAt: at, updatedAt: at } }; + }) }; + // comment-1 counts down first; the person opens it for editing at 1 s. + const props = render({ queue: twoRows, autoSteerDelayMs: delay }); + await advance(1_000); + await act(async () => { render({ ...props, editingCommentId: "comment-1" }); }); + // comment-2 counts down to start + 9 s and keeps the clock ticking past + // comment-1's old deadline. + await advance(5_000); + expect(props.onSteer).not.toHaveBeenCalled(); + const saved = { ...twoRows, revision: "rev-2", entries: twoRows.entries.map((entry, index) => index === 0 + ? { ...entry, comment: { ...entry.comment, body: "Edited", updatedAt: new Date(Date.now()) } } : entry) }; + await act(async () => { render({ ...props, queue: saved, editingCommentId: null }); }); + await advance(500); + expect(props.onSteer).not.toHaveBeenCalledWith("comment-1", expect.anything()); + expect(countdown()?.textContent).toContain("Sending in"); + await advance(5_000); + expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-2"); + }); + + it("keeps a cancelled message in the queue with its explicit Steer button", async () => { + const props = render({ queue: fresh(), autoSteerDelayMs: delay }); + await act(async () => { + container.querySelector('[data-testid="task-chat-queued-auto-steer-cancel-comment-1"]')?.click(); + await Promise.resolve(); + }); + await advance(10_000); + expect(props.onSteer).not.toHaveBeenCalled(); + expect(countdown()).toBeNull(); + expect(container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')).not.toBeNull(); + }); + + it("waits while the message is being edited and restarts from the saved revision", async () => { + const queued = fresh(); + const props = render({ queue: queued, autoSteerDelayMs: delay }); + await act(async () => { + container.querySelector('[data-testid="task-chat-queued-auto-steer-edit-comment-1"]')?.click(); + await Promise.resolve(); + }); + expect(props.onEdit).toHaveBeenCalledWith("comment-1"); + render({ ...props, editingCommentId: "comment-1" }); + await advance(10_000); + expect(props.onSteer).not.toHaveBeenCalled(); + // The save changes the comment; its countdown starts again from that change. + const saved = { ...queued, revision: "rev-2", entries: queued.entries.map(entry => ({ + ...entry, comment: { ...entry.comment, body: "Edited", updatedAt: new Date(Date.now()) } })) }; + await act(async () => { render({ ...props, queue: saved, editingCommentId: null }); }); + expect(countdown()?.textContent).toContain("Sending in 5s"); + await advance(5_100); + expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-2"); + }); + + it("keeps the ordinary queue while steering is unavailable or unsupported", async () => { + for (const steeringDisposition of ["temporarily_unavailable", "unsupported"] as const) { + const props = render({ queue: fresh({ steeringDisposition }), autoSteerDelayMs: delay }); + expect(countdown()).toBeNull(); + await advance(10_000); + expect(props.onSteer).not.toHaveBeenCalled(); + } + }); + + it("tries each message revision once, then leaves it queued after a failure", async () => { + const props = render({ queue: fresh(), autoSteerDelayMs: delay, onSteer: vi.fn().mockRejectedValue(new Error("rejected")) }); + await advance(5_100); + await advance(10_000); + expect(props.onSteer).toHaveBeenCalledOnce(); + expect(container.textContent).toContain("Couldn’t steer. Message is still queued."); + expect(countdown()).toBeNull(); + }); + + it("never sends someone else’s message or a saved card response automatically", async () => { + const notMine = fresh(); + notMine.entries = notMine.entries.map(entry => ({ ...entry, canEdit: false })); + const card = fresh(); + card.entries = card.entries.map(entry => ({ ...entry, source: { kind: "interaction", summary: "Approved" } as never })); + for (const candidate of [notMine, card]) { + const props = render({ queue: candidate, autoSteerDelayMs: delay }); + await advance(10_000); + expect(props.onSteer).not.toHaveBeenCalled(); + } + }); + + it("stays off unless the caller enables it", async () => { + const props = render({ queue: fresh() }); + expect(countdown()).toBeNull(); + await advance(10_000); + expect(props.onSteer).not.toHaveBeenCalled(); + }); + }); }); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index bbc39af7902..12067e4fae8 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -30,6 +30,11 @@ import type { } from "@paperclipai/shared"; import { cn } from "@/lib/utils"; import { TaskChatComposerBar } from "./TaskChatComposerBar"; +import { + queuedMessageAutoSteerDeadline, + queuedMessageAutoSteerEligible, + queuedMessageAutoSteerKey, +} from "./queued-message-auto-steer"; import { DropdownMenu, DropdownMenuContent, @@ -83,6 +88,10 @@ export interface TaskChatQueuedMessagesProps { onSteer: (commentId: string, revision: string) => Promise; onInterrupt?: () => Promise; onDiscard: (commentId: string, revision: string) => Promise; + /** The queued message open in the editor; its automatic send waits for the save. */ + editingCommentId?: string | null; + /** Steer eligible queued messages automatically after this delay; null keeps every message queued. */ + autoSteerDelayMs?: number | null; } function SortableQueuedMessage({ @@ -91,20 +100,25 @@ function SortableQueuedMessage({ busy, queueMutationDisabled, action, + autoSteerSecondsLeft, onEdit, onSteer, onInterrupt, onDiscard, + onCancelAutoSteer, }: { entry: IssueQueuedCommentEntry; queue: IssueQueuedCommentQueue; busy: boolean; queueMutationDisabled: boolean; action: QueueAction; + /** Seconds until this message is steered automatically, or null when it waits in the queue. */ + autoSteerSecondsLeft: number | null; onEdit: () => void; onSteer: () => void; onInterrupt?: () => void; onDiscard: () => void; + onCancelAutoSteer: () => void; }) { const immutableResponse = entry.source?.kind === "interaction"; const label = queuedMessageLabel(entry); @@ -156,7 +170,32 @@ function SortableQueuedMessage({ {label} - {(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession || + {autoSteerSecondsLeft !== null ? ( + + + Sending in {autoSteerSecondsLeft}s + + + + + ) : (queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession || (queue.steeringDisposition !== "available" && onInterrupt) ? (