Skip to content

Commit f509963

Browse files
committed
fix(session): latch abort across compaction controller reset
Rebuild reset() mints a fresh signal, so re-reading the live controller after wrapCompactor discarded a certified stub would record a phantom compaction-handoff.
1 parent c7d7932 commit f509963

5 files changed

Lines changed: 133 additions & 16 deletions

File tree

‎src/session/assemble-runtime.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -557,11 +557,11 @@ export interface ChatAgentWiring {
557557
*/
558558
getCompactor: (wrapPruning?: (pruning: Compactor) => Compactor) => Compactor;
559559
/**
560-
* Bound to the TUI compaction lifecycle abort signal. The completeness
561-
* gate runs inside wrapCompactor's race, so a discarded certified stub
562-
* must not persist a compaction-handoff for a fold that never landed.
560+
* Bound to the TUI compaction lifecycle abort signal. Captured at
561+
* completeness-gate apply start so onBuilt reset() cannot un-abort an
562+
* in-flight fold that wrapCompactor already discarded.
563563
*/
564-
isCompactionAborted?: () => boolean;
564+
getCompactionAbortSignal?: () => AbortSignal;
565565
/** Experimental Anthropic prompt shrink. Default off when omitted. */
566566
anthropicCachePrompt?: () => boolean;
567567
/**
@@ -795,9 +795,9 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
795795
wrapCompactorWithCompletenessGate(
796796
pruning,
797797
primaryArchive,
798-
wiring.isCompactionAborted === undefined
798+
wiring.getCompactionAbortSignal === undefined
799799
? undefined
800-
: { isAborted: wiring.isCompactionAborted },
800+
: { getSignal: wiring.getCompactionAbortSignal },
801801
),
802802
),
803803
},

‎src/session/compaction-archive.ts‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -952,16 +952,21 @@ async function recordFreshHandoffOutput(
952952
* drop it even when the summarizer does not echo it verbatim. `isAborted`
953953
* skips that record: the TUI wrapCompactor race can discard a certified
954954
* stub, and a handoff for a fold that never landed is a phantom.
955+
* `getSignal` is captured at apply start so onBuilt `reset()` cannot
956+
* un-abort an in-flight fold that wrapCompactor already discarded.
955957
*/
956958
export function wrapCompactorWithCompletenessGate(
957959
inner: Compactor,
958960
archive: CompactionArchive,
959-
opts?: { isAborted?: () => boolean },
961+
opts?: { isAborted?: () => boolean; getSignal?: () => AbortSignal },
960962
): Compactor {
961963
return {
962964
name: inner.name,
963965
version: inner.version,
964966
async apply(turns: ConversationTurn[], ctx: StrategyContext) {
967+
const abortSignal = opts?.getSignal?.();
968+
const isAborted = () =>
969+
abortSignal?.aborted === true || opts?.isAborted?.() === true;
965970
await archive.awaitPendingWrites();
966971
const proposed = await inner.apply(turns, ctx);
967972
const units = uncoveredContentUnits(turns, proposed.output);
@@ -988,14 +993,14 @@ export function wrapCompactorWithCompletenessGate(
988993
if (certificate.status !== "complete") {
989994
return incompleteIdentity(inner, turns);
990995
}
991-
if (opts?.isAborted?.() === true) {
996+
if (isAborted()) {
992997
return proposed;
993998
}
994999
await recordFreshHandoffOutput(
9951000
archive,
9961001
turns,
9971002
proposed.output,
998-
opts?.isAborted,
1003+
isAborted,
9991004
);
10001005
return proposed;
10011006
},

‎src/session/runtime-assembly.test.ts‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -985,6 +985,111 @@ describe("createSessionPruningCompactor stub fallback", () => {
985985
);
986986
expect(handoffs).toEqual([]);
987987
});
988+
989+
test("abort then reset during certifyRange does not record a phantom handoff", async () => {
990+
const notices: string[] = [];
991+
const folds: { stub: boolean }[] = [];
992+
const summarize = createModelSummarizer({
993+
getSource: () =>
994+
({
995+
id: "test",
996+
provider: "openai",
997+
model: "test-model",
998+
baseURL: "http://localhost:1",
999+
credentialId: "test",
1000+
}) as never,
1001+
complete: async () => {
1002+
throw new Error("model unreachable");
1003+
},
1004+
});
1005+
const dir = await mkdtemp(join(tmpdir(), "compaction-gate-abort-reset-"));
1006+
const blobs = new Map<string, Uint8Array>();
1007+
const archive = createCompactionArchive({
1008+
sessionId: "sess-gate-abort-reset",
1009+
contextDir: dir,
1010+
writeBlob: async (key, bytes) => {
1011+
blobs.set(key, bytes);
1012+
},
1013+
readBlob: async (key) => {
1014+
const bytes = blobs.get(key);
1015+
if (bytes === undefined) throw new Error(`missing ${key}`);
1016+
return bytes;
1017+
},
1018+
});
1019+
const now = Date.now();
1020+
const many = Array.from({ length: 8 }, (_, i) => ({
1021+
role: (i % 2 === 0 ? "user" : "assistant") as "user" | "assistant",
1022+
content: [{ type: "text" as const, text: `t${i}` }],
1023+
timestamp: now,
1024+
}));
1025+
for (const turn of many) {
1026+
const block = turn.content[0];
1027+
if (block?.type !== "text") continue;
1028+
await archive.recordAuthorizedPayload({
1029+
kind: turn.role === "assistant" ? "assistant_text" : "user_message",
1030+
payload: block.text,
1031+
});
1032+
}
1033+
const lifecycle = createCompactionLifecycle();
1034+
const getSignal = () => lifecycle.getSignal();
1035+
const isAborted = () => lifecycle.getSignal().aborted;
1036+
let releaseGated: () => void = () => undefined;
1037+
const gatedFinished = new Promise<void>((resolve) => {
1038+
releaseGated = resolve;
1039+
});
1040+
const abortingArchive = {
1041+
...archive,
1042+
certifyRange: async (ids: readonly string[]) => {
1043+
const certificate = await archive.certifyRange(ids);
1044+
lifecycle.abortCompaction("operator interrupt");
1045+
// onBuilt reset() mints a fresh controller; the in-flight apply
1046+
// must still treat this compact as aborted.
1047+
lifecycle.reset();
1048+
return certificate;
1049+
},
1050+
};
1051+
const wrapped = lifecycle.wrapCompactor(
1052+
createSessionPruningCompactor({
1053+
summarize,
1054+
onFolded: (info) => folds.push(info),
1055+
onFailure: (text) => notices.push(text),
1056+
getSignal,
1057+
isAborted,
1058+
compactionShape: { tailBudgetTokens: 1 },
1059+
wrapPruning: (pruning) => {
1060+
const gated = wrapCompactorWithCompletenessGate(
1061+
pruning,
1062+
abortingArchive,
1063+
{ getSignal, isAborted },
1064+
);
1065+
return {
1066+
name: gated.name,
1067+
version: gated.version,
1068+
apply: async (turns, ctx) => {
1069+
try {
1070+
return await gated.apply(turns, ctx);
1071+
} finally {
1072+
releaseGated();
1073+
}
1074+
},
1075+
};
1076+
},
1077+
}),
1078+
);
1079+
const result = await wrapped.apply(many as never, {
1080+
state: {} as never,
1081+
trigger: "test",
1082+
});
1083+
await gatedFinished;
1084+
expect(result.output).toBe(many);
1085+
expect(result.record.reason).toBe(COMPACTION_ABORTED_REASON);
1086+
expect(folds).toEqual([]);
1087+
expect(notices).toEqual([]);
1088+
const handoffs = (await archive.listOccurrences()).filter(
1089+
(occurrence) => occurrence.provenance === "compaction-handoff",
1090+
);
1091+
expect(handoffs).toEqual([]);
1092+
});
9881093
});
9891094

9901095
describe("buildCompactionContinuationMessage", () => {

‎src/session/runtime-assembly.ts‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -441,6 +441,12 @@ export interface SessionPruningCompactorArgs {
441441
* no onFolded side effects for work that never landed.
442442
*/
443443
isAborted?: () => boolean;
444+
/**
445+
* Captured at apply start. Prefer this over a live `isAborted` that
446+
* re-reads getSignal(): onBuilt reset() replaces the controller, and a
447+
* discarded in-flight fold must still look aborted.
448+
*/
449+
getSignal?: () => AbortSignal;
444450
/**
445451
* Wraps the inner pruning apply before fold-commit side effects.
446452
* assembleChatAgent passes the completeness gate here so a discarded
@@ -513,15 +519,15 @@ export function createSessionPruningCompactor(
513519
pendingStubNotice = undefined;
514520
const turnsBefore = turns.length;
515521
const startedAt = Date.now();
522+
const abortSignal = args.getSignal?.();
523+
const isAborted = () =>
524+
abortSignal?.aborted === true || args.isAborted?.() === true;
516525
const result = await compactor.apply(turns, ctx);
517526
// A discarded compact reports nothing. When the lifecycle abort wins
518527
// the outer race, this inner run may still complete with a genuine
519528
// fold — but the reactor threw that output away, so emitting telemetry
520529
// or onFolded would describe work that never landed (phantom fold).
521-
if (
522-
args.isAborted?.() === true ||
523-
result.record.reason === COMPACTION_ABORTED_REASON
524-
) {
530+
if (isAborted() || result.record.reason === COMPACTION_ABORTED_REASON) {
525531
return result;
526532
}
527533
// summarizedTurnCount is only set on the branch that actually folded

‎src/tui/runner/session.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -697,7 +697,7 @@ export async function assembleTUISession(
697697
? state.liveDefaultSource
698698
: state.liveSource.id,
699699
anthropicCachePrompt: () => config.anthropicCachePrompt,
700-
isCompactionAborted: () => compactionLifecycle.getSignal().aborted,
700+
getCompactionAbortSignal: () => compactionLifecycle.getSignal(),
701701
getCompactor: (wrapPruning) =>
702702
compactionLifecycle.wrapCompactor(
703703
createSessionPruningCompactor({
@@ -710,8 +710,9 @@ export async function assembleTUISession(
710710
),
711711
// The outer abort race discards this run's output — a fold that
712712
// still completes underneath must not report telemetry or side
713-
// effects for work that never landed.
714-
isAborted: () => compactionLifecycle.getSignal().aborted,
713+
// effects for work that never landed. Capture the signal at apply
714+
// start: onBuilt reset() replaces the live controller.
715+
getSignal: () => compactionLifecycle.getSignal(),
715716
// Stub notice waits until the fold commits (verify abort and
716717
// completeness-gate discard stay silent).
717718
onFailure: (text) => state.systemNotice?.(text),

0 commit comments

Comments
 (0)