fix(tasks): recover compact history after cancellation and expansion

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 06:44:41 -05:00
1 parent 9f8a6b399c
commit 971a762f80
5 files changed
+106 -7

No files matched your search

+6 -6
View File
@@ -1042,7 +1042,7 @@ test.describe("Board send delivery refresh", () => {
); );
await page.route( await page.route(
new RegExp( new RegExp(
`/api/issues/(${issue.id}|${issue.identifier})/comments(?:\\?|$)`, `/api/issues/(${issue.id}|${issue.identifier})/(?:recent-comments|comments)(?:\\?|$)`,
), ),
async (route) => { async (route) => {
if ( if (
@@ -1050,9 +1050,7 @@ test.describe("Board send delivery refresh", () => {
route.request().url().includes(`/issues/${issue.identifier}/`) route.request().url().includes(`/issues/${issue.identifier}/`)
) )
canonicalCommentReadsAfterSend += 1; canonicalCommentReadsAfterSend += 1;
await fulfill( const comments = sends
route,
sends
? [ ? [
{ {
id: publicationId, id: publicationId,
@@ -1066,8 +1064,10 @@ test.describe("Board send delivery refresh", () => {
updatedAt: new Date().toISOString(), updatedAt: new Date().toISOString(),
}, },
] ]
: [], : [];
); await fulfill(route, route.request().url().includes("/recent-comments")
? { comments, latestCommentId: comments.at(-1)?.id ?? null, hasMore: false }
: comments);
}, },
); );
await page.route( await page.route(
@@ -130,6 +130,41 @@ describe("useLiveRunTranscripts", () => {
} }
}); });
it("expands a compact log tail without hiding or duplicating the latest output", async () => {
const row = (seq: number, chunk: string) => JSON.stringify({ seq, chunk, ts: `2026-10-24T00:00:0${seq}Z`, stream: "stdout" });
const result = (content: string) => ({ runId: "run-1", store: "memory", logRef: "log-1", content, nextOffset: 100_000 });
let resolveExpanded!: (value: ReturnType<typeof result>) => void;
logMock.mockResolvedValueOnce(result(`${row(2, "Latest answer")}\n`));
logMock.mockImplementationOnce(() => new Promise(resolve => { resolveExpanded = resolve; }));
function Harness({ limit, bytes = 100_000 }: { limit: number; bytes?: number }) {
useLiveRunTranscripts({
runs: [{ id: "run-1", status: "succeeded", adapterType: "codex_local", logBytes: bytes }],
enableRealtimeUpdates: false,
logReadLimitBytes: limit,
});
return null;
}
const root = createRoot(document.createElement("div"));
const chunks = () => buildTranscriptMock.mock.calls.at(-1)?.[0];
try {
await act(async () => root.render(<Harness limit={32_000} />));
expect(logMock).toHaveBeenLastCalledWith("run-1", 68_000, 32_000, expect.anything());
expect(chunks()).toEqual([expect.objectContaining({ chunk: "Latest answer" })]);
await act(async () => root.render(<Harness limit={256_000} />));
expect(logMock).toHaveBeenLastCalledWith("run-1", 0, 256_000, expect.anything());
expect(chunks()).toEqual([expect.objectContaining({ chunk: "Latest answer" })]);
await act(async () => resolveExpanded(result(`${row(1, "Earlier activity")}\n${row(2, "Latest answer")}\n`)));
expect(chunks()).toEqual([
expect.objectContaining({ chunk: "Earlier activity" }),
expect.objectContaining({ chunk: "Latest answer" }),
]);
await act(async () => root.render(<Harness limit={256_000} bytes={120_000} />));
expect(logMock).toHaveBeenLastCalledWith("run-1", 100_000, 256_000, expect.anything());
} finally {
await act(async () => root.unmount());
}
});
it("waits for a connecting socket to open before closing it during cleanup", async () => { it("waits for a connecting socket to open before closing it during cleanup", async () => {
function Harness() { function Harness() {
useLiveRunTranscripts({ useLiveRunTranscripts({
@@ -143,6 +143,7 @@ export function useLiveRunTranscripts({
const trimmedSeqFloorByRunRef = useRef(new Map<string, number>()); const trimmedSeqFloorByRunRef = useRef(new Map<string, number>());
const pendingLogRowsByRunRef = useRef(new Map<string, string>()); const pendingLogRowsByRunRef = useRef(new Map<string, string>());
const logOffsetByRunRef = useRef(new Map<string, number>()); const logOffsetByRunRef = useRef(new Map<string, number>());
const logReadLimitByRunRef = useRef(new Map<string, number>());
const missingTerminalLogRunIdsRef = useRef(new Set<string>()); const missingTerminalLogRunIdsRef = useRef(new Set<string>());
// PAP-462 B3: buffered runs that dropped out of the `runs` list, mapped to the // PAP-462 B3: buffered runs that dropped out of the `runs` list, mapped to the
// wall-clock deadline (ms) after which their buffer may be pruned. A run still // wall-clock deadline (ms) after which their buffer may be pruned. A run still
@@ -269,6 +270,7 @@ export function useLiveRunTranscripts({
for (const runId of logOffsetByRunRef.current.keys()) { for (const runId of logOffsetByRunRef.current.keys()) {
if (!retainedRunIds.has(runId)) { if (!retainedRunIds.has(runId)) {
logOffsetByRunRef.current.delete(runId); logOffsetByRunRef.current.delete(runId);
logReadLimitByRunRef.current.delete(runId);
} }
} }
for (const runId of trimmedSeqFloorByRunRef.current.keys()) { for (const runId of trimmedSeqFloorByRunRef.current.keys()) {
@@ -312,6 +314,19 @@ export function useLiveRunTranscripts({
return; return;
} }
inFlightRunIds.add(run.id); inFlightRunIds.add(run.id);
const previousReadLimit = logReadLimitByRunRef.current.get(run.id);
if (previousReadLimit !== undefined && logReadLimitBytes > previousReadLimit) {
// Expanding a compact tail must reread the earlier range. Retain the
// visible chunks; merging by sequence deduplicates the overlapping tail.
const expandedOffset = resolveInitialLogOffset(run, logReadLimitBytes);
const previousOffset = logOffsetByRunRef.current.get(run.id);
if (previousOffset !== undefined && expandedOffset < previousOffset) {
logOffsetByRunRef.current.set(run.id, expandedOffset);
pendingLogRowsByRunRef.current.delete(run.id);
pendingLogRowsByRunRef.current.delete(`${run.id}:records`);
}
}
logReadLimitByRunRef.current.set(run.id, logReadLimitBytes);
const offset = logOffsetByRunRef.current.get(run.id) ?? resolveInitialLogOffset(run, logReadLimitBytes); const offset = logOffsetByRunRef.current.get(run.id) ?? resolveInitialLogOffset(run, logReadLimitBytes);
try { try {
const result = await readTranscriptRequest( const result = await readTranscriptRequest(
@@ -34,3 +34,39 @@ it.each([true, false])("retains a burst larger than the preview window (initiall
expect(new Set(current!.recent.data?.comments.map(c=>c.id)).size).toBe(empty ? 60 : 61); expect(new Set(current!.recent.data?.comments.map(c=>c.id)).size).toBe(empty ? 60 : 61);
} finally { await act(async () => root.unmount()); client.clear(); } } finally { await act(async () => root.unmount()); client.clear(); }
}); });
it.each([true, false])("catches up after canceled input and a reply burst (earlier anchor survives: %s)", async (earlierAnchorSurvives) => {
const client = new QueryClient({ defaultOptions: { queries: { retry: false } } });
const root = createRoot(document.createElement("div"));
const comment = (n: number) => ({ id: `comment-${n}`, createdAt: new Date(n * 1000), body: `Message ${n}` } as IssueComment);
let current: ReturnType<typeof useIssueHistoryWindow>;
function Probe() {
current = useIssueHistoryWindow("one", false);
return <span>{current.recent.data?.comments.length}</span>;
}
api.recentComments.mockResolvedValue({ comments: [comment(0), comment(1)], latestCommentId: "comment-1", hasMore: true });
api.listComments.mockReset();
const wait = () => act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); });
try {
await act(async () => root.render(<QueryClientProvider client={client}><Probe /></QueryClientProvider>));
for (let i = 0; i < 20 && !current!.recent.isSuccess; i++) await wait();
expect(current!.recent.isSuccess).toBe(true);
api.recentComments.mockResolvedValue({ comments: [58, 59, 60, 61].map(comment), latestCommentId: "comment-61", hasMore: true });
api.listComments.mockImplementation(async (_issue, options) => {
if (options.after === "comment-1") return []; // Canceled queued input.
if (options.after === "comment-0" && !earlierAnchorSurvives) return [];
const start = options.after === "comment-51" ? 52 : 2;
return Array.from({ length: Math.min(50, 62 - start) }, (_, i) => comment(start + i));
});
await act(async () => { await client.invalidateQueries({ queryKey: queryKeys.issues.comments("one") }); });
for (let i = 0; i < 20 && current!.recent.data?.latestCommentId !== "comment-61"; i++) await wait();
expect(current!.recent.error).toBeNull();
expect(current!.recent.data?.comments.filter(c => Number(c.id.split("-")[1]) >= 2).map(c => c.id))
.toEqual(Array.from({ length: 60 }, (_, i) => `comment-${i + 2}`));
expect(current!.recent.data?.latestCommentId).toBe("comment-61");
expect(api.listComments).toHaveBeenCalledTimes(earlierAnchorSurvives ? 3 : 4);
} finally {
await act(async () => root.unmount());
client.clear();
}
});
+14 -1
View File
@@ -46,6 +46,13 @@ export function useIssueHistoryWindow(
// saved cursor instead of dropping messages between two invalidations. // saved cursor instead of dropping messages between two invalidations.
let cursor = previous.latestCommentId; let cursor = previous.latestCommentId;
const incoming: IssueComment[] = []; const incoming: IssueComment[] = [];
// Canceling queued input deletes its comment, including a saved cursor.
// Retry from an earlier observed row, then from the start if that row
// was also deleted. Never advance to the preview past unread replies.
const recoveryAnchors = [
previous.comments.findLast((comment) => comment.id !== cursor)?.id ?? null,
null,
];
while (cursor !== snapshot.latestCommentId) { while (cursor !== snapshot.latestCommentId) {
const page = await issuesApi.listComments(issueRef!, { const page = await issuesApi.listComments(issueRef!, {
order: "asc", order: "asc",
@@ -53,7 +60,13 @@ export function useIssueHistoryWindow(
limit: 50, limit: 50,
}, { signal }); }, { signal });
if (signal.aborted) throw new DOMException("Aborted", "AbortError"); if (signal.aborted) throw new DOMException("Aborted", "AbortError");
if (!page.length) break; if (!page.length) {
if (cursor && recoveryAnchors.length) {
cursor = recoveryAnchors.shift() ?? null;
continue;
}
break;
}
incoming.push(...page); incoming.push(...page);
const next = page.at(-1)!.id; const next = page.at(-1)!.id;
if (next === cursor) break; if (next === cursor) break;