From f18d523d4efd3d8cd98dcf7555211deae674dd88 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 15 Aug 2026 23:29:36 +0200 Subject: [PATCH] feat(agent): add runtime2 fault propagation --- .../agent/src/harness/runtime2/harness.ts | 22 +++++- packages/agent/src/harness/runtime2/lane.ts | 23 ++++-- .../test/harness/runtime2/harness.test.ts | 73 ++++++++++++++++++- .../agent/test/harness/runtime2/lane.test.ts | 4 +- 4 files changed, 110 insertions(+), 12 deletions(-) diff --git a/packages/agent/src/harness/runtime2/harness.ts b/packages/agent/src/harness/runtime2/harness.ts index 677a77d08..2726fadf5 100644 --- a/packages/agent/src/harness/runtime2/harness.ts +++ b/packages/agent/src/harness/runtime2/harness.ts @@ -17,11 +17,12 @@ export class Harness extends Lane implement readonly seed: LaneConfiguration; readonly lanesByName = new Map(); closePromise: Promise | undefined; + faultError: HarnessFault | undefined; constructor(options: AgentHarnessOptions, seed: LaneConfiguration, restored: Map) { const main = restored.get("main"); if (main === undefined) throw new SessionInvariantError("Session is missing main lane"); - super("main", options.session, options.models, main); + super("main", options.session, options.models, main, (cause) => this.fault(cause)); this.session = options.session; this.seed = seed; this.events = new HarnessEventBus(); @@ -37,7 +38,12 @@ export class Harness extends Lane implement ); this.lanesByName.set("main", this); for (const [name, state] of restored) { - if (name !== "main") this.lanesByName.set(name, new Lane(name, options.session, options.models, state)); + if (name !== "main") { + this.lanesByName.set( + name, + new Lane(name, options.session, options.models, state, (cause) => this.fault(cause)), + ); + } } } @@ -129,6 +135,18 @@ export class Harness extends Lane implement throw new RuntimeSliceNotImplemented("watchSession"); } + fault(cause: unknown): Error { + if (this.faultError !== undefined) return this.faultError; + if (this.closedError !== undefined) return this.closedError; + const normalized = cause instanceof Error ? cause : new Error(String(cause)); + const fault = new HarnessFault("AgentHarness storage or invariant fault", normalized); + this.faultError = fault; + for (const lane of this.lanesByName.values()) lane.seal(fault); + this.hooks.close(fault); + this.events.close(fault); + return fault; + } + close(): Promise { if (this.closePromise !== undefined) return this.closePromise; /* diff --git a/packages/agent/src/harness/runtime2/lane.ts b/packages/agent/src/harness/runtime2/lane.ts index 409fce6e1..35ce4225b 100644 --- a/packages/agent/src/harness/runtime2/lane.ts +++ b/packages/agent/src/harness/runtime2/lane.ts @@ -5,6 +5,8 @@ import { RuntimeSliceNotImplemented } from "../runtime/types.ts"; import type { Session, SessionTree, Transaction } from "../session/types.ts"; import type { LaneState } from "./types.ts"; +type FaultHandler = (cause: unknown) => Error; + interface LaneTransition { transaction: Transaction; next: LaneState; @@ -17,15 +19,17 @@ export class Lane implements AgentLane { readonly sessionTree: SessionTree; readonly session: Session; readonly models: Models; + readonly onFault: FaultHandler; state: LaneState; closedError: Error | undefined; - constructor(name: string, session: Session, models: Models, state: LaneState) { + constructor(name: string, session: Session, models: Models, state: LaneState, onFault: FaultHandler) { this.session = session; this.models = models; this.name = name; this.sessionTree = session.view(name); this.state = state; + this.onFault = onFault; } async getLeafId(): Promise { @@ -40,12 +44,17 @@ export class Lane implements AgentLane { async transition(plan: (state: LaneState) => LaneTransition): Promise { this.assertOpen(); - return this.session.mutate(this.name, async (mutator) => { - const transition = plan(this.state); - await mutator.commit(transition.transaction); - this.state = transition.next; - return transition.result; - }); + try { + return await this.session.mutate(this.name, async (mutator) => { + const transition = plan(this.state); + await mutator.commit(transition.transaction); + this.state = transition.next; + return transition.result; + }); + } catch (error) { + if (this.closedError !== undefined) throw this.closedError; + throw this.onFault(error); + } } async accept(): Promise { diff --git a/packages/agent/test/harness/runtime2/harness.test.ts b/packages/agent/test/harness/runtime2/harness.test.ts index 28edc7e82..61f39dab0 100644 --- a/packages/agent/test/harness/runtime2/harness.test.ts +++ b/packages/agent/test/harness/runtime2/harness.test.ts @@ -5,10 +5,26 @@ import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/com import { RuntimeSliceNotImplemented } from "../../../src/harness/runtime/types.ts"; import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; import { Lane } from "../../../src/harness/runtime2/lane.ts"; -import { MemorySessionRepo } from "../../../src/harness/session/memory.ts"; -import type { LaneConfiguration, OperationMeta, RunState, Session } from "../../../src/harness/session/types.ts"; +import { MemorySessionRepo, MemoryStorage } from "../../../src/harness/session/memory.ts"; +import { StorageBackedSession } from "../../../src/harness/session/session.ts"; +import type { + LaneConfiguration, + OperationMeta, + RunState, + Session, + Transaction, +} from "../../../src/harness/session/types.ts"; + +class FailingMemoryStorage extends MemoryStorage { + failure: Error | undefined; + + override commit(transaction: Transaction) { + return this.failure === undefined ? super.commit(transaction) : Promise.reject(this.failure); + } +} const repos: MemorySessionRepo[] = []; +const standaloneSessions: Session[] = []; const configuredMain: LaneConfiguration = { model: { provider: "configured", modelId: "main" }, thinkingLevel: "low", @@ -21,6 +37,30 @@ async function createSession(): Promise { return repo.create({}); } +async function createFailingSession(): Promise<{ session: Session; storage: FailingMemoryStorage }> { + const storage = new FailingMemoryStorage(); + const session = new StorageBackedSession( + { id: `runtime2-fault-${standaloneSessions.length}`, createdAt: 1, storageVersion: 1 }, + storage, + ); + standaloneSessions.push(session); + await session.mutate("main", (mutator) => + mutator.commit({ + writes: [ + { kind: "register", op: "set", namespace: "lane.leaf", key: "main", value: null }, + { + kind: "register", + op: "set", + namespace: "lane.state", + key: "main", + value: { currentOperationId: null, pendingNextRun: [] }, + }, + ], + }), + ); + return { session, storage }; +} + function modelOptions(session: Session) { const faux = fauxProvider(); const models = createModels(); @@ -50,6 +90,7 @@ function runState(triggerEntryId: string): RunState { afterEach(async () => { for (const repo of repos.splice(0)) await repo.close(); + for (const session of standaloneSessions.splice(0)) await session.close(); }); describe("runtime2 AgentHarness", () => { @@ -177,6 +218,34 @@ describe("runtime2 AgentHarness", () => { await expect(harness.getTools()).rejects.toBeInstanceOf(RuntimeSliceNotImplemented); }); + it("faults every lane when an admitted commit fails", async () => { + const { session, storage } = await createFailingSession(); + await session.createLane("worker", null, configuredMain); + const { harness } = await createAgentHarness(modelOptions(session)); + if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); + const worker = await harness.lane("worker"); + if (!(worker instanceof Lane)) throw new Error("missing runtime2 worker lane"); + const failure = new Error("commit failed"); + storage.failure = failure; + + let rejected: unknown; + try { + await worker.setThinkingLevel("high"); + } catch (error) { + rejected = error; + } + + expect(rejected).toBeInstanceOf(HarnessFault); + if (!(rejected instanceof HarnessFault)) throw new Error("missing harness fault"); + expect(rejected.cause).toBe(failure); + await expect(harness.getLeafId()).rejects.toBe(rejected); + await expect(worker.getThinkingLevel()).rejects.toBe(rejected); + expect(harness.fault(new Error("later"))).toBe(rejected); + expect(worker.state.configuration).toEqual(configuredMain); + expect((await session.getRegister("lane.config", "worker"))?.value).toEqual(configuredMain); + await harness.close(); + }); + it("wraps initialization invariant failures as harness faults", async () => { const session = await createSession(); await session.mutate("main", (mutator) => diff --git a/packages/agent/test/harness/runtime2/lane.test.ts b/packages/agent/test/harness/runtime2/lane.test.ts index a57a5996e..ff49b135c 100644 --- a/packages/agent/test/harness/runtime2/lane.test.ts +++ b/packages/agent/test/harness/runtime2/lane.test.ts @@ -64,7 +64,9 @@ async function createLane(): Promise<{ const models = createModels(); models.setProvider(faux.provider); return { - lane: new Lane("main", session, models, await restoreLane(session, "main")), + lane: new Lane("main", session, models, await restoreLane(session, "main"), (cause) => + cause instanceof Error ? cause : new Error(String(cause)), + ), model: faux.getModel(), session, storage,