mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(agent): add runtime2 fault propagation
This commit is contained in:
parent
980edc9d2b
commit
f18d523d4e
4 changed files with 110 additions and 12 deletions
|
|
@ -17,11 +17,12 @@ export class Harness<TContext extends object | undefined> extends Lane implement
|
|||
readonly seed: LaneConfiguration;
|
||||
readonly lanesByName = new Map<string, Lane>();
|
||||
closePromise: Promise<void> | undefined;
|
||||
faultError: HarnessFault | undefined;
|
||||
|
||||
constructor(options: AgentHarnessOptions<TContext>, seed: LaneConfiguration, restored: Map<string, LaneState>) {
|
||||
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<TContext extends object | undefined> 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<TContext extends object | undefined> 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<void> {
|
||||
if (this.closePromise !== undefined) return this.closePromise;
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -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<TResult> {
|
||||
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<string | null> {
|
||||
|
|
@ -40,12 +44,17 @@ export class Lane implements AgentLane {
|
|||
|
||||
async transition<TResult>(plan: (state: LaneState) => LaneTransition<TResult>): Promise<TResult> {
|
||||
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<never> {
|
||||
|
|
|
|||
|
|
@ -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<Session> {
|
|||
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) =>
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue