Skip to content

🤖 refactor: move attempt identity and settlement receipts into AttemptLedger #7677

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ThomasK33 merged 2 commits into from
Oct 27, 2026
Merged

🤖 refactor: move attempt identity and settlement receipts into AttemptLedger #7677

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b54a81c
🤖 refactor: move attempt identity and settlement receipts into Attemp…
ThomasK33 Oct 27, 2026
2f0833f
🤖 tests: let the await-thenable lint rule see bun:test async matchers…
ThomasK33 Oct 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
654 changes: 654 additions & 0 deletions src/node/services/attemptLedger.test.ts
View file
Open in desktop

Large diffs are not rendered by default.

695 changes: 695 additions & 0 deletions src/node/services/attemptLedger.ts
View file
Open in desktop

Large diffs are not rendered by default.

18 changes: 10 additions & 8 deletions src/node/services/taskService.aiSettings.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1938,8 +1938,10 @@ describe("TaskService", () => {
identity: { attemptId: string | undefined; receiptEligible: boolean }
): unknown;
settleOwnedTaskAttempt(taskId: string, attempt: unknown, source: string): void;
ownedAttemptByTaskId: Map<string, unknown>;
attemptSettlementByTaskId: Map<string, { attempt: unknown; source: string }>;
attemptLedger: {
ownedAttemptByTaskId: Map<string, unknown>;
attemptSettlementByTaskId: Map<string, { attempt: unknown; source: string }>;
};
}

