feat(harness): add loop lifecycle seams and snapshot every turn

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`. */

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,

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";

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))),

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.
  */

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<PluginRef>, pool: Pool) {
 	const operations = new Map<string, { enabled: boolean; index: number; options: PluginOptions }>();

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<Admitted>;
+}
+
+/** One provider request, as the loop built it. */
+export type Request = Omit<LLM.Input, "sessionId">;
+
+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<State.Snapshot>;
+	/** Before every provider attempt, retries included; may replace the request. */
+	readonly prepareRequest: (request: Request, attempt: Attempt) => Effect.Effect<Request>;
+	/**
+	 * 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<Decision | undefined>;
+}
+
+export class Service extends Context.Service<Service, Interface>()("@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";

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) =>

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.

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<Reloaded>;
 	/**
-	 * 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<string, { readonly message: string }>();