Skip to content

🤖 refactor: move stream-end decisions into StreamEndLedger #7679

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 1 commit into from
Oct 27, 2026
Merged

🤖 refactor: move stream-end decisions into StreamEndLedger #7679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 2 additions & 2 deletions src/node/services/attemptLedger.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,9 +194,9 @@ export interface AttemptLedgerDeps {
) => Promise<boolean>;
settleChildGoalPause: (workspaceId: string) => Promise<void>;
findStreamEndDecision: (taskId: string, attemptId: string) => StreamEndDecision | undefined;
/** Drop the stream-end decisions no send obligation can still read (TaskService). */
/** Drop the stream-end decisions no send obligation can still read (StreamEndLedger). */
pruneStreamEndDecisions: (taskId: string) => void;
/** Drop every stream-end decision of the task (TaskService.streamEndDecisionsByTaskId). */
/** Drop every stream-end decision of the task (StreamEndLedger). */
discardStreamEndDecisions: (taskId: string) => void;
inspectAttemptOutcome: (
taskId: string,
Expand Down
451 changes: 451 additions & 0 deletions src/node/services/streamEndLedger.test.ts
View file
Open in desktop

Large diffs are not rendered by default.

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

Large diffs are not rendered by default.

66 changes: 39 additions & 27 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 @@ -1209,7 +1209,7 @@ describe("TaskService child goals", () => {
expect(await t.taskService.markInterruptedTaskRunning(childId)).toBe(true);
const internal = t.taskService as unknown as {
attemptLedger: { ownedAttemptByTaskId: Map<string, { receiptEligible: boolean }> };
streamEndDecisionsByTaskId: Map<string, Array<{ outcome: string }>>;
streamEndLedger: { streamEndDecisionsByTaskId: Map<string, Array<{ outcome: string }>> };
};
expect(internal.attemptLedger.ownedAttemptByTaskId.get(childId)?.receiptEligible).toBe(true);
await t.setChildGoal();
Expand All @@ -1220,9 +1220,9 @@ describe("TaskService child goals", () => {
t.goals.beginStreamAccountingReceipt(childId, "assistant-1");
await streamEnd(t.taskService, t.proseEnd("assistant-1"));
expect(t.child()?.taskStatus).toBe("running");
expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
return { t, internal };
}

Expand All @@ -1239,7 +1239,7 @@ describe("TaskService child goals", () => {
data: { stoppedTaskIds: [childId] },
});
expect(t.taskService.isWorkspaceStopInProgress(childId)).toBe(false);
expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(t.child()?.taskStatus).toBe("interrupted");
});

Expand All @@ -1259,7 +1259,7 @@ describe("TaskService child goals", () => {
await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId), {
timeoutMs: 1_000,
});
expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
});

// #6920 control: a terminate whose removal succeeds. The removal deregisters the workspace and
Expand Down Expand Up @@ -1309,7 +1309,7 @@ describe("TaskService child goals", () => {
expect(await waiter).toBe("rejected: Task terminated");
expect(rejectWaiters.mock.calls.filter((call) => call[0] === childId)).toHaveLength(1);
await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId));
expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(removeQueuedMessagesByDedupeKeyPrefix).not.toHaveBeenCalled();
expect(t.sends()).toHaveLength(0);
expect(await t.parentReports()).toHaveLength(0);
Expand Down Expand Up @@ -1342,7 +1342,7 @@ describe("TaskService child goals", () => {
await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId), {
timeoutMs: 1_000,
});
expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
}
);

Expand All @@ -1361,9 +1361,12 @@ describe("TaskService child goals", () => {

expect(report.reportMarkdown).toBe("Report from another backend.");
expect(t.child()?.taskStatus).toBe("reported");
await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), {
timeoutMs: 1_000,
});
await waitForCondition(
() => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
{
timeoutMs: 1_000,
}
);
// Settled, not continued: no goal turn, recovery prompt or report send for the child.
expect(t.sends()).toHaveLength(0);
});
Expand Down Expand Up @@ -1476,9 +1479,9 @@ describe("TaskService child goals", () => {

expect(waited).toBe("still waiting");
expect(t.child()?.taskStatus).toBe("running");
expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
});

// #7003 legacy artifacts over an interrupted row keep today's self-heal: adopted as the report.
Expand Down Expand Up @@ -1514,17 +1517,20 @@ describe("TaskService child goals", () => {

expect(report.reportMarkdown).toBe("Legacy report.");
expect(t.child()?.taskStatus).toBe("reported");
await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), {
timeoutMs: 1_000,
});
await waitForCondition(
() => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
{
timeoutMs: 1_000,
}
);
});

