From 2c7929ac2f89879b215ebf8dd1ea9078456ba46e Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Sun, 8 Nov 2026 06:20:42 +0530 Subject: [PATCH 1/7] fix(harness): keep the recovery mark for a prompt that wakes an ending run A run that ends visibly (truncated, interrupted, crashloop) cleared the thread's recovery mark unconditionally. A prompt admitted while that run was still going woke the thread again, so its run followed in the same busy period without the mark, and a crash in it was never resumed. idle now keeps the mark while a prompt admitted after the run began is pending, sharing endExecution's pending-prompt rule. A queue that was already parked when the run began stays parked. Co-Authored-By: Claude Opus 5.5 --- .../test/__artifacts__/recovery.ending.json | 274 ++++++++++++++++++ packages/codework/test/recovery.test.ts | 56 ++++ packages/harness/src/runner/loop.ts | 19 +- packages/harness/src/thread/thread.ts | 29 +- 4 files changed, 362 insertions(+), 16 deletions(-) create mode 100644 packages/codework/test/__artifacts__/recovery.ending.json diff --git a/packages/codework/test/__artifacts__/recovery.ending.json b/packages/codework/test/__artifacts__/recovery.ending.json new file mode 100644 index 00000000..d06c8125 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.ending.json @@ -0,0 +1,274 @@ +{ + "log": [ + { + "seq": 1, + "type": "session.created" + }, + { + "seq": 2, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 4, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 5, + "type": "session.thread.system.changed" + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 7, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 8, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 9, + "type": "session.turn.ended" + }, + { + "seq": 10, + "type": "session.llm.started" + }, + { + "seq": 11, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Then answer this." + }, + { + "seq": 12, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 13, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 14, + "type": "session.turn.ended" + }, + { + "seq": 15, + "type": "session.run.failed", + "reason": "truncated", + "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" + }, + { + "seq": 16, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Then answer this." + }, + { + "seq": 17, + "type": "session.llm.started" + }, + { + "seq": 18, + "type": "session.execution.interrupted", + "reason": "crash" + }, + { + "seq": 19, + "type": "session.execution.started", + "resumed": true + }, + { + "seq": 20, + "type": "session.llm.failed", + "reason": "error", + "errorMessage": "The host stopped while the model was answering, on attempt 1", + "parts": [] + }, + { + "seq": 21, + "type": "session.retry.scheduled", + "attempt": 1, + "maxRetries": 3, + "delayMs": 10, + "message": "The host stopped while the model was answering, on attempt 1" + }, + { + "seq": 22, + "type": "session.llm.started" + }, + { + "seq": 23, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answered after the kill." + } + ] + }, + { + "seq": 24, + "type": "session.turn.ended" + }, + { + "seq": 25, + "type": "session.execution.succeeded" + } + ], + "rows": [ + { + "step": "in flight", + "state": "running", + "stateAt": "17 session.llm.started", + "startedAt": "3 session.execution.started", + "endedAt": "15 session.run.failed", + "outcome": "failed", + "error": { + "_tag": "runFailed", + "reason": "truncated", + "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" + }, + "execution": "running" + }, + { + "step": "resumed", + "state": "running", + "stateAt": "22 session.llm.started", + "startedAt": "19 session.execution.started", + "endedAt": null, + "outcome": null, + "error": null, + "execution": "running" + }, + { + "step": "finished", + "state": "idle", + "stateAt": "25 session.execution.succeeded", + "startedAt": "19 session.execution.started", + "endedAt": "25 session.execution.succeeded", + "outcome": "succeeded", + "error": null, + "execution": null + } + ], + "requests": [ + [ + { + "role": "user", + "text": "Run a long command." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "user", + "text": "Then answer this." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "user", + "text": "Then answer this." + } + ] + ] +} diff --git a/packages/codework/test/recovery.test.ts b/packages/codework/test/recovery.test.ts index 6a3e0d30..19185a5a 100644 --- a/packages/codework/test/recovery.test.ts +++ b/packages/codework/test/recovery.test.ts @@ -565,6 +565,62 @@ describe("recovery after the host is killed", () => { ).toMatchFileSnapshot("./__artifacts__/recovery.truncated.json"); }, 90_000); + it("a prompt sent while a run is ending as truncated keeps its run resumable across a kill", async () => { + const cut = gate(); + const resend = gate(); + const truncate = (response: Reply) => { + response.send(toolCalls([{ name: "bash", arguments: { command: "echo cut" } }])); + response.finish("length"); + }; + const fake = await streaming((_body, response, index) => { + if (index === 0) return truncate(response); + // The second cut-off ends the run: held until the prompt is in. + if (index === 1) return void cut.opened.then(() => truncate(response)); + // The run the prompt woke, killed with its request in flight; then its resend. + if (index === 2) return hang(response); + void resend.opened.then(() => answer(response, "Answered after the kill.")); + }); + const { home, start } = setup(fake.url, retry); + + const first = await start(); + const created = await call(first.url, (rpc) => rpc["session.create"]({})); + const ids = { sessionId: created.session.id, threadId: created.id }; + const rows: Array<{ step: string } & ReturnType> = []; + const record = (step: string, events: ReadonlyArray) => + rows.push({ step, ...runRow(home, ids.threadId, events) }); + + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...ids, text: "Run a long command." })); + await logged(first.url, ids, "session.llm.started", 2); + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...ids, text: "Then answer this." })); + await logged(first.url, ids, "session.prompt.admitted", 2); + cut.open(); + await logged(first.url, ids, "session.run.failed"); + const inFlight = await logged(first.url, ids, "session.llm.started", 3); + record("in flight", inFlight); + await first.kill(); + + const second = await start(); + record("resumed", await logged(second.url, ids, "session.llm.started", 4)); + resend.open(); + await logged(second.url, ids, "session.execution.succeeded"); + await call(second.url, (rpc) => rpc["thread.wait"](ids)); + const events = await log(second.url, ids); + record("finished", events); + + expect(rows.find((row) => row.step === "in flight")).toMatchObject({ execution: "running" }); + expect(fake.requests).toHaveLength(4); + expect(sent(fake.requests[3]!)).toEqual(sent(fake.requests[2]!)); + + await expect( + artifact({ + log: events.map((event) => ({ seq: event.durable?.seq, ...summary(event) })), + // Read from the database file: the woken run in flight when the host died, resumed, finished. + rows, + requests: fake.requests.map(sent), + }), + ).toMatchFileSnapshot("./__artifacts__/recovery.ending.json"); + }, 90_000); + it("a deferred tool loaded before the host died is still offered to the resent request", async () => { const tools = (request: ChatRequest) => (request.tools ?? []).map((tool) => tool.function.name).sort(); const fake = await streaming((_body, response, index) => { diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index d77256f1..651bbe66 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -258,10 +258,14 @@ export const layer = (options: Options = {}) => yield* Effect.sleep(Duration.millis(delayMs)); }); - /** Ends a run that is not resumed, leaving the thread idle with its queue kept. */ + /** + * Ends a run that is not resumed, leaving the thread idle with its queue kept. `since` is the log + * position the run began at: a prompt admitted after it woke the thread for another run. + */ const endRun = Effect.fn("Loop.endRun")(function* ( sessionId: SessionSchema.ID, threadId: ThreadSchema.ID, + since: number, reason: EventList.RunFailed["data"]["reason"], message: string, ) { @@ -274,7 +278,7 @@ export const layer = (options: Options = {}) => reason, message, }); - yield* threads.idle(threadId); + yield* threads.idle(threadId, since); }), ); return "ended" as const; @@ -602,10 +606,11 @@ export const layer = (options: Options = {}) => readonly force: boolean; readonly recovered?: boolean; }) { + const thread = yield* threads.get(input.threadId); + if (Option.isNone(thread)) return "drained" as const; + const since = yield* events.latestSequence(thread.value.sessionId); // Only a dead host leaves a request marked in flight: every run that ends clears it. - const crashed = Option.getOrUndefined( - Option.flatMap(yield* threads.get(input.threadId), (row) => row.attempt), - ); + const crashed = Option.getOrUndefined(thread.value.attempt); const carried: Crashed | undefined = crashed === undefined ? undefined @@ -648,7 +653,7 @@ export const layer = (options: Options = {}) => if (carried !== undefined) { const delayMs = Retry.delay(start.retry, carried.attempt, hostStopped(start, carried.message)); if (delayMs === undefined) { - return yield* endRun(start.sessionId, input.threadId, "interrupted", carried.message); + return yield* endRun(start.sessionId, input.threadId, since, "interrupted", carried.message); } yield* scheduleRetry(start, carried.attempt, delayMs, carried.message); } @@ -680,6 +685,7 @@ export const layer = (options: Options = {}) => return yield* endRun( snapshot.sessionId, input.threadId, + since, "truncated", `The response hit the output token limit with tool calls ${truncations} times in a row; the run is ended`, ); @@ -714,6 +720,7 @@ export const layer = (options: Options = {}) => return yield* endRun( SessionSchema.ID.make(thread.value.sessionId), input.threadId, + yield* events.latestSequence(thread.value.sessionId), "crashloop", input.message, ); diff --git a/packages/harness/src/thread/thread.ts b/packages/harness/src/thread/thread.ts index 4f09192c..3c896ab5 100644 --- a/packages/harness/src/thread/thread.ts +++ b/packages/harness/src/thread/thread.ts @@ -234,10 +234,12 @@ export interface Interface { /** Records the 1-based provider request now in flight, or none (null) once its response ended. */ readonly setAttempt: (threadId: ThreadSchema.ID, attempt: number | null) => Effect.Effect; /** - * Leaves a run that is not resumed idle: no mark, no restarts, no request in flight, whatever is - * queued. Its queued prompts wait for the next prompt to wake the thread. + * Leaves a run that is not resumed idle: no restarts, no request in flight, whatever is queued. Its + * queued prompts wait for the next prompt to wake the thread. The mark stays while a prompt + * admitted after `since`, the log position the run began at, is pending: it woke the thread again, + * so its run follows in this process and must survive a crash. */ - readonly idle: (threadId: ThreadSchema.ID) => Effect.Effect; + readonly idle: (threadId: ThreadSchema.ID, since: number) => Effect.Effect; /** Root threads with a run left unfinished, oldest first. */ readonly unfinished: Effect.Effect>; /** @@ -955,6 +957,14 @@ export const layer = Layer.effect( Effect.orDie, ); + /** Whether the thread has a prompt admitted after log position `after` still pending. */ + const pendingAfter = (threadId: ThreadSchema.ID, after: number) => sql` + EXISTS ( + SELECT 1 FROM inbox + WHERE thread_id = ${threadId} AND promoted_seq IS NULL AND cancelled_seq IS NULL AND admitted_seq > ${after} + ) + `; + const markRunning = Effect.fn("Thread.markRunning")( function* (threadId: ThreadSchema.ID) { yield* sql`UPDATE thread SET execution = 'running', rev = ${yield* store.rev} WHERE id = ${threadId}`; @@ -977,11 +987,7 @@ export const layer = Layer.effect( const after = end === "stopped" ? (stoppedAt ?? Number.MAX_SAFE_INTEGER) : -1; yield* sql` UPDATE thread SET execution = NULL, restarts = 0, rev = ${rev} - WHERE id = ${threadId} AND NOT EXISTS ( - SELECT 1 FROM inbox - WHERE thread_id = ${threadId} AND promoted_seq IS NULL AND cancelled_seq IS NULL - AND admitted_seq > ${after} - ) + WHERE id = ${threadId} AND NOT ${pendingAfter(threadId, after)} `; }, store.write, @@ -1020,9 +1026,12 @@ export const layer = Layer.effect( ); const idle = Effect.fn("Thread.idle")( - function* (threadId: ThreadSchema.ID) { + function* (threadId: ThreadSchema.ID, since: number) { yield* sql` - UPDATE thread SET execution = NULL, restarts = 0, attempt = NULL, rev = ${yield* store.rev} WHERE id = ${threadId} + UPDATE thread + SET execution = CASE WHEN ${pendingAfter(threadId, since)} THEN execution ELSE NULL END, + restarts = 0, attempt = NULL, rev = ${yield* store.rev} + WHERE id = ${threadId} `; }, store.write, From b36b59a841d1b97ed7beb0a864a92eaa330a54f0 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Sun, 8 Nov 2026 11:28:07 +0530 Subject: [PATCH 2/7] fix(harness): derive resumable runs from log positions The recovery mark on thread rows was a stored copy of "will another run follow?", and it drifted: a run that ended visibly cleared it while a prompt that had already woken the thread ran next, unmarked, and a crash in that run was never resumed. Replace it with two positions on the session log. inbox.tail is the seq of a prompt's admission or latest move to steer; thread.head is the seq of the last unanswered run end (run.failed, execution.failed, a user stop), written by the projector in that event's commit. A pending prompt past the head is runnable; a thread is unfinished when its last run was cut off by a crash or shutdown, or it has a runnable prompt. - One run per drain that runs: execution.started before anything is promoted, one terminal decided from the stored run state. - A run that ends without an answer never starts the next one: prompts queued before or during its end stay parked for the next prompt. A user stop is recorded even before its run opened. - Crash vs shutdown and restart counting come from the run row; the execution column, markRunning, endExecution, idle and the in-memory stop position are gone. - codework run --server ends on session.run.failed, and attributes a failure before promotion by log seq, not notification order. E2E coverage in recovery.test.ts, cli.test.ts and client.test.ts, plus a real-machine suite on OrbStack (recovery.machines.live.test.ts). Co-Authored-By: Claude Opus 5.5 --- .../recovery.machines.crash.json | 83 ++++ .../recovery.machines.crashloop.json | 125 ++++++ .../recovery.machines.network.json | 99 +++++ .../recovery.machines.shutdown.json | 118 +++++ .../__artifacts__/recovery.machines.stop.json | 105 +++++ .../__artifacts__/sync.machines.crash.json | 4 +- .../test/recovery.machines.live.test.ts | 419 ++++++++++++++++++ apps/server/test/sync.machines.live.test.ts | 2 +- packages/codework/src/server/client.ts | 43 +- packages/codework/src/server/server.ts | 5 +- .../test/__artifacts__/cli.run-failed.json | 4 + ...nding.json => recovery.after-failure.json} | 128 ++++-- .../__artifacts__/recovery.crashloop.json | 2 +- .../__artifacts__/recovery.interrupted.json | 2 +- .../__artifacts__/recovery.lifecycle.json | 10 +- .../test/__artifacts__/recovery.moved.json | 312 +++++++++++++ .../test/__artifacts__/recovery.parked.json | 313 +++++++++++++ .../test/__artifacts__/recovery.shutdown.json | 84 ++++ .../test/__artifacts__/recovery.stop.json | 300 +++++++++++++ .../__artifacts__/recovery.truncated.json | 2 +- .../test/__artifacts__/sync.thread.json | 14 +- packages/codework/test/cli.test.ts | 25 +- packages/codework/test/client.test.ts | 83 ++++ packages/codework/test/fixtures/cli.ts | 9 +- packages/codework/test/recovery.test.ts | 322 ++++++++++++-- packages/codework/test/server.test.ts | 12 +- packages/harness/src/control.ts | 25 +- packages/harness/src/db/migrations.ts | 6 +- packages/harness/src/db/schema.sql.ts | 7 +- packages/harness/src/event/list.ts | 12 +- packages/harness/src/inbox/inbox.ts | 16 +- packages/harness/src/runner/coordinator.ts | 3 - packages/harness/src/runner/execute.ts | 120 +++-- packages/harness/src/runner/loop.ts | 31 +- packages/harness/src/session/projector.ts | 62 ++- packages/harness/src/thread/schema.ts | 8 - packages/harness/src/thread/thread.ts | 134 +++--- .../test/__artifacts__/store.revision.json | 90 ++-- .../test/__artifacts__/thread.lifecycle.json | 45 +- .../test/__artifacts__/thread.recovery.json | 28 +- packages/harness/test/control.prompt.test.ts | 76 ++-- .../harness/test/runner.coordinator.test.ts | 5 +- packages/harness/test/runner.loop.test.ts | 149 +++++-- packages/harness/test/thread.test.ts | 28 +- 44 files changed, 2921 insertions(+), 549 deletions(-) create mode 100644 apps/server/test/__artifacts__/recovery.machines.crash.json create mode 100644 apps/server/test/__artifacts__/recovery.machines.crashloop.json create mode 100644 apps/server/test/__artifacts__/recovery.machines.network.json create mode 100644 apps/server/test/__artifacts__/recovery.machines.shutdown.json create mode 100644 apps/server/test/__artifacts__/recovery.machines.stop.json create mode 100644 apps/server/test/recovery.machines.live.test.ts create mode 100644 packages/codework/test/__artifacts__/cli.run-failed.json rename packages/codework/test/__artifacts__/{recovery.ending.json => recovery.after-failure.json} (80%) create mode 100644 packages/codework/test/__artifacts__/recovery.moved.json create mode 100644 packages/codework/test/__artifacts__/recovery.parked.json create mode 100644 packages/codework/test/__artifacts__/recovery.shutdown.json create mode 100644 packages/codework/test/__artifacts__/recovery.stop.json create mode 100644 packages/codework/test/client.test.ts diff --git a/apps/server/test/__artifacts__/recovery.machines.crash.json b/apps/server/test/__artifacts__/recovery.machines.crash.json new file mode 100644 index 00000000..2cc9e300 --- /dev/null +++ b/apps/server/test/__artifacts__/recovery.machines.crash.json @@ -0,0 +1,83 @@ +{ + "log": [ + "session.prompt.admitted steer \"gate\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"gate\"", + "session.llm.started", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.llm.started", + "session.execution.succeeded" + ], + "steps": [ + { + "step": "in flight", + "head": null, + "restarts": 0, + "runState": "running", + "outcome": null, + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + } + ] + }, + { + "step": "killed", + "head": null, + "restarts": 0, + "runState": "running", + "outcome": null, + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + } + ] + }, + { + "step": "resumed", + "head": null, + "restarts": 1, + "runState": "running", + "outcome": null, + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + } + ] + }, + { + "step": "finished", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + } + ] + } + ] +} diff --git a/apps/server/test/__artifacts__/recovery.machines.crashloop.json b/apps/server/test/__artifacts__/recovery.machines.crashloop.json new file mode 100644 index 00000000..af000175 --- /dev/null +++ b/apps/server/test/__artifacts__/recovery.machines.crashloop.json @@ -0,0 +1,125 @@ +{ + "log": [ + "session.prompt.admitted steer \"hold\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"hold\"", + "session.llm.started", + "session.prompt.admitted followUp \"parked by the loop\"", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.llm.started", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.llm.started", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.llm.started", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.llm.started", + "session.execution.interrupted crash", + "session.execution.started resumed=true", + "session.llm.failed", + "session.run.failed crashloop", + "session.prompt.admitted steer \"next\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"next\"", + "session.llm.started", + "session.prompt.promoted followUp \"parked by the loop\"", + "session.llm.started", + "session.execution.succeeded" + ], + "steps": [ + { + "step": "crash loop", + "head": "session.run.failed crashloop", + "restarts": 0, + "runState": "idle", + "outcome": "failed", + "error": { + "_tag": "runFailed", + "reason": "crashloop", + "message": "The run was interrupted 5 times in a row; it is not resumed again" + }, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "parked by the loop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"parked by the loop\"", + "promoted": null, + "runnable": false + } + ] + }, + { + "step": "restarted", + "head": "session.run.failed crashloop", + "restarts": 0, + "runState": "idle", + "outcome": "failed", + "error": { + "_tag": "runFailed", + "reason": "crashloop", + "message": "The run was interrupted 5 times in a row; it is not resumed again" + }, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "parked by the loop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"parked by the loop\"", + "promoted": null, + "runnable": false + } + ] + }, + { + "step": "next prompt", + "head": "session.run.failed crashloop", + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "parked by the loop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"parked by the loop\"", + "promoted": "session.prompt.promoted followUp \"parked by the loop\"", + "runnable": false + }, + { + "text": "next", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"next\"", + "promoted": "session.prompt.promoted steer \"next\"", + "runnable": false + } + ] + } + ] +} diff --git a/apps/server/test/__artifacts__/recovery.machines.network.json b/apps/server/test/__artifacts__/recovery.machines.network.json new file mode 100644 index 00000000..34a12680 --- /dev/null +++ b/apps/server/test/__artifacts__/recovery.machines.network.json @@ -0,0 +1,99 @@ +{ + "unreachable": true, + "log": [ + "session.prompt.admitted steer \"gate\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"gate\"", + "session.llm.started", + "session.prompt.admitted followUp \"sent while cut\"", + "session.prompt.promoted followUp \"sent while cut\"", + "session.llm.started", + "session.execution.succeeded", + "session.prompt.admitted steer \"on a slow link\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"on a slow link\"", + "session.llm.started", + "session.execution.succeeded" + ], + "steps": [ + { + "step": "cut", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "sent while cut", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"sent while cut\"", + "promoted": "session.prompt.promoted followUp \"sent while cut\"", + "runnable": false + } + ] + }, + { + "step": "retried", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "sent while cut", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"sent while cut\"", + "promoted": "session.prompt.promoted followUp \"sent while cut\"", + "runnable": false + } + ] + }, + { + "step": "slow link", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "sent while cut", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"sent while cut\"", + "promoted": "session.prompt.promoted followUp \"sent while cut\"", + "runnable": false + }, + { + "text": "on a slow link", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"on a slow link\"", + "promoted": "session.prompt.promoted steer \"on a slow link\"", + "runnable": false + } + ] + } + ] +} diff --git a/apps/server/test/__artifacts__/recovery.machines.shutdown.json b/apps/server/test/__artifacts__/recovery.machines.shutdown.json new file mode 100644 index 00000000..72ed6094 --- /dev/null +++ b/apps/server/test/__artifacts__/recovery.machines.shutdown.json @@ -0,0 +1,118 @@ +{ + "reason": "shutdown", + "log": [ + "session.prompt.admitted steer \"gate\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"gate\"", + "session.llm.started", + "session.prompt.admitted followUp \"after the boot\"", + "session.llm.failed", + "session.execution.interrupted shutdown", + "session.execution.started resumed=true", + "session.llm.started", + "session.prompt.promoted followUp \"after the boot\"", + "session.llm.started", + "session.execution.succeeded" + ], + "steps": [ + { + "step": "in flight", + "head": null, + "restarts": 0, + "runState": "running", + "outcome": null, + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "after the boot", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the boot\"", + "promoted": null, + "runnable": true + } + ] + }, + { + "step": "booted", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "interrupted", + "error": { + "_tag": "interrupted", + "reason": "shutdown" + }, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "after the boot", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the boot\"", + "promoted": null, + "runnable": true + } + ] + }, + { + "step": "resumed", + "head": null, + "restarts": 0, + "runState": "running", + "outcome": null, + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "after the boot", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the boot\"", + "promoted": null, + "runnable": true + } + ] + }, + { + "step": "finished", + "head": null, + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "gate", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"gate\"", + "promoted": "session.prompt.promoted steer \"gate\"", + "runnable": false + }, + { + "text": "after the boot", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the boot\"", + "promoted": "session.prompt.promoted followUp \"after the boot\"", + "runnable": false + } + ] + } + ] +} diff --git a/apps/server/test/__artifacts__/recovery.machines.stop.json b/apps/server/test/__artifacts__/recovery.machines.stop.json new file mode 100644 index 00000000..afcb7ff8 --- /dev/null +++ b/apps/server/test/__artifacts__/recovery.machines.stop.json @@ -0,0 +1,105 @@ +{ + "log": [ + "session.prompt.admitted steer \"hold\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"hold\"", + "session.llm.started", + "session.prompt.admitted followUp \"after the stop\"", + "session.llm.failed", + "session.execution.interrupted user", + "session.prompt.admitted steer \"next\"", + "session.execution.started resumed=false", + "session.prompt.promoted steer \"next\"", + "session.llm.started", + "session.prompt.promoted followUp \"after the stop\"", + "session.llm.started", + "session.execution.succeeded" + ], + "steps": [ + { + "step": "stopped", + "head": "session.execution.interrupted user", + "restarts": 0, + "runState": "idle", + "outcome": "interrupted", + "error": { + "_tag": "interrupted", + "reason": "user" + }, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "after the stop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the stop\"", + "promoted": null, + "runnable": false + } + ] + }, + { + "step": "restarted", + "head": "session.execution.interrupted user", + "restarts": 0, + "runState": "idle", + "outcome": "interrupted", + "error": { + "_tag": "interrupted", + "reason": "user" + }, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "after the stop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the stop\"", + "promoted": null, + "runnable": false + } + ] + }, + { + "step": "next prompt", + "head": "session.execution.interrupted user", + "restarts": 0, + "runState": "idle", + "outcome": "succeeded", + "error": null, + "prompts": [ + { + "text": "hold", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"hold\"", + "promoted": "session.prompt.promoted steer \"hold\"", + "runnable": false + }, + { + "text": "after the stop", + "delivery": "followUp", + "tail": "session.prompt.admitted followUp \"after the stop\"", + "promoted": "session.prompt.promoted followUp \"after the stop\"", + "runnable": false + }, + { + "text": "next", + "delivery": "steer", + "tail": "session.prompt.admitted steer \"next\"", + "promoted": "session.prompt.promoted steer \"next\"", + "runnable": false + } + ] + } + ] +} diff --git a/apps/server/test/__artifacts__/sync.machines.crash.json b/apps/server/test/__artifacts__/sync.machines.crash.json index ac144cad..a4a5a89e 100644 --- a/apps/server/test/__artifacts__/sync.machines.crash.json +++ b/apps/server/test/__artifacts__/sync.machines.crash.json @@ -4,13 +4,13 @@ "run_state": "running", "run_outcome": null, "run_error": null, - "execution": "running" + "head": 0 }, "swept": { "run_state": "idle", "run_outcome": "interrupted", "run_error": "{\"_tag\":\"interrupted\",\"reason\":\"crash\"}", - "execution": "running" + "head": 0 }, "events": [ { diff --git a/apps/server/test/recovery.machines.live.test.ts b/apps/server/test/recovery.machines.live.test.ts new file mode 100644 index 00000000..04308e6f --- /dev/null +++ b/apps/server/test/recovery.machines.live.test.ts @@ -0,0 +1,419 @@ +/* @effect-diagnostics nodeBuiltinImport:off -- the suite runs a CodeWork server on an OrbStack Linux machine. */ +import { Effect, Stream } from "effect"; +import { afterAll, beforeAll, describe, expect, it } from "vite-plus/test"; +import { Control, Session } from "../../../packages/harness/src/effect.ts"; +import { snapshot as match, waitFor } from "./fixtures/apps.ts"; +import { available, ensure, machine, orb } from "./machines/orb.ts"; + +/* + * Durable runs across real failures of a real computer: the fixture CodeWork server (scripted model) + * runs on the OrbStack machine cw-a while the suite SIGKILLs it, shuts the whole machine down, + * crash-loops it, and cuts and throttles its network. What the store holds is read inside the + * machine with sqlite3; `recovery.machines..json` records the log and the rows across each + * failure, with log positions named by the event at them. + */ + +const cw = machine("cw-a"); +const work = `${cw.cwHome}/work`; +type Ids = { readonly sessionId: string; readonly threadId: string }; +type Row = Record; + +/** + * Waits until the Mac reaches the server: after the machine boots, OrbStack takes up to a minute to + * route `.orb.local` to it again. + */ +const reachable = () => + waitFor( + () => + fetch(cw.url.replace(/^ws/, "http"), { signal: AbortSignal.timeout(3_000) }).then( + () => true, + () => undefined, + ), + "the server from the Mac", + 180_000, + ); + +/** A fresh store, with retries fast and many so a crash loop is the host's, not the request's. */ +const fresh = async () => { + await cw.net("restore"); + await cw.net("fast"); + await cw.stop("KILL"); + await cw.wipe(); + await cw.sh( + `mkdir -p ${work} && printf '%s' '{"retry":{"maxRetries":10,"baseDelayMs":10,"maxDelayMs":1000}}' > ${cw.cwHome}/settings.jsonc`, + ); + await cw.start(); + await reachable(); +}; + +const create = () => + cw.run((rpc) => + rpc["session.create"]({ directory: work }).pipe( + Effect.map((created): Ids => ({ sessionId: created.session.id, threadId: created.id })), + ), + ); + +const prompt = (ids: Ids, text: string, options: { delivery?: "steer" | "followUp"; id?: string } = {}) => + cw.run((rpc) => + rpc["thread.prompt"]({ + sessionId: Session.SessionSchema.ID.make(ids.sessionId), + threadId: Session.ThreadSchema.ID.make(ids.threadId), + text, + ...(options.delivery === undefined ? {} : { delivery: options.delivery }), + ...(options.id === undefined ? {} : { id: Session.SessionMessageSchema.ID.make(options.id) }), + }), + ); + +const ref = (ids: Ids) => ({ + sessionId: Session.SessionSchema.ID.make(ids.sessionId), + threadId: Session.ThreadSchema.ID.make(ids.threadId), +}); +const idle = (ids: Ids) => cw.run((rpc) => rpc["thread.wait"](ref(ids))); +const queue = (ids: Ids) => + cw.run((rpc) => rpc["thread.queue"](ref(ids)).pipe(Effect.map(({ entries }) => entries.map((one) => one.text)))); + +type Event = { readonly seq: number; readonly type: string; readonly data: Row }; + +/** The session's durable log, read in the machine; the type without its version. */ +const events = async (ids: Ids): Promise> => + (await cw.sql(`SELECT seq, type, data FROM event WHERE aggregate_id = '${ids.sessionId}' ORDER BY seq`)).map( + (row) => ({ + seq: Number(row["seq"]), + type: (row["type"] as string).replace(/\.\d+$/, ""), + data: JSON.parse(row["data"] as string) as Row, + }), + ); + +const textOf = (data: Row) => + ((data["prompt"] as { parts?: ReadonlyArray<{ text?: string }> } | undefined)?.parts ?? []) + .map((part) => part.text ?? "") + .join(""); + +const shown = new Set([ + "session.prompt.admitted", + "session.prompt.promoted", + "session.prompt.updated", + "session.execution.started", + "session.execution.succeeded", + "session.execution.failed", + "session.execution.interrupted", + "session.run.failed", + "session.llm.started", + "session.llm.failed", +]); + +/** What an event says, without ids, times or messages that carry them. */ +const describeEvent = (event: Event) => { + const { type, data } = event; + if (type === "session.prompt.admitted" || type === "session.prompt.promoted") + return `${type} ${data["delivery"] as string} "${textOf(data)}"`; + if (type === "session.prompt.updated") return `${type} ${(data["delivery"] as string | undefined) ?? ""}`.trim(); + if (type === "session.execution.started") return `${type} resumed=${String(data["resumed"] as boolean)}`; + if (type === "session.execution.interrupted" || type === "session.run.failed") + return `${type} ${data["reason"] as string}`; + return type; +}; + +const count = (log: ReadonlyArray, type: string) => log.filter((event) => event.type === type).length; + +/** The thread's row and its prompts, log positions named by the event at them. */ +const rows = async (ids: Ids) => { + const log = await events(ids); + const at = (seq: unknown) => { + if (seq === null || seq === undefined || seq === 0) return null; + const event = log.find((one) => one.seq === Number(seq)); + return event === undefined ? `unmatched ${seq as number}` : describeEvent(event); + }; + const [thread] = await cw.sql( + `SELECT head, restarts, run_state, run_outcome, run_error FROM thread WHERE id = '${ids.threadId}'`, + ); + const prompts = await cw.sql( + `SELECT data, delivery, admitted_seq, tail, promoted_seq FROM inbox WHERE thread_id = '${ids.threadId}' ORDER BY admitted_seq`, + ); + const head = Number(thread!["head"]); + return { + head: at(head), + restarts: thread!["restarts"], + runState: thread!["run_state"], + outcome: thread!["run_outcome"], + error: thread!["run_error"] === null ? null : (JSON.parse(thread!["run_error"] as string) as Row), + prompts: prompts.map((row) => ({ + text: textOf({ prompt: JSON.parse(row["data"] as string) }), + delivery: row["delivery"], + tail: at(row["tail"]), + promoted: at(row["promoted_seq"]), + runnable: row["promoted_seq"] === null && Number(row["tail"]) > head, + })), + }; +}; + +/** The lifecycle part of the log, in order. */ +const lifecycle = async (ids: Ids) => + (await events(ids)).filter((event) => shown.has(event.type)).map((event) => describeEvent(event)); + +/** Waits until the log holds `n` events of `type`, and returns it. */ +const logged = (ids: Ids, type: string, n = 1, timeout = 120_000) => + waitFor( + async () => { + const log = await events(ids); + return count(log, type) >= n ? log : undefined; + }, + `${n} × ${type}`, + timeout, + ); + +/** Waits until the thread's request is out, as the store records it. */ +const inFlight = (ids: Ids, timeout = 120_000) => + waitFor( + async () => { + const [row] = await cw.sql(`SELECT run_state, attempt FROM thread WHERE id = '${ids.threadId}'`); + return row?.["run_state"] === "running" && row["attempt"] !== null ? row : undefined; + }, + "the request in flight", + timeout, + ); + +/** Gives a server that just started time to resume what it owes, then says what ran since `from`. */ +const settledSince = async (ids: Ids, from: number) => { + await new Promise((resolve) => setTimeout(resolve, 3_000)); + await idle(ids); + return (await events(ids)).filter((event) => event.seq > from && shown.has(event.type)).map(describeEvent); +}; + +const lastSeq = async (ids: Ids) => (await events(ids)).at(-1)?.seq ?? 0; + +const releases = async (n: number) => { + for (let step = 0; step < n; step++) await cw.release(); +}; + +const snapshot = (name: string, artifact: unknown) => match(`recovery.machines.${name}`, artifact); + +const describeMachines = available ? describe : describe.skip; + +describeMachines("durable runs across real machine failures", () => { + beforeAll(async () => { + await ensure(cw.name); + await cw.setup(); + }, 1_800_000); + + afterAll(async () => { + await cw.net("fast").catch(() => undefined); + await cw.net("restore").catch(() => undefined); + await cw.stop("KILL").catch(() => undefined); + }, 120_000); + + it("1. a host SIGKILLed mid-request is swept as a crash, resumed with one restart, and finishes", async () => { + await fresh(); + const ids = await create(); + const steps: Array<{ step: string } & Awaited>> = []; + const record = async (step: string) => steps.push({ step, ...(await rows(ids)) }); + + await prompt(ids, "gate"); + await inFlight(ids); + await record("in flight"); + await cw.stop("KILL"); + await record("killed"); + + await cw.start(); + await logged(ids, "session.execution.started", 2); + await inFlight(ids); + await record("resumed"); + expect(steps.at(-1)).toMatchObject({ restarts: 1, runState: "running" }); + // The resumed request is the scripted `gate` again: one step at once, four held. + await releases(4); + await logged(ids, "session.execution.succeeded"); + await idle(ids); + await record("finished"); + + expect(steps.at(-1)).toMatchObject({ restarts: 0, outcome: "succeeded" }); + const log = await lifecycle(ids); + expect(log).toContain("session.execution.interrupted crash"); + expect(log).toContain("session.execution.started resumed=true"); + await snapshot("crash", { log, steps }); + }, 300_000); + + it("2. a machine shut down mid-request and booted again resumes the run, its queue intact", async () => { + await fresh(); + const ids = await create(); + const steps: Array<{ step: string } & Awaited>> = []; + const record = async (step: string) => steps.push({ step, ...(await rows(ids)) }); + + await prompt(ids, "gate"); + await inFlight(ids); + await prompt(ids, "after the boot", { delivery: "followUp" }); + await record("in flight"); + + const init = () => cw.sh("ps -o lstart= -p 1"); + const bootedAt = await init(); + await orb(["stop", cw.name], { timeout: 300_000 }); + const machines = JSON.parse(await orb(["list", "--format", "json"])) as Array<{ name: string; state: string }>; + expect(machines.find((one) => one.name === cw.name)?.state).toBe("stopped"); + await orb(["start", cw.name], { timeout: 300_000 }); + expect(await init()).not.toBe(bootedAt); + // A boot starts nothing: the server runs only once start.sh starts it. + expect((await cw.sh(`pgrep -f '[f]ixtures/codework.ts' || true`)).trim()).toBe(""); + await cw.sh(`rm -f $HOME/cw.pid`); + await record("booted"); + const ended = steps.at(-1)!.error as { reason?: string } | null; + + await cw.start(); + await reachable(); + await logged(ids, "session.execution.started", 2); + await inFlight(ids); + await record("resumed"); + await releases(4); + await logged(ids, "session.execution.succeeded", 1, 120_000); + await waitFor(async () => ((await queue(ids)).length === 0 ? true : undefined), "the follow-up to run", 120_000); + await logged(ids, "session.execution.succeeded"); + await idle(ids); + await record("finished"); + + // A shutdown that published its own end resumes uncounted; one the OS cut off first is a crash. + const reason = ended === null ? "crash" : ended.reason; + expect(["shutdown", "crash"]).toContain(reason); + expect(steps.find((step) => step.step === "resumed")).toMatchObject({ + restarts: reason === "shutdown" ? 0 : 1, + }); + const finished = steps.at(-1)!; + expect(finished).toMatchObject({ restarts: 0, outcome: "succeeded" }); + expect(finished.prompts.map((one) => [one.text, one.promoted !== null])).toEqual([ + ["gate", true], + ["after the boot", true], + ]); + await snapshot("shutdown", { reason, log: await lifecycle(ids), steps }); + }, 900_000); + + it("3. a user stop parks the queue: a SIGKILL and restart run nothing, the next prompt runs it all", async () => { + await fresh(); + const ids = await create(); + const steps: Array<{ step: string } & Awaited>> = []; + const record = async (step: string) => steps.push({ step, ...(await rows(ids)) }); + + await prompt(ids, "hold"); + await inFlight(ids); + await prompt(ids, "after the stop", { delivery: "followUp" }); + await cw.run((rpc) => rpc["thread.interrupt"](ref(ids))); + await logged(ids, "session.execution.interrupted"); + await idle(ids); + await record("stopped"); + const stopped = steps.at(-1)!; + expect(stopped.head).toBe("session.execution.interrupted user"); + expect(stopped.prompts.find((one) => one.text === "after the stop")).toMatchObject({ runnable: false }); + + const before = await lastSeq(ids); + await cw.stop("KILL"); + await cw.start(); + expect(await settledSince(ids, before)).toEqual([]); + expect(await queue(ids)).toEqual(["after the stop"]); + await record("restarted"); + + await prompt(ids, "next"); + await waitFor(async () => ((await queue(ids)).length === 0 ? true : undefined), "the queue to run", 120_000); + await idle(ids); + await record("next prompt"); + const ran = (await lifecycle(ids)).filter((line) => line.startsWith("session.prompt.promoted")); + expect(ran).toEqual([ + 'session.prompt.promoted steer "hold"', + 'session.prompt.promoted steer "next"', + 'session.prompt.promoted followUp "after the stop"', + ]); + await snapshot("stop", { log: await lifecycle(ids), steps }); + }, 300_000); + + it("4. a host that keeps dying ends the run as a crash loop: its follow-up stays parked across another restart", async () => { + await fresh(); + const ids = await create(); + const steps: Array<{ step: string } & Awaited>> = []; + const record = async (step: string) => steps.push({ step, ...(await rows(ids)) }); + + await prompt(ids, "hold"); + await inFlight(ids); + await prompt(ids, "parked by the loop", { delivery: "followUp" }); + // The first run, then a resume each until the limit: each dies with its request in flight. + for (let started = 1; started <= Control.RESTART_LIMIT; started++) { + await logged(ids, "session.llm.started", started); + await inFlight(ids); + await cw.stop("KILL"); + await cw.start(); + } + await logged(ids, "session.run.failed"); + await idle(ids); + await record("crash loop"); + const looped = steps.at(-1)!; + expect(looped.error).toMatchObject({ _tag: "runFailed", reason: "crashloop" }); + expect(looped.head).toBe("session.run.failed crashloop"); + expect(looped.prompts.find((one) => one.text === "parked by the loop")).toMatchObject({ runnable: false }); + + const before = await lastSeq(ids); + await cw.stop("KILL"); + await cw.start(); + expect(await settledSince(ids, before)).toEqual([]); + await record("restarted"); + + await prompt(ids, "next"); + await waitFor(async () => ((await queue(ids)).length === 0 ? true : undefined), "the queue to run", 120_000); + await idle(ids); + await record("next prompt"); + expect(steps.at(-1)).toMatchObject({ restarts: 0, outcome: "succeeded" }); + const log = await lifecycle(ids); + expect(count(await events(ids), "session.llm.started")).toBe(Control.RESTART_LIMIT + 2); + await snapshot("crashloop", { log, steps }); + }, 900_000); + + it("5. a cut network leaves the run going; a retried prompt id lands once; the client catches up, also on a slow link", async () => { + await fresh(); + const ids = await create(); + const steps: Array<{ step: string } & Awaited>> = []; + const record = async (step: string) => steps.push({ step, ...(await rows(ids)) }); + const sent = Session.SessionMessageSchema.ID.create(); + + await prompt(ids, "gate", { id: sent }); + await inFlight(ids); + await cw.net("cut"); + const unreachable = await cw + .run((rpc) => rpc["thread.info"](ref(ids))) + .then( + () => false, + () => true, + ); + expect(unreachable).toBe(true); + + // Inside the machine, over loopback: a follow-up, and the held reply let through. + await cw.drive([["thread.prompt", { ...ref(ids), text: "sent while cut", delivery: "followUp" }]]); + await releases(4); + await logged(ids, "session.execution.succeeded"); + await waitFor( + async () => ((await rows(ids)).prompts.every((one) => one.promoted !== null) ? true : undefined), + "the follow-up to run", + 120_000, + ); + await record("cut"); + await cw.net("restore"); + + // The client never heard back about its prompt before the cut; retrying it is the same prompt. + await prompt(ids, "gate", { id: sent }); + await idle(ids); + await record("retried"); + const log = await events(ids); + expect( + log.filter((event) => event.type === "session.prompt.admitted" && event.data["messageId"] === sent), + ).toHaveLength(1); + expect(steps.at(-1)!.prompts).toEqual(steps.at(-2)!.prompts); + + const succeeded = count(await events(ids), "session.execution.succeeded"); + await cw.net("slow"); + await prompt(ids, "on a slow link"); + await logged(ids, "session.execution.succeeded", succeeded + 1); + await idle(ids); + const caught = await cw.run((rpc) => + rpc["session.log"]({ sessionId: Session.SessionSchema.ID.make(ids.sessionId) }).pipe( + Stream.runCollect, + Effect.map((items) => Array.from(items).filter((item) => item.type !== "log.synced").length), + ), + ); + await cw.net("fast"); + await record("slow link"); + expect(caught).toBe((await events(ids)).length); + await snapshot("network", { unreachable, log: await lifecycle(ids), steps }); + }, 600_000); +}); diff --git a/apps/server/test/sync.machines.live.test.ts b/apps/server/test/sync.machines.live.test.ts index ccbf0967..db292855 100644 --- a/apps/server/test/sync.machines.live.test.ts +++ b/apps/server/test/sync.machines.live.test.ts @@ -471,7 +471,7 @@ describeMachines("sync across OrbStack machines", () => { const crashedAt = lastSeq(appHome); await cwA.stop("KILL"); - const harnessRow = `SELECT run_state, run_outcome, run_error, execution FROM thread WHERE id = '${ref.threadId}'`; + const harnessRow = `SELECT run_state, run_outcome, run_error, head FROM thread WHERE id = '${ref.threadId}'`; const open = await cwA.sql(harnessRow); await cwA.sweep(); const swept = await cwA.sql(harnessRow); diff --git a/packages/codework/src/server/client.ts b/packages/codework/src/server/client.ts index 43948386..ed90907d 100644 --- a/packages/codework/src/server/client.ts +++ b/packages/codework/src/server/client.ts @@ -41,9 +41,20 @@ export const run = Effect.fn("Client.run")(function* ( let admitted = false; let promoted = false; let settled = false; - let unattributed: ExecutionError | undefined; - // Raised once a busy period ends without our prompt in it: the session may be - // about to go idle, and only idleness proves our prompt is never running. + // By log position, since notifications may arrive out of order: a run that started after our + // admission is one our prompt woke, so its failure before promoting us is ours. A run already going + // fails on its own, and parks our prompt behind it. + let admittedAt: number | undefined; + const starts: Array = []; + // Runs that ended before promoting us, failed ones with their error. + const ends: Array<{ readonly at: number; readonly error: ExecutionError | undefined }> = []; + const ours = () => + ends.findLast(({ at, error }) => { + const start = Math.max(...starts.filter((seq) => seq < at)); + return error !== undefined && admittedAt !== undefined && start > admittedAt; + })?.error; + // Raised once a run ends after our admission without our prompt in it: the session may be about to + // go idle, and only idleness proves our prompt is never running. const stalled = yield* Deferred.make(); const terminal = (event: EventSchema.Payload): ExecutionError | undefined => { @@ -52,6 +63,8 @@ export const run = Effect.fn("Client.run")(function* ( type: "aborted", message: event.data.reason === "shutdown" ? "Server stopped the session" : "Session interrupted", }); + if (Schema.is(EventList.RunFailed)(event)) + return new ExecutionError({ type: `run.${event.data.reason}`, message: event.data.message }); if (Schema.is(EventList.ExecutionFailed)(event)) return new ExecutionError({ type: event.data.error.type, @@ -72,21 +85,22 @@ export const run = Effect.fn("Client.run")(function* ( Effect.catchTag("UnknownEventTypeError", () => Effect.void), ); if (event === undefined) return; - if (Schema.is(EventList.PromptAdmitted)(event) && event.data.messageId === id) admitted = true; - if (Schema.is(EventList.Prompted)(event) && event.data.messageId === id) { - promoted = true; - // Our prompt is running after all; whatever ended before it was not ours. - unattributed = undefined; + const seq = event.durable?.seq; + if (Schema.is(EventList.ExecutionStarted)(event) && seq !== undefined) starts.push(seq); + if (Schema.is(EventList.PromptAdmitted)(event) && event.data.messageId === id) { + admitted = true; + admittedAt = seq; } - if (!admitted) return; + if (Schema.is(EventList.Prompted)(event) && event.data.messageId === id) promoted = true; const ended = terminal(event); const succeeded = Schema.is(EventList.ExecutionSucceeded)(event); if (!promoted) { - if (ended !== undefined) unattributed = ended; - // A busy period that ended without us: wait for idle to find out - // whether a successor still picks our prompt up. - if (ended !== undefined || succeeded) yield* Deferred.succeed(stalled, undefined); + // Kept whatever arrives first: our admission comes from another fiber than the runs, so a + // run's end may be notified before it. + if ((ended !== undefined || succeeded) && seq !== undefined) ends.push({ at: seq, error: ended }); + if (admitted && ends.some(({ at }) => admittedAt === undefined || at > admittedAt)) + yield* Deferred.succeed(stalled, undefined); return; } if (ended !== undefined) return yield* ended; @@ -121,8 +135,7 @@ export const run = Effect.fn("Client.run")(function* ( yield* Effect.raceFirst(consume, reconcile); if (settled) return; - if (abandoned) - return yield* unattributed ?? new ExecutionError({ type: "unknown", message: "Prompt was not promoted" }); + if (abandoned) return yield* ours() ?? new ExecutionError({ type: "unknown", message: "Prompt was not promoted" }); return yield* new ExecutionError({ type: "transport.disconnected", message: "Event stream ended before the prompt finished; its output was not rendered", diff --git a/packages/codework/src/server/server.ts b/packages/codework/src/server/server.ts index 282baeb4..f65e4033 100644 --- a/packages/codework/src/server/server.ts +++ b/packages/codework/src/server/server.ts @@ -57,8 +57,9 @@ const managedShutdown = Layer.unwrap( const sweep = Layer.effectDiscard(Effect.flatMap(Control.Service, (control) => control.sweep)); /** - * Resumes the runs the last server left unfinished, whether it crashed or shut down: the session - * table records them as they start, so this needs nothing from the previous process. + * Resumes the runs the last server left unfinished, whether it crashed or shut down, and the prompts + * woken past a thread's head: the log and its projections record both, so this needs nothing from the + * previous process. */ const resumeUnfinished = Layer.effectDiscard( Effect.gen(function* () { diff --git a/packages/codework/test/__artifacts__/cli.run-failed.json b/packages/codework/test/__artifacts__/cli.run-failed.json new file mode 100644 index 00000000..a168a07a --- /dev/null +++ b/packages/codework/test/__artifacts__/cli.run-failed.json @@ -0,0 +1,4 @@ +{ + "status": 1, + "stderr": "error[run.truncated]: The response hit the output token limit with tool calls 2 times in a row; the run is ended\n" +} diff --git a/packages/codework/test/__artifacts__/recovery.ending.json b/packages/codework/test/__artifacts__/recovery.after-failure.json similarity index 80% rename from packages/codework/test/__artifacts__/recovery.ending.json rename to packages/codework/test/__artifacts__/recovery.after-failure.json index d06c8125..e86f7f3d 100644 --- a/packages/codework/test/__artifacts__/recovery.ending.json +++ b/packages/codework/test/__artifacts__/recovery.after-failure.json @@ -59,12 +59,6 @@ }, { "seq": 11, - "type": "session.prompt.admitted", - "delivery": "steer", - "text": "Then answer this." - }, - { - "seq": 12, "type": "session.llm.ended", "reason": "length", "parts": [ @@ -75,7 +69,7 @@ ] }, { - "seq": 13, + "seq": 12, "type": "session.tool.settled", "part": { "tool": "bash", @@ -84,44 +78,55 @@ } }, { - "seq": 14, + "seq": 13, "type": "session.turn.ended" }, { - "seq": 15, + "seq": 14, "type": "session.run.failed", "reason": "truncated", "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" }, + { + "seq": 15, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Now this." + }, { "seq": 16, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 17, "type": "session.prompt.promoted", "delivery": "steer", - "text": "Then answer this." + "text": "Now this." }, { - "seq": 17, + "seq": 18, "type": "session.llm.started" }, { - "seq": 18, + "seq": 19, "type": "session.execution.interrupted", "reason": "crash" }, { - "seq": 19, + "seq": 20, "type": "session.execution.started", "resumed": true }, { - "seq": 20, + "seq": 21, "type": "session.llm.failed", "reason": "error", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, { - "seq": 21, + "seq": 22, "type": "session.retry.scheduled", "attempt": 1, "maxRetries": 3, @@ -129,11 +134,11 @@ "message": "The host stopped while the model was answering, on attempt 1" }, { - "seq": 22, + "seq": 23, "type": "session.llm.started" }, { - "seq": 23, + "seq": 24, "type": "session.llm.ended", "reason": "stop", "parts": [ @@ -143,48 +148,77 @@ ] }, { - "seq": 24, + "seq": 25, "type": "session.turn.ended" }, { - "seq": 25, + "seq": 26, "type": "session.execution.succeeded" } ], - "rows": [ + "steps": [ { "step": "in flight", - "state": "running", - "stateAt": "17 session.llm.started", - "startedAt": "3 session.execution.started", - "endedAt": "15 session.run.failed", - "outcome": "failed", - "error": { - "_tag": "runFailed", - "reason": "truncated", - "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" - }, - "execution": "running" + "head": 14, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Now this.", + "delivery": "steer", + "admittedSeq": 15, + "tail": 15, + "promotedSeq": 17 + } + ] }, { "step": "resumed", - "state": "running", - "stateAt": "22 session.llm.started", - "startedAt": "19 session.execution.started", - "endedAt": null, - "outcome": null, - "error": null, - "execution": "running" + "head": 14, + "restarts": 1, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Now this.", + "delivery": "steer", + "admittedSeq": 15, + "tail": 15, + "promotedSeq": 17 + } + ] }, { "step": "finished", - "state": "idle", - "stateAt": "25 session.execution.succeeded", - "startedAt": "19 session.execution.started", - "endedAt": "25 session.execution.succeeded", - "outcome": "succeeded", - "error": null, - "execution": null + "head": 14, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Now this.", + "delivery": "steer", + "admittedSeq": 15, + "tail": 15, + "promotedSeq": 17 + } + ] } ], "requests": [ @@ -237,7 +271,7 @@ }, { "role": "user", - "text": "Then answer this." + "text": "Now this." } ], [ @@ -267,7 +301,7 @@ }, { "role": "user", - "text": "Then answer this." + "text": "Now this." } ] ] diff --git a/packages/codework/test/__artifacts__/recovery.crashloop.json b/packages/codework/test/__artifacts__/recovery.crashloop.json index a6202630..d3e6d136 100644 --- a/packages/codework/test/__artifacts__/recovery.crashloop.json +++ b/packages/codework/test/__artifacts__/recovery.crashloop.json @@ -236,7 +236,7 @@ "reason": "crashloop", "message": "The run was interrupted 5 times in a row; it is not resumed again" }, - "execution": null + "head": 30 }, "requests": [ [ diff --git a/packages/codework/test/__artifacts__/recovery.interrupted.json b/packages/codework/test/__artifacts__/recovery.interrupted.json index 88ba0a37..27a6215e 100644 --- a/packages/codework/test/__artifacts__/recovery.interrupted.json +++ b/packages/codework/test/__artifacts__/recovery.interrupted.json @@ -110,7 +110,7 @@ "reason": "interrupted", "message": "The host stopped while the model was answering, on attempt 2" }, - "execution": null + "head": 16 }, "requests": [ [ diff --git a/packages/codework/test/__artifacts__/recovery.lifecycle.json b/packages/codework/test/__artifacts__/recovery.lifecycle.json index 1f1c8114..a9838497 100644 --- a/packages/codework/test/__artifacts__/recovery.lifecycle.json +++ b/packages/codework/test/__artifacts__/recovery.lifecycle.json @@ -50,7 +50,7 @@ "endedAt": "9 session.execution.succeeded", "outcome": "succeeded", "error": null, - "execution": null + "head": 0 }, { "step": "in flight", @@ -60,7 +60,7 @@ "endedAt": null, "outcome": null, "error": null, - "execution": "running" + "head": 0 }, { "step": "killed", @@ -70,7 +70,7 @@ "endedAt": null, "outcome": null, "error": null, - "execution": "running" + "head": 0 }, { "step": "resumed", @@ -80,7 +80,7 @@ "endedAt": null, "outcome": null, "error": null, - "execution": "running" + "head": 0 }, { "step": "finished", @@ -90,7 +90,7 @@ "endedAt": "21 session.execution.succeeded", "outcome": "succeeded", "error": null, - "execution": null + "head": 0 } ], "requests": [ diff --git a/packages/codework/test/__artifacts__/recovery.moved.json b/packages/codework/test/__artifacts__/recovery.moved.json new file mode 100644 index 00000000..ca4cb7b0 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.moved.json @@ -0,0 +1,312 @@ +{ + "log": [ + { + "seq": 1, + "type": "session.created" + }, + { + "seq": 2, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 4, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 5, + "type": "session.thread.system.changed" + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 7, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 8, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 9, + "type": "session.turn.ended" + }, + { + "seq": 10, + "type": "session.llm.started" + }, + { + "seq": 11, + "type": "session.prompt.admitted", + "delivery": "followUp", + "text": "Then this." + }, + { + "seq": 12, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 13, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 14, + "type": "session.turn.ended" + }, + { + "seq": 15, + "type": "session.run.failed", + "reason": "truncated", + "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" + }, + { + "seq": 16, + "type": "session.prompt.updated" + }, + { + "seq": 17, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 18, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Then this." + }, + { + "seq": 19, + "type": "session.llm.started" + }, + { + "seq": 20, + "type": "session.execution.interrupted", + "reason": "crash" + }, + { + "seq": 21, + "type": "session.execution.started", + "resumed": true + }, + { + "seq": 22, + "type": "session.llm.failed", + "reason": "error", + "errorMessage": "The host stopped while the model was answering, on attempt 1", + "parts": [] + }, + { + "seq": 23, + "type": "session.retry.scheduled", + "attempt": 1, + "maxRetries": 3, + "delayMs": 10, + "message": "The host stopped while the model was answering, on attempt 1" + }, + { + "seq": 24, + "type": "session.llm.started" + }, + { + "seq": 25, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answered after the kill." + } + ] + }, + { + "seq": 26, + "type": "session.turn.ended" + }, + { + "seq": 27, + "type": "session.execution.succeeded" + } + ], + "steps": [ + { + "step": "resent", + "head": 15, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then this.", + "delivery": "followUp", + "admittedSeq": 11, + "tail": 11, + "promotedSeq": null + } + ] + }, + { + "step": "moved", + "head": 15, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then this.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 16, + "promotedSeq": 18 + } + ] + }, + { + "step": "finished", + "head": 15, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then this.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 16, + "promotedSeq": 18 + } + ] + } + ], + "requests": [ + [ + { + "role": "user", + "text": "Run a long command." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "user", + "text": "Then this." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "user", + "text": "Then this." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.parked.json b/packages/codework/test/__artifacts__/recovery.parked.json new file mode 100644 index 00000000..09615d41 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.parked.json @@ -0,0 +1,313 @@ +{ + "log": [ + { + "seq": 1, + "type": "session.created" + }, + { + "seq": 2, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 4, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Run a long command." + }, + { + "seq": 5, + "type": "session.thread.system.changed" + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 7, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 8, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 9, + "type": "session.turn.ended" + }, + { + "seq": 10, + "type": "session.llm.started" + }, + { + "seq": 11, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Then answer this." + }, + { + "seq": 12, + "type": "session.prompt.admitted", + "delivery": "followUp", + "text": "And this." + }, + { + "seq": 13, + "type": "session.prompt.updated" + }, + { + "seq": 14, + "type": "session.llm.ended", + "reason": "length", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 15, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "skipped", + "result": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + }, + { + "seq": 16, + "type": "session.turn.ended" + }, + { + "seq": 17, + "type": "session.run.failed", + "reason": "truncated", + "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" + }, + { + "seq": 18, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Next." + }, + { + "seq": 19, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 20, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Then answer this." + }, + { + "seq": 21, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "And this." + }, + { + "seq": 22, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Next." + }, + { + "seq": 23, + "type": "session.llm.started" + }, + { + "seq": 24, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answered with the queue." + } + ] + }, + { + "seq": 25, + "type": "session.turn.ended" + }, + { + "seq": 26, + "type": "session.execution.succeeded" + } + ], + "steps": [ + { + "step": "parked", + "head": 17, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then answer this.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 11, + "promotedSeq": null + }, + { + "text": "And this.", + "delivery": "steer", + "admittedSeq": 12, + "tail": 13, + "promotedSeq": null + } + ] + }, + { + "step": "reopened", + "head": 17, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then answer this.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 11, + "promotedSeq": null + }, + { + "text": "And this.", + "delivery": "steer", + "admittedSeq": 12, + "tail": 13, + "promotedSeq": null + } + ] + }, + { + "step": "next prompt", + "head": 17, + "restarts": 0, + "prompts": [ + { + "text": "Run a long command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Then answer this.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 11, + "promotedSeq": 20 + }, + { + "text": "And this.", + "delivery": "steer", + "admittedSeq": 12, + "tail": 13, + "promotedSeq": 21 + }, + { + "text": "Next.", + "delivery": "steer", + "admittedSeq": 18, + "tail": 18, + "promotedSeq": 22 + } + ] + } + ], + "requests": [ + [ + { + "role": "user", + "text": "Run a long command." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + } + ], + [ + { + "role": "user", + "text": "Run a long command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool call \"bash\" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments." + }, + { + "role": "user", + "text": "Then answer this." + }, + { + "role": "user", + "text": "And this." + }, + { + "role": "user", + "text": "Next." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.shutdown.json b/packages/codework/test/__artifacts__/recovery.shutdown.json new file mode 100644 index 00000000..33cd96c0 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.shutdown.json @@ -0,0 +1,84 @@ +{ + "log": [ + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 8, + "type": "session.execution.interrupted", + "reason": "shutdown" + }, + { + "seq": 9, + "type": "session.execution.started", + "resumed": true + }, + { + "seq": 10, + "type": "session.llm.started" + }, + { + "seq": 13, + "type": "session.execution.succeeded" + } + ], + "stopped": { + "head": 0, + "restarts": 0, + "prompts": [ + { + "text": "First.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + } + ] + }, + "resumed": { + "head": 0, + "restarts": 0, + "prompts": [ + { + "text": "First.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + } + ] + }, + "finished": { + "head": 0, + "restarts": 0, + "prompts": [ + { + "text": "First.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + } + ] + }, + "requests": [ + [ + { + "role": "user", + "text": "First." + } + ], + [ + { + "role": "user", + "text": "First." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.stop.json b/packages/codework/test/__artifacts__/recovery.stop.json new file mode 100644 index 00000000..72298146 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.stop.json @@ -0,0 +1,300 @@ +{ + "log": [ + { + "seq": 1, + "type": "session.created" + }, + { + "seq": 2, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Run the stubborn command." + }, + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 4, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Run the stubborn command." + }, + { + "seq": 5, + "type": "session.thread.system.changed" + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 7, + "type": "session.llm.ended", + "reason": "toolUse", + "parts": [ + { + "tool": "bash", + "status": "pending" + } + ] + }, + { + "seq": 8, + "type": "session.prompt.admitted", + "delivery": "followUp", + "text": "Queued before the stop." + }, + { + "seq": 9, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Sent while stopping." + }, + { + "seq": 10, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "aborted", + "result": "Tool Execution Interrupted" + } + }, + { + "seq": 11, + "type": "session.turn.ended" + }, + { + "seq": 12, + "type": "session.execution.interrupted", + "reason": "user" + }, + { + "seq": 13, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Next." + }, + { + "seq": 14, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 15, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Sent while stopping." + }, + { + "seq": 16, + "type": "session.prompt.promoted", + "delivery": "steer", + "text": "Next." + }, + { + "seq": 17, + "type": "session.llm.started" + }, + { + "seq": 18, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answer 1." + } + ] + }, + { + "seq": 19, + "type": "session.turn.ended" + }, + { + "seq": 20, + "type": "session.prompt.promoted", + "delivery": "followUp", + "text": "Queued before the stop." + }, + { + "seq": 21, + "type": "session.llm.started" + }, + { + "seq": 22, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answer 2." + } + ] + }, + { + "seq": 23, + "type": "session.turn.ended" + }, + { + "seq": 24, + "type": "session.execution.succeeded" + } + ], + "steps": [ + { + "step": "stopped", + "head": 12, + "restarts": 0, + "prompts": [ + { + "text": "Run the stubborn command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Queued before the stop.", + "delivery": "followUp", + "admittedSeq": 8, + "tail": 8, + "promotedSeq": null + }, + { + "text": "Sent while stopping.", + "delivery": "steer", + "admittedSeq": 9, + "tail": 9, + "promotedSeq": null + } + ] + }, + { + "step": "reopened", + "head": 12, + "restarts": 0, + "prompts": [ + { + "text": "Run the stubborn command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Queued before the stop.", + "delivery": "followUp", + "admittedSeq": 8, + "tail": 8, + "promotedSeq": null + }, + { + "text": "Sent while stopping.", + "delivery": "steer", + "admittedSeq": 9, + "tail": 9, + "promotedSeq": null + } + ] + }, + { + "step": "next prompt", + "head": 12, + "restarts": 0, + "prompts": [ + { + "text": "Run the stubborn command.", + "delivery": "steer", + "admittedSeq": 2, + "tail": 2, + "promotedSeq": 4 + }, + { + "text": "Queued before the stop.", + "delivery": "followUp", + "admittedSeq": 8, + "tail": 8, + "promotedSeq": 20 + }, + { + "text": "Sent while stopping.", + "delivery": "steer", + "admittedSeq": 9, + "tail": 9, + "promotedSeq": 15 + }, + { + "text": "Next.", + "delivery": "steer", + "admittedSeq": 13, + "tail": 13, + "promotedSeq": 16 + } + ] + } + ], + "requests": [ + [ + { + "role": "user", + "text": "Run the stubborn command." + } + ], + [ + { + "role": "user", + "text": "Run the stubborn command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool Execution Interrupted" + }, + { + "role": "user", + "text": "Sent while stopping." + }, + { + "role": "user", + "text": "Next." + } + ], + [ + { + "role": "user", + "text": "Run the stubborn command." + }, + { + "role": "assistant", + "calls": [ + "bash" + ] + }, + { + "role": "tool", + "text": "Tool Execution Interrupted" + }, + { + "role": "user", + "text": "Sent while stopping." + }, + { + "role": "user", + "text": "Next." + }, + { + "role": "assistant", + "text": "Answer 1." + }, + { + "role": "user", + "text": "Queued before the stop." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.truncated.json b/packages/codework/test/__artifacts__/recovery.truncated.json index 6da9d032..45823938 100644 --- a/packages/codework/test/__artifacts__/recovery.truncated.json +++ b/packages/codework/test/__artifacts__/recovery.truncated.json @@ -118,7 +118,7 @@ "reason": "truncated", "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended" }, - "execution": null + "head": 14 }, "requests": [ [ diff --git a/packages/codework/test/__artifacts__/sync.thread.json b/packages/codework/test/__artifacts__/sync.thread.json index c5724a60..50148d8c 100644 --- a/packages/codework/test/__artifacts__/sync.thread.json +++ b/packages/codework/test/__artifacts__/sync.thread.json @@ -195,7 +195,7 @@ { "_tag": "Synced", "storeId": "", - "rev": "" + "rev": "" } ], "settled": [ @@ -221,7 +221,7 @@ ], "size": 89 }, - "rev": "" + "rev": "" }, { "_tag": "Entry", @@ -241,7 +241,7 @@ ], "size": 38 }, - "rev": "" + "rev": "" }, { "_tag": "Inbox", @@ -253,7 +253,7 @@ "createdAt": "