From c5a35eabbfcb706fc240ac9417a1815aec4b13ff Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 26 Aug 2026 16:46:39 +0200 Subject: [PATCH] feat(agent): fuse structural boundary routing --- .../work-packages/05-direct-durable-drive.md | 4 +- .../src/harness/runtime/drive/structural.ts | 159 ++++++++++++++---- .../harness/runtime/drive-structural.test.ts | 144 +++++++++++++--- 3 files changed, 253 insertions(+), 54 deletions(-) diff --git a/packages/agent/docs/work-packages/05-direct-durable-drive.md b/packages/agent/docs/work-packages/05-direct-durable-drive.md index 47bd1e931..2d6220b98 100644 --- a/packages/agent/docs/work-packages/05-direct-durable-drive.md +++ b/packages/agent/docs/work-packages/05-direct-durable-drive.md @@ -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 diff --git a/packages/agent/src/harness/runtime/drive/structural.ts b/packages/agent/src/harness/runtime/drive/structural.ts index f08a54e8f..809f7f9c8 100644 --- a/packages/agent/src/harness/runtime/drive/structural.ts +++ b/packages/agent/src/harness/runtime/drive/structural.ts @@ -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( 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], + }, + ]), ], }; } diff --git a/packages/agent/test/harness/runtime/drive-structural.test.ts b/packages/agent/test/harness/runtime/drive-structural.test.ts index 7d5857942..a2a401270 100644 --- a/packages/agent/test/harness/runtime/drive-structural.test.ts +++ b/packages/agent/test/harness/runtime/drive-structural.test.ts @@ -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);