// #7003: another backend's `reported` for the deferred stream's own attempt lands, and then any
// local notification re-runs the deferral before this backend reads the artifact. The re-run
// must settle the decision as indeterminate (held follow-ups refused), not as nonreport.
test("a deferral re-run after another backend reported its attempt settles as indeterminate", async () => {
const { t, internal } = await deferStreamEndOfOwnedAttempt();
const decision = internal.streamEndDecisionsByTaskId.get(childId)?.[0];
const decision = internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0];
expect(decision?.outcome).toBe("pending");
await t.editChild((workspace) => {
workspace.taskStatus = "reported";
Expand All @@ -1535,9 +1541,12 @@ describe("TaskService child goals", () => {
t.taskService as unknown as { notifyAttemptSettlementListeners: (taskId: string) => void }
).notifyAttemptSettlementListeners(childId);

await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), {
timeoutMs: 1_000,
});
await waitForCondition(
() => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
{
timeoutMs: 1_000,
}
);
expect(decision?.outcome).toBe("indeterminate");
expect(t.sends()).toHaveLength(0);
});
Expand Down Expand Up @@ -1579,9 +1588,9 @@ describe("TaskService child goals", () => {

expect(listeners.get(childId)?.size).toBe(1);
expect(t.child()?.taskStatus).toBe("running");
expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
});

// #7003: a row moved to awaiting_report while the stream end is deferred (no notify on that
Expand All @@ -1597,9 +1606,12 @@ describe("TaskService child goals", () => {
t.taskService as unknown as { notifyAttemptSettlementListeners: (taskId: string) => void }
).notifyAttemptSettlementListeners(childId);

await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), {
timeoutMs: 1_000,
});
await waitForCondition(
() => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
{
timeoutMs: 1_000,
}
);
await t.untilStatus("reported");
expect(t.child()?.taskStatus).toBe("reported");
// The parent's delivery lands after the status write.
Expand Down
74 changes: 47 additions & 27 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 @@ -54,7 +54,9 @@ interface Internals {
attemptSettlementByTaskId: Map<string, { attemptId?: string; phase: string }>;
admittedSendsByTaskId: Map<string, Set<{ state: string; attemptId: string }>>;
};
streamEndDecisionsByTaskId: Map<string, Array<{ attemptId: string; outcome: string }>>;
streamEndLedger: {
streamEndDecisionsByTaskId: Map<string, Array<{ attemptId: string; outcome: string }>>;
};
taskStopLatch: { workspaceStopRecords: Map<string, unknown> };
emitWorkspaceMetadata: (workspaceId: string) => Promise<void>;
workspaceEventLocks: { withLock<T>(key: string, operation: () => Promise<T>): Promise<T> };
Expand Down Expand Up @@ -346,7 +348,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// The session's terminal drain runs while TaskService still decides: the entry is HELD
// — still queued, no turn claimed, nothing dispatched.
await yieldMacrotasks(3);
expect(svc.streamEndDecisionsByTaskId.get(childId)?.[0]).toMatchObject({
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]).toMatchObject({
attemptId: attemptA,
});
}
Expand All @@ -368,7 +370,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
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);
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
// The follow-up is held by the session (nothing is pushed into the composer), refused
// because the task reported — and published with that reason.
expect(stack.heldTexts()).toEqual(["follow-up text"]);
Expand Down Expand Up @@ -507,7 +509,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// Its obligation was discharged, and the decision is gone once the reporting turn settled
// (bounded: nothing retained past the turn it was decided for).
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
// After the turn settled, a manual send is a new admission that mints a fresh attempt.
expect(await taskService.markInterruptedTaskRunning(childId)).toBe(false);
expect(entryOf(config, childId)?.taskAttemptId).not.toBe(attemptA);
Expand Down Expand Up @@ -855,7 +857,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
);
// The decision reads `nonreport` for as long as a queued reader (the nudge) is bound to A.
expect(
(svc.streamEndDecisionsByTaskId.get(childId) ?? []).every(
(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).every(
(decision) => decision.outcome === "nonreport"
)
).toBe(true);
Expand Down Expand Up @@ -896,7 +898,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// Refused as indeterminate — not as a completed report and not as a plain stale attempt.
expect(onCanceled).toHaveBeenCalledWith(TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE);
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
} finally {
await stack.cleanup();
}
Expand Down Expand Up @@ -951,13 +953,14 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
).toEqual(Ok(undefined));
stack.endStream(0, {}, true);
await until(
() => svc.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome === "pending",
() =>
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome === "pending",
"deferred decision"
);
await yieldMacrotasks(30);
expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
expect(completions).toHaveLength(1);

