mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
feat: make chat-style tasks the default experience (#11101)
This commit is contained in:
1 parent
2da6a248c3
commit
815e49bb7c
80 files changed
+3858
-1497
No files matched your search
@@ -6,7 +6,7 @@ import { parseAcpxStdoutLine } from "@paperclipai/adapter-utils/acpx-engine/ui";
|
||||
import { buildTranscript, type RunLogChunk, type TranscriptEntry } from "../../adapters";
|
||||
import type { ToolRunDecision } from "@paperclipai/shared";
|
||||
import { ThemeProvider } from "../../context/ThemeContext";
|
||||
import { RunTranscriptView, normalizeTranscript } from "./RunTranscriptView";
|
||||
import { RunTranscriptView, keyTranscriptBlocks, normalizeTranscript } from "./RunTranscriptView";
|
||||
|
||||
describe("RunTranscriptView", () => {
|
||||
it("folds repeated tool_call status updates for the same toolUseId into one block", () => {
|
||||
@@ -322,4 +322,48 @@ describe("RunTranscriptView", () => {
|
||||
expect(html).not.toContain("line-250");
|
||||
expect(html).not.toContain("line-499");
|
||||
});
|
||||
|
||||
it("keeps the streaming tail block's key stable as deltas merge in", () => {
|
||||
// A streaming assistant message accumulates deltas; each delta advances the
|
||||
// block's `ts`. The React key must stay anchored to the block's opening
|
||||
// timestamp so the tail does not unmount/remount (restarting its fade).
|
||||
const firstDelta: TranscriptEntry[] = [
|
||||
{ kind: "assistant", ts: "2026-03-12T00:00:00.000Z", text: "Hel", delta: true },
|
||||
];
|
||||
const withMoreDeltas: TranscriptEntry[] = [
|
||||
...firstDelta,
|
||||
{ kind: "assistant", ts: "2026-03-12T00:00:00.400Z", text: "lo the", delta: true },
|
||||
{ kind: "assistant", ts: "2026-03-12T00:00:00.900Z", text: "re", delta: true },
|
||||
];
|
||||
|
||||
const keyOf = (entries: TranscriptEntry[]) => {
|
||||
const keyed = keyTranscriptBlocks(normalizeTranscript(entries, true));
|
||||
return keyed[keyed.length - 1]!.key;
|
||||
};
|
||||
|
||||
// Same opening timestamp → identical key even though `ts` advanced.
|
||||
expect(keyOf(withMoreDeltas)).toBe(keyOf(firstDelta));
|
||||
// And that key is anchored to the opening ts, not the latest one.
|
||||
expect(keyOf(withMoreDeltas)).toContain("2026-03-12T00:00:00.000Z");
|
||||
expect(keyOf(withMoreDeltas)).not.toContain("2026-03-12T00:00:00.900Z");
|
||||
});
|
||||
|
||||
it("assigns unique keys and does not remount earlier blocks when a new tail arrives", () => {
|
||||
const base: TranscriptEntry[] = [
|
||||
{ kind: "assistant", ts: "2026-03-12T00:00:00.000Z", text: "first message", delta: true },
|
||||
{ kind: "thinking", ts: "2026-03-12T00:00:01.000Z", text: "pondering", delta: true },
|
||||
];
|
||||
const withNewTail: TranscriptEntry[] = [
|
||||
...base,
|
||||
{ kind: "assistant", ts: "2026-03-12T00:00:02.000Z", text: "second message", delta: true },
|
||||
];
|
||||
|
||||
const baseKeys = keyTranscriptBlocks(normalizeTranscript(base, true)).map((b) => b.key);
|
||||
const nextKeys = keyTranscriptBlocks(normalizeTranscript(withNewTail, true)).map((b) => b.key);
|
||||
|
||||
// Keys are unique within a render.
|
||||
expect(new Set(nextKeys).size).toBe(nextKeys.length);
|
||||
// Existing blocks keep their identity when a new block appends after them.
|
||||
expect(nextKeys.slice(0, baseKeys.length)).toEqual(baseKeys);
|
||||
});
|
||||
});
|
||||
@@ -48,12 +48,18 @@ type TranscriptBlock =
|
||||
type: "message";
|
||||
role: "assistant" | "user";
|
||||
ts: string;
|
||||
// Timestamp of the first entry that opened this block. `ts` tracks the
|
||||
// latest merged delta and mutates every chunk; `startTs` stays fixed so
|
||||
// the React key is stable and the streaming block does not remount (and
|
||||
// restart its fade) on each delta.
|
||||
startTs: string;
|
||||
text: string;
|
||||
streaming: boolean;
|
||||
}
|
||||
| {
|
||||
type: "thinking";
|
||||
ts: string;
|
||||
startTs: string;
|
||||
text: string;
|
||||
streaming: boolean;
|
||||
}
|
||||
@@ -73,6 +79,7 @@ type TranscriptBlock =
|
||||
| {
|
||||
type: "activity";
|
||||
ts: string;
|
||||
startTs: string;
|
||||
activityId?: string;
|
||||
name: string;
|
||||
status: "running" | "completed";
|
||||
@@ -125,6 +132,7 @@ type TranscriptBlock =
|
||||
| {
|
||||
type: "stdout";
|
||||
ts: string;
|
||||
startTs: string;
|
||||
text: string;
|
||||
}
|
||||
| {
|
||||
@@ -550,6 +558,7 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
||||
type: "message",
|
||||
role: entry.kind,
|
||||
ts: entry.ts,
|
||||
startTs: entry.ts,
|
||||
text: entry.text,
|
||||
streaming: isStreaming,
|
||||
});
|
||||
@@ -567,6 +576,7 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
||||
blocks.push({
|
||||
type: "thinking",
|
||||
ts: entry.ts,
|
||||
startTs: entry.ts,
|
||||
text: entry.text,
|
||||
streaming: isStreaming,
|
||||
});
|
||||
@@ -691,6 +701,7 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
||||
const block: Extract<TranscriptBlock, { type: "activity" }> = {
|
||||
type: "activity",
|
||||
ts: entry.ts,
|
||||
startTs: entry.ts,
|
||||
activityId: activity.activityId,
|
||||
name: activity.name,
|
||||
status: activity.status,
|
||||
@@ -758,6 +769,7 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
||||
blocks.push({
|
||||
type: "stdout",
|
||||
ts: entry.ts,
|
||||
startTs: entry.ts,
|
||||
text: entry.text,
|
||||
});
|
||||
}
|
||||
@@ -766,6 +778,55 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
||||
return groupToolBlocks(groupCommandBlocks(blocks));
|
||||
}
|
||||
|
||||
/**
|
||||
* Stable identity for a block's React key. Anchored to the block's opening
|
||||
* timestamp (`startTs`) or a durable id (`toolUseId`, `activityId`) rather than
|
||||
* its latest `ts`, which mutates on every streamed delta. A mutating key
|
||||
* remounts the block, restarting its 300ms fade-in so the text visibly blinks
|
||||
* out and back each chunk; a stable one keeps the streaming tail mounted.
|
||||
*/
|
||||
function transcriptBlockIdentity(block: TranscriptBlock): string {
|
||||
switch (block.type) {
|
||||
case "message":
|
||||
return `message:${block.role}:${block.startTs}`;
|
||||
case "thinking":
|
||||
return `thinking:${block.startTs}`;
|
||||
case "stdout":
|
||||
return `stdout:${block.startTs}`;
|
||||
case "activity":
|
||||
return `activity:${block.activityId ?? block.startTs}`;
|
||||
case "tool":
|
||||
return `tool:${block.toolUseId ?? block.ts}`;
|
||||
case "command_group":
|
||||
return `command_group:${block.ts}`;
|
||||
case "tool_group":
|
||||
return `tool_group:${block.ts}`;
|
||||
case "stderr_group":
|
||||
return `stderr_group:${block.ts}`;
|
||||
case "system_group":
|
||||
return `system_group:${block.ts}`;
|
||||
case "diff_group":
|
||||
return `diff_group:${block.ts}`;
|
||||
case "event":
|
||||
return `event:${block.label}:${block.ts}`;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Assign each block a stable, unique React key. Identity is position-independent
|
||||
* so earlier blocks collapsing (truncation) does not remount the survivors; a
|
||||
* per-render occurrence counter disambiguates the rare identity collision.
|
||||
*/
|
||||
export function keyTranscriptBlocks(blocks: TranscriptBlock[]): Array<{ block: TranscriptBlock; key: string }> {
|
||||
const seen = new Map<string, number>();
|
||||
return blocks.map((block) => {
|
||||
const identity = transcriptBlockIdentity(block);
|
||||
const occurrence = seen.get(identity) ?? 0;
|
||||
seen.set(identity, occurrence + 1);
|
||||
return { block, key: occurrence === 0 ? identity : `${identity}#${occurrence}` };
|
||||
});
|
||||
}
|
||||
|
||||
function TranscriptMessageBlock({
|
||||
block,
|
||||
density,
|
||||
@@ -789,7 +850,9 @@ function TranscriptMessageBlock({
|
||||
<MarkdownBody
|
||||
className={cn(
|
||||
"[&>*:first-child]:mt-0 [&>*:last-child]:mb-0",
|
||||
compact ? "text-xs leading-5 text-foreground/85" : "text-sm",
|
||||
// Match the default view's chat message body (IssueChatThread:
|
||||
// `text-sm leading-6`) so streamed text reads identically (PAP-461, A2).
|
||||
compact ? "text-xs leading-5 text-foreground/85" : "text-sm leading-6",
|
||||
)}
|
||||
externalReferences={externalReferences}
|
||||
>
|
||||
@@ -798,7 +861,7 @@ function TranscriptMessageBlock({
|
||||
{block.streaming && (
|
||||
<div className="mt-2 inline-flex items-center gap-1 text-(length:--text-nano) font-medium italic text-muted-foreground">
|
||||
<span className="relative flex h-1.5 w-1.5">
|
||||
<span className="absolute inline-flex h-full w-full animate-ping rounded-full bg-current opacity-70" />
|
||||
<span className="tc-live-ping absolute inline-flex h-full w-full rounded-full bg-current opacity-70" />
|
||||
<span className="relative inline-flex h-1.5 w-1.5 rounded-full bg-current" />
|
||||
</span>
|
||||
Streaming
|
||||
@@ -822,8 +885,11 @@ function TranscriptThinkingBlock({
|
||||
return (
|
||||
<MarkdownBody
|
||||
className={cn(
|
||||
"italic text-foreground/70 [&>*:first-child]:mt-0 [&>*:last-child]:mb-0",
|
||||
density === "compact" ? "text-(length:--text-micro) leading-5" : "text-sm leading-6",
|
||||
// Match the default view's chain-of-thought text (IssueChatThread:
|
||||
// `text-(length:--text-compact) italic leading-5 text-muted-foreground/70`)
|
||||
// so streamed thinking reads identically across both views (PAP-461, A2).
|
||||
"italic text-muted-foreground/70 [&>*:first-child]:mt-0 [&>*:last-child]:mb-0",
|
||||
density === "compact" ? "text-(length:--text-micro) leading-5" : "text-(length:--text-compact) leading-5",
|
||||
className,
|
||||
)}
|
||||
externalReferences={externalReferences}
|
||||
@@ -1304,7 +1370,7 @@ function TranscriptActivityRow({
|
||||
<Check className="mt-0.5 h-3.5 w-3.5 shrink-0 text-emerald-600 dark:text-emerald-300" />
|
||||
) : (
|
||||
<span className="relative mt-1 flex h-2.5 w-2.5 shrink-0">
|
||||
<span className="absolute inline-flex h-full w-full animate-ping rounded-full bg-blue-400 opacity-70" />
|
||||
<span className="tc-live-ping absolute inline-flex h-full w-full rounded-full bg-blue-400 opacity-70" />
|
||||
<span className="relative inline-flex h-2.5 w-2.5 rounded-full bg-blue-500" />
|
||||
</span>
|
||||
)}
|
||||
@@ -1716,6 +1782,7 @@ export function RunTranscriptView({
|
||||
[entries, mode, streaming],
|
||||
);
|
||||
const visibleBlocks = limit ? blocks.slice(-limit) : blocks;
|
||||
const keyedBlocks = useMemo(() => keyTranscriptBlocks(visibleBlocks), [visibleBlocks]);
|
||||
const visibleEntries = limit ? entries.slice(-limit) : entries;
|
||||
|
||||
if (entries.length === 0) {
|
||||
@@ -1736,10 +1803,10 @@ export function RunTranscriptView({
|
||||
|
||||
return (
|
||||
<div className={cn("space-y-3", className)}>
|
||||
{visibleBlocks.map((block, index) => (
|
||||
{keyedBlocks.map(({ block, key }, index) => (
|
||||
<div
|
||||
key={`${block.type}-${block.ts}-${index}`}
|
||||
className={cn(index === visibleBlocks.length - 1 && streaming && "animate-in fade-in slide-in-from-bottom-1 duration-300")}
|
||||
key={key}
|
||||
className={cn(index === keyedBlocks.length - 1 && streaming && "tc-stream-block-enter")}
|
||||
>
|
||||
{block.type === "message" && (
|
||||
<TranscriptMessageBlock
|
||||
|
||||
@@ -515,4 +515,112 @@ describe("useLiveRunTranscripts", () => {
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("retains an accumulated buffer through a transient empty poll (PAP-462 B3)", async () => {
|
||||
const ts = "2026-08-08T00:00:00.000Z";
|
||||
const row = JSON.stringify({
|
||||
ts,
|
||||
stream: "stdout",
|
||||
chunk: '{"type":"acpx.text_delta","text":"hello"}\n',
|
||||
seq: 1,
|
||||
});
|
||||
// Serve the buffered chunk on the FIRST persisted-log read only; every later
|
||||
// read (including after the run reappears) returns nothing new. So the chunk
|
||||
// can only be present the second time if the buffer survived the gap.
|
||||
logMock.mockResolvedValue({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 100 });
|
||||
logMock.mockResolvedValueOnce({ runId: "run-1", store: "memory", logRef: "log-1", content: `${row}\n`, nextOffset: 100 });
|
||||
|
||||
const captured: { value: ReturnType<typeof useLiveRunTranscripts> | null } = { value: null };
|
||||
function Harness({ runs }: { runs: Array<{ id: string; status: string; adapterType: string }> }) {
|
||||
captured.value = useLiveRunTranscripts({ companyId: "company-1", runs, enableRealtimeUpdates: false });
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
const runList = [{ id: "run-1", status: "running", adapterType: "gemini_local" }];
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={runList} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(captured.value?.transcriptByRun.get("run-1")).toHaveLength(1);
|
||||
|
||||
// Transient empty poll: run momentarily absent from the list.
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={[]} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
// Run reappears within the grace window — the buffer must survive rather than
|
||||
// re-hydrate from a (now empty) truncated read.
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={runList} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(captured.value?.transcriptByRun.get("run-1")).toHaveLength(1);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("prunes a buffer once a run stays absent past the grace window (PAP-462 B3)", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const ts = "2026-08-08T00:00:00.000Z";
|
||||
const row = JSON.stringify({
|
||||
ts,
|
||||
stream: "stdout",
|
||||
chunk: '{"type":"acpx.text_delta","text":"hello"}\n',
|
||||
seq: 1,
|
||||
});
|
||||
logMock.mockResolvedValue({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 100 });
|
||||
logMock.mockResolvedValueOnce({ runId: "run-1", store: "memory", logRef: "log-1", content: `${row}\n`, nextOffset: 100 });
|
||||
|
||||
const captured: { value: ReturnType<typeof useLiveRunTranscripts> | null } = { value: null };
|
||||
function Harness({ runs }: { runs: Array<{ id: string; status: string; adapterType: string }> }) {
|
||||
captured.value = useLiveRunTranscripts({ companyId: "company-1", runs, enableRealtimeUpdates: false });
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
const runList = [{ id: "run-1", status: "running", adapterType: "gemini_local" }];
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={runList} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(captured.value?.transcriptByRun.get("run-1")).toHaveLength(1);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={[]} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
// Stay absent long enough for the grace window to lapse and the deferred
|
||||
// prune to fire.
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(25_000);
|
||||
});
|
||||
|
||||
// On reappear the buffer is gone, so the (now empty) read rebuilds nothing.
|
||||
await act(async () => {
|
||||
root.render(<Harness runs={runList} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(captured.value?.transcriptByRun.get("run-1") ?? []).toHaveLength(0);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
mergeRunLogChunks,
|
||||
parsePersistedLogContent,
|
||||
readChunkSeq,
|
||||
type ChunkRetentionBudget,
|
||||
} from "../../lib/run-log-chunks";
|
||||
|
||||
// TODO(perf): this whole hook polls the log/runs endpoints on an interval. The
|
||||
@@ -24,6 +25,20 @@ const LOG_READ_LIMIT_BYTES = 256_000;
|
||||
// gaps and reconnects instead of polling every couple of seconds.
|
||||
const REALTIME_FALLBACK_POLL_INTERVAL_MS = 30_000;
|
||||
const EMPTY_RUN_LOG_CHUNKS: RunLogChunk[] = [];
|
||||
// Retained transcript payload budget for full task views. A byte budget (rather
|
||||
// than a tiny chunk count) keeps the whole streamed scrollback intact — a
|
||||
// delta-streaming run emits thousands of one-token chunks in seconds, and the
|
||||
// old 200-chunk cap discarded just-rendered messages off the top irreversibly.
|
||||
// If a run genuinely exceeds this, the oldest output collapses behind a visible
|
||||
// marker instead of vanishing (see `applyRetentionBudget`).
|
||||
const TASK_VIEW_MAX_BYTES_PER_RUN = 2_000_000;
|
||||
// Grace period before an accumulated transcript buffer is pruned for a run that
|
||||
// has vanished from the `runs` list. The parent refetches runs on its own
|
||||
// interval, and a single transient empty/errored poll would otherwise wipe the
|
||||
// buffer — forcing a rehydration that skips to the last `LOG_READ_LIMIT_BYTES`
|
||||
// and silently drops already-rendered scrollback (PAP-462 B3). A run that is
|
||||
// genuinely gone stays absent past this window and is then pruned as before.
|
||||
const RUN_ABSENCE_PRUNE_GRACE_MS = 20_000;
|
||||
|
||||
export interface RunTranscriptSource {
|
||||
id: string;
|
||||
@@ -37,7 +52,13 @@ export interface RunTranscriptSource {
|
||||
interface UseLiveRunTranscriptsOptions {
|
||||
runs: RunTranscriptSource[];
|
||||
companyId?: string | null;
|
||||
/**
|
||||
* Compact chunk-count cap for ticker-style consumers (dashboard). When set,
|
||||
* trimming is silent — the historical behavior. Full task views omit this and
|
||||
* use the byte budget below instead.
|
||||
*/
|
||||
maxChunksPerRun?: number;
|
||||
maxBytesPerRun?: number;
|
||||
logPollIntervalMs?: number;
|
||||
logReadLimitBytes?: number;
|
||||
enableRealtimeUpdates?: boolean;
|
||||
@@ -67,11 +88,21 @@ export function resolveInitialLogOffset(run: RunTranscriptSource, limitBytes: nu
|
||||
export function useLiveRunTranscripts({
|
||||
runs,
|
||||
companyId,
|
||||
maxChunksPerRun = 200,
|
||||
maxChunksPerRun,
|
||||
maxBytesPerRun = TASK_VIEW_MAX_BYTES_PER_RUN,
|
||||
logPollIntervalMs = LOG_POLL_INTERVAL_MS,
|
||||
logReadLimitBytes = LOG_READ_LIMIT_BYTES,
|
||||
enableRealtimeUpdates = true,
|
||||
}: UseLiveRunTranscriptsOptions) {
|
||||
// Ticker consumers opt into the silent chunk-count cap; full task views use a
|
||||
// byte budget that collapses (not discards) the oldest output when exceeded.
|
||||
const retentionBudget: ChunkRetentionBudget = useMemo(
|
||||
() =>
|
||||
typeof maxChunksPerRun === "number"
|
||||
? { maxChunks: maxChunksPerRun }
|
||||
: { maxBytes: maxBytesPerRun, collapseTrimmed: true },
|
||||
[maxChunksPerRun, maxBytesPerRun],
|
||||
);
|
||||
const runsKey = useMemo(
|
||||
() =>
|
||||
runs
|
||||
@@ -95,6 +126,14 @@ export function useLiveRunTranscripts({
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
const missingTerminalLogRunIdsRef = useRef(new Set<string>());
|
||||
// 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
|
||||
// inside its grace window is retained across the empty poll; `pruneTick` fires
|
||||
// the effect again once the nearest deadline elapses so a run that stays gone
|
||||
// is eventually cleaned up even if the `runs` list never changes again.
|
||||
const absenceDeadlineByRunRef = useRef(new Map<string, number>());
|
||||
const prevKnownRunIdsRef = useRef(new Set<string>());
|
||||
const [pruneTick, setPruneTick] = useState(0);
|
||||
const transcriptCacheRef = useRef(new Map<string, {
|
||||
adapterType: string;
|
||||
chunks: RunLogChunk[];
|
||||
@@ -134,7 +173,7 @@ export function useLiveRunTranscripts({
|
||||
seenChunkKeys: seenChunkKeysRef.current,
|
||||
trimmedSeqFloorByRun: trimmedSeqFloorByRunRef.current,
|
||||
},
|
||||
maxChunksPerRun,
|
||||
retentionBudget,
|
||||
);
|
||||
if (!changed) return prev;
|
||||
const next = new Map(prev);
|
||||
@@ -145,10 +184,40 @@ export function useLiveRunTranscripts({
|
||||
|
||||
useEffect(() => {
|
||||
const knownRunIds = new Set(normalizedRuns.map((run) => run.id));
|
||||
const now = Date.now();
|
||||
const deadlines = absenceDeadlineByRunRef.current;
|
||||
|
||||
// PAP-462 B3: a run that just disappeared from the list starts its grace
|
||||
// clock; one that reappeared clears any pending deadline. Comparing against
|
||||
// the previous known set means a transient empty poll only *arms* the timer
|
||||
// rather than pruning the buffer outright.
|
||||
for (const runId of prevKnownRunIdsRef.current) {
|
||||
if (!knownRunIds.has(runId) && !deadlines.has(runId)) {
|
||||
deadlines.set(runId, now + RUN_ABSENCE_PRUNE_GRACE_MS);
|
||||
}
|
||||
}
|
||||
for (const runId of knownRunIds) {
|
||||
deadlines.delete(runId);
|
||||
}
|
||||
prevKnownRunIdsRef.current = knownRunIds;
|
||||
|
||||
// Retain known runs plus any absent run still inside its grace window; only
|
||||
// runs absent past their deadline are actually pruned.
|
||||
const retainedRunIds = new Set(knownRunIds);
|
||||
let soonestExpiryMs = Number.POSITIVE_INFINITY;
|
||||
for (const [runId, deadline] of deadlines) {
|
||||
if (deadline > now) {
|
||||
retainedRunIds.add(runId);
|
||||
soonestExpiryMs = Math.min(soonestExpiryMs, deadline);
|
||||
} else {
|
||||
deadlines.delete(runId);
|
||||
}
|
||||
}
|
||||
|
||||
setChunksByRun((prev) => {
|
||||
const next = new Map<string, RunLogChunk[]>();
|
||||
for (const [runId, chunks] of prev) {
|
||||
if (knownRunIds.has(runId)) {
|
||||
if (retainedRunIds.has(runId)) {
|
||||
next.set(runId, chunks);
|
||||
}
|
||||
}
|
||||
@@ -157,7 +226,7 @@ export function useLiveRunTranscripts({
|
||||
setHydratedRunIds((prev) => {
|
||||
const next = new Set<string>();
|
||||
for (const runId of prev) {
|
||||
if (knownRunIds.has(runId)) {
|
||||
if (retainedRunIds.has(runId)) {
|
||||
next.add(runId);
|
||||
}
|
||||
}
|
||||
@@ -166,31 +235,41 @@ export function useLiveRunTranscripts({
|
||||
|
||||
for (const key of pendingLogRowsByRunRef.current.keys()) {
|
||||
const runId = key.replace(/:records$/, "");
|
||||
if (!knownRunIds.has(runId)) {
|
||||
if (!retainedRunIds.has(runId)) {
|
||||
pendingLogRowsByRunRef.current.delete(key);
|
||||
}
|
||||
}
|
||||
for (const runId of logOffsetByRunRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
if (!retainedRunIds.has(runId)) {
|
||||
logOffsetByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of trimmedSeqFloorByRunRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
if (!retainedRunIds.has(runId)) {
|
||||
trimmedSeqFloorByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of missingTerminalLogRunIdsRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
if (!retainedRunIds.has(runId)) {
|
||||
missingTerminalLogRunIdsRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of transcriptCacheRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
if (!retainedRunIds.has(runId)) {
|
||||
transcriptCacheRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
}, [normalizedRuns]);
|
||||
|
||||
// Re-run once the nearest grace window elapses so a run that stays gone is
|
||||
// pruned even if `normalizedRuns` never changes again.
|
||||
if (soonestExpiryMs !== Number.POSITIVE_INFINITY) {
|
||||
const timer = window.setTimeout(
|
||||
() => setPruneTick((tick) => tick + 1),
|
||||
Math.max(0, soonestExpiryMs - now) + 50,
|
||||
);
|
||||
return () => window.clearTimeout(timer);
|
||||
}
|
||||
}, [normalizedRuns, pruneTick]);
|
||||
|
||||
useEffect(() => {
|
||||
if (normalizedRuns.length === 0) return;
|
||||
|
||||
Reference in new issue
Block a user