diff --git a/packages/codework/src/acp/agent.ts b/packages/codework/src/acp/agent.ts index 7593306b..8614928e 100644 --- a/packages/codework/src/acp/agent.ts +++ b/packages/codework/src/acp/agent.ts @@ -5,10 +5,12 @@ import { Cause, Deferred, Effect, Exit, Option, Schema, type Scope, type Stdio, export class RequestError extends Schema.TaggedError()("AcpRequestError", { code: Schema.Int, message: Schema.String, + data: Schema.optionalKey(Schema.Json), }) { static invalidRequest = (message: string) => new RequestError({ code: -32600, message }); static invalidParams = (message: string) => new RequestError({ code: -32602, message }); - static internalError = (message: string) => new RequestError({ code: -32603, message }); + static internalError = (message: string, data?: Schema.Json) => + new RequestError({ code: -32603, message, ...(data === undefined ? {} : { data }) }); static resourceNotFound = (message: string) => new RequestError({ code: -32002, message }); } @@ -60,7 +62,7 @@ export const serve = Effect.fn("ACP.serve")(function* ( // A JSON-RPC answer for a handler that didn't succeed; the SDK sends what `handle` throws. const toWire = (cause: Cause.Cause): Acp.RequestError => { const error = Cause.findErrorOption(cause); - if (Option.isSome(error)) return new Acp.RequestError(error.value.code, error.value.message); + if (Option.isSome(error)) return new Acp.RequestError(error.value.code, error.value.message, error.value.data); if (Cause.hasInterruptsOnly(cause)) return Acp.RequestError.requestCancelled(); return Acp.RequestError.internalError(undefined, String(Cause.squash(cause))); }; diff --git a/packages/codework/src/acp/handlers.ts b/packages/codework/src/acp/handlers.ts index d624adce..b166c6be 100644 --- a/packages/codework/src/acp/handlers.ts +++ b/packages/codework/src/acp/handlers.ts @@ -2,7 +2,6 @@ import type * as Acp from "@agentclientprotocol/sdk"; import { Control, EventList, - type EventSchema, type Harness, PromptSchema, Session, @@ -22,10 +21,9 @@ const isNotFound = Schema.is(SessionStore.SessionNotFoundError); const isConflict = Schema.is(Control.PromptConflictError); const isInvalidPrompt = Schema.is(PromptSchema.InvalidPromptError); const isLLMEnded = Schema.is(EventList.LLMEnded); -const isSucceeded = Schema.is(EventList.ExecutionSucceeded); const isFailed = Schema.is(EventList.ExecutionFailed); const isInterrupted = Schema.is(EventList.ExecutionInterrupted); -const isSettled = (event: EventSchema.Payload) => isSucceeded(event) || isFailed(event) || isInterrupted(event); +const isRunFailed = Schema.is(EventList.RunFailed); /** Harness failures as JSON-RPC errors; handlers raise protocol-level ones directly. */ const toRequestError = (error: { readonly message: string }): Agent.RequestError => { @@ -271,7 +269,7 @@ export const make = Effect.gen(function* () { onSome: (update) => client.sessionUpdate({ sessionId: id, update }).pipe(Effect.ignore), }); }), - Stream.filter(isSettled), + Stream.filter(EventList.isRunEnded), Stream.runHead, Effect.forkScoped({ startImmediately: true }), ); @@ -289,6 +287,10 @@ export const make = Effect.gen(function* () { if (Option.isNone(end) || isInterrupted(end.value)) return { stopReason: "cancelled" as const }; if (isFailed(end.value)) return yield* Agent.RequestError.internalError(end.value.data.error.message); + if (isRunFailed(end.value)) { + const { reason, message } = end.value.data; + return yield* Agent.RequestError.internalError(message, { reason }); + } return { stopReason: reason === "length" ? ("max_tokens" as const) : ("end_turn" as const) }; }).pipe(Effect.scoped), ), diff --git a/packages/codework/src/cli/error.ts b/packages/codework/src/cli/error.ts index 0f889e16..eceb0e13 100644 --- a/packages/codework/src/cli/error.ts +++ b/packages/codework/src/cli/error.ts @@ -1,4 +1,12 @@ -import { ModelCatalog, Plugin, Runner, SandboxError, SessionStore, Settings } from "@codeworksh/harness/effect"; +import { + Database, + ModelCatalog, + Plugin, + Runner, + SandboxError, + SessionStore, + Settings, +} from "@codeworksh/harness/effect"; import { SandboxProvider } from "@codeworksh/harness/sandbox"; import { Duration, Effect, Schema } from "effect"; import { Client } from "../server/client.ts"; @@ -38,6 +46,7 @@ const isSettingsError = Schema.is(Settings.SettingsError); const isSandboxDriverNotRegisteredError = Schema.is(SandboxError.SandboxDriverNotRegisteredError); const isSandboxDriverRegistrationError = Schema.is(SandboxError.SandboxDriverRegistrationError); const isExecutionError = Schema.is(Client.ExecutionError); +const isDatabaseOutdatedError = Schema.is(Database.OutdatedError); const credentialHint = (provider: string): string => { switch (provider) { @@ -196,6 +205,9 @@ export const renderError = (error: unknown): string => { if (isInvalidInputError(error)) { return `error: ${error.message}\n`; } + if (isDatabaseOutdatedError(error)) { + return `error[database-outdated]: ${error.message}\n`; + } if (isOAuthError(error)) { return ( [ diff --git a/packages/codework/src/server/client.ts b/packages/codework/src/server/client.ts index ed90907d..96e79d7a 100644 --- a/packages/codework/src/server/client.ts +++ b/packages/codework/src/server/client.ts @@ -57,7 +57,8 @@ export const run = Effect.fn("Client.run")(function* ( // go idle, and only idleness proves our prompt is never running. const stalled = yield* Deferred.make(); - const terminal = (event: EventSchema.Payload): ExecutionError | undefined => { + /** How a run ended, as this client reports it; `undefined` for a success. */ + const failure = (event: EventList.RunEnded): ExecutionError | undefined => { if (Schema.is(EventList.ExecutionInterrupted)(event)) return new ExecutionError({ type: "aborted", @@ -93,18 +94,18 @@ export const run = Effect.fn("Client.run")(function* ( } if (Schema.is(EventList.Prompted)(event) && event.data.messageId === id) promoted = true; - const ended = terminal(event); - const succeeded = Schema.is(EventList.ExecutionSucceeded)(event); + const ended = EventList.isRunEnded(event); + const error = ended ? failure(event) : undefined; if (!promoted) { // 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 (ended && seq !== undefined) ends.push({ at: seq, error }); if (admitted && ends.some(({ at }) => admittedAt === undefined || at > admittedAt)) yield* Deferred.succeed(stalled, undefined); return; } - if (ended !== undefined) return yield* ended; - if (succeeded) { + if (error !== undefined) return yield* error; + if (ended) { settled = true; return; } diff --git a/packages/codework/test/__artifacts__/acp.run-failed.json b/packages/codework/test/__artifacts__/acp.run-failed.json new file mode 100644 index 00000000..a7dd67dc --- /dev/null +++ b/packages/codework/test/__artifacts__/acp.run-failed.json @@ -0,0 +1,10 @@ +{ + "answer": { + "code": -32603, + "message": "The response hit the output token limit with tool calls 2 times in a row; the run is ended", + "data": { + "reason": "truncated" + } + }, + "requests": 2 +} diff --git a/packages/codework/test/__artifacts__/cli.old-database.json b/packages/codework/test/__artifacts__/cli.old-database.json new file mode 100644 index 00000000..bebe5c5c --- /dev/null +++ b/packages/codework/test/__artifacts__/cli.old-database.json @@ -0,0 +1,4 @@ +{ + "status": 1, + "stderr": "error[database-outdated]: The database at /data/codework.db was created by an older version and lacks thread.head, inbox.tail; delete it, with its -wal and -shm files, to recreate it\n" +} diff --git a/packages/codework/test/__artifacts__/recovery.after-failure.json b/packages/codework/test/__artifacts__/recovery.after-failure.json index e86f7f3d..2adcb7c0 100644 --- a/packages/codework/test/__artifacts__/recovery.after-failure.json +++ b/packages/codework/test/__artifacts__/recovery.after-failure.json @@ -121,7 +121,7 @@ { "seq": 21, "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, diff --git a/packages/codework/test/__artifacts__/recovery.backoff.json b/packages/codework/test/__artifacts__/recovery.backoff.json new file mode 100644 index 00000000..724eb59e --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.backoff.json @@ -0,0 +1,111 @@ +{ + "log": [ + { + "seq": 3, + "type": "session.execution.started", + "resumed": false + }, + { + "seq": 6, + "type": "session.llm.started" + }, + { + "seq": 7, + "type": "session.llm.failed", + "reason": "error", + "errorMessage": "cannot connect to API: connect ECONNREFUSED ", + "parts": [] + }, + { + "seq": 8, + "type": "session.retry.scheduled", + "attempt": 1, + "maxRetries": 3, + "delayMs": 60000, + "message": "mcp/mcp-model: cannot connect to API: connect ECONNREFUSED " + }, + { + "seq": 9, + "type": "session.execution.interrupted", + "reason": "crash" + }, + { + "seq": 10, + "type": "session.execution.started", + "resumed": true + }, + { + "seq": 11, + "type": "session.retry.scheduled", + "attempt": 1, + "maxRetries": 3, + "delayMs": 60000, + "message": "The host stopped while waiting to retry, before attempt 2" + }, + { + "seq": 12, + "type": "session.execution.interrupted", + "reason": "shutdown" + }, + { + "seq": 13, + "type": "session.execution.started", + "resumed": true + }, + { + "seq": 14, + "type": "session.llm.started" + }, + { + "seq": 15, + "type": "session.llm.ended", + "reason": "stop", + "parts": [ + { + "text": "Answered." + } + ] + }, + { + "seq": 17, + "type": "session.execution.succeeded" + } + ], + "live": [ + { + "type": "session.retry.finished", + "attempt": 1, + "success": true + } + ], + "steps": [ + { + "step": "backoff", + "restarts": 0, + "attempt": 2 + }, + { + "step": "resumed", + "restarts": 0, + "attempt": 2 + }, + { + "step": "stopped", + "restarts": 0, + "attempt": 2 + }, + { + "step": "finished", + "restarts": 0, + "attempt": null + } + ], + "requests": [ + [ + { + "role": "user", + "text": "Wait it out." + } + ] + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.crashloop.json b/packages/codework/test/__artifacts__/recovery.crashloop.json index d3e6d136..37fef6e2 100644 --- a/packages/codework/test/__artifacts__/recovery.crashloop.json +++ b/packages/codework/test/__artifacts__/recovery.crashloop.json @@ -33,7 +33,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, @@ -57,7 +57,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 2", "parts": [] }, @@ -81,7 +81,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 3", "parts": [] }, @@ -105,7 +105,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 4", "parts": [] }, @@ -129,7 +129,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The run was interrupted 5 times in a row; it is not resumed again", "parts": [] }, diff --git a/packages/codework/test/__artifacts__/recovery.interrupted.json b/packages/codework/test/__artifacts__/recovery.interrupted.json index 27a6215e..05010d2f 100644 --- a/packages/codework/test/__artifacts__/recovery.interrupted.json +++ b/packages/codework/test/__artifacts__/recovery.interrupted.json @@ -38,7 +38,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, @@ -62,7 +62,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 2", "parts": [] }, diff --git a/packages/codework/test/__artifacts__/recovery.isolation.json b/packages/codework/test/__artifacts__/recovery.isolation.json new file mode 100644 index 00000000..bf729347 --- /dev/null +++ b/packages/codework/test/__artifacts__/recovery.isolation.json @@ -0,0 +1,42 @@ +{ + "broken": [ + { + "type": "session.execution.started", + "resumed": false + }, + { + "type": "session.execution.interrupted", + "reason": "crash" + }, + { + "type": "session.execution.started", + "resumed": true + }, + { + "type": "session.execution.failed" + } + ], + "healthy": [ + { + "type": "session.execution.started", + "resumed": false + }, + { + "type": "session.execution.interrupted", + "reason": "crash" + }, + { + "type": "session.execution.started", + "resumed": true + }, + { + "type": "session.execution.succeeded" + } + ], + "resent": [ + { + "role": "user", + "text": "Healthy." + } + ] +} diff --git a/packages/codework/test/__artifacts__/recovery.mid-stream.json b/packages/codework/test/__artifacts__/recovery.mid-stream.json index ef60a11c..29c2d81e 100644 --- a/packages/codework/test/__artifacts__/recovery.mid-stream.json +++ b/packages/codework/test/__artifacts__/recovery.mid-stream.json @@ -45,7 +45,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [ { diff --git a/packages/codework/test/__artifacts__/recovery.moved.json b/packages/codework/test/__artifacts__/recovery.moved.json index ca4cb7b0..d9c203ce 100644 --- a/packages/codework/test/__artifacts__/recovery.moved.json +++ b/packages/codework/test/__artifacts__/recovery.moved.json @@ -125,7 +125,7 @@ { "seq": 22, "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, diff --git a/packages/codework/test/__artifacts__/recovery.shutdown.json b/packages/codework/test/__artifacts__/recovery.shutdown.json index 33cd96c0..97d7c326 100644 --- a/packages/codework/test/__artifacts__/recovery.shutdown.json +++ b/packages/codework/test/__artifacts__/recovery.shutdown.json @@ -39,7 +39,8 @@ "tail": 2, "promotedSeq": 4 } - ] + ], + "attempt": 1 }, "resumed": { "head": 0, diff --git a/packages/codework/test/__artifacts__/recovery.stop.json b/packages/codework/test/__artifacts__/recovery.stop.json index 72298146..0601cf60 100644 --- a/packages/codework/test/__artifacts__/recovery.stop.json +++ b/packages/codework/test/__artifacts__/recovery.stop.json @@ -84,59 +84,41 @@ { "seq": 15, "type": "session.prompt.promoted", - "delivery": "steer", - "text": "Sent while stopping." + "delivery": "followUp", + "text": "Queued before the stop." }, { "seq": 16, "type": "session.prompt.promoted", "delivery": "steer", - "text": "Next." + "text": "Sent while stopping." }, { "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." + "delivery": "steer", + "text": "Next." }, { - "seq": 21, + "seq": 18, "type": "session.llm.started" }, { - "seq": 22, + "seq": 19, "type": "session.llm.ended", "reason": "stop", "parts": [ { - "text": "Answer 2." + "text": "Answer 1." } ] }, { - "seq": 23, + "seq": 20, "type": "session.turn.ended" }, { - "seq": 24, + "seq": 21, "type": "session.execution.succeeded" } ], @@ -214,21 +196,21 @@ "delivery": "followUp", "admittedSeq": 8, "tail": 8, - "promotedSeq": 20 + "promotedSeq": 15 }, { "text": "Sent while stopping.", "delivery": "steer", "admittedSeq": 9, "tail": 9, - "promotedSeq": 15 + "promotedSeq": 16 }, { "text": "Next.", "delivery": "steer", "admittedSeq": 13, "tail": 13, - "promotedSeq": 16 + "promotedSeq": 17 } ] } @@ -257,27 +239,7 @@ }, { "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" + "text": "Queued before the stop." }, { "role": "user", @@ -286,14 +248,6 @@ { "role": "user", "text": "Next." - }, - { - "role": "assistant", - "text": "Answer 1." - }, - { - "role": "user", - "text": "Queued before the stop." } ] ] diff --git a/packages/codework/test/__artifacts__/recovery.tools.json b/packages/codework/test/__artifacts__/recovery.tools.json index 16e0c616..64ae6143 100644 --- a/packages/codework/test/__artifacts__/recovery.tools.json +++ b/packages/codework/test/__artifacts__/recovery.tools.json @@ -60,7 +60,7 @@ }, { "type": "session.llm.failed", - "reason": "error", + "reason": "aborted", "errorMessage": "The host stopped while the model was answering, on attempt 1", "parts": [] }, diff --git a/packages/codework/test/acp.test.ts b/packages/codework/test/acp.test.ts index 7e15bada..be7a3fb6 100644 --- a/packages/codework/test/acp.test.ts +++ b/packages/codework/test/acp.test.ts @@ -13,6 +13,7 @@ import { crc32, deflateSync } from "node:zlib"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import { afterAll, describe, expect, it } from "vite-plus/test"; +import { streaming, toolCalls } from "./fixtures/cli.ts"; const cli = fileURLToPath(new URL("../src/index.ts", import.meta.url)); const models = fileURLToPath(new URL("../../../models.gen.json", import.meta.url)); @@ -167,7 +168,11 @@ const settle = (effect: Effect.Effect) => if (Exit.isSuccess(exit)) return { ok: exit.value }; const failure = EffectCause.findErrorOption(exit.cause); return Option.isSome(failure) - ? { code: failure.value.code, message: failure.value.message } + ? { + code: failure.value.code, + message: failure.value.message, + ...(failure.value.data === undefined ? {} : { data: failure.value.data }), + } : { defect: String(exit.cause) }; }); @@ -365,6 +370,42 @@ describe("codework acp", () => { ); }); + // A run can end in `session.run.failed` rather than an execution terminal: the prompt must answer it. + it("answers a prompt whose run fails with an internal error carrying the reason", () => { + const { home, project } = workspace(); + return run( + Effect.gen(function* () { + const provider = yield* Effect.promise(() => + streaming((_body, response) => { + response.send(toolCalls([{ name: "bash", arguments: { command: "echo cut" } }])); + response.finish("length"); + }), + ); + writeSettings(join(home, "settings.jsonc"), { + model: { provider: "fake", id: "echo" }, + models: [newModel({ id: "fake", name: "Fake" }, "echo", { baseUrl: `${provider.url}/v1` })], + }); + yield* withAgent( + home, + (acp) => + Effect.gen(function* () { + const { sessionId } = yield* acp.agent.createSession({ cwd: project, mcpServers: [] }); + const answer = yield* settle( + acp.agent.prompt({ sessionId, prompt: [{ type: "text", text: "Run a long command." }] }), + ); + expect(answer).toMatchObject({ code: -32603, data: { reason: "truncated" } }); + yield* Effect.promise(() => + expect(artifact({ answer, requests: provider.requests.length })).toMatchFileSnapshot( + "./__artifacts__/acp.run-failed.json", + ), + ); + }), + withoutProviderKeys(), + ); + }), + ); + }, 60_000); + it("offers the models its credentials reach, and starts from the project's model", () => { const { home, project } = workspace(); // A Copilot login in the home's own credential file: `--home` must be where it is looked for. diff --git a/packages/codework/test/cli.test.ts b/packages/codework/test/cli.test.ts index ac59fb4d..c62a2451 100644 --- a/packages/codework/test/cli.test.ts +++ b/packages/codework/test/cli.test.ts @@ -3,6 +3,7 @@ import { spawnSync } from "node:child_process"; import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; +import { DatabaseSync } from "node:sqlite"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { describe, expect, it, onTestFinished } from "vite-plus/test"; @@ -503,6 +504,34 @@ describe("codework CLI", () => { ).toMatchFileSnapshot("./__artifacts__/cli.run-failed.json"); }, 60_000); + // `head` and `tail` were folded into the applied `init` migration, so the migrator never adds them. + it("refuses a database created before folded columns, naming the file, and exits 1", async () => { + const home = mkdtempSync(join(tmpdir(), "codework-cli-old-database-")); + onTestFinished(() => rmSync(home, { recursive: true, force: true })); + const fake = await streaming((_body, response) => { + response.send({ role: "assistant", content: "hello" }); + response.finish("stop"); + }); + writeSettings(home, fake.url); + const options = { cwd: home, env: { ...process.env, HOME: home, CODEWORK_MODELS_FILE: models } }; + const args = ["--home", home, "--user-config-dir", home, "run", "Say hello."]; + const created = await runCli(args, options); + expect(created.status, created.stderr).toBe(0); + + const file = join(home, "data", "codework.db"); + const db = new DatabaseSync(file); + db.exec("ALTER TABLE thread DROP COLUMN head; ALTER TABLE inbox DROP COLUMN tail;"); + db.close(); + const result = await runCli(args, options); + + expect(result.status, result.stderr).toBe(1); + expect(fake.requests).toHaveLength(1); + expect(result.stderr).toContain(file); + await expect( + `${JSON.stringify({ status: result.status, stderr: result.stderr.replaceAll(home, "") }, null, "\t")}\n`, + ).toMatchFileSnapshot("./__artifacts__/cli.old-database.json"); + }, 60_000); + it("reports a failed refresh", () => { const result = runIsolated( { CODEWORK_MODELS_FILE: undefined, CODEWORK_CATALOG_URL: "http://127.0.0.1:1/v1/models.json" }, diff --git a/packages/codework/test/recovery.test.ts b/packages/codework/test/recovery.test.ts index 8bfc8bdc..3e7b7c1d 100644 --- a/packages/codework/test/recovery.test.ts +++ b/packages/codework/test/recovery.test.ts @@ -4,6 +4,8 @@ import { Control, Session } from "@codeworksh/harness/effect"; import { Effect, Fiber, Layer, Predicate, Stream } from "effect"; import { existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; import { tmpdir } from "node:os"; import { DatabaseSync } from "node:sqlite"; import { join } from "node:path"; @@ -258,6 +260,23 @@ const truncate = (response: Reply) => { response.finish("length"); }; +/** The thread's provider request in flight, or the last one failed while a retry waits, read from the database file. */ +const attemptOf = (home: string, threadId: string) => { + const db = new DatabaseSync(join(home, "data", "codework.db")); + const row = db.prepare("SELECT attempt FROM thread WHERE id = ?").get(threadId)!; + db.close(); + return row["attempt"]; +}; + +/** A provider address nothing listens on: every request fails to connect, a transient failure. */ +const refused = async () => { + const server = createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + await new Promise((resolve) => server.close(() => resolve())); + return `http://127.0.0.1:${port}`; +}; + /** The thread's head and restart count, and each prompt's tail, read from the database file. */ const positions = (home: string, threadId: string) => { const db = new DatabaseSync(join(home, "data", "codework.db")); @@ -383,7 +402,7 @@ describe("recovery after the host is killed", () => { const failed = events.find((event) => event.type === "session.llm.failed"); expect(summary(failed!)).toMatchObject({ - reason: "error", + reason: "aborted", errorMessage: "The host stopped while the model was answering, on attempt 1", parts: [{ text: "Partial answer, streamed before the host died." }], }); @@ -764,12 +783,17 @@ describe("recovery after the host is killed", () => { await call(second.url, (rpc) => rpc["thread.wait"](ids)); record("next prompt"); - // Steers with the next prompt, the follow-up once they are answered. + // One request: the steers and the oldest follow-up go in with the next prompt, in the order they were + // sent, so the follow-up queued before the stop comes first. expect(fake.requests.map((request) => sent(request).at(-1)?.text)).toEqual([ "Run the stubborn command.", "Next.", - "Queued before the stop.", ]); + expect( + sent(fake.requests[1]!) + .filter((message) => message.role === "user") + .map((message) => message.text), + ).toEqual(["Run the stubborn command.", "Queued before the stop.", "Sent while stopping.", "Next."]); const events = await log(second.url, ids); await expect( artifact({ @@ -855,7 +879,9 @@ describe("recovery after the host is killed", () => { await logged(first.url, ids, "session.llm.started"); await until("the request", () => fake.requests.length === 1); await first.stop(); - const stopped = positions(home, ids.threadId); + const stopped = { ...positions(home, ids.threadId), attempt: attemptOf(home, ids.threadId) }; + // The request the shutdown ended keeps its attempt: the resume goes on counting from it. + expect(stopped.attempt).toBe(1); const second = await start(); await logged(second.url, ids, "session.llm.started", 2); @@ -930,4 +956,114 @@ describe("recovery after the host is killed", () => { }), ).toMatchFileSnapshot("./__artifacts__/recovery.tools.json"); }, 90_000); + + it("a thread that cannot be resumed fails alone: the other thread cut off with it still resumes", async () => { + let answering = false; + const fake = await streaming((_body, response) => (answering ? answer(response, "Resumed.") : hang(response))); + const { home, start } = setup(fake.url, retry); + + const first = await start(); + const broken = await call(first.url, (rpc) => rpc["session.create"]({})); + const healthy = await call(first.url, (rpc) => rpc["session.create"]({})); + const brokenIds = { sessionId: broken.session.id, threadId: broken.id }; + const healthyIds = { sessionId: healthy.session.id, threadId: healthy.id }; + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...brokenIds, text: "Broken." })); + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...healthyIds, text: "Healthy." })); + await logged(first.url, brokenIds, "session.llm.started"); + await logged(first.url, healthyIds, "session.llm.started"); + await until("both requests", () => fake.requests.length === 2); + await first.kill(); + // A draft no resume can read: healing it dies. + const db = new DatabaseSync(join(home, "data", "codework.db")); + db.prepare( + "UPDATE entry SET data = 'not json', rev = (SELECT rev FROM store) WHERE thread_id = ? AND state = 'draft'", + ).run(brokenIds.threadId); + db.close(); + + answering = true; + const second = await start(); + await logged(second.url, brokenIds, "session.execution.failed"); + const events = await logged(second.url, healthyIds, "session.execution.succeeded"); + await call(second.url, (rpc) => rpc["thread.wait"](healthyIds)); + await call(second.url, (rpc) => rpc["thread.wait"](brokenIds)); + + expect(fake.requests).toHaveLength(3); + const lifecycle = (log: ReadonlyArray) => + log + .filter((event) => event.type.startsWith("session.execution.") || event.type === "session.run.failed") + .map(summary); + await expect( + artifact({ + broken: lifecycle(await log(second.url, brokenIds)), + healthy: lifecycle(events), + resent: sent(fake.requests[2]!), + }), + ).toMatchFileSnapshot("./__artifacts__/recovery.isolation.json"); + }, 90_000); + + it("a run killed, then shut down, while it waits out a retry is resumed uncounted, its attempt carried", async () => { + const resend = gate(); + const fake = await streaming((_body, response) => void resend.opened.then(() => answer(response, "Answered."))); + const dead = await refused(); + // The backoff outlasts the test: the host is killed and stopped while it sleeps. + const asleep = { retry: { maxRetries: 3, baseDelayMs: 60_000, maxDelayMs: 60_000 } }; + const { home, start } = setup(dead, asleep); + const steps: Array<{ step: string; restarts: number; attempt: unknown }> = []; + let threadId = ""; + const record = (step: string) => + steps.push({ step, restarts: positions(home, threadId).restarts, attempt: attemptOf(home, threadId) }); + + const first = await start(); + const created = await call(first.url, (rpc) => rpc["session.create"]({})); + const ids = { sessionId: created.session.id, threadId: created.id }; + threadId = ids.threadId; + const live = subscribe(first.url, ids.threadId); + await live.ready; + await call(first.url, (rpc) => rpc["thread.prompt"]({ ...ids, text: "Wait it out." })); + await until("the retry", () => live.events.find((event) => event.type === "session.retry.scheduled")); + record("backoff"); + await first.kill(); + + // The resume waits out the carried retry again: a kill in backoff counted no restart. + const second = await start(); + await until("the resumed run", async () => + (await log(second.url, ids)).find( + (event) => event.type === "session.execution.started" && (event.data as Json)["resumed"] === true, + ), + ); + record("resumed"); + await second.stop(); + record("stopped"); + + writeSettings(home, fake.url, retry); + const third = await start(); + const resumed = subscribe(third.url, ids.threadId); + await resumed.ready; + await until("the resend", () => fake.requests.length === 1); + resend.open(); + await until("retry.finished", () => resumed.events.find((event) => event.type === "session.retry.finished")); + const events = await logged(third.url, ids, "session.execution.succeeded"); + await call(third.url, (rpc) => rpc["thread.wait"](ids)); + record("finished"); + + expect(steps.map((step) => step.restarts)).toEqual([0, 0, 0, 0]); + // The attempt the backoff waits for, past the one that failed: the shutdown keeps it, and it is sent next. + expect(steps.find((step) => step.step === "stopped")?.attempt).toBe(2); + await expect( + artifact({ + log: events + .filter((event) => + ["session.execution.", "session.llm.", "session.retry."].some((prefix) => + event.type.startsWith(prefix), + ), + ) + .map((event) => ({ seq: event.durable?.seq, ...summary(event) })), + // The resend goes on from the attempt the host stopped in. + live: resumed.events.filter((event) => event.type === "session.retry.finished").map(summary), + steps, + requests: fake.requests.map(sent), + // The refused address is a fresh port each run. + }).replaceAll(new URL(dead).host, ""), + ).toMatchFileSnapshot("./__artifacts__/recovery.backoff.json"); + }, 120_000); }); diff --git a/packages/codework/test/server.test.ts b/packages/codework/test/server.test.ts index 136581c1..6d96ac7a 100644 --- a/packages/codework/test/server.test.ts +++ b/packages/codework/test/server.test.ts @@ -357,14 +357,16 @@ describe("server", () => { Effect.gen(function* () { const rpc = yield* RpcTest.makeClient(Contract.Api); const events = yield* Event.Service; + // A run end for a thread that does not exist rolls back; the thread must be real. + const created = yield* rpc["session.create"]({}); const queue = yield* rpc["event.subscribe"]({}).pipe(Stream.toQueue({ capacity: 16 })); expect((yield* Queue.take(queue)).type).toBe(Envelope.Connected.type); yield* events.publish(EventSchema.define({ type: "plugin.test.custom", schema: { value: Schema.String } }), { value: "hello", }); yield* events.publish(EventList.ExecutionSucceeded, { - sessionId: Session.SessionSchema.ID.create(), - threadId: Session.ThreadSchema.ID.create(), + sessionId: Session.SessionSchema.ID.make(created.session.id), + threadId: Session.ThreadSchema.ID.make(created.id), timestamp: yield* DateTime.now, }); expect((yield* Queue.take(queue)).type).toBe(EventList.ExecutionSucceeded.type); diff --git a/packages/harness/src/control.ts b/packages/harness/src/control.ts index 54820869..6f28a554 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 { Context, DateTime, Effect, Layer, Option, Schema } from "effect"; +import { Cause, 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"; @@ -82,8 +82,9 @@ export interface Interface { /** Starts a drain while the thread is idle, or joins the one already running. */ readonly resume: (threadId: ThreadSchema.ID) => Effect.Effect; /** - * Stops work owned by this process. Idle is a no-op. Resolves once the - * interruption is accepted and returns whether there was work to accept it; + * Stops work owned by this process. Idle is a no-op, except that a user stop still ends a run the + * last process cut off, or parks a runnable prompt, when no execution here owns the thread. Resolves + * once the interruption is accepted and returns whether there was work to accept it; * `awaitSettlement` additionally waits for that execution's cleanup, without * following work another caller admits in the meantime. */ @@ -102,9 +103,10 @@ export interface Interface { readonly active: Effect.Effect>; /** * Resumes the runs 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, a shutdown - * does not; at {@link RESTART_LIMIT} the run ends visibly instead, its queue parked for the next - * prompt. Resolves once every resumed run has settled. + * 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. */ readonly resumeUnfinished: Effect.Effect; /** @@ -218,11 +220,49 @@ export const layer = Layer.effect( yield* execution.resume(threadId); }); + const isThreadNotFound = Schema.is(Thread.ThreadNotFoundError); + + /** + * A user stop that found no execution here still ends what durable state says is owed a run: a run + * cut off by the last process and not resumed yet, or a prompt left runnable. It is recorded as the + * stop it is, and lands on an execution a wake started meanwhile too. + */ + const stopIdle = Effect.fnUntraced(function* ( + threadId: ThreadSchema.ID, + options?: { readonly awaitSettlement?: boolean }, + ) { + const thread = yield* threads.get(threadId); + if (Option.isNone(thread) || !(yield* threads.resumable(threadId))) return false; + yield* events + .publish(EventList.ExecutionInterrupted, { + timestamp: yield* DateTime.now, + sessionId: SessionSchema.ID.make(thread.value.sessionId), + threadId, + reason: "user", + }) + .pipe( + // Deleted between the read and the commit: the projector rolled the event back. + Effect.catchDefect((defect) => (isThreadNotFound(defect) ? Effect.void : Effect.die(defect))), + ); + if ((yield* execution.active).has(threadId)) yield* execution.interrupt(threadId, "user", options); + return true; + }); + const interrupt = Effect.fn("Control.interrupt")( ( threadId: ThreadSchema.ID, options?: { readonly reason?: SessionSchema.InterruptReason; readonly awaitSettlement?: boolean }, - ) => Effect.uninterruptible(execution.interrupt(threadId, options?.reason ?? "user", options)), + ) => + Effect.uninterruptible( + Effect.gen(function* () { + const reason = options?.reason ?? "user"; + if (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; + return yield* stopIdle(threadId, options); + }), + ), ); const interruptSession = Effect.fn("Control.interruptSession")(function* ( @@ -274,34 +314,37 @@ export const layer = Layer.effect( const unfinished = (yield* threads.unfinished).filter((row) => !running.has(ThreadSchema.ID.make(row.id))); yield* Effect.forEach( unfinished, - Effect.fnUntraced(function* (row) { + (row) => { const threadId = ThreadSchema.ID.make(row.id); - 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; - const recovery = - restarts.value >= RESTART_LIMIT + 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); - yield* recovery.pipe( + }).pipe( + // Defects included: one thread that cannot be resumed must not cancel the others. Effect.tapCause((cause) => - Effect.logWarning("Resuming an unfinished thread failed", cause).pipe( - Effect.annotateLogs({ threadId }), - ), + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logWarning("Resuming an unfinished thread failed", cause).pipe( + Effect.annotateLogs({ threadId }), + ), ), - Effect.ignore, + Effect.ignoreCause, ); - }), + }, { concurrency: "unbounded", discard: true }, ); }).pipe(Effect.withSpan("Control.resumeUnfinished")); diff --git a/packages/harness/src/db/db.ts b/packages/harness/src/db/db.ts index a8a12c91..c625f9af 100644 --- a/packages/harness/src/db/db.ts +++ b/packages/harness/src/db/db.ts @@ -4,10 +4,11 @@ import { Migrator, SqlClient } from "effect/sql"; import { Global } from "../global.ts"; import { fileSystem } from "../host.ts"; import { posix } from "../util/posix.ts"; -import { migrations } from "./migrations.ts"; +import { migrations, OutdatedError, verify } from "./migrations.ts"; import { Store } from "./store.ts"; export { SqlClient, SqlSchema } from "effect/sql"; +export { OutdatedError }; // PRAGMAs and migrations run once during layer construction; the node:sqlite // client holds a single connection for the lifetime of the layer, so the settings @@ -45,6 +46,13 @@ export function layer(location: string) { return client.pipe( Layer.tap((context) => setup.pipe(Effect.provide(context))), Layer.orDie, + // Typed, unlike the setup above: an outdated database is the user's to delete, not a bug. + Layer.tap((context) => + verify(location).pipe( + Effect.catchTag("SqlError", (error) => Effect.die(error)), + Effect.provide(context), + ), + ), (sqlite) => Layer.provideMerge(Store.layer, sqlite), ); } diff --git a/packages/harness/src/db/migrations.ts b/packages/harness/src/db/migrations.ts index 176d11e8..c82f383a 100644 --- a/packages/harness/src/db/migrations.ts +++ b/packages/harness/src/db/migrations.ts @@ -1,4 +1,4 @@ -import { Effect } from "effect"; +import { Effect, Schema } from "effect"; import { SqlClient } from "effect/sql"; import { uuidv7 } from "uuidv7"; @@ -316,3 +316,33 @@ export const migrations = { } }), }; + +/** + * Columns added to `init` after databases were already created from it. The migrator never reruns an + * applied id, so an older database lacks them, and the first query reading one fails obscurely. + */ +const folded: ReadonlyArray = [ + ["thread", "head"], + ["inbox", "tail"], +]; + +/** The database predates columns this version reads; it is recreated, not upgraded. */ +export class OutdatedError extends Schema.TaggedError()("Database.OutdatedError", { + location: Schema.String, + missing: Schema.Array(Schema.String), +}) { + override get message(): string { + return `The database at ${this.location} was created by an older version and lacks ${this.missing.join(", ")}; delete it, with its -wal and -shm files, to recreate it`; + } +} + +/** Fails with `OutdatedError` when the migrated database at `location` lacks a folded column. */ +export const verify = Effect.fn("Database.verify")(function* (location: string) { + const sql = yield* SqlClient.SqlClient; + const missing: Array = []; + for (const [table, column] of folded) { + const columns = yield* sql<{ readonly name: string }>`SELECT name FROM pragma_table_info(${table})`; + if (!columns.some((row) => row.name === column)) missing.push(`${table}.${column}`); + } + if (missing.length > 0) return yield* new OutdatedError({ location, missing }); +}); diff --git a/packages/harness/src/event/list.ts b/packages/harness/src/event/list.ts index 00d07336..9eac010b 100644 --- a/packages/harness/src/event/list.ts +++ b/packages/harness/src/event/list.ts @@ -373,6 +373,15 @@ export const RunFailed = EventSchema.define({ }); export type RunFailed = typeof RunFailed.Type; +/** + * The one way a run ends, whatever the outcome: a consumer waiting for a run reads this and nothing + * else, as an io_uring consumer waits on one completion queue. A terminal left out of it is the hang + * a client cannot see. + */ +export const RunEnded = Schema.Union([ExecutionSucceeded, ExecutionFailed, ExecutionInterrupted, RunFailed]); +export type RunEnded = typeof RunEnded.Type; +export const isRunEnded = Schema.is(RunEnded); + /** * Kernel lifecycle notice for one plugin module. * The `plugin..` namespace belongs to plugin-declared types, so this kernel type can never collide with one. diff --git a/packages/harness/src/inbox/inbox.ts b/packages/harness/src/inbox/inbox.ts index d8298180..86855347 100644 --- a/packages/harness/src/inbox/inbox.ts +++ b/packages/harness/src/inbox/inbox.ts @@ -107,6 +107,9 @@ export interface ProjectPromptedInput extends AdmitInput { readonly timeCreated: DateTime.Utc; } +/** Where a run promotes: before a reply without tool results (`final`), or after tool results. */ +export type Boundary = "final" | "postTools"; + export interface Interface { readonly find: (id: SessionMessageSchema.ID) => Effect.Effect>; /** Records a prompt durably. Re-admitting the same id returns the stored row. */ @@ -139,22 +142,18 @@ export interface Interface { /** Whether the lane holds an input that has not been promoted yet. */ readonly hasPending: (threadId: ThreadSchema.ID, delivery: Delivery) => Effect.Effect; /** - * Promotes every pending steer admitted at or before `cutoff`, in admission - * order. The cutoff keeps a steer that lands mid-promotion out of a request - * that is already being assembled. + * Promotes, in admission order, every pending input admitted at or before `cutoff` that the boundary + * takes: at `final` (a run's first promotion, or a reply without tool results) every steer and the + * oldest follow-up, placed together; at `postTools` (tool results waiting for a reply) steers only. + * One follow-up per boundary is what makes that lane turn-at-a-time. The cutoff keeps an input that + * lands mid-promotion out of a request that is already being assembled. * * Counts promotions *this* caller won. A row another drainer promoted first * is not counted: callers treat a positive count as new work to answer, and * counting a promotion someone else committed would have two workers answer * one prompt. */ - readonly promoteSteers: (threadId: ThreadSchema.ID, cutoff: number) => Effect.Effect; - /** - * Promotes the single oldest pending follow-up, if any. One per run, unlike - * steers, which is what makes the follow-up lane strictly turn-at-a-time. - * False when there was none, or when another caller promoted it first. - */ - readonly promoteFollowUp: (threadId: ThreadSchema.ID) => Effect.Effect; + readonly promote: (threadId: ThreadSchema.ID, cutoff: number, boundary: Boundary) => Effect.Effect; } export const make = Effect.gen(function* () { @@ -277,28 +276,20 @@ export const make = Effect.gen(function* () { `, }); - const selectPendingSteers = SqlSchema.findAll({ - Request: Schema.Struct({ threadId: Schema.String, cutoff: Schema.Int }), - Result: InboxRow, - execute: ({ threadId, cutoff }) => sql` - SELECT * FROM inbox - WHERE thread_id = ${threadId} - AND promoted_seq IS NULL - AND cancelled_seq IS NULL - AND delivery = 'steer' - AND admitted_seq <= ${cutoff} - ORDER BY admitted_seq - `, - }); - - const selectPendingFollowUp = SqlSchema.findAll({ - Request: Schema.String, + const selectBoundary = SqlSchema.findAll({ + Request: Schema.Struct({ threadId: Schema.String, cutoff: Schema.Int, final: Schema.Boolean }), Result: InboxRow, - execute: (threadId) => sql` + execute: ({ threadId, cutoff, final }) => sql` SELECT * FROM inbox - WHERE thread_id = ${threadId} AND promoted_seq IS NULL AND cancelled_seq IS NULL AND delivery = 'followUp' + WHERE thread_id = ${threadId} AND promoted_seq IS NULL AND cancelled_seq IS NULL AND admitted_seq <= ${cutoff} + AND (delivery = 'steer' OR (${final ? 1 : 0} = 1 AND id = ( + SELECT id FROM inbox + WHERE thread_id = ${threadId} AND promoted_seq IS NULL AND cancelled_seq IS NULL + AND admitted_seq <= ${cutoff} AND delivery = 'followUp' + ORDER BY admitted_seq + LIMIT 1 + ))) ORDER BY admitted_seq - LIMIT 1 `, }); @@ -584,17 +575,15 @@ export const make = Effect.gen(function* () { return rows.length > 0; }, Effect.orDie); - const promoteSteers = Effect.fn("Inbox.promoteSteers")(function* (threadId: ThreadSchema.ID, cutoff: number) { - const rows = yield* selectPendingSteers({ threadId, cutoff }).pipe(Effect.orDie); + const promote = Effect.fn("Inbox.promote")(function* ( + threadId: ThreadSchema.ID, + cutoff: number, + boundary: Boundary, + ) { + const rows = yield* selectBoundary({ threadId, cutoff, final: boundary === "final" }).pipe(Effect.orDie); return yield* publishPrompted(rows); }); - const promoteFollowUp = Effect.fn("Inbox.promoteFollowUp")(function* (threadId: ThreadSchema.ID) { - const rows = yield* selectPendingFollowUp(threadId).pipe(Effect.orDie); - if (rows.length === 0) return false; - return (yield* publishPrompted(rows)) > 0; - }); - return { find, admit, @@ -606,8 +595,7 @@ export const make = Effect.gen(function* () { cancel, pending, hasPending, - promoteSteers, - promoteFollowUp, + promote, } satisfies Interface; }); diff --git a/packages/harness/src/runner/coordinator.ts b/packages/harness/src/runner/coordinator.ts index 3cc994e7..b5f4d181 100644 --- a/packages/harness/src/runner/coordinator.ts +++ b/packages/harness/src/runner/coordinator.ts @@ -120,7 +120,8 @@ export const make = (options: { if (execution === undefined || execution.stopping) return Effect.succeed(false); if (execution.owner === undefined) { // The terminal exit is already decided. Claim earlier wakes so settlement - // does not restart work for the interrupted intent. + // does not restart work for the interrupted intent; the caller decides the + // rest from durable state. execution.pendingWake = false; return Effect.succeed(false); } diff --git a/packages/harness/src/runner/execute.ts b/packages/harness/src/runner/execute.ts index b3a593ec..cd54cc3e 100644 --- a/packages/harness/src/runner/execute.ts +++ b/packages/harness/src/runner/execute.ts @@ -3,7 +3,7 @@ * Connects the process coordinator to the durable input/output loop. */ -import { Cause, Context, DateTime, Effect, Exit, Layer, Option } from "effect"; +import { Cause, Context, DateTime, Effect, Exit, Layer, Option, Schema } from "effect"; import { SqlClient } from "effect/sql"; import { Store } from "../db/store.ts"; import { Event } from "../event/event.ts"; @@ -55,6 +55,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 sessionOf = (threadId: ThreadId) => threads.get(threadId).pipe(Effect.map(Option.map((thread) => SessionSchema.ID.make(thread.sessionId)))); const publishLifecycle = ( @@ -68,7 +69,10 @@ 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 }); + yield* publish({ sessionId: sessionId.value, threadId, timestamp: yield* DateTime.now }).pipe( + // Deleted between the read and the commit: the projector rolled the event back. + Effect.catchDefect((defect) => (isThreadNotFound(defect) ? Effect.void : Effect.die(defect))), + ); }); // Threads being recovered: their run is announced as resumed. @@ -106,102 +110,118 @@ 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. - drain: Effect.fnUntraced(function* (threadId: ThreadId, force) { - 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; - 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; - // 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 }), - ), + // Level-triggered: work still runnable as a drain ends (queued behind a hook's `end`, or left by a + // 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.tap(() => + Effect.gen(function* () { + if (yield* threads.runnable(threadId)) yield* coordinator.wake(threadId); + }), + ), + ), + }); + + // 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. + 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 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; + // 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 }), ), ), ), - ); - // `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. - 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)); - // 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. - const placement = yield* sessions.placement(threadId); - if (Option.isNone(placement)) { - return yield* new Session.SessionLinkedSpaceNotFoundError({ sessionId, threadId }); + ), + ); + // `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. + 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)); + // 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. + const placement = yield* sessions.placement(threadId); + if (Option.isNone(placement)) { + return yield* new Session.SessionLinkedSpaceNotFoundError({ sessionId, threadId }); + } + const instanceId = placement.value.space.env; + const mount = sandbox.mount(instanceId, { cwd: placement.value.directory }); + + const scopedRun = Effect.gen(function* () { + const mountContext = yield* Layer.build(mount); + const current = Context.get(mountContext, SandboxIO.Current); + const fs = Context.get(mountContext, SandboxIO.FileSystem); + if (!(yield* fs.exists(current.cwd))) { + return yield* new Runner.SandboxDirectoryNotFoundError({ + threadId, + sandboxInstanceId: instanceId, + directory: current.cwd, + }); } - const instanceId = placement.value.space.env; - const mount = sandbox.mount(instanceId, { cwd: placement.value.directory }); - - const scopedRun = Effect.gen(function* () { - const mountContext = yield* Layer.build(mount); - const current = Context.get(mountContext, SandboxIO.Current); - const fs = Context.get(mountContext, SandboxIO.FileSystem); - if (!(yield* fs.exists(current.cwd))) { - return yield* new Runner.SandboxDirectoryNotFoundError({ - threadId, - sandboxInstanceId: instanceId, - directory: current.cwd, - }); - } - - const location = Location.layerMounted().pipe( - Layer.provide(Layer.succeedContext(mountContext)), - Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), - Layer.provide(Layer.succeed(Store.Service, store)), - ); - const locationContext = yield* Layer.build(location).pipe( - Effect.catchTag( - ["Location.DirectoryNotFoundError", "Location.NotDirectoryError"], - (error) => - new Runner.SandboxDirectoryNotFoundError({ - threadId, - sandboxInstanceId: error.sandboxInstanceId, - directory: error.directory, - }), - ), - ); - - // 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 }) - .pipe(Effect.provideContext(Context.merge(locationContext, mountContext)), Effect.flatMap(close)); - }); - - return yield* scopedRun.pipe( - Effect.scoped, - Effect.tapCause((cause) => - Cause.hasDies(cause) - ? Effect.logError("defect", cause).pipe(Effect.annotateLogs({ threadId })) - : Effect.void, + + const location = Location.layerMounted().pipe( + Layer.provide(Layer.succeedContext(mountContext)), + Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), + Layer.provide(Layer.succeed(Store.Service, store)), + ); + const locationContext = yield* Layer.build(location).pipe( + Effect.catchTag( + ["Location.DirectoryNotFoundError", "Location.NotDirectoryError"], + (error) => + new Runner.SandboxDirectoryNotFoundError({ + threadId, + sandboxInstanceId: error.sandboxInstanceId, + directory: error.directory, + }), ), ); - }), + + // 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* scopedRun.pipe( + Effect.scoped, + Effect.tapCause((cause) => + Cause.hasDies(cause) + ? Effect.logError("defect", cause).pipe(Effect.annotateLogs({ threadId })) + : Effect.void, + ), + ); }); const interrupt: RunnerExecution.Interface["interrupt"] = (threadId, reason, options) => diff --git a/packages/harness/src/runner/hooks.ts b/packages/harness/src/runner/hooks.ts index 4a6168bb..2e57e0b1 100644 --- a/packages/harness/src/runner/hooks.ts +++ b/packages/harness/src/runner/hooks.ts @@ -20,8 +20,8 @@ export interface Completed { /** `toolUse` means tool results wait for a reply. */ readonly reason: "stop" | "length" | "toolUse"; /** - * What the next run would promote, read without promoting it: the queued steers, or the oldest - * follow-up when there are none. As of the commit; input admitted later is not in it. + * What the next run would promote, read without promoting it: the queued steers and the oldest + * follow-up, in admission order. As of the commit; input admitted later is not in it. */ readonly pending: ReadonlyArray; } @@ -44,9 +44,9 @@ export interface Interface { /** Before every provider attempt, retries included; may replace the request. */ readonly prepareRequest: (request: Request, attempt: Attempt) => Effect.Effect; /** - * After a turn commits. `end` stops the run and leaves queued steers and follow-ups for the - * next one; `continue` asks for one more request, which the loop only sends when something - * was appended after the assistant message. + * After a turn commits. `end` ends this run; queued items start the next one. `continue` asks + * for one more request, which the loop only sends when something was appended after the + * assistant message. */ readonly finishTurn: (turn: Completed) => Effect.Effect; } diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index 6ba3d41a..d878cb8b 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -29,7 +29,7 @@ import { RunnerPart } from "./part.ts"; import { Retry } from "./retry.ts"; import { Runner } from "./run.ts"; -export type Promotion = "steer" | "followUp" | undefined; +export type Promotion = Inbox.Boundary | undefined; /** A turn that committed an assistant message; `undefined` when it sent no request. */ export type TurnResult = @@ -82,10 +82,11 @@ const hostStopped = (snapshot: State.Snapshot, message: string) => reason: new Runner.ProviderTransportError({ message, isRetryable: true }), }); -/** A request still in flight when the host died, as recovery finds it. */ -interface Crashed { - readonly attempt: number; - readonly message: string; +/** Where a resumed run's retries start: the attempts already failed, and why the last one did. */ +interface Carried { + readonly retries: number; + /** Set when the host's death failed an attempt: the resend waits out its retry first. */ + readonly failed?: string; } const errorPart = ( @@ -164,13 +165,21 @@ export const layer = (options: Options = {}) => ); }); - const publishFailure = Effect.fn("Loop.publishFailure")(function* ( + const publishFailure = ( draft: Thread.HydratedEntry, cause: EventList.TurnAbortCause, partial?: Message.AssistantMessage, + ) => + cause._tag === "interrupted" + ? publishAborted(draft, "aborted", "turn interrupted before commit", partial) + : publishAborted(draft, "error", cause.message, partial); + + const publishAborted = Effect.fn("Loop.publishAborted")(function* ( + draft: Thread.HydratedEntry, + reason: "aborted" | "error", + message: string, + partial: Message.AssistantMessage | undefined, ) { - const reason = cause._tag === "interrupted" ? "aborted" : "error"; - const message = cause._tag === "interrupted" ? "turn interrupted before commit" : cause.message; yield* events.publish(EventList.LLMFailed, { timestamp: yield* DateTime.now, sessionId: draft.entry.sessionId, @@ -211,8 +220,8 @@ export const layer = (options: Options = {}) => }); /** - * Settles the draft a dead process left. One whose request was in flight (`inFlight`, its - * failure) is aborted with what streamed. One whose response ended with tool calls had them + * Settles the draft a dead process left. One whose request was in flight (`inFlight`, why it + * stopped) is aborted with what streamed: the host stopped it, the provider did not fail it. One whose response ended with tool calls had them * running: those not settled are aborted, never run again, and the turn commits so the model * sees what ran. Anything else is aborted as interrupted. */ @@ -222,7 +231,7 @@ export const layer = (options: Options = {}) => ) { const draft = yield* threads.draft(threadId); if (Option.isNone(draft)) return; - if (inFlight !== undefined) return yield* publishFailure(draft.value, { _tag: "error", message: inFlight }); + if (inFlight !== undefined) return yield* publishAborted(draft.value, "aborted", inFlight, undefined); const decoded = yield* decodeDraft(draft.value); if (!hasToolCalls(decoded)) return yield* publishFailure(draft.value, { _tag: "interrupted" }); const ids = { sessionId: draft.value.entry.sessionId, threadId }; @@ -246,6 +255,9 @@ export const layer = (options: Options = {}) => delayMs: number, message: string, ) { + // The attempt the sleep waits for: a resume after a shutdown sends it at once, one after a crash + // waits again first. + yield* threads.setAttempt(snapshot.threadId, attempt + 1); yield* events.publish(EventList.RetryScheduled, { timestamp: yield* DateTime.now, sessionId: snapshot.sessionId, @@ -436,16 +448,15 @@ export const layer = (options: Options = {}) => promotion: Promotion, snapshot: State.Snapshot, continuing: boolean, - carried: Crashed | undefined, + carried: Carried | undefined, ) { const { sessionId, threadId } = snapshot; if (promotion !== undefined) { const cutoff = yield* events.latestSequence(sessionId); - const followUp = promotion === "followUp" && (yield* inputs.promoteFollowUp(threadId)); - const steers = yield* inputs.promoteSteers(threadId, cutoff); + const promoted = yield* inputs.promote(threadId, cutoff, promotion); // The pending input this turn was started for was cancelled before promotion: with no // earlier work to continue there is nothing to answer, so no provider request. - if (!continuing && !followUp && steers === 0) return undefined; + if (!continuing && promoted === 0) return undefined; } const assembled = yield* context.assemble(threadId); @@ -456,8 +467,8 @@ export const layer = (options: Options = {}) => let publisher: LLMEventPublisher.Publisher | undefined; let committed = false; // A recovered run goes on counting from the request the host died with. - let retries = carried?.attempt ?? 0; - let failed: string | undefined = carried?.message; + let retries = carried?.retries ?? 0; + let failed: string | undefined = carried?.failed; /** Abort the latest attempt's draft, unless its own terminal event already did. */ const settleDraft = (cause: EventList.TurnAbortCause) => @@ -592,7 +603,7 @@ export const layer = (options: Options = {}) => promotion: Promotion, snapshot: State.Snapshot, continuing: boolean, - carried: Crashed | undefined, + carried: Carried | undefined, ) { return yield* runTurnAttempt(promotion, snapshot, continuing, carried); }); @@ -601,22 +612,36 @@ export const layer = (options: Options = {}) => readonly threadId: ThreadSchema.ID; readonly force: boolean; readonly recovered?: boolean; + readonly shutdown?: boolean; }) { - // Only a dead host leaves a request marked in flight: every run that ends clears it. - const crashed = Option.getOrUndefined( + // Only a host that stopped leaves an attempt marked: every run that ends otherwise clears it. It is + // the request in flight, or the one a retry backoff waits for. A shutdown failed nothing: that + // request goes again as itself. A crash with the request streaming (its draft open) failed that + // attempt; one in backoff failed only the attempts before it. + const stored = Option.getOrUndefined( Option.flatMap(yield* threads.get(input.threadId), (row) => row.attempt), ); - const carried: Crashed | undefined = - crashed === undefined + const streaming = Option.isSome(yield* threads.draft(input.threadId)); + const carried: Carried | undefined = + stored === undefined ? undefined - : { - attempt: crashed, - message: `The host stopped while the model was answering, on attempt ${crashed}`, - }; - yield* healDraft(input.threadId, carried?.message); + : input.shutdown === true + ? { retries: stored - 1 } + : streaming + ? { + retries: stored, + failed: `The host stopped while the model was answering, on attempt ${stored}`, + } + : stored === 1 + ? { retries: 0 } + : { + retries: stored - 1, + failed: `The host stopped while waiting to retry, before attempt ${stored}`, + }; + yield* healDraft(input.threadId, carried?.failed); const hasSteer = yield* inputs.hasPending(input.threadId, "steer"); - const hasFollowUp = hasSteer ? false : yield* inputs.hasPending(input.threadId, "followUp"); + const hasFollowUp = yield* inputs.hasPending(input.threadId, "followUp"); if (!input.force && !hasSteer && !hasFollowUp) return "drained" as const; yield* Effect.annotateCurrentSpan({ "gen_ai.operation.name": "invoke_agent", @@ -629,28 +654,27 @@ export const layer = (options: Options = {}) => // join it either way. A wake from a new prompt is neither: its input goes in where it lands. const cutOff = (input.recovered === true || input.force) && - !hasSteer && hasFollowUp && (yield* context .assemble(input.threadId) .pipe(Effect.map((assembled) => assembled.messages.length > 0 && !leafAnswered(assembled)))); // A cut-off run still promotes steers that land while it loads: they join the reply being answered. - let promotion: Promotion = hasSteer || cutOff ? "steer" : hasFollowUp ? "followUp" : undefined; - /** Steers when there are any, else the oldest follow-up: what a run promotes first. */ + let promotion: Promotion = cutOff ? "postTools" : hasSteer || hasFollowUp ? "final" : undefined; + /** The steers and the oldest follow-up, in admission order: what a run promotes first. */ const peekPending = (threadId: ThreadSchema.ID) => Effect.map(inputs.pending(threadId), (queued) => { - const steers = queued.filter((item) => item.delivery === "steer"); - return steers.length > 0 ? steers : queued.slice(0, 1); + const followUp = queued.find((item) => item.delivery === "followUp"); + return queued.filter((item) => item.delivery === "steer" || item === followUp); }); const drain = Effect.gen(function* () { const start = yield* state.snapshot(input.threadId); - if (carried !== undefined) { - const delayMs = Retry.delay(start.retry, carried.attempt, hostStopped(start, carried.message)); + 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.message); + return yield* endRun(start.sessionId, input.threadId, "interrupted", carried.failed); } - yield* scheduleRetry(start, carried.attempt, delayMs, carried.message); + yield* scheduleRetry(start, carried.retries, delayMs, carried.failed); } let snapshot = start; let continuing = input.force; @@ -674,7 +698,7 @@ export const layer = (options: Options = {}) => pending: yield* peekPending(input.threadId), }; const decision = completed === undefined ? undefined : yield* hooks.finishTurn(completed); - // Queued steers and follow-ups stay for the next run. + // Queued steers and follow-ups start the next run. if (decision === "end") return "drained" as const; if (truncations >= TRUNCATION_LIMIT) { return yield* endRun( @@ -689,11 +713,13 @@ export const layer = (options: Options = {}) => // `runTurnAttempt` ends the turn unless something was appended after it. continuing = toolUse || decision === "continue"; const steer = yield* inputs.hasPending(input.threadId, "steer"); - // Only tool results outrank a follow-up. A forced continuation rides along with it: the - // follow-up is the natural next request, and the continuation only its fallback. - const followUp = !toolUse && !steer && (yield* inputs.hasPending(input.threadId, "followUp")); + // Only tool results hold a follow-up back: past them steers alone join. A reply without them is + // a final boundary, where steers and the oldest follow-up go in together, in admission order. A + // forced continuation rides along: the follow-up is the natural next request, the continuation + // only its fallback. + const followUp = !toolUse && (yield* inputs.hasPending(input.threadId, "followUp")); if (!continuing && !steer && !followUp) return "drained" as const; - promotion = followUp ? "followUp" : "steer"; + promotion = toolUse ? "postTools" : "final"; const next = holdServers(start, yield* state.snapshot(input.threadId)); snapshot = completed === undefined ? next : yield* hooks.prepareNextTurn(completed, next); diff --git a/packages/harness/src/runner/run.ts b/packages/harness/src/runner/run.ts index 4b3a9d67..16d20021 100644 --- a/packages/harness/src/runner/run.ts +++ b/packages/harness/src/runner/run.ts @@ -195,6 +195,8 @@ export interface Interface { readonly force: boolean; /** Resuming a run a previous process left unfinished: its cut-off reply is answered first. */ readonly recovered?: boolean; + /** The last run was ended by a clean shutdown: the request it stopped is resent at once, as the same attempt. */ + readonly shutdown?: boolean; }) => Effect.Effect; /** * Ends a run the server no longer resumes: settles what the dead process left of its reply, diff --git a/packages/harness/src/session/projector.ts b/packages/harness/src/session/projector.ts index adf99892..7c83ba09 100644 --- a/packages/harness/src/session/projector.ts +++ b/packages/harness/src/session/projector.ts @@ -120,7 +120,7 @@ export const layer = Layer.effectDiscard( ); // Publishing `Prompted` is what promotes an input, so this registration is - // what makes `promoteSteers` / `promoteFollowUp` take effect at all. + // what makes `promote` take effect at all. yield* events.project(EventList.Prompted, (event) => Effect.gen(function* () { const seq = yield* seqOf(event); diff --git a/packages/harness/src/thread/thread.ts b/packages/harness/src/thread/thread.ts index 688d3281..d25cb1e8 100644 --- a/packages/harness/src/thread/thread.ts +++ b/packages/harness/src/thread/thread.ts @@ -218,8 +218,8 @@ export interface Interface { 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 - * dead in it (a shutdown is not counted). The new count, or none when nothing is unfinished any - * more, say because the user stopped it since. + * 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. */ readonly claimRestart: (threadId: ThreadSchema.ID) => Effect.Effect>; /** A committed turn is progress: the restart count starts over. Runs inside `TurnEnded`'s commit. */ @@ -233,14 +233,15 @@ export interface Interface { /** 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; /** - * Root 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. + * 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. */ readonly unfinished: Effect.Effect>; /** * Projector-only: the `run_*` columns, at the time `at` of the event that changed them, and what a * run's end settles: `head` moves to an unanswered end (a failure or a user stop), which parks the - * queue; the restart count and the request in flight clear, except where the run resumes. + * queue; the restart count and the request in flight clear, except where the run resumes. An end + * landing on a run already ended only moves `head`. Dies when the thread is gone, rolling the event back. */ readonly projectRun: (threadId: ThreadSchema.ID, at: DateTime.Utc, change: RunChange) => Effect.Effect; /** Threads whose stored run is not idle: at startup, the runs a dead process left open. */ @@ -959,12 +960,23 @@ export const layer = Layer.effect( const interruptedOf = (reasons: ReadonlyArray) => sql`(thread.run_outcome = 'interrupted' AND json_extract(thread.run_error, '$.reason') IN ${sql.in(reasons)})`; const unfinishedOf = sql`(${interruptedOf(["crash", "shutdown"])} OR ${runnableOf})`; + /* + * Over a `thread` row: its run sleeps in retry backoff, its attempt set and no draft open (the failed + * attempt's draft was aborted before the sleep). A crash there is no restart: a receive count, as SQS + * keeps one per message, counts only deliveries being worked on, and a sleeping one is not; its + * carried attempt bounds it instead. The moment between setting the attempt and opening its draft + * reads as asleep too, bounded the same way. + */ + const asleepOf = sql`(thread.attempt IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM entry WHERE entry.thread_id = thread.id AND entry.state = 'draft' + ))`; const claimRestart = Effect.fn("Thread.claimRestart")( function* (threadId: ThreadSchema.ID) { const rows = yield* sql<{ restarts: number }>` UPDATE thread - SET restarts = restarts + (CASE WHEN ${interruptedOf(["crash"])} THEN 1 ELSE 0 END), rev = ${yield* store.rev} + SET restarts = restarts + (CASE WHEN ${interruptedOf(["crash"])} AND NOT ${asleepOf} THEN 1 ELSE 0 END), + rev = ${yield* store.rev} WHERE id = ${threadId} AND ${unfinishedOf} RETURNING restarts `; @@ -1003,7 +1015,7 @@ export const layer = Layer.effect( Request: Schema.Void, Result: ThreadRow, execute: () => sql` - SELECT * FROM thread WHERE origin_kind <> 'subagent' AND ${unfinishedOf} ORDER BY updated_at + SELECT * FROM thread WHERE ${unfinishedOf} ORDER BY updated_at `, }); const unfinished = findUnfinished(undefined).pipe(Effect.orDie, Effect.withSpan("Thread.unfinished")); @@ -1019,34 +1031,53 @@ export const layer = Layer.effect( change._tag === "started" ? "starting" : change._tag === "running" ? "running" : "idle"; // SET reads the old row, so `run_state_at` moves only when the state does. const stateAt = sql`CASE WHEN run_state = ${state} THEN run_state_at ELSE ${time} END`; + // Checked inside the event's commit: a lifecycle event for a thread deleted since its publisher read it + // must roll back, or it recreates the session's sequence row. + const found = (rows: ReadonlyArray) => + rows.length > 0 ? Effect.void : Effect.die(new ThreadNotFoundError({ threadId })); if (change._tag === "running") { - yield* sql`UPDATE thread SET run_state_at = ${stateAt}, run_state = ${state}, rev = ${rev} WHERE id = ${threadId}`; - return; + return yield* found( + yield* sql`UPDATE thread SET run_state_at = ${stateAt}, run_state = ${state}, rev = ${rev} WHERE id = ${threadId} RETURNING id`, + ); } if (change._tag === "started") { - yield* sql` + return yield* found( + yield* sql` UPDATE thread SET run_state_at = ${stateAt}, run_state = ${state}, run_started_at = ${time}, run_ended_at = NULL, run_outcome = NULL, run_error = NULL, rev = ${rev} WHERE id = ${threadId} - `; - return; + RETURNING id + `, + ); } const error = change.error === undefined ? null : yield* encodeRunError(change.error); const interrupted = change.error?._tag === "interrupted" ? change.error.reason : undefined; // An unanswered end parks what is queued: a prompt woken before it waits for the next one. const head = change.outcome === "failed" || interrupted === "user" ? sql`MAX(head, ${change.seq})` : sql`head`; - // A run the host cut off resumes: a crash keeps its count for the resume to claim and its request - // in flight for the retry; a shutdown ended its request itself. - const restarts = interrupted === "crash" || interrupted === "shutdown" ? sql`restarts` : sql`0`; - const attempt = interrupted === "crash" ? sql`attempt` : sql`NULL`; - yield* sql` + // A run already ended keeps the outcome of the terminal that ended it; a later one only parks. One the + // host cut off is still owed its end: a stop before its resume is that end. + const parked = yield* sql` + UPDATE thread SET head = ${head}, rev = ${rev} + WHERE id = ${threadId} AND run_state = 'idle' AND NOT ${interruptedOf(["crash", "shutdown"])} + RETURNING id + `; + if (parked.length > 0) return; + // 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"; + const restarts = cutOff ? sql`restarts` : sql`0`; + const attempt = cutOff ? sql`attempt` : sql`NULL`; + yield* found( + yield* sql` UPDATE thread SET run_state_at = ${stateAt}, run_state = ${state}, run_ended_at = ${time}, run_outcome = ${change.outcome}, run_error = ${error}, head = ${head}, restarts = ${restarts}, attempt = ${attempt}, rev = ${rev} WHERE id = ${threadId} - `; + RETURNING id + `, + ); }, Effect.orDie); const findOpen = SqlSchema.findAll({ diff --git a/packages/harness/test/control.prompt.test.ts b/packages/harness/test/control.prompt.test.ts index adf67179..cb74a98b 100644 --- a/packages/harness/test/control.prompt.test.ts +++ b/packages/harness/test/control.prompt.test.ts @@ -116,8 +116,8 @@ describe("Control.prompt", () => { const counts = yield* Effect.all( [ - inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER), - inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER), + inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"), + inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"), ], { concurrency: "unbounded" }, ); @@ -146,9 +146,15 @@ describe("Control.prompt", () => { const inputs = yield* Inbox.make; yield* control.prompt({ sessionId, prompt: prompt("Later"), delivery: "followUp" }); - const won = yield* Effect.all([inputs.promoteFollowUp(threadId), inputs.promoteFollowUp(threadId)], { - concurrency: "unbounded", - }); + const won = yield* Effect.all( + [ + inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "final"), + inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "final"), + ], + { + concurrency: "unbounded", + }, + ); expect(won.filter(Boolean).length).toBe(1); expect(yield* eventCount("session.prompt.promoted.1")).toBe(1); @@ -163,7 +169,7 @@ describe("Control.prompt", () => { const cutoff = first.admittedSeq; const second = yield* control.prompt({ sessionId, prompt: prompt("After cutoff") }); - yield* inputs.promoteSteers(threadId, cutoff); + yield* inputs.promote(threadId, cutoff, "postTools"); expect(some(yield* inputs.find(first.id)).promotedSeq).toBeDefined(); expect(some(yield* inputs.find(second.id)).promotedSeq).toBeUndefined(); @@ -178,7 +184,7 @@ describe("Control threads", () => { const inputs = yield* Inbox.make; const first = yield* control.prompt({ sessionId, prompt: prompt("On main") }); expect(first.threadId).toBe(threadId); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); const forked = yield* threads.fork({ threadId, entryId: first.id }); const fork = ThreadSchema.ID.make(forked.thread.id); @@ -221,6 +227,59 @@ describe("Control threads", () => { })); }); +describe("Control lifecycle", () => { + // The publisher read the thread, then the session was deleted before the commit: the event must roll back + // with its projection, or it recreates the session's sequence row. + it("rolls back a lifecycle event for a thread deleted since, leaving no sequence row", () => + Effect.gen(function* () { + const { sessions, sessionId, threadId } = yield* setup; + const events = yield* Event.Service; + const sql = yield* SqlClient.SqlClient; + yield* sessions.delete(sessionId); + + const exit = yield* events + .publish(EventList.ExecutionStarted, { + timestamp: yield* DateTime.now, + sessionId, + threadId, + resumed: false, + }) + .pipe(Effect.exit); + + expect(Exit.isFailure(exit) && Schema.is(Thread.ThreadNotFoundError)(Cause.squash(exit.cause))).toBe(true); + expect(yield* sql`SELECT seq FROM sequence WHERE aggregate_id = ${sessionId}`).toEqual([]); + expect(yield* sql`SELECT id FROM event WHERE aggregate_id = ${sessionId}`).toEqual([]); + })); +}); + +describe("Control.interrupt", () => { + // No execution here owns the thread (the next start has not resumed it yet), but durable state says its + // run is owed a resume: the stop must stand, not be dropped. + it("ends a run the last process cut off when no execution here owns it", () => + Effect.gen(function* () { + const { threads, control, sessionId, threadId } = yield* setup; + const events = yield* Event.Service; + yield* events.publish(EventList.ExecutionStarted, { + timestamp: yield* DateTime.now, + sessionId, + threadId, + resumed: false, + }); + yield* events.publish(EventList.ExecutionInterrupted, { + timestamp: yield* DateTime.now, + sessionId, + threadId, + reason: "crash", + }); + + expect(yield* control.interrupt(threadId)).toBe(true); + expect(yield* threads.resumable(threadId)).toBe(false); + expect((yield* threads.unfinished).map((row) => row.id)).not.toContain(threadId); + // Nothing is owed any more: a second stop has nothing to end. + expect(yield* control.interrupt(threadId)).toBe(false); + })); +}); + describe("Control.cancel", () => { it("lists pending prompts of both lanes in admission order, without promoted or cancelled ones", () => Effect.gen(function* () { @@ -239,7 +298,7 @@ describe("Control.cancel", () => { ]); yield* control.cancel(threadId, dropped.id); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); expect((yield* control.pending(threadId)).map((input) => input.id)).toEqual([later.id]); })); @@ -257,7 +316,7 @@ describe("Control.cancel", () => { expect(yield* inputs.hasPending(threadId, "steer")).toBe(false); expect(some(yield* inputs.find(admitted.id)).cancelledSeq).toBeDefined(); // Nothing is left to promote, and nothing reaches the conversation. - expect(yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)).toBe(0); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")).toBe(0); expect(yield* threads.line(threadId)).toEqual([]); })); @@ -285,7 +344,7 @@ describe("Control.cancel", () => { const { threads, control, sessionId, threadId } = yield* setup; const inputs = yield* Inbox.make; const admitted = yield* control.prompt({ sessionId, prompt: prompt(text) }); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); expect(yield* control.cancel(threadId, admitted.id)).toBe(false); @@ -319,7 +378,7 @@ describe("Control.cancel", () => { }), ).toEqual(retried); - expect(yield* inputs.promoteFollowUp(threadId)).toBe(true); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "final")).toBe(1); const line = yield* threads.line(threadId); expect(line.map((h) => h.entry.id)).toEqual([messageId]); expect(ContextCodec.parts(yield* ContextCodec.decodeMessage(line[0]!))[0]).toMatchObject({ @@ -359,7 +418,7 @@ describe("Control.cancel", () => { const admitted = yield* control.prompt({ sessionId, prompt: prompt(text) }); const [cancelled, promoted] = yield* Effect.all( - [control.cancel(threadId, admitted.id), inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)], + [control.cancel(threadId, admitted.id), inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")], { concurrency: "unbounded" }, ); @@ -460,7 +519,7 @@ describe("Control.update", () => { expect(yield* control.update(threadId, first.id, { prompt: prompt("Fix the flaky test only") })).toBe(true); expect((yield* control.pending(threadId)).map((input) => input.id)).toEqual([first.id, second.id]); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); const line = yield* threads.line(threadId); expect(line.map((h) => h.entry.id)).toEqual([first.id, second.id]); expect(ContextCodec.parts(yield* ContextCodec.decodeMessage(line[0]!))[0]).toMatchObject({ @@ -480,7 +539,7 @@ describe("Control.update", () => { tag: "test", }); const promoted = yield* control.prompt({ sessionId, prompt: prompt("Promoted") }); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); const cancelled = yield* control.prompt({ sessionId, prompt: prompt("Cancelled") }); yield* control.cancel(threadId, cancelled.id); const pending = yield* control.prompt({ sessionId, prompt: prompt("Pending"), delivery: "followUp" }); @@ -504,7 +563,7 @@ describe("Control.update", () => { const [updated, promoted] = yield* Effect.all( [ control.update(threadId, admitted.id, { prompt: prompt("Rewritten") }), - inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER), + inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"), ], { concurrency: "unbounded" }, ); @@ -575,11 +634,11 @@ describe("Control queue races", () => { : Effect.void, ); - expect(yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)).toBe(1); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")).toBe(1); expect(rewritten).toBe(true); expect(yield* control.pending(threadId)).toMatchObject([{ id: second.id }]); - expect(yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)).toBe(1); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")).toBe(1); const line = yield* threads.line(threadId); expect(line.map((h) => h.entry.id)).toEqual([first.id, second.id]); expect(ContextCodec.parts(yield* ContextCodec.decodeMessage(line[1]!))[0]).toMatchObject({ @@ -610,13 +669,13 @@ describe("Control queue races", () => { : Effect.void, ); - expect(yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)).toBe(1); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")).toBe(1); expect(resent).toBe(true); const [pending] = yield* control.pending(threadId); expect(pending?.id).toBe(second.id); expect(pending!.admittedSeq).toBeGreaterThan(before.admittedSeq); - expect(yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER)).toBe(1); + expect(yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools")).toBe(1); expect((yield* threads.line(threadId)).map((h) => h.entry.id)).toEqual([first.id, second.id]); })); diff --git a/packages/harness/test/runner.loop.test.ts b/packages/harness/test/runner.loop.test.ts index 85f02e1f..3314a9d8 100644 --- a/packages/harness/test/runner.loop.test.ts +++ b/packages/harness/test/runner.loop.test.ts @@ -157,7 +157,7 @@ describe("runner loop — aikit input/output", () => { const { effect: it } = testEffect(runtime({ contexts })); it( - "drains steers before follow-ups and persists each terminal assistant", + "places a follow-up older than a steer first, the two in the run's first request", Effect.gen(function* () { contexts.length = 0; const execution = yield* RunnerExecution.Service; @@ -170,20 +170,15 @@ describe("runner loop — aikit input/output", () => { const rows = yield* delivered(threadId); const bySeq = [...rows].sort((left, right) => (left.promotedSeq ?? 0) - (right.promotedSeq ?? 0)); - expect(bySeq.map((row) => row.id)).toEqual(["msg_steer", "msg_follow"]); + expect(bySeq.map((row) => row.id)).toEqual(["msg_follow", "msg_steer"]); expect((yield* threads.line(threadId)).map((item) => item.entry.type)).toEqual([ "user", - "system", - "assistant", "user", + "system", "assistant", ]); - - // Every request is rebuilt from durable history. The second request sees - // the first terminal assistant and the newly promoted follow-up. expect(contexts.map((context) => context.messages.map((message) => message.role))).toEqual([ - ["user", "system"], - ["user", "system", "assistant", "user"], + ["user", "user", "system"], ]); }), ); @@ -210,7 +205,7 @@ describe("runner loop — aikit input/output", () => { ); it( - "promotes all current steers as one request, then follow-ups one per request", + "promotes all current steers with the oldest follow-up as one request, then follow-ups one per request", Effect.gen(function* () { const execution = yield* RunnerExecution.Service; const threads = yield* Thread.Service; @@ -228,18 +223,17 @@ describe("runner loop — aikit input/output", () => { "user", "user", "user", - "system", - "assistant", "user", + "system", "assistant", "user", "assistant", ]); expect(path.filter((item) => item.entry.type === "user").map((item) => item.entry.id)).toEqual([ "msg_s1", + "msg_f1", "msg_s2", "msg_s3", - "msg_f1", "msg_f2", ]); }), @@ -1042,7 +1036,7 @@ describe("runner loop — queued input", () => { ); // A failed run does not start the next one: what was sent while it failed is parked, and the next - // prompt places it, steers with that prompt and follow-ups after. + // prompt places it: the steers and the oldest follow-up with that prompt, in admission order. it( "keeps prompts sent while a run fails queued until the next prompt places them", Effect.gen(function* () { @@ -1067,10 +1061,7 @@ describe("runner loop — queued input", () => { yield* settle(threadId); expect(seen).toEqual(["failed", "succeeded"]); - expect(contexts.map(texts).slice(1)).toEqual([ - ["A", "S1", "N"], - ["A", "S1", "N", "F1"], - ]); + expect(contexts.map(texts).slice(1)).toEqual([["A", "F1", "S1", "N"]]); }), ); @@ -1125,7 +1116,7 @@ describe("runner loop — queued input", () => { ); // A stop claims the wakes of what was sent during the run: it stays queued, and the next prompt - // places it, steers with that prompt and follow-ups after. + // places it: the steers and the oldest follow-up with that prompt, in admission order. it( "keeps prompts sent before a stop queued until the next prompt places them", Effect.gen(function* () { @@ -1160,10 +1151,7 @@ describe("runner loop — queued input", () => { yield* settle(threadId); expect(seen).toEqual(["interrupted", "succeeded"]); - expect(contexts.map(texts).slice(1)).toEqual([ - ["A", "S1", "N"], - ["A", "S1", "N", "F1"], - ]); + expect(contexts.map(texts).slice(1)).toEqual([["A", "S1", "F1", "N"]]); }), ); }); @@ -1171,7 +1159,8 @@ describe("runner loop — queued input", () => { describe("runner loop — resume after restart", () => { const provider = scriptedProvider(); const { contexts, script } = provider; - const { effect: it } = testEffect(runtime({ open: provider.open })); + // Live, with a short backoff: a resumed request whose attempt was carried waits out its retry first. + const { live: it } = testEffect(runtime({ open: provider.open, settings: settings({ baseDelayMs: 1 }) })); /** Whether the next server start would resume the thread, and its restart count. */ const stateOf = (threadId: ThreadSchema.ID) => @@ -1221,7 +1210,7 @@ describe("runner loop — resume after restart", () => { const events = yield* Event.Service; const inputs = yield* Inbox.make; yield* admit({ id, threadId, delivery: "steer" }); - yield* inputs.promoteSteers(threadId, Number.MAX_SAFE_INTEGER); + yield* inputs.promote(threadId, Number.MAX_SAFE_INTEGER, "postTools"); const input: LLM.Input = { resolvedModel: yield* LLM.resolve({ provider: "openai", model: "gpt-4o-mini" }), threadId, @@ -1498,6 +1487,85 @@ describe("runner loop — resume after restart", () => { }), ); + // The drain is gone and its `settled` hook is running: a stop finds nothing to interrupt, yet a prompt sent + // in that window would otherwise run on its own at the next start. The stop is decided from durable state. + it( + "parks a prompt sent while a failed run settles when the stop finds no drain to interrupt", + Effect.gen(function* () { + yield* provider.reset; + const control = yield* Control.Service; + const events = yield* Event.Service; + const seen = yield* outcomes; + const { threadId } = yield* seedSession(); + const services = yield* Effect.context(); + const stopped: boolean[] = []; + yield* events.listen((event) => + event.type === EventList.ExecutionFailed.type + ? send(threadId, "Q", "followUp").pipe( + Effect.andThen(control.interrupt(threadId, { reason: "user" })), + Effect.map((accepted) => void stopped.push(accepted)), + Effect.orDie, + Effect.provideContext(services), + ) + : Effect.void, + ); + script.push("fail"); + + yield* send(threadId, "A"); + yield* settle(threadId); + yield* control.resumeUnfinished; + + // No drain was left to interrupt: the stop is recorded against the runnable prompt instead. + expect(stopped).toEqual([true]); + expect(seen).toEqual(["failed", "interrupted"]); + expect(contexts.map(texts)).toEqual([["A"]]); + expect(yield* stateOf(threadId)).toEqual({ unfinished: false, restarts: 0 }); + }), + ); + + // A stop after `run.failed` ends nothing: the run row keeps the failure, and the stop only parks what was + // sent in between. + it( + "keeps a run's failure when a stop after its run.failed parks a prompt sent in between", + Effect.gen(function* () { + yield* provider.reset; + const control = yield* Control.Service; + const events = yield* Event.Service; + const execution = yield* RunnerExecution.Service; + const sql = yield* SqlClient.SqlClient; + const { threadId } = yield* seedSession(); + const services = yield* Effect.context(); + let failedSeq = 0; + yield* events.listen((event) => + event.type === EventList.RunFailed.type + ? Effect.sync(() => (failedSeq = event.durable?.seq ?? 0)).pipe( + Effect.andThen(send(threadId, "Q", "followUp")), + Effect.andThen(execution.interrupt(threadId, "user")), + Effect.asVoid, + Effect.orDie, + Effect.provideContext(services), + ) + : Effect.void, + ); + yield* crashedMidReply(threadId, "A"); + yield* sql`UPDATE thread SET restarts = ${Control.RESTART_LIMIT - 1}, rev = (SELECT rev FROM store) WHERE id = ${threadId}`; + + yield* Effect.exit(control.resumeUnfinished); + yield* settle(threadId); + + const [row] = yield* sql<{ + runOutcome: string; + runError: string; + head: number; + }>`SELECT run_outcome, run_error, head FROM thread WHERE id = ${threadId}`; + expect(row!.runOutcome).toBe("failed"); + expect(JSON.parse(row!.runError)).toMatchObject({ _tag: "runFailed", reason: "crashloop" }); + expect(row!.head).toBeGreaterThan(failedSeq); + expect(yield* stateOf(threadId)).toEqual({ unfinished: false, restarts: 0 }); + expect(contexts).toEqual([]); + }), + ); + // A thread a crash cut off, stopped before a drain opened a run: nothing is runnable, but the stop must // still stand, or the next start resumes the run the user stopped. it( @@ -1722,7 +1790,7 @@ describe("runner loop — lifecycle hooks", () => { ); it( - "shows the queued steers, else the oldest follow-up, without promoting them", + "shows the queued steers and the oldest follow-up, without promoting them", Effect.gen(function* () { const execution = yield* RunnerExecution.Service; const { threadId } = yield* seedSession(); @@ -1732,8 +1800,8 @@ describe("runner loop — lifecycle hooks", () => { yield* execution.resume(threadId); - // Turn 1 (toolUse), turn 2 (stop), turn 3 (first follow-up), turn 4 (second follow-up). - expect(seen).toEqual([["followUp:msg_later"], ["followUp:msg_later"], ["followUp:msg_next"], []]); + // Turn 1 (the steer and the first follow-up; toolUse), turn 2 (stop), turn 3 (second follow-up). + expect(seen).toEqual([["followUp:msg_next"], ["followUp:msg_next"], []]); }), ); }); @@ -1741,54 +1809,46 @@ describe("runner loop — lifecycle hooks", () => { describe("finishTurn end", () => { const contexts: Message.Context[] = []; let prepared = 0; + /** Runs as the hook ends a run. */ + const ending = { run: Effect.void as Effect.Effect }; const { effect: it } = testEffect( runtime({ open: recording(contexts), hooks: hooksOf({ - finishTurn: (turn) => Effect.succeed(turn.reason === "toolUse" ? "end" : undefined), + finishTurn: (turn) => + turn.reason === "toolUse" ? ending.run.pipe(Effect.as("end" as const)) : Effect.undefined, prepareNextTurn: (_turn, next) => Effect.sync(() => ((prepared += 1), next)), }), plugins: [...plugins, { id: "test.tool.echo", kind: "tool", setup: (ctx) => ctx.plugin.tools.add(echo) }], }), ); + // Level-triggered: the drain ends with a prompt still runnable, so it drains again with no new bell. it( - "stops the run after the committed turn and leaves queued input for the next run", + "ends the run after the committed turn, and what is queued starts the next run without a new prompt", Effect.gen(function* () { const execution = yield* RunnerExecution.Service; + const seen = yield* outcomes; const { threadId } = yield* seedSession(); + const services = yield* Effect.context>>(); + ending.run = admit({ id: "msg_queued", threadId, delivery: "followUp" }).pipe( + Effect.asVoid, + Effect.provideContext(services), + ); yield* admit({ id: "msg_first", threadId, delivery: "steer" }); - yield* admit({ id: "msg_queued", threadId, delivery: "followUp" }); yield* execution.resume(threadId); - expect(contexts).toHaveLength(1); expect(prepared).toBe(0); - const threads = yield* Thread.Service; - const path = yield* threads.line(threadId); - expect(path.map((item) => [item.entry.type, item.entry.state])).toEqual([ - ["user", "committed"], - ["system", "committed"], - ["assistant", "committed"], + expect(seen).toEqual(["succeeded", "succeeded"]); + // The next run places the follow-up where it lands, after the tool results. + expect(contexts.map((context) => context.messages.map((message) => message.role))).toEqual([ + ["user", "system"], + ["user", "system", "assistant", "user"], ]); - const tool = path[2]!.parts.find((part) => part.type === "toolCall"); - expect(tool).toBeDefined(); + expect(contexts.map(texts)).toEqual([["msg_first"], ["msg_first", "msg_queued"]]); const rows = yield* delivered(threadId); - expect(rows.find((row) => row.id === "msg_queued")?.promotedSeq).toBeNull(); - - // The next run answers the tool results first, then picks the queued follow-up up. - yield* execution.resume(threadId); - expect(contexts).toHaveLength(3); - expect(contexts[1]!.messages.map((message) => message.role)).toEqual(["user", "system", "assistant"]); - expect(contexts[2]!.messages.map((message) => message.role)).toEqual([ - "user", - "system", - "assistant", - "assistant", - "user", - ]); - const next = yield* delivered(threadId); - expect(next.find((row) => row.id === "msg_queued")?.promotedSeq).not.toBeNull(); + expect(rows.find((row) => row.id === "msg_queued")?.promotedSeq).not.toBeNull(); }), ); }); @@ -1811,8 +1871,9 @@ describe("runner loop — lifecycle hooks", () => { const execution = yield* RunnerExecution.Service; const { threadId } = yield* seedSession(); yield* admit({ id: "msg_first", threadId, delivery: "steer" }); - yield* admit({ id: "msg_queued", threadId, delivery: "followUp" }); yield* execution.resume(threadId); + // Queued with no wake: the explicit resume below finds it behind the unanswered tool results. + yield* admit({ id: "msg_queued", threadId, delivery: "followUp" }); const services = yield* Effect.context>>(); gate.run = admit({ id: "msg_late", threadId, delivery: "steer" }).pipe( diff --git a/packages/harness/test/thread.test.ts b/packages/harness/test/thread.test.ts index a8043af5..e83ba90a 100644 --- a/packages/harness/test/thread.test.ts +++ b/packages/harness/test/thread.test.ts @@ -769,3 +769,77 @@ describe("threads", () => { }), ); }); + +describe("thread runs left to resume", () => { + /** The thread's stored run, as a dead host left it once the next start swept it. */ + const crashed = Effect.fnUntraced(function* (sessionId: SessionSchema.ID, threadId: ThreadSchema.ID) { + const events = yield* Event.Service; + yield* events.publish(EventList.ExecutionStarted, { + timestamp: yield* DateTime.now, + sessionId, + threadId, + resumed: false, + }); + yield* events.publish(EventList.ExecutionInterrupted, { + timestamp: yield* DateTime.now, + sessionId, + threadId, + reason: "crash", + }); + }); + + it( + "lists a subagent thread whose run a crash cut off", + Effect.gen(function* () { + yield* reset; + const threads = yield* Thread.Service; + const { sessionId, main } = yield* createSession; + const subagent = ThreadSchema.ID.create(); + yield* threads.insert({ + id: subagent, + sessionId, + origin: { kind: "subagent", threadId: main, callId: "call_1" }, + seq: 1, + }); + yield* crashed(sessionId, subagent); + + expect((yield* threads.unfinished).map((row) => row.id)).toContain(subagent); + }), + ); + + // A host that keeps dying at the same point between requests must reach the restart limit; only a run + // asleep in retry backoff is spared. + it( + "counts a crash as a restart unless the run slept in retry backoff", + Effect.gen(function* () { + yield* reset; + const threads = yield* Thread.Service; + const events = yield* Event.Service; + const { sessionId, main } = yield* createSession; + yield* crashed(sessionId, main); + + // Between requests: no attempt marked, no draft open. + expect(yield* threads.claimRestart(main)).toEqual(Option.some(1)); + // Asleep in backoff: the attempt it waits for is marked, the failed one's draft already aborted. + yield* threads.setAttempt(main, 2); + expect(yield* threads.claimRestart(main)).toEqual(Option.some(1)); + // Streaming that attempt: its draft is open. + const input: LLM.Input = { + resolvedModel: yield* LLM.resolve({ provider: "openai", model: "gpt-4o-mini" }), + threadId: main, + context: { messages: [] }, + provider: "openai", + model: "gpt-4o-mini", + }; + const draft = assistant(input, 1, { parts: [{ type: "text", text: "half" }] }); + yield* events.publish(EventList.LLMStarted, { + timestamp: yield* DateTime.now, + sessionId, + threadId: main, + messageId: SessionMessageSchema.ID.from(draft.messageId), + message: draft, + }); + expect(yield* threads.claimRestart(main)).toEqual(Option.some(2)); + }), + ); +});