diff --git a/apps/web/src/widgets/session/schema.ts b/apps/web/src/widgets/session/schema.ts index 70246f04..eb242af1 100644 --- a/apps/web/src/widgets/session/schema.ts +++ b/apps/web/src/widgets/session/schema.ts @@ -12,7 +12,7 @@ export const topics = { /** Requests: create a session and send its first prompt; answered with its id. */ create: "codework/session.create", prompt: "codework/session.prompt", - /** Requests: choose a session's model or thinking level; they apply from its next exchange. */ + /** Requests: choose a session's model or thinking level; they apply from its next turn. */ configure: "codework/session.configure", interrupt: "codework/session.interrupt", /** Requests: withdraw a queued prompt before the agent takes it in; answered with `Cancelled`. */ diff --git a/apps/widget/src/codework.ts b/apps/widget/src/codework.ts index 4265e94c..6588eef7 100644 --- a/apps/widget/src/codework.ts +++ b/apps/widget/src/codework.ts @@ -242,7 +242,7 @@ export const Api = RpcGroup.make( payload: {}, success: Schema.Array(SessionInfo), }), - /** Sets the session's model and thinking level over its settings; they apply from its next exchange. */ + /** Sets the session's model and thinking level over its settings; they apply from its next turn. */ Rpc.make("session.configure", { payload: { sessionId: Schema.String, runtime: RuntimeConfig }, success: SessionInfo, diff --git a/packages/harness/README.md b/packages/harness/README.md index 084ecb6a..b5822184 100644 --- a/packages/harness/README.md +++ b/packages/harness/README.md @@ -22,7 +22,7 @@ that owns the plugin, tool and sandbox contract this package consumes — a plug on it rather than on the whole harness. The same types are re-exported here as `Plugin` and `Tool` for embedders who already hold the harness. -Pass a plugin list when constructing the harness. Each plugin declares an ID and the domain it extends — `tool` or `prompt` — and contributes during setup. Setup runs once per exchange, after model resolution; its tools, hooks, and prompt remain pinned through tool continuations. +Pass a plugin list when constructing the harness. Each plugin declares an ID and the domain it extends — `tool` or `prompt` — and contributes during setup. Setup runs before every turn, after model resolution; its tools, hooks, and prompt are pinned for that turn. ```ts import { Effect, Schema } from "effect"; diff --git a/packages/harness/src/effect/harness.ts b/packages/harness/src/effect/harness.ts index a9dd1878..e7e2649f 100644 --- a/packages/harness/src/effect/harness.ts +++ b/packages/harness/src/effect/harness.ts @@ -14,6 +14,7 @@ import { PluginSource } from "../plugin/source.ts"; import { PluginStore } from "../plugin/store.ts"; import { RunnerExecute } from "../runner/execute.ts"; import { LLM } from "../runner/llm.ts"; +import { Hooks } from "../runner/hooks.ts"; import { Loop } from "../runner/loop.ts"; import { SandboxController } from "../sandbox/control.ts"; import { SandboxDriver } from "../sandbox/driver.ts"; @@ -248,7 +249,7 @@ export const layer = (options: Options = {}) => ); const authFile = Global.authFile(options.home === undefined ? undefined : paths.home); const open = options.llm ?? LLM.openWith(authFile === undefined ? {} : { authFile }); - const loop = Loop.layer({ request: LLM.make(open) }); + const loop = Loop.layer({ request: LLM.make(open) }).pipe(Layer.provide(Hooks.layer)); return Control.layer.pipe( Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(loop))), diff --git a/packages/harness/src/effect/session.ts b/packages/harness/src/effect/session.ts index 90696c56..68019552 100644 --- a/packages/harness/src/effect/session.ts +++ b/packages/harness/src/effect/session.ts @@ -346,7 +346,7 @@ export const attach = Effect.fn("Session.attach")(function* (input: AttachInput) }); /** - * The model and thinking level the session's next exchange runs with. A choice made through + * The model and thinking level the session's next turn runs with. A choice made through * {@link create} or {@link attach} is saved with the session and outlives the process; without * one, the session follows its settings. */ diff --git a/packages/harness/src/plugin/catalog.ts b/packages/harness/src/plugin/catalog.ts index c4293cd6..b1964340 100644 --- a/packages/harness/src/plugin/catalog.ts +++ b/packages/harness/src/plugin/catalog.ts @@ -197,7 +197,7 @@ export interface Selected { * is ignored rather than failing the boot — a typo costs a debug line, never a fetch. * * Nothing here touches the disk, which is the whole point of the split: an `options` edit, an - * `enabled` flip or a reordering applies at the next exchange without importing anything. + * `enabled` flip or a reordering applies at the next turn without importing anything. */ export const select = Effect.fn("PluginCatalog.select")(function* (references: ReadonlyArray, pool: Pool) { const operations = new Map(); diff --git a/packages/harness/src/runner/hooks.ts b/packages/harness/src/runner/hooks.ts new file mode 100644 index 00000000..3ccb3d3d --- /dev/null +++ b/packages/harness/src/runner/hooks.ts @@ -0,0 +1,65 @@ +/* + * @file Lifecycle seams of the session drain. + * + * The loop calls these and knows nothing about what sits behind them. The live layer is the one + * place that fixes the order consumers run in (compaction before a loadout rebuild, context + * projection before routing), so adding a consumer means editing it, not registering from afar. + * Harness-internal: plugins get nothing from here. + */ + +import { Context, Effect, Layer } from "effect"; +import type { Admitted } from "../session/input/schema.ts"; +import type { State } from "../state/state.ts"; +import type { LLM } from "./llm.ts"; + +/** A committed turn. */ +export interface Completed { + /** The snapshot the turn ran on. */ + readonly snapshot: State.Snapshot; + readonly messageId: string; + readonly reason: "stop" | "length" | "toolUse"; + readonly needsContinuation: boolean; + /** + * 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. + */ + readonly pending: ReadonlyArray; +} + +/** One provider request, as the loop built it. */ +export type Request = Omit; + +export interface Attempt { + /** 0 for the first attempt of a turn. */ + readonly retry: number; + /** Why the previous attempt failed. */ + readonly failed?: string; +} + +export type Decision = "end" | "continue"; + +export interface Interface { + /** Before every turn but the first of a run, with the fresh snapshot; may replace it. */ + readonly prepareNextTurn: (turn: Completed, next: State.Snapshot) => Effect.Effect; + /** 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. + */ + readonly finishTurn: (turn: Completed) => Effect.Effect; +} + +export class Service extends Context.Service()("@codeworksh/harness/runner/hooks/Service") {} + +export const layer = Layer.succeed( + Service, + Service.of({ + prepareNextTurn: (_turn, next) => Effect.succeed(next), + prepareRequest: (request) => Effect.succeed(request), + finishTurn: () => Effect.undefined, + }), +); + +export * as Hooks from "./hooks.ts"; diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index 25a7fd01..83808a11 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -19,6 +19,7 @@ import { State } from "../state/state.ts"; import { Telemetry } from "../telemetry/telemetry.ts"; import { errorMessage } from "../util/error.ts"; import { LLMEventPublisher } from "./event.ts"; +import { Hooks } from "./hooks.ts"; import { LLM } from "./llm.ts"; import { Retry } from "./retry.ts"; import { Runner } from "./run.ts"; @@ -27,7 +28,9 @@ export type Promotion = "steer" | "followUp" | undefined; export interface TurnResult { readonly needsContinuation: boolean; + /** Set with `reason` when the turn committed an assistant message. */ readonly messageId?: string; + readonly reason?: Hooks.Completed["reason"]; } export interface Options { @@ -89,6 +92,7 @@ export const layer = (options: Options = {}) => Effect.gen(function* () { const context = yield* Context.Service; const events = yield* Event.Service; + const hooks = yield* Hooks.Service; const inputs = yield* SessionInput.make; const sessions = yield* Session.Service; const state = yield* State.Service; @@ -291,6 +295,7 @@ export const layer = (options: Options = {}) => let publisher: LLMEventPublisher.Publisher | undefined; let committed = false; let retries = 0; + let failed: string | undefined; /** Abort the latest attempt's draft, unless its own terminal event already did. */ const settleDraft = (cause: EventList.TurnAbortCause) => @@ -306,26 +311,28 @@ export const layer = (options: Options = {}) => publisher = yield* LLMEventPublisher.make({ sessionId }).pipe( Effect.provideService(Event.Service, events), ); - const terminal = yield* requestLLM({ - sessionId, - context: { - systemPrompt: snapshot.systemPrompt, - messages: [...assembled.messages], - tools: [...snapshot.tools.wire], + const prepared = yield* hooks.prepareRequest( + { + context: { + systemPrompt: snapshot.systemPrompt, + messages: [...assembled.messages], + tools: [...snapshot.tools.wire], + }, + provider: snapshot.provider, + model: snapshot.model, + resolvedModel: snapshot.resolvedModel, + thinkingLevel: snapshot.thinkingLevel, + options: snapshot.request, + settings: snapshot.settings, }, - provider: snapshot.provider, - model: snapshot.model, - resolvedModel: snapshot.resolvedModel, - thinkingLevel: snapshot.thinkingLevel, - options: snapshot.request, - settings: snapshot.settings, - publisher, - }); + { retry: retries, ...(failed === undefined ? {} : { failed }) }, + ); + const terminal = yield* requestLLM({ sessionId, ...prepared, publisher }); if (terminal.outcome === "failed") { if (terminal.reason === "aborted") return yield* Effect.interrupt; return yield* LLM.providerError( - { provider: snapshot.provider, model: snapshot.model }, + { provider: prepared.provider, model: prepared.model }, LLM.messageFailure(terminal.message), ); } @@ -349,6 +356,7 @@ export const layer = (options: Options = {}) => if (delayMs === undefined) return Effect.fail(error); return Effect.gen(function* () { retries += 1; + failed = error.message; yield* Effect.uninterruptible(settleDraft({ _tag: "error", message: error.message })); yield* events.publish(EventList.RetryScheduled, { timestamp: yield* DateTime.now, @@ -395,6 +403,7 @@ export const layer = (options: Options = {}) => return { needsContinuation: terminal.reason === "toolUse", messageId: terminal.message.messageId, + reason: terminal.reason, } satisfies TurnResult; }); @@ -443,10 +452,11 @@ export const layer = (options: Options = {}) => "gen_ai.conversation.id": input.sessionId, }); - // A recovered run first answers the reply it was cut off in, from the same committed messages, - // before a queued follow-up goes in; steers join it either way. + // A recovered or explicitly resumed run first answers the reply it stopped in (cut off, or ended + // after a tool batch), from the same committed messages, before a queued follow-up goes in; steers + // 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.recovered === true || input.force) && !hasSteer && hasFollowUp && (yield* context.assemble(input.sessionId).pipe( @@ -460,23 +470,48 @@ export const layer = (options: Options = {}) => ); }), )); - let promotion: Promotion = hasSteer ? "steer" : hasFollowUp && !cutOff ? "followUp" : undefined; + // 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. */ + const peekPending = (sessionId: SessionSchema.ID) => + Effect.map(inputs.pending(sessionId), (queued) => { + const steers = queued.filter((item) => item.delivery === "steer"); + return steers.length > 0 ? steers : queued.slice(0, 1); + }); + const drain = Effect.gen(function* () { - let shouldRun = true; + let snapshot = yield* state.snapshot(input.sessionId); let continuing = input.force; - while (shouldRun) { - const snapshot = yield* state.snapshot(input.sessionId); - let needsContinuation = true; - while (needsContinuation) { - const result = yield* runTurn(promotion, snapshot, continuing); - promotion = "steer"; - continuing = result.needsContinuation; - needsContinuation = - result.needsContinuation || (yield* inputs.hasPending(input.sessionId, "steer")); - } - continuing = false; - shouldRun = yield* inputs.hasPending(input.sessionId, "followUp"); - promotion = shouldRun ? "followUp" : undefined; + // Tool results and steers continue first; a follow-up only once both are exhausted. + while (true) { + const result: TurnResult = yield* runTurn(promotion, snapshot, continuing); + const completed: Hooks.Completed | undefined = + result.messageId === undefined || result.reason === undefined + ? undefined + : { + snapshot, + messageId: result.messageId, + reason: result.reason, + needsContinuation: result.needsContinuation, + pending: yield* peekPending(input.sessionId), + }; + const decision = completed === undefined ? undefined : yield* hooks.finishTurn(completed); + // Queued steers and follow-ups stay for the next run. + if (decision === "end") return; + + // A forced continuation after a finished reply sends nothing: the leaf guard in + // `runTurnAttempt` ends the turn unless something was appended after it. + continuing = result.needsContinuation || decision === "continue"; + const steer = yield* inputs.hasPending(input.sessionId, "steer"); + // Only tool results outrank a follow-up. A forced continuation rides along with it, as in + // Pi, where the follow-up is the natural next request and the continuation its fallback. + const followUp = + !result.needsContinuation && !steer && (yield* inputs.hasPending(input.sessionId, "followUp")); + if (!continuing && !steer && !followUp) return; + promotion = followUp ? "followUp" : "steer"; + + const next = yield* state.snapshot(input.sessionId); + snapshot = completed === undefined ? next : yield* hooks.prepareNextTurn(completed, next); } }); yield* Telemetry.timed(drain, (seconds, exit) => diff --git a/packages/harness/src/settings/settings.ts b/packages/harness/src/settings/settings.ts index 0465dee4..7802e2a2 100644 --- a/packages/harness/src/settings/settings.ts +++ b/packages/harness/src/settings/settings.ts @@ -28,7 +28,7 @@ * directory, because that would answer with the wrong project rather than with no project. * * `load` re-reads all layers on every call. There is no cache to invalidate and no reload - * API: an edit lands at the next exchange capture because the next capture goes to disk. + * API: an edit lands at the next turn capture because the next capture goes to disk. * * A missing file is ordinary; a file that exists and cannot be used is not. An unreadable, * unparseable or invalid file fails the read with the path, the reason and the offending key. diff --git a/packages/harness/src/state/state.ts b/packages/harness/src/state/state.ts index 6b460f39..357f535c 100644 --- a/packages/harness/src/state/state.ts +++ b/packages/harness/src/state/state.ts @@ -1,10 +1,10 @@ /* - * @file Runtime state snapshot for one exchange. + * @file Runtime state snapshot for one turn. * * State.Snapshot answers "what is this agent, right now": its system prompt, its tool - * set, and the provider options a request is built from. It is captured once per - * exchange and pinned for every turn inside it, so a turn's continuations all - * see the same prompt and the same tools. + * set, and the provider options a request is built from. It is captured before every + * turn and pinned for that turn, so a change (a model switch, a tool declared mid-chain) + * lands on the next request and never inside a turn's tool batch. * * State.Snapshot should read the mount; it never acquires one. `snapshot` therefore declares * `SandboxIO.Provides | Location.Service` in its requirements, which makes @@ -39,7 +39,7 @@ import type { Resolved } from "../tool/registry.ts"; * How a turn's tool calls are scheduled once the array has been re-read. * * Pinned in the snapshot so a configuration change cannot switch scheduling - * halfway through an exchange. State supplies the value; Loop implements it. + * halfway through a turn. State supplies the value; Loop implements it. */ export type ToolExecutionMode = "sequential" | "parallel"; @@ -155,12 +155,12 @@ export interface Interface { * is reported: a long-running server that empties its tool registry because someone typo'd a * settings file is worse than one that keeps working and says so. * - * A reload lands at the next exchange, never inside one, because `snapshot` reads the loaded + * A reload lands at the next turn, never inside one, because `snapshot` reads the loaded * set once at its top. */ readonly reload: Effect.Effect; /** - * The model and thinking level a session's next exchange runs with: its runtime bindings over + * The model and thinking level a session's next turn runs with: its runtime bindings over * its chosen `SessionSchema.Config` over the caller's options over settings, resolved the * way {@link snapshot} resolves them. The level is the one the model will actually run at. */ @@ -171,7 +171,7 @@ export interface Interface { Settings.SettingsError | Runner.ModelCatalogError | Runner.ModelNotFoundError | Runner.ProviderError >; /** - * Capture runtime state for one exchange. Called inside a session drain, + * Capture runtime state for one turn. Called inside a session drain, * where the mount it reads is already open. */ readonly snapshot: ( @@ -199,7 +199,7 @@ const resolver = (input: string | PromptResolver): PromptResolver => (typeof inp * modules already in memory, so it happens at every exchange. That is why editing `options` in a * settings file takes effect at the next turn without anything being re-imported. * - * `snapshot` reads the `Ref` once, at the top, so a reload lands on the next exchange and never + * `snapshot` reads the `Ref` once, at the top, so a reload lands on the next turn and never * mid-turn. Combined with generations, a reload that brings new bytes for a loaded plugin imports * a genuinely different URL, so the module registry cannot hand back the old one. * @@ -331,9 +331,9 @@ export const layer = ( ...("reference" in cause && typeof cause.reference === "string" ? { reference: cause.reference } : {}), ...(sessionId === undefined ? {} : { sessionId }), }); - /** Settings and composed configuration for a session's next exchange. */ + /** Settings and composed configuration for a session's next turn. */ const configure = Effect.fnUntraced(function* (sessionId: SessionId) { - // Read per exchange, not captured at creation, so `Session.link` takes effect at + // Read per turn, not captured at creation, so `Session.link` takes effect at // the next one. A session with none discovers no project layer: there is no // fallback to the process's directory, which would hand it a stranger's project // (it is a long-running server; one process serves sessions in many projects, or @@ -365,7 +365,7 @@ export const layer = ( return Service.of({ reload: Effect.gen(function* () { // Swaps the module set of every view. Which of those a given session runs stays - // that session's own question, answered by its config pass at the next exchange. + // that session's own question, answered by its config pass at the next turn. // A view that fails keeps its previous set; the others still swap. A broken user-layer // entry breaks every view the same way, and is one failure, reported once. const failures = new Map(); diff --git a/packages/harness/test/__artifacts__/runner.hooks.json b/packages/harness/test/__artifacts__/runner.hooks.json new file mode 100644 index 00000000..b1229a17 --- /dev/null +++ b/packages/harness/test/__artifacts__/runner.hooks.json @@ -0,0 +1,40 @@ +{ + "calls": [ + { + "hook": "prepareRequest", + "retry": 0 + }, + { + "hook": "finishTurn", + "reason": "toolUse", + "needsContinuation": true + }, + { + "hook": "prepareNextTurn", + "completed": "toolUse", + "needsContinuation": true, + "tools": [ + "echo", + "echo_late" + ] + }, + { + "hook": "prepareRequest", + "retry": 0 + }, + { + "hook": "finishTurn", + "reason": "stop", + "needsContinuation": false + } + ], + "requestTools": [ + [ + "echo" + ], + [ + "echo", + "echo_late" + ] + ] +} diff --git a/packages/harness/test/fixtures/runner.cycle.spec.ts b/packages/harness/test/fixtures/runner.cycle.spec.ts index 174ec4dc..6cf7b694 100644 --- a/packages/harness/test/fixtures/runner.cycle.spec.ts +++ b/packages/harness/test/fixtures/runner.cycle.spec.ts @@ -18,6 +18,7 @@ import { Database } from "../../src/db/db.ts"; import { Event } from "../../src/event/event.ts"; import { EventRegistry } from "../../src/event/registry.ts"; import { RunnerExecute } from "../../src/runner/execute.ts"; +import { Hooks } from "../../src/runner/hooks.ts"; import { Loop } from "../../src/runner/loop.ts"; import { SandboxController } from "../../src/sandbox/control.ts"; import { SandboxDriverRegistry } from "../../src/sandbox/registry.ts"; @@ -38,7 +39,7 @@ const sandbox = SandboxController.layer({ hostCwd: "/" }).pipe( ); const runtime = (root: string, custom: string) => Control.layer.pipe( - Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(Loop.layer()))), + Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(Loop.layer().pipe(Layer.provide(Hooks.layer))))), Layer.provideMerge( ((seeded) => State.layer({}, seeded.ref, seeded.references, seeded.follow, seeded.rebuild))(pooled(builtins)), ), diff --git a/packages/harness/test/runner.loop.test.ts b/packages/harness/test/runner.loop.test.ts index 2e0fb59f..dbfaee49 100644 --- a/packages/harness/test/runner.loop.test.ts +++ b/packages/harness/test/runner.loop.test.ts @@ -19,6 +19,7 @@ import { EventRegistry } from "../src/event/registry.ts"; import type { EventSchema } from "../src/event/schema.ts"; import { RunnerExecute } from "../src/runner/execute.ts"; import { RunnerExecution } from "../src/runner/execution.ts"; +import { Hooks } from "../src/runner/hooks.ts"; import { LLM } from "../src/runner/llm.ts"; import { Loop } from "../src/runner/loop.ts"; import { SandboxController } from "../src/sandbox/control.ts"; @@ -33,7 +34,7 @@ import { SessionRuntime } from "../src/session/runtime.ts"; import { State } from "../src/state/state.ts"; import { pooled } from "./fixtures/pool.ts"; import * as Tool from "../src/tool/tool.ts"; -import { assistant, immediateOpen } from "./fixtures/llm.ts"; +import { assistant, immediateOpen, toolTurn } from "./fixtures/llm.ts"; import { seedSpace } from "./fixtures/space.ts"; import { testEffect } from "./utils/effect.ts"; @@ -44,6 +45,7 @@ const runtime = ( readonly state?: State.Options; readonly plugins?: ReadonlyArray; readonly settings?: Info; + readonly hooks?: Hooks.Interface; /** Runs as each turn's snapshot loads settings: after the loop checks for pending input, before it promotes. */ readonly beforeSettings?: { run: Effect.Effect }; } = {}, @@ -55,7 +57,17 @@ const runtime = ( Layer.provideMerge(database), ); return Control.layer.pipe( - Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(Loop.layer({ request })))), + Layer.provideMerge( + RunnerExecute.layer.pipe( + Layer.provide( + Loop.layer({ request }).pipe( + Layer.provide( + options.hooks === undefined ? Hooks.layer : Layer.succeed(Hooks.Service, options.hooks), + ), + ), + ), + ), + ), Layer.provideMerge( ((seeded) => State.layer(options.state ?? {}, seeded.ref, seeded.references, seeded.follow, seeded.rebuild))( pooled(options.plugins ?? plugins), @@ -81,6 +93,14 @@ const runtime = ( ); }; +/** Hooks that do nothing, except where overridden. */ +const hooksOf = (overrides: Partial): Hooks.Interface => ({ + prepareNextTurn: (_turn, next) => Effect.succeed(next), + prepareRequest: (request) => Effect.succeed(request), + finishTurn: () => Effect.undefined, + ...overrides, +}); + const inputsFor = (sql: SqlClient.SqlClient) => SqlSchema.findAll({ Request: Schema.String, @@ -298,7 +318,7 @@ describe("runner loop — tool continuation and lifecycle gate", () => { ); it( - "continues after toolUse with no steer, pins state, and never moves the assistant seq", + "continues after toolUse with no steer, re-snapshots each turn, and never moves the assistant seq", Effect.gen(function* () { contexts.length = 0; snapshots = 0; @@ -312,7 +332,7 @@ describe("runner loop — tool continuation and lifecycle gate", () => { yield* execution.resume(sessionId); expect(contexts).toHaveLength(2); - expect(snapshots).toBe(1); + expect(snapshots).toBe(2); const firstTool = contexts[1]!.messages[1]!.parts.find((part) => part.type === "toolCall"); expect(firstTool).toMatchObject({ callID: "call_echo", status: "completed" }); @@ -635,49 +655,51 @@ describe("runner loop — provider failure", () => { ); }); -describe("runner loop — retry", () => { - type Failure = { readonly _tag: string; readonly retryable: boolean; readonly retryAfterMs?: number }; - - /** Each call takes the next script entry: a failure (optionally after streaming text), else success. */ - const scripted = ( - script: ReadonlyArray<{ readonly failure: Failure; readonly streamed?: string } | "ok">, - contexts: Message.Context[] = [], - ): LLM.Open => { - let index = 0; - return (input) => - Effect.sync(() => { - index += 1; - contexts.push(structuredClone(input.context)); - const step = script[index - 1] ?? "ok"; - const events = createAssistantMessageEventStream(); - if (step === "ok") { - const message = assistant(input, index); - events.push({ type: "start", partial: message }); - events.push({ type: "done", reason: "stop", message }); - return events; - } - const failed = Object.assign( - assistant(input, index, { - stopReason: "error", - errorMessage: `${step.failure._tag} failure`, - parts: step.streamed === undefined ? [] : [{ type: "text", text: step.streamed }], - }), - { failure: { ...step.failure, message: `${step.failure._tag} failure` } }, - ); - events.push({ type: "start", partial: failed }); - if (step.streamed !== undefined) { - events.push({ type: "text.delta", partIndex: 0, delta: step.streamed, partial: failed }); - } - events.push({ type: "error", reason: "error", error: failed }); +type Failure = { readonly _tag: string; readonly retryable: boolean; readonly retryAfterMs?: number }; + +/** Each call takes the next script entry: a failure (optionally after streaming text), else success. */ +const scripted = ( + script: ReadonlyArray<{ readonly failure: Failure; readonly streamed?: string } | "ok">, + contexts: Message.Context[] = [], + models: string[] = [], +): LLM.Open => { + let index = 0; + return (input) => + Effect.sync(() => { + index += 1; + contexts.push(structuredClone(input.context)); + models.push(input.model); + const step = script[index - 1] ?? "ok"; + const events = createAssistantMessageEventStream(); + if (step === "ok") { + const message = assistant(input, index); + events.push({ type: "start", partial: message }); + events.push({ type: "done", reason: "stop", message }); return events; - }); - }; + } + const failed = Object.assign( + assistant(input, index, { + stopReason: "error", + errorMessage: `${step.failure._tag} failure`, + parts: step.streamed === undefined ? [] : [{ type: "text", text: step.streamed }], + }), + { failure: { ...step.failure, message: `${step.failure._tag} failure` } }, + ); + events.push({ type: "start", partial: failed }); + if (step.streamed !== undefined) { + events.push({ type: "text.delta", partIndex: 0, delta: step.streamed, partial: failed }); + } + events.push({ type: "error", reason: "error", error: failed }); + return events; + }); +}; - const settings = (retry: Partial): Info => ({ - ...Settings.defaults, - retry: { ...Settings.defaults.retry, ...retry }, - }); +const settings = (retry: Partial): Info => ({ + ...Settings.defaults, + retry: { ...Settings.defaults.retry, ...retry }, +}); +describe("runner loop — retry", () => { /** The turn's lifecycle as a client sees it, without timestamps or ids. */ const journal = Effect.fnUntraced(function* () { const events = yield* Event.Service; @@ -1327,3 +1349,295 @@ describe("runner loop — cancel and unfinished replies", () => { }), ); }); + +describe("runner loop — lifecycle hooks", () => { + const echo = Tool.register( + Tool.make({ + name: "echo", + description: "echoes text", + parameters: Schema.Struct({ value: Schema.String }), + success: Schema.String, + encodeContent: (value: string) => [{ type: "text", text: value }], + handler: ({ value }) => Effect.succeed(value), + }), + ); + const late = Tool.register( + Tool.make({ + name: "echo_late", + description: "declared from the second snapshot on", + parameters: Schema.Struct({ value: Schema.String }), + success: Schema.String, + encodeContent: (value: string) => [{ type: "text", text: value }], + handler: ({ value }) => Effect.succeed(value), + }), + ); + const echoCall = { + type: "toolCall" as const, + callID: "call_echo", + name: "echo", + arguments: { value: "hello" }, + status: "pending" as const, + time: { start: 1, end: 1 }, + }; + /** A fresh tool turn per test, since it counts its own requests. */ + const recording = (contexts: Message.Context[]): LLM.Open => { + const next = toolTurn(echoCall); + return (input, signal) => { + contexts.push(input.context); + return next(input, signal); + }; + }; + const toolNames = (context: Message.Context) => + (context.tools ?? []).map((tool) => tool.name).filter((name) => name.startsWith("echo")); + + describe("a tool-use chain", () => { + const contexts: Message.Context[] = []; + const calls: Array> = []; + let snapshots = 0; + const hooks: Hooks.Interface = { + prepareNextTurn: (turn, next) => + Effect.sync(() => { + calls.push({ + hook: "prepareNextTurn", + completed: turn.reason, + needsContinuation: turn.needsContinuation, + tools: next.tools.defs.map((def) => def.name).filter((name) => name.startsWith("echo")), + }); + return next; + }), + prepareRequest: (request, attempt) => + Effect.sync(() => { + calls.push({ hook: "prepareRequest", retry: attempt.retry }); + return request; + }), + finishTurn: (turn) => + Effect.sync(() => { + calls.push({ hook: "finishTurn", reason: turn.reason, needsContinuation: turn.needsContinuation }); + return undefined; + }), + }; + const { effect: it } = testEffect( + runtime({ + open: recording(contexts), + hooks, + plugins: [ + ...plugins, + { + id: "test.tool.echo", + kind: "tool", + setup: (ctx) => { + snapshots += 1; + ctx.plugin.tools.add(echo); + // A tool the plugin only declares from the second snapshot on. + if (snapshots > 1) ctx.plugin.tools.add(late); + }, + }, + ], + }), + ); + + it( + "snapshots before each turn, so a tool declared mid-chain reaches the next request", + Effect.gen(function* () { + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_chain", sessionId, delivery: "steer" }); + + yield* execution.resume(sessionId); + + expect(snapshots).toBe(2); + expect(contexts.map(toolNames)).toEqual([["echo"], ["echo", "echo_late"]]); + yield* Effect.promise(() => + expect( + `${JSON.stringify({ calls, requestTools: contexts.map(toolNames) }, null, "\t")}\n`, + ).toMatchFileSnapshot("./__artifacts__/runner.hooks.json"), + ); + }), + ); + }); + + describe("pending input", () => { + const seen: Array> = []; + const { effect: it } = testEffect( + runtime({ + open: recording([]), + hooks: hooksOf({ + finishTurn: (turn) => + Effect.sync(() => { + seen.push(turn.pending.map((item) => `${item.delivery}:${item.id}`)); + return undefined; + }), + }), + plugins: [...plugins, { id: "test.tool.echo", kind: "tool", setup: (ctx) => ctx.plugin.tools.add(echo) }], + }), + ); + + it( + "shows the queued steers, else the oldest follow-up, without promoting them", + Effect.gen(function* () { + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_first", sessionId, delivery: "steer" }); + yield* admit({ id: "msg_later", sessionId, delivery: "followUp" }); + yield* admit({ id: "msg_next", sessionId, delivery: "followUp" }); + + yield* execution.resume(sessionId); + + // 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"], []]); + }), + ); + }); + + describe("finishTurn end", () => { + const contexts: Message.Context[] = []; + const { effect: it } = testEffect( + runtime({ + open: recording(contexts), + hooks: hooksOf({ finishTurn: (turn) => Effect.succeed(turn.reason === "toolUse" ? "end" : undefined) }), + plugins: [...plugins, { id: "test.tool.echo", kind: "tool", setup: (ctx) => ctx.plugin.tools.add(echo) }], + }), + ); + + it( + "stops the run after the committed turn and leaves queued input for the next run", + Effect.gen(function* () { + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_first", sessionId, delivery: "steer" }); + yield* admit({ id: "msg_queued", sessionId, delivery: "followUp" }); + + yield* execution.resume(sessionId); + + expect(contexts).toHaveLength(1); + const sessions = yield* Session.Service; + const path = yield* sessions.path(sessionId); + expect(path.map((item) => [item.entry.type, item.entry.state])).toEqual([ + ["user", "committed"], + ["assistant", "committed"], + ]); + const tool = path[1]!.parts.find((part) => part.type === "toolCall"); + expect(tool).toBeDefined(); + const rows = yield* delivered(sessionId); + 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(sessionId); + expect(contexts).toHaveLength(3); + expect(contexts[1]!.messages).toHaveLength(2); + expect(contexts[2]!.messages).toHaveLength(4); + const next = yield* delivered(sessionId); + expect(next.find((row) => row.id === "msg_queued")?.promotedSeq).not.toBeNull(); + }), + ); + }); + + describe("a steer landing while a resume loads", () => { + const contexts: Message.Context[] = []; + const gate: { run: Effect.Effect } = { run: Effect.void }; + const { effect: it } = testEffect( + runtime({ + open: recording(contexts), + beforeSettings: gate, + hooks: hooksOf({ finishTurn: (turn) => Effect.succeed(turn.reason === "toolUse" ? "end" : undefined) }), + plugins: [...plugins, { id: "test.tool.echo", kind: "tool", setup: (ctx) => ctx.plugin.tools.add(echo) }], + }), + ); + + it( + "joins the reply being answered, ahead of the queued follow-up", + Effect.gen(function* () { + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_first", sessionId, delivery: "steer" }); + yield* admit({ id: "msg_queued", sessionId, delivery: "followUp" }); + yield* execution.resume(sessionId); + + const services = yield* Effect.context>>(); + gate.run = admit({ id: "msg_late", sessionId, delivery: "steer" }).pipe( + Effect.andThen( + Effect.sync(() => { + gate.run = Effect.void; + }), + ), + Effect.provideContext(services), + ); + yield* execution.resume(sessionId); + + expect(contexts.map(texts)).toEqual([ + ["msg_first"], + ["msg_first", "msg_late"], + ["msg_first", "msg_late", "msg_queued"], + ]); + }), + ); + }); + + describe("finishTurn continue", () => { + const contexts: Message.Context[] = []; + const completed: string[] = []; + const { effect: it } = testEffect( + runtime({ + contexts, + hooks: hooksOf({ + finishTurn: () => Effect.succeed("continue"), + prepareNextTurn: (turn, next) => Effect.sync(() => (completed.push(turn.reason), next)), + }), + }), + ); + + it( + "sends nothing after a finished reply when nothing was appended", + Effect.gen(function* () { + contexts.length = 0; + completed.length = 0; + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_done", sessionId, delivery: "steer" }); + + yield* execution.resume(sessionId); + + // The decision was honoured (another turn was scheduled) and the leaf guard sent nothing. + expect(completed).toEqual(["stop"]); + expect(contexts).toHaveLength(1); + }), + ); + }); + + describe("prepareRequest on a retry", () => { + const models: string[] = []; + const attempts: Array<{ readonly retry: number; readonly failed?: string }> = []; + const open = scripted([{ failure: { _tag: "Unavailable", retryable: true } }], [], models); + const { live: it } = testEffect( + runtime({ + open, + settings: settings({ baseDelayMs: 5, maxDelayMs: 8 }), + hooks: hooksOf({ + prepareRequest: (request, attempt) => + Effect.sync(() => { + attempts.push(attempt); + return attempt.retry > 0 ? { ...request, model: "routed-model" } : request; + }), + }), + }), + ); + + it( + "sees the retry number and failure, and its replacement model reaches the request", + Effect.gen(function* () { + const execution = yield* RunnerExecution.Service; + const sessionId = yield* seedSession(); + yield* admit({ id: "msg_route", sessionId, delivery: "steer" }); + + yield* execution.resume(sessionId); + + expect(attempts).toMatchObject([ + { retry: 0 }, + { retry: 1, failed: expect.stringContaining("Unavailable failure") }, + ]); + expect(models[1]).toBe("routed-model"); + expect(models[0]).not.toBe("routed-model"); + }), + ); + }); +}); diff --git a/packages/harness/test/settings.loop.test.ts b/packages/harness/test/settings.loop.test.ts index e590c65c..d40b80cf 100644 --- a/packages/harness/test/settings.loop.test.ts +++ b/packages/harness/test/settings.loop.test.ts @@ -48,8 +48,8 @@ const terminal = (input: LLM.Input, index: number, tool = false) => { return stream; }; -describe("settings at exchange boundaries", () => { - it("pins settings through a blocked request, tool continuation and steer; refreshes follow-up and idle prompts", () => +describe("settings at turn boundaries", () => { + it("pins settings through a blocked request, and applies an edit made during a tool batch at the next turn", () => withSettings(async ({ root, custom, local }) => { await file(custom, "low"); // This belongs to the session directory, not host startup, and must not be loaded. @@ -96,8 +96,10 @@ describe("settings at exchange boundaries", () => { yield* handle.prompt({ text: "later", delivery: "followUp" }); yield* Deferred.succeed(toolRelease, undefined); yield* handle.wait(); - expect(inputs.map((input) => input.thinkingLevel)).toEqual(["low", "low", "high"]); - expect(inputs.map((input) => input.options?.headers?.revision)).toEqual(["low", "low", "high"]); + // The "medium" edit landed while request 1 was blocked; the request already built stays "low". + // The "high" edit landed during the tool batch, so the continuation turn reads it. + expect(inputs.map((input) => input.thinkingLevel)).toEqual(["low", "high", "high"]); + expect(inputs.map((input) => input.options?.headers?.revision)).toEqual(["low", "high", "high"]); expect(inputs[1]?.context.systemPrompt).toBe(inputs[0]?.context.systemPrompt); yield* Effect.promise(() => file(custom, "off")); yield* handle.run("idle steer"); @@ -127,7 +129,7 @@ describe("settings at exchange boundaries", () => { ); })); - it("pins matching parallel tool execution through continuations and refreshes it next exchange", () => + it("pins parallel tool execution for the running batch and refreshes it at the next turn", () => withSettings(async ({ root, custom }) => { const configure = (toolExecution: string) => writeFile( @@ -139,7 +141,7 @@ describe("settings at exchange boundaries", () => { await configure("parallel"); await Effect.runPromise( Effect.gen(function* () { - const gates = [yield* Deferred.make(), yield* Deferred.make()]; + const gates = [yield* Deferred.make()]; const starts = [0, 0, 0]; const trace: string[] = []; let request = 0; @@ -157,7 +159,8 @@ describe("settings at exchange boundaries", () => { if (batch === 0 && starts[batch] === 1) yield* Effect.promise(() => configure("sequential")); const gate = gates[batch]; if (gate !== undefined) { - // A sequential scheduler deadlocks here: both calls must enter before either finishes. + // Only the batch that started parallel has a gate: a sequential scheduler would deadlock + // here, since both calls must enter before either finishes. if (starts[batch] === 2) yield* Deferred.succeed(gate, undefined); yield* Deferred.await(gate); } @@ -193,12 +196,16 @@ describe("settings at exchange boundaries", () => { yield* handle.run("parallel, including continuation"); yield* handle.run("fresh sequential"); expect(starts).toEqual([2, 2, 2]); - expect(trace.filter((item) => item.startsWith("2:"))).toEqual([ - "2:start", - "2:end", - "2:start", - "2:end", - ]); + // Batch 0 ran parallel through the edit made inside it; the next turn and the next + // exchange read the edited file and run sequentially. + for (const batch of [1, 2]) { + expect(trace.filter((item) => item.startsWith(`${batch}:`))).toEqual([ + `${batch}:start`, + `${batch}:end`, + `${batch}:start`, + `${batch}:end`, + ]); + } }).pipe( Effect.provide( Harness.layer({ @@ -322,7 +329,7 @@ describe("settings at exchange boundaries", () => { expect(failures[0]?.data.error.message).toContain("model.options.timeoutMs"); })); - it("applies a session binding change at the next capture, not inside the running exchange", () => + it("applies a session binding change at the next turn, not inside the running one", () => withSettings(async ({ root, custom }) => { await file(custom, "low"); const inputs: LLM.Input[] = []; @@ -353,13 +360,13 @@ describe("settings at exchange boundaries", () => { const handle = yield* Session.create({ directory: root }); yield* handle.prompt("first"); yield* Deferred.await(toolEntered); - // A binding set mid-exchange must not disturb the snapshot already pinned. + // A binding set during a tool batch must not disturb the turn already running. yield* Session.attach({ sessionId: handle.id, thinkingLevel: "max" }); yield* Deferred.succeed(toolRelease, undefined); yield* handle.wait(); - expect(inputs.map((input) => input.thinkingLevel)).toEqual(["low", "low"]); + expect(inputs.map((input) => input.thinkingLevel)).toEqual(["low", "max"]); yield* handle.run("second"); - // The next capture reads it, and a binding outranks the file. + // A binding outranks the file. expect(inputs.at(-1)?.thinkingLevel).toBe("max"); }).pipe( Effect.provide(