if (rowStatus === "reported") {
Expand Down Expand Up @@ -989,7 +992,8 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
});
expect(report.reportMarkdown).toBe("Report from another backend.");
await until(
() => svc.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome !== "pending",
() =>
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome !== "pending",
"decision settled"
);
await until(() => !workspaceService.hasQueuedMessages(childId), "queue drained");
Expand Down Expand Up @@ -1055,7 +1059,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(completions).toHaveLength(1);
expect(stack.heldTexts()).toEqual(["follow-up text"]);
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
} finally {
gate.resolve();
await stack.cleanup();
Expand Down Expand Up @@ -1175,7 +1179,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
// Held: still queued, obligation still enqueued, decision pending.
expect(workspaceService.hasQueuedMessages(childId)).toBe(true);
expect(outstanding(svc, childId).map((send) => send.state)).toEqual(["enqueued"]);
expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual(["pending"]);
expect(
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
return { gate, attemptA, onCanceled };
}

Expand Down Expand Up @@ -1205,13 +1211,16 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(onCanceled).toHaveBeenCalledWith("Queued message cleared before dispatch.");
// The obligation is discharged by the clear; the decision still belongs to the handler.
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
expect(completions).toHaveLength(1);
// 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.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
"decision pruned"
);
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released");
await yieldMacrotasks(5);
expect(entryOf(config, childId)).toMatchObject({
Expand Down Expand Up @@ -1256,12 +1265,15 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.taskStopLatch.workspaceStopRecords.has(childId)).toBe(false);
expect(taskService.isWorkspaceStopInProgress(childId)).toBe(false);
expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([
"pending",
]);
expect(
svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)
).toEqual(["pending"]);
expect(completions).toHaveLength(1);
gate.resolve();
await until(() => !svc.streamEndDecisionsByTaskId.has(childId), "decision pruned");
await until(
() => !svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId),
"decision pruned"
);
await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released");
await yieldMacrotasks(5);
expect(entryOf(config, childId)).toMatchObject({
Expand Down Expand Up @@ -1351,7 +1363,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
expect(completions).toHaveLength(1);
expect(stack.heldTexts()).toEqual(["follow-up text"]);
expect(outstanding(svc, childId)).toHaveLength(0);
expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false);
expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false);
// The hold came from the decision alone: no ownership (settlement/receipt authority) granted.
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
expect(entryOf(config, childId)?.taskAttemptId).toBe(redriven);
Expand Down Expand Up @@ -1394,7 +1406,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
stack.endStream(0, { report: "done by the re-drive" }, true);
// No decision of the ended stream is ever keyed by B.
expect(
(svc.streamEndDecisionsByTaskId.get(childId) ?? []).map((decision) => decision.attemptId)
(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).map(
(decision) => decision.attemptId
)
).not.toContain(foreign);
await until(() => completions.length === 2, "B's follow-up dispatched");
await yieldMacrotasks(20);
Expand Down Expand Up @@ -1457,7 +1471,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await until(() => rotated, "the publication write");
await until(
() =>
!(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some((d) => d.outcome === "pending"),
!(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some(
(d) => d.outcome === "pending"
),
"the decision resolved"
);
await yieldMacrotasks(20);
Expand All @@ -1472,7 +1488,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
).toBeNull();
// The re-drive's decision stayed keyed by its own attempt, and no ownership was granted.
expect(
(svc.streamEndDecisionsByTaskId.get(childId) ?? []).map((decision) => decision.attemptId)
(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).map(
(decision) => decision.attemptId
)
).not.toContain(foreign);
expect(redriven).not.toBe(foreign);
expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false);
Expand Down Expand Up @@ -1533,7 +1551,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await until(() => rotated, "the rotation");
await until(
() =>
!(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some(
!(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some(
(decision) => decision.outcome === "pending"
),
"the decision resolved"
Expand Down Expand Up @@ -1658,7 +1676,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
await rotation;
await until(
() =>
!(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some((d) => d.outcome === "pending"),
!(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some(
(d) => d.outcome === "pending"
),
"the decision resolved"
);
await yieldMacrotasks(20);
Expand Down
Loading
Loading