mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 19:35:13 +01:00
fix(tasks): recover compact history after cancellation and expansion
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9f8a6b399c
commit
971a762f80
5 files changed
+106
-7
No files matched your search
@@ -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();
|
||||||
|
}
|
||||||
|
});
|
||||||
@@ -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;
|
||||||
|
|||||||
Reference in new issue
Block a user