/** Queued admission: returns at once and hands onAccepted to the test once armed. */
Expand Down Expand Up @@ -1974,7 +1976,7 @@ describe("TaskService", () => {
await setDelegatedExec(config, { modelString: MODEL_B });
const busy = admitQueued(workspaceService, childId);
const ledger = taskService as unknown as AttemptLedger;
const previousAttempt = ledger.ownedAttemptByTaskId.get(childId);
const previousAttempt = ledger.attemptLedger.ownedAttemptByTaskId.get(childId);
const store = workspaceTurnManagerInternals(taskService).taskHandleStore;
const originalUpsert = store.upsertWorkspaceTurn.bind(store);
let reactivationAttempt: unknown;
Expand All @@ -1988,7 +1990,7 @@ describe("TaskService", () => {
child.taskExecutionStatus === "running"
) {
// The claim-and-settings write is committed; the next acceptance step fails.
reactivationAttempt = ledger.ownedAttemptByTaskId.get(childId);
reactivationAttempt = ledger.attemptLedger.ownedAttemptByTaskId.get(childId);
if (variant === "settled meanwhile") {
ledger.settleOwnedTaskAttempt(childId, reactivationAttempt, "test-settlement");
} else if (variant === "successor meanwhile") {
Expand All @@ -2011,15 +2013,15 @@ describe("TaskService", () => {
// Existing settlement settled the claimed mirror to a terminal status.
expect(child?.taskExecutionStatus).toBeDefined();
expect(isActiveWorkspaceTurnTaskStatus(child?.taskExecutionStatus)).toBe(false);
const owned = ledger.ownedAttemptByTaskId.get(childId);
const owned = ledger.attemptLedger.ownedAttemptByTaskId.get(childId);
if (variant === "no settlement") {
// A published attempt is never rolled back (#4308): the refused reactivation's
// attempt stays owned until a Stop settles it.
expect(reactivationAttempt).not.toBe(previousAttempt);
expect(owned).toBe(reactivationAttempt);
} else if (variant === "settled meanwhile") {
expect(owned).toBe(reactivationAttempt);
expect(ledger.attemptSettlementByTaskId.get(childId)?.attempt).toBe(
expect(ledger.attemptLedger.attemptSettlementByTaskId.get(childId)?.attempt).toBe(
reactivationAttempt
);
} else {
Expand Down Expand Up @@ -2193,7 +2195,7 @@ describe("TaskService", () => {
report: { reportMarkdown: string },
attempt: unknown
): Promise<unknown>;
ownedAttemptByTaskId: Map<string, unknown>;
attemptLedger: { ownedAttemptByTaskId: Map<string, unknown> };
};

expect(
Expand All @@ -2207,7 +2209,7 @@ describe("TaskService", () => {
childId,
findWorkspaceEntry(config.loadConfigOrDefault(), childId),
{ reportMarkdown: "done" },
internal.ownedAttemptByTaskId.get(childId)
internal.attemptLedger.ownedAttemptByTaskId.get(childId)
);
const report = await readSubagentReportArtifact(
path.join(config.sessionsDir, parentId),
Expand Down
271 changes: 148 additions & 123 deletions src/node/services/taskService.attemptAdmission.test.ts
View file
Open in desktop

Large diffs are not rendered by default.

8 changes: 4 additions & 4 deletions src/node/services/taskService.childGoals.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1208,10 +1208,10 @@ describe("TaskService child goals", () => {
expect(seeded.success).toBe(true);
expect(await t.taskService.markInterruptedTaskRunning(childId)).toBe(true);
const internal = t.taskService as unknown as {
ownedAttemptByTaskId: Map<string, { receiptEligible: boolean }>;
attemptLedger: { ownedAttemptByTaskId: Map<string, { receiptEligible: boolean }> };
streamEndDecisionsByTaskId: Map<string, Array<{ outcome: string }>>;
};
expect(internal.ownedAttemptByTaskId.get(childId)?.receiptEligible).toBe(true);
expect(internal.attemptLedger.ownedAttemptByTaskId.get(childId)?.receiptEligible).toBe(true);
await t.setChildGoal();
const realWait = t.goals.waitForStreamAccountingReceipt.bind(t.goals);
spyOn(t.goals, "waitForStreamAccountingReceipt").mockImplementation((workspaceId, messageId) =>
Expand Down Expand Up @@ -1566,10 +1566,10 @@ describe("TaskService child goals", () => {

const listeners = (
t.taskService as unknown as {
attemptSettlementListenersByTaskId: Map<string, Set<() => void>>;
attemptLedger: { attemptSettlementListenersByTaskId: Map<string, Set<() => void>> };
notifyAttemptSettlementListeners: (taskId: string) => void;
}
).attemptSettlementListenersByTaskId;
).attemptLedger.attemptSettlementListenersByTaskId;
expect(listeners.get(childId)?.size).toBe(1);
(
t.taskService as unknown as { notifyAttemptSettlementListeners: (taskId: string) => void }
Expand Down
48 changes: 25 additions & 23 deletions src/node/services/taskService.reportDecisionHold.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,10 @@ const rootId = "root-hold";
const model = "openai:gpt-5.2";

interface Internals {
ownedAttemptByTaskId: Map<string, { attemptId?: string }>;
attemptSettlementByTaskId: Map<string, { attemptId?: string; phase: string }>;
attemptLedger: {
ownedAttemptByTaskId: Map<string, { attemptId?: string }>;
attemptSettlementByTaskId: Map<string, { attemptId?: string; phase: string }>;
};
admittedSendsByTaskId: Map<string, Set<{ state: string; attemptId: string }>>;
streamEndDecisionsByTaskId: Map<string, Array<{ attemptId: string; outcome: string }>>;
taskStopLatch: { workspaceStopRecords: Map<string, unknown> };
Expand Down Expand Up @@ -213,7 +215,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
messageId,
row: row?.taskAttemptId,
status: row?.taskStatus,
owner: ledger.svc?.ownedAttemptByTaskId.get(childId)?.attemptId,
owner: ledger.svc?.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId,
});
aiEmitter.emit("stream-start", {
type: "stream-start",
Expand Down Expand Up @@ -349,7 +351,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await until(() => entryOf(config, childId)?.taskStatus === "reported", "report");
if (!ordering.startsWith("drain before")) {
// The decision landed before the session finished the turn: the drain reads it late.
await until(() => !svc.ownedAttemptByTaskId.has(childId), "release");
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "release");
stack.completeStream(0, event);
}
await until(() => !workspaceService.hasQueuedMessages(childId), "queue drained");
Expand All @@ -361,7 +363,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
taskStatus: "reported",
taskAttemptId: attemptA,
});
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
// The follow-up's obligation was discharged exactly once and the decision pruned.
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
Expand Down Expand Up @@ -416,7 +418,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// still live, so the queued follow-up has not drained.
const event = stack.endStream(0, { report: "done" }, false);
await until(() => entryOf(config, childId)?.taskStatus === "reported", "report");
await until(() => !svc.ownedAttemptByTaskId.has(childId), "attempt released");
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released");
expect(workspaceService.hasQueuedMessages(childId)).toBe(true);
expect(stack.heldInputs()).toHaveLength(0);

Expand Down Expand Up @@ -477,7 +479,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// The report turn ends on a terminal agent_report; the session has not finished the turn.
const event = stack.endStream(0, { report: "done" }, false);
await until(() => entryOf(config, childId)?.taskStatus === "reported", "report");
await until(() => !svc.ownedAttemptByTaskId.has(childId), "release");
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "release");
// The decision resolved with no reader queued yet. The user's follow-up arrives now, while
// the reporting turn is still live: it targets the completed attempt, so it is refused in
// its preflight (fail closed, text stays with the caller) instead of queueing behind the
Expand All @@ -496,7 +498,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
taskStatus: "reported",
taskAttemptId: attemptA,
});
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
// Nor was it accepted as a user turn of that attempt.
const history = await fixture.historyService.getLastMessages(childId, 10);
expect(history.success && history.data.map((message) => message.role)).toEqual(["user"]);
Expand Down Expand Up @@ -886,7 +888,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(completions).toHaveLength(1);
// Not released (no durable report), not continued: fail closed, the follow-up held — as
// indeterminate, never as if a report had been confirmed.
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(stack.heldTexts()).toEqual(["follow-up text"]);
expect(stack.heldInputs().map((held) => held.reason)).toEqual(["indeterminate"]);
// Refused as indeterminate — not as a completed report and not as a plain stale attempt.
Expand Down Expand Up @@ -1208,7 +1210,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// Handler resumes: publishes and releases A; no reader is left, so the decision is dropped.
gate.resolve();
await until(() => !svc.streamEndDecisionsByTaskId.has(childId), "decision pruned");
await until(() => !svc.ownedAttemptByTaskId.has(childId), "attempt released");
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released");
await yieldMacrotasks(5);
expect(entryOf(config, childId)).toMatchObject({
taskStatus: "reported",
Expand Down Expand Up @@ -1258,7 +1260,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(completions).toHaveLength(1);
gate.resolve();
await until(() => !svc.streamEndDecisionsByTaskId.has(childId), "decision pruned");
await until(() => !svc.ownedAttemptByTaskId.has(childId), "attempt released");
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released");
await yieldMacrotasks(5);
expect(entryOf(config, childId)).toMatchObject({
taskStatus: "reported",
Expand Down Expand Up @@ -1305,15 +1307,15 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
try {
if (status === "reported") {
// The released shape of an ordinary reported child (no owner, no settlement entry).
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
}
const result = await workspaceService.sendMessage(childId, "manual follow-up", sendOptions);
expect(result.success).toBe(false);
// Nothing streams under the winner's attempt here and no obligation is left bound to it.
await yieldMacrotasks(3);
expect(completions).toHaveLength(0);
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).not.toBe(winner);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).not.toBe(winner);
expect(entryOf(config, childId)?.taskAttemptId).toBe(winner);
} finally {
await stack.cleanup();
Expand All @@ -1332,7 +1334,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await until(() => completions.length === 1, "the re-drive's stream");
const redriven = entryOf(config, childId)?.taskAttemptId;
expect(redriven).not.toBe("att_00000000000000a4");
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
// The user's follow-up arrives mid-stream: queued, bound to the re-driven attempt.
expect(await workspaceService.sendMessage(childId, "follow-up text", sendOptions)).toEqual(
Ok(undefined)
Expand All @@ -1349,7 +1351,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
// The hold came from the decision alone: no ownership (settlement/receipt authority) granted.
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(entryOf(config, childId)?.taskAttemptId).toBe(redriven);
} finally {
await stack.cleanup();
Expand All @@ -1368,7 +1370,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await until(() => completions.length === 1, "the re-drive's stream");
const redriven = entryOf(config, childId)?.taskAttemptId;
expect(redriven).not.toBe("att_00000000000000a4");
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
// Backend B admits the row under its own attempt while this backend's stream still runs.
await otherBackend.editConfig((cfg) => {
for (const project of cfg.projects.values()) {
Expand Down Expand Up @@ -1407,7 +1409,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(
await readSubagentReportArtifact(path.join(config.sessionsDir, rootId), childId)
).toBeNull();
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
} finally {
await stack.cleanup();
}
Expand All @@ -1423,7 +1425,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await taskService.recoverInterruptedTasks();
await until(() => completions.length === 1, "the re-drive's stream");
const redriven = entryOf(config, childId)?.taskAttemptId;
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
// Backend B re-admits the row right before the publication's status write, i.e. after
// every earlier check of the stream-end handler read the re-driven attempt.
let rotated = false;
Expand Down Expand Up @@ -1471,7 +1473,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
(svc.streamEndDecisionsByTaskId.get(childId) ?? []).map((decision) => decision.attemptId)
).not.toContain(foreign);
expect(redriven).not.toBe(foreign);
expect(svc.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
} finally {
await stack.cleanup();
}
Expand Down Expand Up @@ -1499,7 +1501,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
try {
expect(await taskService.markInterruptedTaskRunning(childId)).toBe(true);
const attemptA = entryOf(config, childId)!.taskAttemptId!;
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(await workspaceService.sendMessage(childId, "work", sendOptions)).toEqual(
Ok(undefined)
);
Expand Down Expand Up @@ -1559,7 +1561,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
try {
expect(await taskService.markInterruptedTaskRunning(childId)).toBe(true);
const attemptA = entryOf(config, childId)!.taskAttemptId!;
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
// Backend B re-admits the idle row under its own attempt; this process still owns A.
await otherBackend.editConfig((cfg) => {
for (const project of cfg.projects.values()) {
Expand All @@ -1577,7 +1579,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
Ok(undefined)
);
expect(completions).toHaveLength(1);
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).toBe(attemptA);
stack.endStream(0, { report: "done by B" }, true);
await until(() => entryOf(config, childId)?.taskStatus === "reported", "B's report");
expect(entryOf(config, childId)?.taskAttemptId).toBe(foreign);
Expand Down Expand Up @@ -1669,7 +1671,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(
await readSubagentReportArtifact(path.join(config.sessionsDir, rootId), childId)
).toBeNull();
expect(svc.ownedAttemptByTaskId.get(childId)?.attemptId).not.toBe(foreign);
expect(svc.attemptLedger.ownedAttemptByTaskId.get(childId)?.attemptId).not.toBe(foreign);
} finally {
await stack.cleanup();
}
Expand Down
Loading
Loading