sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(agent): fuse structural boundary routing

Browse source
This commit is contained in:
Mario Zechner 2026-08-26 16:46:39 +02:00
parent c60dd63fdf
commit c5a35eabbf
3 changed files with 253 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/agent/docs/work-packages/05-direct-durable-drive.md
View file

@ -1,6 +1,6 @@
# WP05 — Direct durable drive
**Status: M5 complete; M6 structural foundation committed; R1a lane-owned inbox relocation, R2 record-only outcomes, and reviewed R3 family-neutral leaves implemented; R1b next, then M7 (reconciliation), M8 (public surfaces), M9 (documentation reconciliation), and M10 (provider KV-cache identity review).**
**Status: M5 complete; M6 structural foundation committed; R1a lane-owned inbox relocation, R2 record-only outcomes, reviewed R3 family-neutral leaves, and R1b-1 structural boundary fusion implemented; R1b-2 next, then M7 (reconciliation), M8 (public surfaces), M9 (documentation reconciliation), and M10 (provider KV-cache identity review).**
WP06's Session/Branch/Lane separation is already part of the foundation. Public drive remains disabled until R1–R3, cancellation reconciliation, the total drive switch, every public execution surface, and the provider KV-cache identity review are complete.
@ -244,7 +244,7 @@ Crash between TX4..n and TX-pub → structural recovery (attempt unknown, per ex
Slices, in order (the order is load-bearing — deleting the marker while decline still re-entered `checkpoint` would live-loop):
1. **R1b-1**: extract the shared planner; fuse it into the structural `resume_checkpoint` boundary (`threshold: "skip"`); structural success/decline stop committing back into `checkpoint` except the success/`may_finish` resting case. Both patch fields still written and checked by the old checkpoint pass — harmless.
1. **R1b-1**: fuse the future shared planner into the structural `resume_checkpoint` boundary (`threshold: "skip"`); structural success/decline route directly except for a `may_finish` resting checkpoint, which remains safe for either outcome while the durable marker still exists. Keep the route inline until R1b-2 creates its second caller. Both patch fields are still written and checked by the old checkpoint pass — harmless.
2. **R1b-2**: convert the `checkpoint` pass to single-commit exits with the recency guard, implement the three-phase finish arm, delete `skipInboxOnce` and `thresholdCheckedTriggerEntryId`.
### Rules

159
packages/agent/src/harness/runtime/drive/structural.ts
View file

@ -17,6 +17,7 @@ import { insertEntry, insertUsage } from "../../session/commit.ts";
import { SessionInvariantError } from "../../session/session.ts";
import {
type AssistantEffectPendingOperation,
type AssistantReadyOperation,
type BranchSummaryEntry,
type CheckpointOperation,
type CommitResult,
@ -40,10 +41,17 @@ import {
type UsageRow,
type Write,
} from "../../session/types.ts";
import { branchTip, entryLabel, operationPreparation, setValue } from "../../session/values.ts";
import {
branchTip,
deleteValue,
entryLabel,
operationPreparation,
pendingEntry,
setValue,
} from "../../session/values.ts";
import type { AgentHarnessStreamOptions } from "../../types.ts";
import type { Lane } from "../lane.ts";
import { committedEntryEvents, readBoundedEntries } from "../transcript.ts";
import { committedEntryEvents, readBoundedEntries, readPendingMessages } from "../transcript.ts";
import type { ContinueOperationResult, Drive, ProcedureResult } from "../types.ts";
import { retryDelay, retryNotBefore, waitUntil } from "./retry.ts";
import { operationCleanupWrites, operationResultRecord } from "./terminal.ts";
@ -296,38 +304,129 @@ async function publishStructuralOutcome<TContext extends object | undefined>(
if (outcome.kind === "branch_summary") {
throw new SessionInvariantError("Run compaction boundary received a branch summary");
}
if (outcome.kind === "compaction") {
if (
outcome.kind === "compaction" ||
(outcome.kind === "declined" && current.task.reason === "threshold")
) {
if (terminalTipId === null) throw new SessionInvariantError("Run compaction has no Branch tip");
const steer = state.inbox.filter((item) => item.kind === "steer");
const selectedSteer = new Set(
(current.settings.steeringMode === "all" ? steer : steer.slice(0, 1)).map((item) => item.entryId),
);
const selected = state.inbox.filter(
(item) => item.kind === "write" || (item.kind === "steer" && selectedSteer.has(item.entryId)),
);
const pending = await Promise.all(
selected.map(async (item) => {
const stored = await reader.getValue(pendingEntry(item.entryId), drive.context);
if (stored === undefined) {
throw new SessionInvariantError(
`Pending ${item.kind} entry ${item.entryId} is missing its payload`,
);
}
if (item.kind === "steer" && stored.value.type !== "message") {
throw new SessionInvariantError(`Queued steer entry ${item.entryId} is not a message`);
}
return { item, pending: stored.value };
}),
);
let parentId = terminalTipId;
let triggerEntryId: string | undefined;
const entries: NewEntry[] = pending.map(({ item, pending: value }) => {
const entry: NewEntry =
value.type === "message"
? { id: item.entryId, parentId, type: "message", message: value.payload }
: {
id: item.entryId,
parentId,
type: "custom",
customType: value.customType,
...(value.payload === undefined ? {} : { data: value.payload }),
};
parentId = item.entryId;
if (
value.type === "message" ||
lane.readConfig().entryProjectors[value.customType] !== undefined
) {
triggerEntryId = item.entryId;
}
return entry;
});
const selectedIds = new Set(selected.map((item) => item.entryId));
const inbox = state.inbox.filter((item) => !selectedIds.has(item.entryId));
const queueUpdate =
selectedSteer.size === 0
? undefined
: await Promise.all(
(["steer", "followUp", "nextRun"] as const).map((kind) =>
readPendingMessages(
reader,
inbox.filter((item) => item.kind === kind).map((item) => item.entryId),
`Pending ${kind} entry`,
drive.context,
),
),
);
const entryWriteIndex = writes.length;
writes.push(
...entries.map((entry) => insertEntry(entry)),
...selected.map((item) => deleteValue(pendingEntry(item.entryId))),
...(entries.length === 0 ? [] : [setValue(branchTip(lane.name), parentId)]),
);
const continuation = current.task.boundary.resumeAfter.continuation;
const operationState: AssistantReadyOperation | CheckpointOperation =
triggerEntryId !== undefined || continuation.kind === "need_assistant"
? {
...operationScopeOf(current),
at: "assistant.ready",
generationContext: {
stepId: lane.session.idGenerator.next(),
triggerEntryId: triggerEntryId ?? current.task.boundary.resumeAfter.triggerEntryId,
configuration: state.configuration,
streamOptions: lane.readConfig().streamOptions,
retryPolicy: normalizedRetryPolicy(lane),
overflowRecoveryUsed:
triggerEntryId === undefined && continuation.kind === "need_assistant"
? continuation.overflowRecoveryUsed
: false,
},
nextAttempt: 1,
}
: {
...operationScopeOf(current),
at: "checkpoint",
...current.task.boundary.resumeAfter,
};
return {
kind: "commit",
writes,
operationState: {
...operationScopeOf(current),
at: "checkpoint",
...current.task.boundary.resumeAfter,
},
lane: { tipId: terminalTipId },
operationState,
lane: { tipId: parentId, inbox },
materialize: () => ({ kind: "continue" }) as const,
events,
};
}
if (outcome.kind === "declined" && current.task.reason === "threshold") {
return {
kind: "commit",
writes: [],
operationState: {
...operationScopeOf(current),
at: "checkpoint",
...current.task.boundary.resumeAfter,
},
materialize: () => ({ kind: "continue" }) as const,
events: () => [
{
type: "compaction_end",
lane: lane.name,
runId: drive.operationId,
reason: "threshold",
status: "declined",
},
events: (commit) => [
...(outcome.kind === "compaction"
? events(commit)
: [
{
type: "compaction_end" as const,
lane: lane.name,
runId: drive.operationId,
reason: "threshold" as const,
status: "declined" as const,
},
]),
...committedEntryEvents(entries, commit, lane.name, drive.operationId, entryWriteIndex),
...(queueUpdate === undefined
? []
: [
{
type: "queue_update" as const,
lane: lane.name,
steer: queueUpdate[0],
followUp: queueUpdate[1],
nextRun: queueUpdate[2],
},
]),
],
};
}

144
packages/agent/test/harness/runtime/drive-structural.test.ts
View file

@ -291,7 +291,7 @@ afterEach(async () => {
});
describe("runtime structural drive", () => {
it("marks a threshold once and restores that checkpoint when the hook declines", async () => {
it("routes a declined threshold directly to assistant generation", async () => {
const fixture = await createFixture();
const model = fixture.faux.getModel();
const settings = {
@ -331,9 +331,12 @@ describe("runtime structural drive", () => {
fixture.hooks.on("before_compaction", () => ({ decline: true }));
expect(await runStructuralDecision(fixture.lane, fixture.drive, deciding)).toEqual({ kind: "continue" });
const restored = currentState(fixture);
if (restored.at !== "checkpoint") throw new Error("threshold decline did not restore checkpoint");
expect(restored.thresholdCheckedTriggerEntryId).toBe(restored.triggerEntryId);
const routed = currentState(fixture);
if (routed.at !== "assistant.ready") throw new Error("threshold decline did not route to generation");
expect(routed.generationContext).toMatchObject({
triggerEntryId: "assistant",
overflowRecoveryUsed: false,
});
expect(fixture.events.filter((event) => event.type === "compaction_start")).toHaveLength(1);
expect(fixture.events.filter((event) => event.type === "compaction_end")).toHaveLength(1);
});
@ -399,19 +402,114 @@ describe("runtime structural drive", () => {
releaseHook.resolve();
expect(await running).toEqual({ kind: "continue" });
let checkpoint = currentState(fixture);
if (checkpoint.at !== "checkpoint") throw new Error("compaction did not restore checkpoint");
expect(fixture.lane.state.inbox).toEqual([{ entryId: "queued", kind: queue }]);
expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
checkpoint = currentState(fixture);
if (checkpoint.at !== "checkpoint") throw new Error("queued input did not reach checkpoint");
expect(checkpoint.continuation).toEqual({ kind: "need_assistant", overflowRecoveryUsed: false });
expect(fixture.lane.state.inbox).toEqual([]);
expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
expect(currentState(fixture).at).toBe("assistant.ready");
const routed = currentState(fixture);
if (queue === "steer") {
if (routed.at !== "assistant.ready") throw new Error("steer did not route directly to generation");
expect(routed.generationContext).toMatchObject({
triggerEntryId: "queued",
overflowRecoveryUsed: false,
});
expect(fixture.lane.state.inbox).toEqual([]);
expect(
await fixture.session.getValue(storedValues.pendingEntry("queued"), BACKGROUND_CONTEXT),
).toBeUndefined();
const entry = await fixture.session.getEntry("queued", BACKGROUND_CONTEXT);
expect(entry).toMatchObject({ parentId: expect.any(String), type: "message" });
expect((await fixture.session.getEntry(entry!.parentId!, BACKGROUND_CONTEXT))?.type).toBe("compaction");
} else {
if (routed.at !== "checkpoint") throw new Error("follow-up did not reach the finish checkpoint");
expect(fixture.lane.state.inbox).toEqual([{ entryId: "queued", kind: "followUp" }]);
expect(await runCheckpoint(fixture.lane, fixture.drive, routed)).toEqual({ kind: "continue" });
const checkpoint = currentState(fixture);
if (checkpoint.at !== "checkpoint") throw new Error("follow-up did not reach checkpoint");
expect(checkpoint.continuation).toEqual({ kind: "need_assistant", overflowRecoveryUsed: false });
expect(fixture.lane.state.inbox).toEqual([]);
expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
expect(currentState(fixture).at).toBe("assistant.ready");
}
},
);
it("publishes structural output and mixed write/steer input in one admission-ordered commit", async () => {
const fixture = await createFixture();
const scope = runScope();
const deciding = {
...scope,
settings: { ...scope.settings, steeringMode: "one-at-a-time" },
at: "summary.deciding",
task: runCompactionTask("threshold", {
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
triggerEntryId: "tip",
thresholdCheckedTriggerEntryId: "tip",
}),
} as const;
await installOperation(
fixture,
deciding,
{ kind: "run", promptEntryIds: ["tip"] },
{
entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }],
preparation: { taskId: "task", value: compactionPreparation() },
},
);
await fixture.lane.command((projection) => {
const inbox = [
{ entryId: "write-1", kind: "write" as const },
{ entryId: "steer", kind: "steer" as const },
{ entryId: "write-2", kind: "write" as const },
{ entryId: "steer-2", kind: "steer" as const },
];
return {
kind: "commit",
writes: [
storedValues.setValue(storedValues.pendingEntry("write-1"), {
type: "custom",
customType: "note",
payload: { order: 1 },
}),
storedValues.setValue(storedValues.pendingEntry("steer"), {
type: "message",
payload: user("steer"),
}),
storedValues.setValue(storedValues.pendingEntry("write-2"), {
type: "custom",
customType: "note",
payload: { order: 2 },
}),
storedValues.setValue(storedValues.pendingEntry("steer-2"), {
type: "message",
payload: user("next steer"),
}),
storedValues.setValue(storedValues.laneState("main"), {
currentOperationId: operationId,
lastOperationId: null,
inbox,
}),
],
next: { ...projection, inbox },
materialize: () => undefined,
};
}, BACKGROUND_CONTEXT);
fixture.storage.clearCommitAttempts();
fixture.hooks.on("before_compaction", () => ({
compaction: { summary: "summary", tokensBefore: 1_000, retainedTail: [user("tail")] },
}));
expect(await runStructuralDecision(fixture.lane, fixture.drive, deciding)).toEqual({ kind: "continue" });
expect(fixture.storage.getCommitAttempts()).toHaveLength(1);
expect(fixture.lane.state.inbox).toEqual([{ entryId: "steer-2", kind: "steer" }]);
expect(await fixture.session.getValue(storedValues.pendingEntry("steer-2"), BACKGROUND_CONTEXT)).toBeDefined();
const routed = currentState(fixture);
if (routed.at !== "assistant.ready") throw new Error("mixed input did not route to generation");
expect(routed.generationContext.triggerEntryId).toBe("steer");
expect(await fixture.session.getEntry("write-1", BACKGROUND_CONTEXT)).toMatchObject({
parentId: expect.any(String),
});
expect(await fixture.session.getEntry("steer", BACKGROUND_CONTEXT)).toMatchObject({ parentId: "write-1" });
expect(await fixture.session.getEntry("write-2", BACKGROUND_CONTEXT)).toMatchObject({ parentId: "steer" });
expect(fixture.lane.state.tipId).toBe("write-2");
});
it("queues writes during standalone structural work without changing the operation", async () => {
const fixture = await createFixture();
const deciding = {
@ -510,6 +608,7 @@ describe("runtime structural drive", () => {
preparation: { taskId: "task", value: compactionPreparation() },
},
);
const queuedId = await fixture.lane.appendCustomEntry("retained", { after: "compaction" }, BACKGROUND_CONTEXT);
fixture.hooks.on("before_compaction", () => ({
compaction: {
summary: "hook summary",
@ -525,6 +624,8 @@ describe("runtime structural drive", () => {
outcome: { operationId, kind: "compaction", status: "completed" },
});
expect(fixture.lane.state.operation).toBeNull();
expect(fixture.lane.state.inbox).toEqual([{ entryId: queuedId, kind: "write" }]);
expect(await fixture.session.getValue(storedValues.pendingEntry(queuedId), BACKGROUND_CONTEXT)).toBeDefined();
const entry = await fixture.session.getEntry(fixture.lane.state.tipId!, BACKGROUND_CONTEXT);
expect(entry).toMatchObject({
type: "compaction",
@ -657,7 +758,7 @@ describe("runtime structural drive", () => {
expect(fixture.events.filter((event) => event.type === "usage")).toHaveLength(2);
});
it("restores the captured run checkpoint after generated threshold compaction", async () => {
it("preserves the overflow recovery bound when compaction resumes generation", async () => {
const fixture = await createFixture();
const resumeAfter: CheckpointOperation = {
...runScope(),
@ -668,8 +769,8 @@ describe("runtime structural drive", () => {
};
const ready = summaryReady(
runScope(),
runCompactionTask("threshold", {
continuation: resumeAfter.continuation,
runCompactionTask("overflow", {
continuation: { kind: "need_assistant", overflowRecoveryUsed: true },
triggerEntryId: resumeAfter.triggerEntryId,
thresholdCheckedTriggerEntryId: resumeAfter.thresholdCheckedTriggerEntryId,
}),
@ -687,12 +788,11 @@ describe("runtime structural drive", () => {
fixture.faux.setResponses([fauxAssistantMessage("generated summary")]);
expect(await runStructuralGeneration(fixture.lane, fixture.drive, ready)).toEqual({ kind: "continue" });
const restored = currentState(fixture);
if (restored.at !== "checkpoint") throw new Error("generated compaction did not restore checkpoint");
expect(restored).toMatchObject({
continuation: resumeAfter.continuation,
const routed = currentState(fixture);
if (routed.at !== "assistant.ready") throw new Error("generated compaction did not route to generation");
expect(routed.generationContext).toMatchObject({
triggerEntryId: "tip",
thresholdCheckedTriggerEntryId: "tip",
overflowRecoveryUsed: true,
});
expect(fixture.lane.state.tipId).toBe("summary-entry");
const entry = await fixture.session.getEntry("summary-entry", BACKGROUND_CONTEXT);