From 4bd9ca782a8e1568f4d46442ffcf86db6340346d Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 10 Nov 2026 16:31:01 +0530 Subject: [PATCH 1/5] fix(harness): record a stop before interrupting, derive recovery in the drain A user stop on an open run now publishes session.execution.stopping before anything is interrupted, so a host killed while the stop winds down ends the run as stopped instead of resuming it as a crash. A shutdown that reaches a run already asked to stop ends it as the user's stop too. Recovery of a run the host cut off is decided by the drain from the thread row, whichever wake reaches it first: a prompt that gets there before startup recovery no longer leaves the restart uncounted. resumeUnfinished only wakes. Every run terminal is published by the executor, joined to the run's last step, and a run whose execution.started cannot be published does not run. Co-Authored-By: Claude Fable 5.2 --- .../__artifacts__/recovery.stop-killed.json | 200 ++++++++++++++++++ .../test/__artifacts__/recovery.stop.json | 56 ++--- packages/codework/test/recovery.test.ts | 58 +++++ packages/harness/src/control.ts | 79 +++---- packages/harness/src/event/list.ts | 14 ++ packages/harness/src/runner/execute.ts | 165 +++++++-------- packages/harness/src/runner/execution.ts | 22 +- packages/harness/src/runner/loop.ts | 37 +--- packages/harness/src/runner/run.ts | 8 +- packages/harness/src/session/projector.ts | 3 + packages/harness/src/thread/schema.ts | 5 +- packages/harness/src/thread/thread.ts | 41 +++- .../test/__artifacts__/thread.fork.json | 7 +- .../test/__artifacts__/thread.lifecycle.json | 48 ++--- packages/harness/test/control.prompt.test.ts | 2 - packages/harness/test/runner.loop.test.ts | 94 +++++++- 16 files changed, 598 insertions(+), 241 deletions(-) create mode 100644 packages/codework/test/__artifacts__/recovery.stop-killed.json diff --git a/packages/codework/test/__artifacts__/recovery.stop-killed.json b/packages/codework/test/__artifacts__/recovery.stop-killed.json new file mode 100644 index 00000000..4c8f1f9e --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.stop-killed.json @@ -0,0 +1,200 @@ +{ + "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.execution.stopping" + }, + { + "seq": 10, + "type": "session.execution.interrupted", + "reason": "user" + }, + { + "seq": 11, + "type": "session.prompt.admitted", + "delivery": "steer", + "text": "Next." + }, + { + "seq": 12, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 13, + "type": "session.tool.settled", + "part": { + "tool": "bash", + "status": "aborted", + "result": "Tool Execution Interrupted" + } + }, + { + "seq": 14, + "type": "session.turn.ended" + }, + { + "seq": 15, + "type": "session.prompt.promoted", + "delivery": "followUp", + "text": "Queued before the stop." + }, + { + "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.execution.succeeded" + } + ], + "steps": [ + { + "step": "reopened", + "head": 10, + "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 + } + ] + }, + { + "step": "next prompt", + "head": 10, + "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": 15 + }, + { + "text": "Next.", + "delivery": "steer", + "admittedSeq": 11, + "tail": 11, + "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": "Queued before the stop." + }, + { + "role": "user", + "text": "Next." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.stop.json b/packages/codework/test/__artifacts__/recovery.stop.json index 0601cf60..a2b12d8c 100644 --- a/packages/codework/test/__artifacts__/recovery.stop.json +++ b/packages/codework/test/__artifacts__/recovery.stop.json @@ -48,12 +48,16 @@ }, { "seq": 9, + "type": "session.execution.stopping" + }, + { + "seq": 10, "type": "session.prompt.admitted", "delivery": "steer", "text": "Sent while stopping." }, { - "seq": 10, + "seq": 11, "type": "session.tool.settled", "part": { "tool": "bash", @@ -62,49 +66,49 @@ } }, { - "seq": 11, + "seq": 12, "type": "session.turn.ended" }, { - "seq": 12, + "seq": 13, "type": "session.execution.interrupted", "reason": "user" }, { - "seq": 13, + "seq": 14, "type": "session.prompt.admitted", "delivery": "steer", "text": "Next." }, { - "seq": 14, + "seq": 15, "type": "session.execution.started", "resumed": false }, { - "seq": 15, + "seq": 16, "type": "session.prompt.promoted", "delivery": "followUp", "text": "Queued before the stop." }, { - "seq": 16, + "seq": 17, "type": "session.prompt.promoted", "delivery": "steer", "text": "Sent while stopping." }, { - "seq": 17, + "seq": 18, "type": "session.prompt.promoted", "delivery": "steer", "text": "Next." }, { - "seq": 18, + "seq": 19, "type": "session.llm.started" }, { - "seq": 19, + "seq": 20, "type": "session.llm.ended", "reason": "stop", "parts": [ @@ -114,18 +118,18 @@ ] }, { - "seq": 20, + "seq": 21, "type": "session.turn.ended" }, { - "seq": 21, + "seq": 22, "type": "session.execution.succeeded" } ], "steps": [ { "step": "stopped", - "head": 12, + "head": 13, "restarts": 0, "prompts": [ { @@ -145,15 +149,15 @@ { "text": "Sent while stopping.", "delivery": "steer", - "admittedSeq": 9, - "tail": 9, + "admittedSeq": 10, + "tail": 10, "promotedSeq": null } ] }, { "step": "reopened", - "head": 12, + "head": 13, "restarts": 0, "prompts": [ { @@ -173,15 +177,15 @@ { "text": "Sent while stopping.", "delivery": "steer", - "admittedSeq": 9, - "tail": 9, + "admittedSeq": 10, + "tail": 10, "promotedSeq": null } ] }, { "step": "next prompt", - "head": 12, + "head": 13, "restarts": 0, "prompts": [ { @@ -196,21 +200,21 @@ "delivery": "followUp", "admittedSeq": 8, "tail": 8, - "promotedSeq": 15 + "promotedSeq": 16 }, { "text": "Sent while stopping.", "delivery": "steer", - "admittedSeq": 9, - "tail": 9, - "promotedSeq": 16 + "admittedSeq": 10, + "tail": 10, + "promotedSeq": 17 }, { "text": "Next.", "delivery": "steer", - "admittedSeq": 13, - "tail": 13, - "promotedSeq": 17 + "admittedSeq": 14, + "tail": 14, + "promotedSeq": 18 } ] } diff --git a/packages/codework/test/recovery.test.ts b/packages/codework/test/recovery.test.ts index d1f79cfe..9b81f6c9 100644 --- a/packages/codework/test/recovery.test.ts +++ b/packages/codework/test/recovery.test.ts @@ -804,6 +804,64 @@ describe("recovery after the host is killed", () => { ).toMatchFileSnapshot("./__artifacts__/recovery.stop.json"); }, 90_000); + it("a host killed while a stop winds down keeps the stop: the run is not resumed, its queue parked", async () => { + const marker = join(realpathSync(mkdtempSync(join(tmpdir(), "codework-recovery-marker-"))), "ran"); + onTestFinished(() => rmSync(join(marker, ".."), { recursive: true, force: true })); + const fake = await streaming((_body, response, index) => { + if (index > 0) return answer(response, `Answer ${index}.`); + // Ignores SIGTERM, so the stop is still winding down when the host is killed. + response.send( + toolCalls([{ name: "bash", arguments: { command: `trap '' TERM; echo ran > ${marker}; sleep 20` } }]), + ); + response.finish("tool_calls"); + }); + 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 steps: Array<{ step: string } & ReturnType> = []; + const record = (step: string) => steps.push({ step, ...positions(home, ids.threadId) }); + + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...ids, text: "Run the stubborn command." })); + await until("the tool to run", () => existsSync(marker)); + await call(first.url, (rpc) => + rpc["thread.prompt"]({ ...ids, text: "Queued before the stop.", delivery: "followUp" }), + ); + expect(await call(first.url, (rpc) => rpc["thread.interrupt"](ids))).toEqual({ interrupted: true }); + // The stop was accepted, and its run has not ended yet. + expect(count(await log(first.url, ids), "session.execution.interrupted")).toBe(0); + await first.kill(); + + const second = await start(); + await call(second.url, (rpc) => rpc["thread.wait"](ids)); + record("reopened"); + const reopened = await log(second.url, ids); + expect( + reopened.filter((event) => event.type === "session.execution.interrupted").map((event) => summary(event)), + ).toEqual([{ type: "session.execution.interrupted", reason: "user" }]); + expect(count(reopened, "session.execution.started")).toBe(1); + expect(fake.requests).toHaveLength(1); + + await call(second.url, (rpc) => rpc["thread.prompt"]({ ...ids, text: "Next." })); + await logged(second.url, ids, "session.execution.succeeded"); + await call(second.url, (rpc) => rpc["thread.wait"](ids)); + record("next prompt"); + expect( + sent(fake.requests[1]!) + .filter((message) => message.role === "user") + .map((message) => message.text), + ).toEqual(["Run the stubborn command.", "Queued before the stop.", "Next."]); + const events = await log(second.url, ids); + await expect( + artifact({ + log: events.map((event) => ({ seq: event.durable?.seq, ...summary(event) })), + steps, + requests: fake.requests.map(sent), + }), + ).toMatchFileSnapshot("./__artifacts__/recovery.stop-killed.json"); + }, 90_000); + it("a parked prompt moved to steer runs now and is resumed after a kill; resending it changes nothing", async () => { const cut = gate(); const resend = gate(); diff --git a/packages/harness/src/control.ts b/packages/harness/src/control.ts index 6f28a554..2fa39dae 100644 --- a/packages/harness/src/control.ts +++ b/packages/harness/src/control.ts @@ -14,7 +14,7 @@ * execution and on storage, and neither depends back. */ -import { Cause, Context, DateTime, Effect, Layer, Option, Schema } from "effect"; +import { Context, DateTime, Effect, Layer, Option, Schema } from "effect"; import { Event } from "./event/event.ts"; import { EventList } from "./event/list.ts"; import { Inbox } from "./inbox/inbox.ts"; @@ -56,11 +56,7 @@ export interface PromptInput extends ThreadRef { export type ResolveError = Session.SessionNotFoundError | Thread.ThreadNotFoundError; -/** - * Times in a row a run may be cut off by a dying process before it ends (`session.run.failed`, - * `crashloop`) instead of resuming. A committed turn starts the count over. - */ -export const RESTART_LIMIT = 5; +export const RESTART_LIMIT = RunnerExecution.RESTART_LIMIT; export interface Interface { /** The thread a reference names. A thread of another session is not found. */ @@ -102,16 +98,16 @@ export interface Interface { /** Threads this process is currently draining. */ readonly active: Effect.Effect>; /** - * Resumes the runs the last process left unfinished, as the server starts: a run a crash or a + * Wakes the threads the last process left unfinished, as the server starts: a run a crash or a * shutdown cut off, or a prompt woken past the thread's head. A crash counts a restart unless the * run slept in retry backoff, a shutdown does not; at {@link RESTART_LIMIT} the run ends * visibly instead, its queue parked for the next prompt. One thread failing to resume leaves the - * others resuming. Resolves once every resumed run has settled. + * others resuming. Resolves once every woken thread is idle. */ readonly resumeUnfinished: Effect.Effect; /** - * Ends every stored run the last process left open with a durable `execution.interrupted` - * `crash`: a shutdown ends its runs itself. Runs as the server starts, before it accepts + * Ends every stored run the last process left open with a durable `execution.interrupted`, as a + * `crash`, or as the `user` stop it had been asked for: a shutdown ends its runs itself. Runs as the server starts, before it accepts * connections and before {@link resumeUnfinished}. */ readonly sweep: Effect.Effect; @@ -248,6 +244,18 @@ export const layer = Layer.effect( return true; }); + const requestStop = Effect.fnUntraced(function* (threadId: ThreadSchema.ID) { + const thread = yield* threads.get(threadId); + if (Option.isNone(thread) || thread.value.runState === "idle" || Thread.stopping(thread.value)) return; + yield* events + .publish(EventList.ExecutionStopping, { + timestamp: yield* DateTime.now, + sessionId: SessionSchema.ID.make(thread.value.sessionId), + threadId, + }) + .pipe(Effect.catchDefect((defect) => (isThreadNotFound(defect) ? Effect.void : Effect.die(defect)))); + }); + const interrupt = Effect.fn("Control.interrupt")( ( threadId: ThreadSchema.ID, @@ -256,7 +264,16 @@ export const layer = Layer.effect( Effect.uninterruptible( Effect.gen(function* () { const reason = options?.reason ?? "user"; - if (yield* execution.interrupt(threadId, reason, options)) return true; + // Recorded before anything is interrupted, as a prompt is admitted before its wake: a host + // that dies while the stop winds down ends the run as stopped instead of resuming it. + if (reason === "user") yield* requestStop(threadId); + if (yield* execution.interrupt(threadId, reason)) { + // Checked again, as a futex re-reads before it sleeps: a drain that opened its run + // between the record above and the interrupt would otherwise wind down unmarked. + if (reason === "user") yield* requestStop(threadId); + if (options?.awaitSettlement === true) yield* execution.interrupt(threadId, reason, options); + return true; + } // Nothing here to interrupt: durable state says whether work is still owed, now or on the // next start, and the stop is recorded against it. if (reason !== "user") return false; @@ -309,54 +326,28 @@ export const layer = Layer.effect( ), ); + // Only the bell: each drain reads what its thread is owed, a resume, an end at the restart limit or + // a parked queue, from the thread itself. const resumeUnfinished = Effect.gen(function* () { - const running = yield* execution.active; - const unfinished = (yield* threads.unfinished).filter((row) => !running.has(ThreadSchema.ID.make(row.id))); yield* Effect.forEach( - unfinished, + yield* threads.unfinished, (row) => { const threadId = ThreadSchema.ID.make(row.id); - return Effect.gen(function* () { - if ((yield* execution.active).has(threadId)) return; - // No run was cut off, only a prompt woken past the head: it runs as its wake would have. - if (!(yield* threads.cutOff(threadId))) { - yield* execution.wake(threadId); - return yield* execution.awaitIdle(threadId); - } - // Claimed again here, not from the list: a stop since then cleared it, and must stand. - // Counted before resuming, so a crash inside the resumed run is counted too. - const restarts = yield* threads.claimRestart(threadId); - if (Option.isNone(restarts)) return; - yield* restarts.value >= RESTART_LIMIT - ? execution.abandon( - threadId, - `The run was interrupted ${restarts.value} times in a row; it is not resumed again`, - ) - : execution.recover(threadId); - }).pipe( - // Defects included: one thread that cannot be resumed must not cancel the others. - Effect.tapCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.void - : Effect.logWarning("Resuming an unfinished thread failed", cause).pipe( - Effect.annotateLogs({ threadId }), - ), - ), - Effect.ignoreCause, - ); + return execution.wake(threadId).pipe(Effect.andThen(execution.awaitIdle(threadId))); }, { concurrency: "unbounded", discard: true }, ); }).pipe(Effect.withSpan("Control.resumeUnfinished")); const sweep = Effect.gen(function* () { - // A shutdown ends its runs itself, so a run still open is one the process died in. + // A shutdown ends its runs itself, so a run still open is one the process died in. One the user + // had asked to stop ends as stopped: its host died while the stop wound down. for (const row of yield* threads.open) { yield* events.publish(EventList.ExecutionInterrupted, { timestamp: yield* DateTime.now, sessionId: SessionSchema.ID.make(row.sessionId), threadId: ThreadSchema.ID.make(row.id), - reason: "crash", + reason: Thread.stopping(row) ? "user" : "crash", }); } }).pipe(Effect.withSpan("Control.sweep")); diff --git a/packages/harness/src/event/list.ts b/packages/harness/src/event/list.ts index 9eac010b..9c777274 100644 --- a/packages/harness/src/event/list.ts +++ b/packages/harness/src/event/list.ts @@ -350,6 +350,18 @@ export const ExecutionFailed = EventSchema.define({ }); export type ExecutionFailed = typeof ExecutionFailed.Type; +/** + * A user stop accepted for the run that is open, recorded before anything is interrupted: the run's + * terminal follows once its cleanup settles, and a host that dies in between still ends it as the + * user's stop rather than resuming it as a crash. + */ +export const ExecutionStopping = EventSchema.define({ + type: "session.execution.stopping", + ...durableOptions, + schema: baseOptions, +}); +export type ExecutionStopping = typeof ExecutionStopping.Type; + export const ExecutionInterrupted = EventSchema.define({ type: "session.execution.interrupted", ...durableOptions, @@ -455,6 +467,7 @@ export const Definitions = EventSchema.inventory( ExecutionStarted, ExecutionSucceeded, ExecutionFailed, + ExecutionStopping, ExecutionInterrupted, PromptAdmitted, Prompted, @@ -498,6 +511,7 @@ export const DurableDefinitions = EventSchema.inventory( ExecutionStarted, ExecutionSucceeded, ExecutionFailed, + ExecutionStopping, ExecutionInterrupted, PromptAdmitted, Prompted, diff --git a/packages/harness/src/runner/execute.ts b/packages/harness/src/runner/execute.ts index cd54cc3e..24346856 100644 --- a/packages/harness/src/runner/execute.ts +++ b/packages/harness/src/runner/execute.ts @@ -56,6 +56,7 @@ export const layer = Layer.effect( // The session a thread belongs to, for the lifecycle events its log carries. Read every time, not // cached: the events are durable, and one for a session deleted since would recreate its log. const isThreadNotFound = Schema.is(Thread.ThreadNotFoundError); + const isRunStopping = Schema.is(Thread.RunStoppingError); const sessionOf = (threadId: ThreadId) => threads.get(threadId).pipe(Effect.map(Option.map((thread) => SessionSchema.ID.make(thread.sessionId)))); const publishLifecycle = ( @@ -68,45 +69,49 @@ export const layer = Layer.effect( ) => Effect.gen(function* () { const sessionId = yield* sessionOf(threadId); - if (Option.isNone(sessionId)) return; - yield* publish({ sessionId: sessionId.value, threadId, timestamp: yield* DateTime.now }).pipe( + if (Option.isNone(sessionId)) return false; + return yield* publish({ sessionId: sessionId.value, threadId, timestamp: yield* DateTime.now }).pipe( + Effect.as(true), // Deleted between the read and the commit: the projector rolled the event back. - Effect.catchDefect((defect) => (isThreadNotFound(defect) ? Effect.void : Effect.die(defect))), + Effect.catchDefect((defect) => (isThreadNotFound(defect) ? Effect.succeed(false) : Effect.die(defect))), ); }); - // Threads being recovered: their run is announced as resumed. - const recovering = new Set(); - // Recovered threads whose run is ended instead of resumed, with the message it ends with. - const abandoning = new Map(); - // Threads whose current drain opened a run with `execution.started`. - const opened = new Set(); - const coordinator = yield* RunCoordinator.make({ - // A drain that failed or was interrupted ends the busy period, and the run it opened ends here unless - // it already published its own terminal: the stored run state says which, so a run ends exactly once. + // A drain that failed or was interrupted ends the busy period, and the run it left open ends here: one + // that published its own terminal is idle by now, so a run ends exactly once. settled: (threadId, exit, reason) => reportLifecycle( threadId, Effect.gen(function* () { - const ran = opened.delete(threadId); if (Exit.isSuccess(exit)) return; const interrupted = Cause.hasInterruptsOnly(exit.cause); const thread = yield* threads.get(threadId); if (Option.isNone(thread)) return; + const open = thread.value.runState !== "idle"; + // The stop a run was asked for is on its row, and outlasts whatever interrupted it since. + const stop = interrupted && (reason === "user" || Thread.stopping(thread.value)); // With no run open, before it opened or after it ended, a user stop is recorded only when // something would otherwise still run, now or on the next start. - const open = ran && thread.value.runState !== "idle"; - const stop = interrupted && reason === "user"; if (!open && !(stop && (yield* threads.resumable(threadId)))) return; - yield* publishLifecycle(threadId, (base) => - interrupted - ? // Nobody named a reason, so the process took the decision itself. - events.publish(EventList.ExecutionInterrupted, { ...base, reason: reason ?? "shutdown" }) - : events.publish(EventList.ExecutionFailed, { - ...base, - error: SessionFailure.fromCause(Cause.squash(exit.cause)), - }), + const interruptedAs = (as: SessionSchema.InterruptReason) => + publishLifecycle(threadId, (base) => + events.publish(EventList.ExecutionInterrupted, { ...base, reason: as }), + ); + if (!interrupted) { + return yield* publishLifecycle(threadId, (base) => + events.publish(EventList.ExecutionFailed, { + ...base, + error: SessionFailure.fromCause(Cause.squash(exit.cause)), + }), + ); + } + // Nobody named a reason, so the process took the decision itself. A stop recorded since the + // read above rolls that end back: the run then ends as the user's. + yield* interruptedAs(stop ? "user" : (reason ?? "shutdown")).pipe( + Effect.catchDefect((defect) => + isRunStopping(defect) ? interruptedAs("user") : Effect.die(defect), + ), ); }), ), @@ -114,6 +119,11 @@ export const layer = Layer.effect( // lost lifecycle publish) is drained again rather than left waiting for a bell. drain: (threadId: ThreadId, force: boolean): Effect.Effect => drainOnce(threadId, force).pipe( + Effect.tapCause((cause) => + Cause.hasDies(cause) + ? Effect.logError("defect", cause).pipe(Effect.annotateLogs({ threadId })) + : Effect.void, + ), Effect.tap(() => Effect.gen(function* () { if (yield* threads.runnable(threadId)) yield* coordinator.wake(threadId); @@ -122,52 +132,62 @@ export const layer = Layer.effect( ), }); - // One run per drain that runs, opened before it promotes anything: a process dying anywhere in it - // leaves an open run, which the next server start sweeps as a crash and resumes. + /* + * One run per drain that runs, opened before it promotes anything: a process dying anywhere in it + * leaves an open run, which the next server start sweeps as a crash. What to do is read from the + * thread, never from what woke the drain: a run the host cut off is resumed, or ended at the restart + * limit, by whichever drain gets to it first. + */ const drainOnce = Effect.fnUntraced(function* (threadId: ThreadId, force: boolean) { - opened.delete(threadId); const thread = yield* threads.get(threadId); // Deleted with its session after something woke it: nothing is left to run. if (Option.isNone(thread)) return; const sessionId = thread.value.sessionId; // Read before `execution.started` clears it. + const cut = Option.exists( + thread.value.runError, + (error) => error._tag === "interrupted" && error.reason !== "user", + ); const shutdown = Option.exists( thread.value.runError, (error) => error._tag === "interrupted" && error.reason === "shutdown", ); - const recovered = recovering.delete(threadId); - // A recovery claimed the thread before its drain: a stop since then must stand. - if (recovered && !(yield* threads.resumable(threadId))) return; + // Counted before resuming, so a crash inside the resumed run is counted too. None: a stop since + // the read ended the run, and stands. + const restarts = + cut && thread.value.runState === "idle" ? yield* threads.claimRestart(threadId) : Option.none(); + const recovered = Option.isSome(restarts); // The doorbell is only a hint: an unforced drain runs only for a prompt woken past the head. - if (!force && !(yield* threads.runnable(threadId))) return; - // Whole or not at all: a stop lands either before the run opened or on an open run. - yield* Effect.uninterruptible( - Effect.sync(() => opened.add(threadId)).pipe( - Effect.andThen( - reportLifecycle( - threadId, - publishLifecycle(threadId, (base) => - events.publish(EventList.ExecutionStarted, { ...base, resumed: recovered }), - ), - ), - ), + if (!recovered && !force && !(yield* threads.runnable(threadId))) return; + // Whole or not at all: a stop lands either before the run opened or on an open run. A run that + // cannot be opened does not run. + const started = yield* Effect.uninterruptible( + publishLifecycle(threadId, (base) => + events.publish(EventList.ExecutionStarted, { ...base, resumed: recovered }), ), ); - // `ended`: the run published its own terminal, `session.run.failed`. A stop arriving meanwhile - // waits for the success to land, so `settled` finds the run ended. + if (!started) return; + // Every terminal a run that returned gets is published here, joined to its last step: a stop or a + // shutdown lands on the open run or after its end, never between the two. const close = (outcome: Runner.Outcome) => - outcome === "ended" - ? Effect.void - : Effect.uninterruptible( - reportLifecycle( - threadId, - publishLifecycle(threadId, (base) => events.publish(EventList.ExecutionSucceeded, base)), - ), - ); - // Only the drain the abandon itself started: one it joined is a live run, which must stand. - const abandoned = recovered && force ? abandoning.get(threadId) : undefined; - if (abandoned !== undefined) - return yield* runner.abandon({ threadId, message: abandoned }).pipe(Effect.flatMap(close)); + reportLifecycle( + threadId, + publishLifecycle(threadId, (base) => + outcome === "drained" + ? events.publish(EventList.ExecutionSucceeded, base) + : events.publish(EventList.RunFailed, { ...base, ...outcome }), + ), + ); + const ending = (run: Effect.Effect) => + Effect.uninterruptibleMask((restore) => restore(run).pipe(Effect.flatMap(close))); + if (recovered && restarts.value >= RunnerExecution.RESTART_LIMIT) { + return yield* ending( + runner.abandon({ + threadId, + message: `The run was interrupted ${restarts.value} times in a row; it is not resumed again`, + }), + ); + } // The thread's placement, else the session's space and directory. A placement names its // space in a document no FK guards, so a space gone since is typed: `relink --thread` // is the repair. @@ -209,46 +229,23 @@ export const layer = Layer.effect( // The thread's MCP servers stay open while its drain runs, and idle out after. yield* mcp.lease({ sessionId, threadId }); - return yield* runner - .run({ threadId, force, recovered, shutdown }) - .pipe(Effect.provideContext(Context.merge(locationContext, mountContext)), Effect.flatMap(close)); + return yield* ending( + runner + .run({ threadId, force: force || recovered, recovered, shutdown }) + .pipe(Effect.provideContext(Context.merge(locationContext, mountContext))), + ); }); - return yield* scopedRun.pipe( - Effect.scoped, - Effect.tapCause((cause) => - Cause.hasDies(cause) - ? Effect.logError("defect", cause).pipe(Effect.annotateLogs({ threadId })) - : Effect.void, - ), - ); + return yield* scopedRun.pipe(Effect.scoped); }); const interrupt: RunnerExecution.Interface["interrupt"] = (threadId, reason, options) => coordinator.interrupt(threadId, reason, options); - // Under the thread's execution like any run, so a prompt that wakes it meanwhile cannot heal - // the same draft. - const recover = (threadId: ThreadId, abandon?: string) => - Effect.suspend(() => { - recovering.add(threadId); - if (abandon !== undefined) abandoning.set(threadId, abandon); - return coordinator.run(threadId); - }).pipe( - Effect.ensuring( - Effect.sync(() => { - recovering.delete(threadId); - abandoning.delete(threadId); - }), - ), - ); - return RunnerExecution.Service.of({ active: coordinator.active, interrupt, resume: coordinator.run, - recover: (threadId) => recover(threadId), - abandon: recover, wake: coordinator.wake, awaitIdle: coordinator.awaitIdle, }); diff --git a/packages/harness/src/runner/execution.ts b/packages/harness/src/runner/execution.ts index fc3725f6..5f1e4bdf 100644 --- a/packages/harness/src/runner/execution.ts +++ b/packages/harness/src/runner/execution.ts @@ -8,22 +8,20 @@ import type { SessionSchema } from "../session/schema.ts"; import type { ThreadSchema } from "../thread/schema.ts"; import { Runner } from "./run.ts"; +/** + * Times in a row a run may be cut off by a dying process before it ends (`session.run.failed`, + * `crashloop`) instead of resuming. A committed turn starts the count over. + */ +export const RESTART_LIMIT = 5; + export interface Interface { /** Snapshots active execution owned by this process. */ readonly active: Effect.Effect>; - /** Starts execution while idle or joins the active execution. */ - readonly resume: (threadId: ThreadSchema.ID) => Effect.Effect; /** - * Resumes a run the last process left unfinished. Like `resume`, except that it marks nothing, - * and runs nothing when the thread is no longer marked as it begins: a stop that landed since - * recovery chose the thread stands. + * Starts execution while idle or joins the active execution. A run the last process left cut off is + * resumed by whichever drain reaches it first, this one or a wake's. */ - readonly recover: (threadId: ThreadSchema.ID) => Effect.Effect; - /** - * Ends a run the last process left unfinished instead of resuming it, under the thread's execution - * like `recover`; see `Runner.Interface.abandon`. - */ - readonly abandon: (threadId: ThreadSchema.ID, message: string) => Effect.Effect; + readonly resume: (threadId: ThreadSchema.ID) => Effect.Effect; /** Registers newly recorded work. Repeated wakeups may coalesce. */ readonly wake: (threadId: ThreadSchema.ID) => Effect.Effect; /** @@ -47,8 +45,6 @@ export const noopLayer = Layer.succeed( Service.of({ active: Effect.succeed(new Set()), resume: () => Effect.void, - recover: () => Effect.void, - abandon: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: () => Effect.void, diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index ab09192e..5af75a98 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -270,26 +270,10 @@ export const layer = (options: Options = {}) => yield* Effect.sleep(Duration.millis(delayMs)); }); - /** - * Ends a run that is not resumed. Its terminal parks the queue: a prompt woken before it waits for - * the next one. - */ - const endRun = Effect.fn("Loop.endRun")(function* ( - sessionId: SessionSchema.ID, - threadId: ThreadSchema.ID, - reason: EventList.RunFailed["data"]["reason"], - message: string, - ) { - yield* Effect.uninterruptible( - events.publish(EventList.RunFailed, { - timestamp: yield* DateTime.now, - sessionId, - threadId, - reason, - message, - }), - ); - return "ended" as const; + /** A run that is not resumed: its execution records the end, which parks the queue. */ + const failed = (reason: ThreadSchema.RunFailedReason, message: string): Runner.Outcome => ({ + reason, + message, }); const settleTools = Effect.fn("Loop.settleTools", { level: "Info" })(function* ( @@ -667,7 +651,7 @@ export const layer = (options: Options = {}) => if (carried?.failed !== undefined) { const delayMs = Retry.delay(start.retry, carried.retries, hostStopped(start, carried.failed)); if (delayMs === undefined) { - return yield* endRun(start.sessionId, input.threadId, "interrupted", carried.failed); + return failed("interrupted", carried.failed); } yield* scheduleRetry(start, carried.retries, delayMs, carried.failed); } @@ -696,9 +680,7 @@ export const layer = (options: Options = {}) => // Queued steers and follow-ups start the next run. if (decision === "end") return "drained" as const; if (truncations >= TRUNCATION_LIMIT) { - return yield* endRun( - snapshot.sessionId, - input.threadId, + return failed( "truncated", `The response hit the output token limit with tool calls ${truncations} times in a row; the run is ended`, ); @@ -732,12 +714,7 @@ export const layer = (options: Options = {}) => const thread = yield* threads.get(input.threadId); if (Option.isNone(thread)) return "drained" as const; yield* healDraft(input.threadId, Option.isSome(thread.value.attempt) ? input.message : undefined); - return yield* endRun( - SessionSchema.ID.make(thread.value.sessionId), - input.threadId, - "crashloop", - input.message, - ); + return failed("crashloop", input.message); }); return Runner.Service.of({ run, abandon }); diff --git a/packages/harness/src/runner/run.ts b/packages/harness/src/runner/run.ts index 16d20021..f62d85c0 100644 --- a/packages/harness/src/runner/run.ts +++ b/packages/harness/src/runner/run.ts @@ -185,8 +185,8 @@ export type RunError = | SandboxMountError | SandboxFileSystem.FileSystemError; -/** `ended`: the run published its own terminal, `session.run.failed`; `drained`: it left that to its execution. */ -export type Outcome = "ended" | "drained"; +/** How a run that returned ended: `drained` with its work answered, or failed with why it is not resumed. */ +export type Outcome = "drained" | { readonly reason: ThreadSchema.RunFailedReason; readonly message: string }; export interface Interface { /** Drains eligible durable work. Explicit runs perform one provider attempt even when no work is eligible. */ @@ -199,8 +199,8 @@ export interface Interface { readonly shutdown?: boolean; }) => Effect.Effect; /** - * Ends a run the server no longer resumes: settles what the dead process left of its reply, - * records `session.run.failed` with `message`, and leaves the thread idle, its queue kept. + * Ends a run the server no longer resumes: settles what the dead process left of its reply and + * returns the failure its execution records, `crashloop` with `message`. */ readonly abandon: (input: { readonly threadId: ThreadSchema.ID; diff --git a/packages/harness/src/session/projector.ts b/packages/harness/src/session/projector.ts index 7c83ba09..4361804b 100644 --- a/packages/harness/src/session/projector.ts +++ b/packages/harness/src/session/projector.ts @@ -165,6 +165,9 @@ export const layer = Layer.effectDiscard( yield* events.project(EventList.ExecutionStarted, ({ data }) => threads.projectRun(data.threadId, data.timestamp, { _tag: "started" }), ); + yield* events.project(EventList.ExecutionStopping, ({ data }) => + threads.projectRun(data.threadId, data.timestamp, { _tag: "stopping" }), + ); // A run's end settles the thread at the event's own seq: an unanswered end moves its head. const ended = ( event: { readonly type: string; readonly durable?: { readonly seq: number } }, diff --git a/packages/harness/src/thread/schema.ts b/packages/harness/src/thread/schema.ts index 685958c0..a1bc7e1a 100644 --- a/packages/harness/src/thread/schema.ts +++ b/packages/harness/src/thread/schema.ts @@ -38,7 +38,10 @@ export type RunOutcome = typeof RunOutcome.Type; export const RunFailedReason = Schema.Literals(["crashloop", "interrupted", "truncated"]); export type RunFailedReason = typeof RunFailedReason.Type; -/** Why the last run did not succeed; stored as `thread.run_error` JSON. */ +/** + * Why the last run did not succeed; stored as `thread.run_error` JSON. On a run still open it holds the + * user stop that run was asked for, until its terminal lands. + */ export const RunError = Schema.Union([ Schema.TaggedStruct("failed", { error: SessionFailure.Error }), Schema.TaggedStruct("runFailed", { reason: RunFailedReason, message: Schema.String }), diff --git a/packages/harness/src/thread/thread.ts b/packages/harness/src/thread/thread.ts index fcfc6fb4..2d2bc592 100644 --- a/packages/harness/src/thread/thread.ts +++ b/packages/harness/src/thread/thread.ts @@ -20,6 +20,11 @@ import { EventList } from "../event/list.ts"; import { SessionSchema } from "../session/schema.ts"; import { ThreadSchema } from "./schema.ts"; +/** A host interruption landed on a run the user had asked to stop; see `projectRun`. */ +export class RunStoppingError extends Schema.TaggedError()("RunStoppingError", { + threadId: Schema.String, +}) {} + export class ThreadNotFoundError extends Schema.TaggedError()("ThreadNotFoundError", { threadId: Schema.String, }) { @@ -146,6 +151,8 @@ export interface InsertThread { export type RunChange = | { readonly _tag: "started" } | { readonly _tag: "running" } + /** A user stop accepted for the open run: kept on the row until the run's terminal lands. */ + | { readonly _tag: "stopping" } | { readonly _tag: "ended"; readonly outcome: ThreadSchema.RunOutcome; @@ -154,6 +161,11 @@ export type RunChange = readonly seq: number; }; +/** Whether the row's open run was asked to stop by the user; see `session.execution.stopping`. */ +export const stopping = (row: Pick) => + row.runState !== "idle" && + Option.exists(row.runError, (error) => error._tag === "interrupted" && error.reason === "user"); + // Entry row + its part rows, zipped. Message decode happens above this layer. export interface HydratedEntry { readonly entry: EntryRow; @@ -217,9 +229,9 @@ export interface Interface { /** Names an untitled thread. Returns whether the title changed. */ readonly retitle: (threadId: ThreadSchema.ID, title: string) => Effect.Effect; /** - * Claims a thread's unfinished run for recovery, counting a restart when the sweep found its host + * Claims the run the host cut off for recovery, counting a restart when the sweep found its host * dead in it, unless the run slept in retry backoff (a shutdown is not counted either). The new count, - * or none when nothing is unfinished any more, say because the user stopped it since. + * or none when no run is cut off any more, say because the user stopped it since. */ readonly claimRestart: (threadId: ThreadSchema.ID) => Effect.Effect>; /** A committed turn is progress: the restart count starts over. Runs inside `TurnEnded`'s commit. */ @@ -230,8 +242,6 @@ export interface Interface { readonly runnable: (threadId: ThreadSchema.ID) => Effect.Effect; /** Whether the thread is among {@link unfinished}: still owed a resume, say not stopped since. */ readonly resumable: (threadId: ThreadSchema.ID) => Effect.Effect; - /** Whether the thread's last run was cut off by the host, a crash or a shutdown: a run to resume. */ - readonly cutOff: (threadId: ThreadSchema.ID) => Effect.Effect; /** * Threads with a run left to resume, oldest first: their last run was interrupted by a crash or a * shutdown, or a pending prompt is runnable. Read after the startup sweep has closed open runs. @@ -978,7 +988,7 @@ export const layer = Layer.effect( UPDATE thread SET restarts = restarts + (CASE WHEN ${interruptedOf(["crash"])} AND NOT ${asleepOf} THEN 1 ELSE 0 END), rev = ${yield* store.rev} - WHERE id = ${threadId} AND ${unfinishedOf} + WHERE id = ${threadId} AND ${interruptedOf(["crash", "shutdown"])} RETURNING restarts `; return Option.fromUndefinedOr(rows[0]?.restarts); @@ -1010,7 +1020,6 @@ export const layer = Layer.effect( }, Effect.orDie); const runnable = holds("Thread.runnable", runnableOf); const resumable = holds("Thread.resumable", unfinishedOf); - const cutOff = holds("Thread.cutOff", interruptedOf(["crash", "shutdown"])); const findUnfinished = SqlSchema.findAll({ Request: Schema.Void, @@ -1028,6 +1037,16 @@ export const layer = Layer.effect( ) { const time = DateTime.toEpochMillis(at); const rev = yield* store.rev; + if (change._tag === "stopping") { + // Only an open run is asked to stop: one that ended meanwhile keeps the error it ended with. + const stop = yield* encodeRunError({ _tag: "interrupted", reason: "user" }); + const rows = yield* sql` + UPDATE thread SET run_error = CASE WHEN run_state <> 'idle' THEN ${stop} ELSE run_error END, rev = ${yield* store.rev} + WHERE id = ${threadId} + RETURNING id + `; + return rows.length > 0 ? undefined : yield* Effect.die(new ThreadNotFoundError({ threadId })); + } const state: ThreadSchema.RunState = change._tag === "started" ? "starting" : change._tag === "running" ? "running" : "idle"; // SET reads the old row, so `run_state_at` moves only when the state does. @@ -1067,6 +1086,15 @@ export const layer = Layer.effect( // A run the host cut off resumes, a clean shutdown's too: it keeps its count for the resume to claim and // its attempt, so a restart does not hand a failing request a fresh retry budget. const cutOff = interrupted === "crash" || interrupted === "shutdown"; + // A run asked to stop ends as stopped: a host interruption that lost the race to the stop's record + // rolls back, for its publisher to end the run as the user's. + if (cutOff) { + const asked = yield* sql` + SELECT 1 FROM thread + WHERE id = ${threadId} AND run_state <> 'idle' AND json_extract(run_error, '$.reason') = 'user' + `; + if (asked.length > 0) return yield* Effect.die(new RunStoppingError({ threadId })); + } const restarts = cutOff ? sql`restarts` : sql`0`; const attempt = cutOff ? sql`attempt` : sql`NULL`; yield* found( @@ -1131,7 +1159,6 @@ export const layer = Layer.effect( setAttempt, runnable, resumable, - cutOff, unfinished, projectRun, open, diff --git a/packages/harness/test/__artifacts__/thread.fork.json b/packages/harness/test/__artifacts__/thread.fork.json index 2149cd76..08f32ced 100644 --- a/packages/harness/test/__artifacts__/thread.fork.json +++ b/packages/harness/test/__artifacts__/thread.fork.json @@ -30,8 +30,9 @@ "21 session.execution.started.1 T1", "22 session.prompt.promoted.1 T1", "23 session.llm.started.1 T1", - "24 session.llm.failed.1 T1", - "25 session.execution.interrupted.1 T1" + "24 session.execution.stopping.1 T1", + "25 session.llm.failed.1 T1", + "26 session.execution.interrupted.1 T1" ], "threads": [ { @@ -66,7 +67,7 @@ "E4 T1 user seq=12 committed=12 committed", "E5 T1 assistant seq=13 committed=15 committed", "E6 T1 user seq=22 committed=22 committed", - "E7 T1 assistant seq=23 committed=24 aborted" + "E7 T1 assistant seq=23 committed=25 aborted" ], "documents": [ { diff --git a/packages/harness/test/__artifacts__/thread.lifecycle.json b/packages/harness/test/__artifacts__/thread.lifecycle.json index 0eb22025..bb36e192 100644 --- a/packages/harness/test/__artifacts__/thread.lifecycle.json +++ b/packages/harness/test/__artifacts__/thread.lifecycle.json @@ -5,13 +5,13 @@ "9 session.execution.succeeded.1 T1", "11 session.execution.started.1 T1 {\"resumed\":false}", "13 session.llm.started.1 T1", - "15 session.execution.interrupted.1 T1 {\"reason\":\"user\"}", - "17 session.execution.started.1 T1 {\"resumed\":false}", - "18 session.execution.interrupted.1 T1 {\"reason\":\"crash\"}", - "19 session.execution.started.1 T1 {\"resumed\":true}", - "20 session.llm.started.1 T1", - "24 session.llm.started.1 T1", - "27 session.execution.succeeded.1 T1" + "16 session.execution.interrupted.1 T1 {\"reason\":\"user\"}", + "18 session.execution.started.1 T1 {\"resumed\":false}", + "19 session.execution.interrupted.1 T1 {\"reason\":\"crash\"}", + "20 session.execution.started.1 T1 {\"resumed\":true}", + "21 session.llm.started.1 T1", + "25 session.llm.started.1 T1", + "28 session.execution.succeeded.1 T1" ], "states": [ { @@ -51,62 +51,62 @@ "step": "stopped", "thread": "T1", "state": "idle", - "stateAt": "15 session.execution.interrupted.1", + "stateAt": "16 session.execution.interrupted.1", "startedAt": "11 session.execution.started.1", - "endedAt": "15 session.execution.interrupted.1", + "endedAt": "16 session.execution.interrupted.1", "outcome": "interrupted", "error": { "_tag": "interrupted", "reason": "user" }, - "head": 15 + "head": 16 }, { "step": "died", "thread": "T1", "state": "starting", - "stateAt": "17 session.execution.started.1", - "startedAt": "17 session.execution.started.1", + "stateAt": "18 session.execution.started.1", + "startedAt": "18 session.execution.started.1", "endedAt": null, "outcome": null, "error": null, - "head": 15 + "head": 16 }, { "step": "swept", "thread": "T1", "state": "idle", - "stateAt": "18 session.execution.interrupted.1", - "startedAt": "17 session.execution.started.1", - "endedAt": "18 session.execution.interrupted.1", + "stateAt": "19 session.execution.interrupted.1", + "startedAt": "18 session.execution.started.1", + "endedAt": "19 session.execution.interrupted.1", "outcome": "interrupted", "error": { "_tag": "interrupted", "reason": "crash" }, - "head": 15 + "head": 16 }, { "step": "resumed", "thread": "T1", "state": "running", - "stateAt": "20 session.llm.started.1", - "startedAt": "19 session.execution.started.1", + "stateAt": "21 session.llm.started.1", + "startedAt": "20 session.execution.started.1", "endedAt": null, "outcome": null, "error": null, - "head": 15 + "head": 16 }, { "step": "finished", "thread": "T1", "state": "idle", - "stateAt": "27 session.execution.succeeded.1", - "startedAt": "19 session.execution.started.1", - "endedAt": "27 session.execution.succeeded.1", + "stateAt": "28 session.execution.succeeded.1", + "startedAt": "20 session.execution.started.1", + "endedAt": "28 session.execution.succeeded.1", "outcome": "succeeded", "error": null, - "head": 15 + "head": 16 } ] } diff --git a/packages/harness/test/control.prompt.test.ts b/packages/harness/test/control.prompt.test.ts index cb74a98b..e300fa27 100644 --- a/packages/harness/test/control.prompt.test.ts +++ b/packages/harness/test/control.prompt.test.ts @@ -453,8 +453,6 @@ const watched = Control.layer.pipe( RunnerExecution.Service.of({ active: Effect.succeed(new Set()), resume: () => Effect.void, - recover: () => Effect.void, - abandon: () => Effect.void, wake: (threadId) => Effect.sync(() => wakes.push(threadId)), interrupt: () => Effect.succeed(false), awaitIdle: () => Effect.void, diff --git a/packages/harness/test/runner.loop.test.ts b/packages/harness/test/runner.loop.test.ts index 3314a9d8..624fdae3 100644 --- a/packages/harness/test/runner.loop.test.ts +++ b/packages/harness/test/runner.loop.test.ts @@ -1343,8 +1343,60 @@ describe("runner loop — resume after restart", () => { }), ); - // Recovery chose the session, then the user ran and stopped it: the stop ended the run, and the - // recovered run must find nothing to resume rather than force another turn. + // Whatever wakes a cut-off thread first resumes it: a prompt that gets there before recovery does must + // not make the restart go uncounted, or a host dying in that run is never bounded by the limit. + it( + "counts the restart when a prompt reaches a cut-off thread before recovery does", + Effect.gen(function* () { + yield* provider.reset; + const control = yield* Control.Service; + const events = yield* Event.Service; + const { threadId } = yield* seedSession(); + const resumed: Array = []; + yield* events.listen((event) => + Schema.is(EventList.ExecutionStarted)(event) + ? Effect.sync(() => { + resumed.push(event.data.resumed); + }) + : Effect.void, + ); + yield* crashedMidReply(threadId, "A"); + const atStart = yield* atEachStart(threadId); + + yield* send(threadId, "B"); + yield* settle(threadId); + yield* control.resumeUnfinished; + + expect(atStart).toEqual([{ unfinished: true, restarts: 1 }]); + expect(resumed).toEqual([true]); + expect(contexts.map(texts)).toEqual([["A", "B"]]); + }), + ); + + it( + "ends a run at the restart limit when a prompt reaches it before recovery does", + Effect.gen(function* () { + yield* provider.reset; + const sql = yield* SqlClient.SqlClient; + const threads = yield* Thread.Service; + const { threadId } = yield* seedSession(); + yield* crashedMidReply(threadId, "A"); + yield* sql`UPDATE thread SET restarts = ${Control.RESTART_LIMIT - 1}, rev = (SELECT rev FROM store) WHERE id = ${threadId}`; + + yield* send(threadId, "B"); + yield* settle(threadId); + + // The cut-off run ends as a crashloop, and the prompt that woke it is parked with its queue. + expect(contexts).toEqual([]); + expect(Option.getOrThrow(yield* threads.get(threadId)).runError).toMatchObject( + Option.some({ _tag: "runFailed", reason: "crashloop" }), + ); + expect(yield* stateOf(threadId)).toEqual({ unfinished: false, restarts: 0 }); + }), + ); + + // The user stopped the cut-off run before anything resumed it: the stop ended the run, and a wake + // must find nothing to resume rather than force another turn. it( "runs nothing when a session recovery chose was stopped before its run began", Effect.gen(function* () { @@ -1360,7 +1412,8 @@ describe("runner loop — resume after restart", () => { reason: "user", }); - yield* execution.recover(threadId); + yield* execution.wake(threadId); + yield* settle(threadId); expect(contexts).toEqual([]); expect(yield* stateOf(threadId)).toEqual({ unfinished: false, restarts: 0 }); @@ -1589,6 +1642,41 @@ describe("runner loop — resume after restart", () => { }), ); + // A shutdown that reaches a run the user already asked to stop must not end it as a shutdown, or the + // next start resumes what the user stopped. + it( + "ends a run asked to stop as stopped when a shutdown interrupts it", + Effect.gen(function* () { + yield* provider.reset; + const events = yield* Event.Service; + const execution = yield* RunnerExecution.Service; + const threads = yield* Thread.Service; + const { threadId } = yield* seedSession(); + const base = { sessionId: yield* sessionOf(threadId), threadId }; + const streaming = yield* hanging; + script.push("hang"); + yield* send(threadId, "A"); + yield* streaming; + yield* events.publish(EventList.ExecutionStopping, { ...base, timestamp: yield* DateTime.now }); + + // A shutdown end published against the stop's record rolls back. + const lost = yield* Effect.exit( + events.publish(EventList.ExecutionInterrupted, { + ...base, + timestamp: yield* DateTime.now, + reason: "shutdown", + }), + ); + expect(Exit.isFailure(lost)).toBe(true); + yield* execution.interrupt(threadId, "shutdown", { awaitSettlement: true }); + + expect(Option.getOrThrow(yield* threads.get(threadId)).runError).toEqual( + Option.some({ _tag: "interrupted", reason: "user" }), + ); + expect(yield* stateOf(threadId)).toEqual({ unfinished: false, restarts: 0 }); + }), + ); + // The stop is accepted before the drain opened its run: it is still the user's, so what woke the drain // is parked rather than left to run on the next start. it( From 44e89f940397e839fc2ae04516d107d1081d2816 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 10 Nov 2026 16:31:01 +0530 Subject: [PATCH 2/5] test(harness): bring two live runner tests up to date The thinking test ignores durable llm.progress events, and the retry test expects the mid-stream attempt to keep what streamed. The retry proxy cuts at the first delta that carries text, not at a role-only delta. Co-Authored-By: Claude Fable 5.2 --- packages/harness/test/runner.llm.live.test.ts | 3 ++- .../harness/test/runner.retry.live.test.ts | 18 +++++++++++------- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/packages/harness/test/runner.llm.live.test.ts b/packages/harness/test/runner.llm.live.test.ts index 2b43744d..2558643c 100644 --- a/packages/harness/test/runner.llm.live.test.ts +++ b/packages/harness/test/runner.llm.live.test.ts @@ -104,7 +104,8 @@ describe.each(LIVE)("runner LLM — live $name", (live) => { expect(JSON.parse(path[0]!.entry.data)).toMatchObject({ stopReason: "stop" }); expect(path[0]!.parts.some((part) => part.type === "thinking")).toBe(true); const durable = yield* sql`SELECT type FROM event WHERE aggregate_id = ${sessionId} ORDER BY seq`; - expect(durable.map((row) => row.type)).toEqual([ + // Between the two, only the streamed progress the draft is rebuilt from. + expect(durable.map((row) => row.type).filter((type) => type !== "session.llm.progress.1")).toEqual([ "session.created.1", "session.llm.started.1", "session.llm.ended.1", diff --git a/packages/harness/test/runner.retry.live.test.ts b/packages/harness/test/runner.retry.live.test.ts index f3e8a780..04170c74 100644 --- a/packages/harness/test/runner.retry.live.test.ts +++ b/packages/harness/test/runner.retry.live.test.ts @@ -22,6 +22,9 @@ const flushed = (res: ServerResponse, chunk: Uint8Array) => * client, and every later request is forwarded whole. The harness, AI SDK and aikit see genuine * HTTP failures. Each request's arrival time and body are recorded. */ +/** A streamed field holding text: the provider's delta, content, reasoning or thinking. */ +const content = /"(delta|content|reasoning_content|reasoning|text|thinking)":\s*"[^"]/; + const proxy = async (upstream: string) => { const requests: Array<{ readonly at: number; body: Buffer }> = []; const forward = async (req: IncomingMessage, res: ServerResponse, body: Buffer, cut: boolean) => { @@ -47,10 +50,10 @@ const proxy = async (upstream: string) => { return res.end(); } /* - * A dropped connection mid-stream. Forward exactly through the first SSE event naming a - * `…delta…` (every provider's content events do), then cut: one network chunk can carry the - * whole short response, so cutting at a chunk boundary could deliver it complete. The pause - * lets the client parse the flushed event before the socket goes. + * A dropped connection mid-stream. Forward exactly through the first SSE event whose delta + * carries content (a chat-completions stream opens with a role-only delta), then cut: one + * network chunk can carry the whole short response, so cutting at a chunk boundary could + * deliver it complete. The pause lets the client parse the flushed event before the socket goes. */ const decoder = new TextDecoder(); let pending = ""; @@ -60,7 +63,7 @@ const proxy = async (upstream: string) => { pending = events.pop() ?? ""; for (const event of events) { await flushed(res, new TextEncoder().encode(`${event}\n\n`)); - if (!event.includes("delta")) continue; + if (!event.includes("delta") || !content.test(event)) continue; await new Promise((resolve) => setTimeout(resolve, 100)); return res.socket?.destroy(); } @@ -178,10 +181,11 @@ describe.each(cases)("loop retry against live $name", (live) => { const assistants = path.filter((hydrated) => hydrated.entry.type === "assistant"); expect(assistants.map((hydrated) => hydrated.entry.state)).toEqual(["aborted", "aborted", "committed"]); + // Attempt 1 failed before any byte of a response; attempt 2 failed mid-stream, and keeps + // what streamed on its aborted entry. expect(assistants[0]!.parts).toEqual([]); - expect(assistants[1]!.parts).toEqual([]); - // Attempt 1 failed before any byte of a response; attempt 2 failed mid-stream. expect(fragments.get(assistants[0]!.entry.id) ?? 0).toBe(0); + expect(assistants[1]!.parts.length).toBeGreaterThan(0); expect(fragments.get(assistants[1]!.entry.id) ?? 0).toBeGreaterThan(0); const answer = assistants[2]!.parts .filter((part) => part.type === "text") From 640063678616f1c01453feda1d4151f27299504e Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 10 Nov 2026 16:31:01 +0530 Subject: [PATCH 3/5] fix(web): show a stop CodeWork accepted as stopping A thread whose open run carries the user's stop reads as stopping in every window and after a reload. The real-machine recovery tests record the new event and expect a parked follow-up and the next steer to go in together. Co-Authored-By: Claude Fable 5.2 --- .../__artifacts__/recovery.machines.crashloop.json | 3 +-- .../test/__artifacts__/recovery.machines.stop.json | 4 ++-- apps/server/test/recovery.machines.live.test.ts | 7 +++++-- apps/web/src/widgets/session/backend.ts | 3 ++- apps/web/src/widgets/session/schema.ts | 2 ++ apps/web/src/widgets/session/view.tsx | 7 ++++++- apps/widget/src/codework.ts | 11 ++++++++++- 7 files changed, 28 insertions(+), 9 deletions(-) diff --git a/apps/server/test/__artifacts__/recovery.machines.crashloop.json b/apps/server/test/__artifacts__/recovery.machines.crashloop.json index af000175..a5497c8e 100644 --- a/apps/server/test/__artifacts__/recovery.machines.crashloop.json +++ b/apps/server/test/__artifacts__/recovery.machines.crashloop.json @@ -27,9 +27,8 @@ "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.prompt.promoted steer \"next\"", "session.llm.started", "session.execution.succeeded" ], diff --git a/apps/server/test/__artifacts__/recovery.machines.stop.json b/apps/server/test/__artifacts__/recovery.machines.stop.json index afcb7ff8..a27778eb 100644 --- a/apps/server/test/__artifacts__/recovery.machines.stop.json +++ b/apps/server/test/__artifacts__/recovery.machines.stop.json @@ -5,13 +5,13 @@ "session.prompt.promoted steer \"hold\"", "session.llm.started", "session.prompt.admitted followUp \"after the stop\"", + "session.execution.stopping", "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.prompt.promoted steer \"next\"", "session.llm.started", "session.execution.succeeded" ], diff --git a/apps/server/test/recovery.machines.live.test.ts b/apps/server/test/recovery.machines.live.test.ts index 04308e6f..f50c6f9e 100644 --- a/apps/server/test/recovery.machines.live.test.ts +++ b/apps/server/test/recovery.machines.live.test.ts @@ -96,6 +96,7 @@ const shown = new Set([ "session.execution.started", "session.execution.succeeded", "session.execution.failed", + "session.execution.stopping", "session.execution.interrupted", "session.run.failed", "session.llm.started", @@ -312,10 +313,11 @@ describeMachines("durable runs across real machine failures", () => { await idle(ids); await record("next prompt"); const ran = (await lifecycle(ids)).filter((line) => line.startsWith("session.prompt.promoted")); + // The parked follow-up and the new steer go in together, in the order they were sent. expect(ran).toEqual([ 'session.prompt.promoted steer "hold"', - 'session.prompt.promoted steer "next"', 'session.prompt.promoted followUp "after the stop"', + 'session.prompt.promoted steer "next"', ]); await snapshot("stop", { log: await lifecycle(ids), steps }); }, 300_000); @@ -356,7 +358,8 @@ describeMachines("durable runs across real machine failures", () => { 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); + // One request per run that died, then one for the parked follow-up and the next prompt together. + expect(count(await events(ids), "session.llm.started")).toBe(Control.RESTART_LIMIT + 1); await snapshot("crashloop", { log, steps }); }, 900_000); diff --git a/apps/web/src/widgets/session/backend.ts b/apps/web/src/widgets/session/backend.ts index 6fef8c0b..d1f93ba2 100644 --- a/apps/web/src/widgets/session/backend.ts +++ b/apps/web/src/widgets/session/backend.ts @@ -1,6 +1,6 @@ import { readTopics, SyncChanged, syncTopics } from "@codeworksh/contract"; import { Bus, Codework, defineBackend, handle, type LineThread, Sync } from "@codeworksh/widget/backend"; -import { LiveDelta, QueuedPrompt, UNTITLED } from "@codeworksh/widget/codework"; +import { LiveDelta, QueuedPrompt, stopping, UNTITLED } from "@codeworksh/widget/codework"; import type { Event } from "@codeworksh/widget/event"; import { Deferred, Effect, Fiber, Option, Result, Schema, Stream } from "effect"; @@ -43,6 +43,7 @@ const toInfo = (thread: LineThread): Info => { title: thread.title ?? UNTITLED, updated: thread.updatedAt, runState: thread.runState, + stopping: stopping(thread), project: thread.project, ...(thread.hostDir === undefined ? {} : { hostDir: thread.hostDir }), ...(thread.thinkingLevel === undefined ? {} : { thinkingLevel: thread.thinkingLevel }), diff --git a/apps/web/src/widgets/session/schema.ts b/apps/web/src/widgets/session/schema.ts index 304f89e0..b9f6d4a7 100644 --- a/apps/web/src/widgets/session/schema.ts +++ b/apps/web/src/widgets/session/schema.ts @@ -138,6 +138,8 @@ export const Info = Schema.Struct({ updated: Schema.Number, /** Where the thread's run is, as the harness stores it. */ runState: RunState, + /** The run is still going, but CodeWork has accepted a stop for it. */ + stopping: Schema.Boolean, /** * Why the last run failed: CodeWork's failure category (`sandbox-mount-error` and the like) and * its message. Absent once another run starts, or when it did not fail. diff --git a/apps/web/src/widgets/session/view.tsx b/apps/web/src/widgets/session/view.tsx index 359927ae..f4fc795a 100644 --- a/apps/web/src/widgets/session/view.tsx +++ b/apps/web/src/widgets/session/view.tsx @@ -456,7 +456,12 @@ function Conversation({ // A prompt this window sent is pending until it lands in app.db, as a message or a queued prompt. const key = threadKey(thread); const inFlight = usePendingPrompts(key); - const { landed, forget } = useSent(); + const { landed, forget, stop } = useSent(); + // A stop CodeWork accepted, from any window or before a reload, shows as this window's own until the run ends. + const stopAccepted = loaded.status === "ready" && loaded.info.stopping; + useEffect(() => { + if (stopAccepted) stop(key); + }, [stopAccepted, key, stop]); useEffect(() => { if (loaded.status !== "ready" || inFlight.size === 0) return; const known = new Set([...loaded.messages, ...loaded.queued].map((item) => item.id)); diff --git a/apps/widget/src/codework.ts b/apps/widget/src/codework.ts index 2655e0c9..373fb015 100644 --- a/apps/widget/src/codework.ts +++ b/apps/widget/src/codework.ts @@ -160,7 +160,10 @@ export const RunError = Schema.Union([ ]); export type RunError = typeof RunError.Type; -/** The last run of a thread; `error` is set unless it succeeded or is still going. */ +/** + * The last run of a thread; `error` is set unless it succeeded. While the run is still going it is + * set only to the user's stop that run was asked for, until its end lands. + */ export const ThreadRun = Schema.Struct({ /** Epoch milliseconds. */ startedAt: Schema.Finite, @@ -170,6 +173,12 @@ export const ThreadRun = Schema.Struct({ }); export type ThreadRun = typeof ThreadRun.Type; +/** Whether the thread's open run was asked to stop by the user: `session.execution.stopping` landed, its end not yet. */ +export const stopping = (thread: { readonly runState: RunState; readonly lastRun?: ThreadRun }) => + thread.runState !== "idle" && + thread.lastRun?.error?._tag === "interrupted" && + thread.lastRun.error.reason === "user"; + const ThreadFields = { id: Schema.String, origin: ThreadOrigin, From cccfac941b4cee39409e09717ac3b8be6f77ff5c Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 10 Nov 2026 17:04:27 +0530 Subject: [PATCH 4/5] feat(harness): tell a parked queue from one about to run A thread summary carries `parked`: idle, nothing owed a run, and a prompt pending. It is derived from the head and tail positions on every read, so clients get the fact without the positions. A change to what is pending carries the thread's revision forward, so a client that follows by revision reads the thread again. Co-Authored-By: Claude Fable 5.2 --- packages/codework/src/server/contract.ts | 2 ++ packages/codework/src/server/handlers.ts | 1 + .../codework/test/__artifacts__/sync.index.json | 3 +++ packages/codework/test/recovery.test.ts | 3 +++ packages/harness/src/effect/session.ts | 3 +++ packages/harness/src/inbox/inbox.ts | 13 ++++++++++--- packages/harness/src/thread/thread.ts | 13 +++++++++++++ .../harness/test/__artifacts__/store.revision.json | 1 + 8 files changed, 36 insertions(+), 3 deletions(-) diff --git a/packages/codework/src/server/contract.ts b/packages/codework/src/server/contract.ts index 37ad299f..c3db2716 100644 --- a/packages/codework/src/server/contract.ts +++ b/packages/codework/src/server/contract.ts @@ -143,6 +143,8 @@ const ThreadFields = { runStateAt: optional(Schema.Finite), /** Absent before the first run. */ lastRun: optional(ThreadRun), + /** Whether it holds queued prompts that wait for its next prompt instead of running by themselves. */ + parked: Schema.Boolean, /** Automatic restarts of the current run. */ restarts: Schema.Int, /** The 1-based provider request in flight; absent when none is. */ diff --git a/packages/codework/src/server/handlers.ts b/packages/codework/src/server/handlers.ts index 8dde7c89..42777b52 100644 --- a/packages/codework/src/server/handlers.ts +++ b/packages/codework/src/server/handlers.ts @@ -85,6 +85,7 @@ const toThreadSummary = (info: Session.ThreadSummary): ThreadSummary => ({ runState: info.runState, ...(info.runStateAt === undefined ? {} : { runStateAt: info.runStateAt }), ...(info.lastRun === undefined ? {} : { lastRun: info.lastRun }), + parked: info.parked, restarts: info.restarts, ...(info.attempt === undefined ? {} : { attempt: info.attempt }), createdSeq: info.createdSeq, diff --git a/packages/codework/test/__artifacts__/sync.index.json b/packages/codework/test/__artifacts__/sync.index.json index addaed73..4c1594f9 100644 --- a/packages/codework/test/__artifacts__/sync.index.json +++ b/packages/codework/test/__artifacts__/sync.index.json @@ -20,6 +20,7 @@ "cacheWrite": 0 }, "runState": "idle", + "parked": false, "restarts": 0, "createdSeq": 1, "createdAt": "