Fix live run streaming text readability (#9330)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The issue thread UI renders live agent output from adapter run logs and transcript parsing. > - Some adapter streams emit many small or repeated token chunks, and live UI updates can expose partial words, duplicated slices, or transient markdown placeholders. > - That makes active run updates look like gibberish even when the underlying agent output is valid. > - The fix needs to preserve raw logs while making the live thread view stable, readable, and ordered. > - This pull request adds monotonic run-log sequencing, safer live transcript dedupe/order handling, markdown placeholder hiding, and readable live text stabilization. > - The benefit is a live issue thread that updates smoothly without showing confusing partial parser artifacts. ## Linked Issues or Issue Description No public GitHub issue exists yet, so this PR includes the bug details inline. ### What happened? Live run updates in the issue thread can show confusing repeated or partial text while an adapter is streaming. The visible text appears to lose parsing boundaries during active updates, especially with ACP-style token deltas, so the live output can briefly render duplicated chunks, incomplete words, or HTML-comment placeholders. ### Expected behavior Live text should remain readable while preserving the underlying run output for raw inspection. ### Steps to reproduce 1. Start a live agent run whose adapter emits small stdout token deltas. 2. Watch the issue thread while the run is still active. 3. Observe transient duplicated chunks, incomplete words, or markdown placeholder artifacts in the live rendered text. ### Paperclip version or commit Reproduced against current `master` before this PR branch. ### Deployment mode Local dev issue-thread UI with live local adapter runs. ### Additional context GitHub PR search for `live run streaming text markdown transcript` found one broad merged PR, `#252` (“Dotta updates - sorry it's so large”), but no targeted duplicate for this live streaming readability bug. ## What Changed - Added per-run monotonic sequence numbers to persisted and live run-log chunks. - Dedupe and order live transcript chunks by sequence before falling back to timestamp ordering. - Hide markdown HTML comment placeholder text from rendered markdown output. - Smooth live issue-thread text updates so partial additions reveal at readable word boundaries and sliding-window removals do not produce gibberish. - Added coverage for run-log ordering/deduping, markdown comment hiding, live issue-thread stabilization, and Greptile-reviewed edge cases where overlap rewrites could synthesize text or no-boundary additions could stay hidden. ## Verification - `pnpm --filter @paperclipai/ui exec vitest run src/lib/issue-chat-messages.test.ts src/components/MarkdownBody.test.tsx src/components/transcript/useLiveRunTranscripts.test.tsx` passed before the review fix: 3 files, 86 tests. - `pnpm --filter @paperclipai/ui exec vitest run src/lib/issue-chat-messages.test.ts` passed after the review fix: 1 file, 30 tests. - `pnpm --filter @paperclipai/ui exec vitest run src/lib/issue-chat-messages.test.ts src/components/transcript/useLiveRunTranscripts.test.tsx` passed after the final Greptile overlap fix: 2 files, 40 tests. - `pnpm check:token-gates` passed. - Local PII/secret scan of touched files found only expected code/test words such as `secret`, `token`, and redaction-related strings; no literal credentials found. - `pnpm -r typecheck` passed after restoring declared dependencies with `CI=1 pnpm install --frozen-lockfile` and running with a short `TMPDIR` because `tsx` IPC sockets fail under the long sandbox temp path. - `pnpm build` passed with existing Vite CSS/font/chunk warnings. - GitHub PR checks passed on head `4c052dfe86aecb5feb73504e6b48843f68fce813`: build, typecheck/release registry, server and workspace test shards, serialized server suites, e2e, canary dry run, policy, review, Socket, Superagent, Snyk, and verify. - Greptile review passed on head `4c052dfe86aecb5feb73504e6b48843f68fce813` with confidence score 5/5 and no blocking issues found. - `pnpm test:run` failed in unrelated server workspace tests on this macOS local environment: - `server/src/__tests__/heartbeat-workspace-branch-containment.test.ts`: two assertions compare `/tmp/...` with `/private/tmp/...`. - `server/src/__tests__/heartbeat-worktree-suppression.test.ts`: expected one heartbeat run but observed two, followed by cleanup fallout in the full run. - Isolated rerun of those two server suites reproduced the same three failures. ## Risks - Low product risk for the UI changes: the readable smoothing only affects active live-run display stabilization, not stored comments or raw run logs. - Moderate verification risk: local full Vitest did not pass because of unrelated server workspace tests. Targeted tests for this change, typecheck, token gates, and build passed. - Run-log sequence fields are optional for compatibility with older log rows that do not include `seq`. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, GPT-5-based coding agent, tool-using local workspace execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I searched the GitHub PR list for similar PRs and confirmed this is not a duplicate - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Devin Foley
GitHub
parent
991279f52c
commit
a4993a72a6
|
|
@ -12306,19 +12306,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk);
|
||||
const ts = new Date().toISOString();
|
||||
|
||||
outputSeq += 1;
|
||||
const chunkSeq = outputSeq;
|
||||
let appendedBytes = 0;
|
||||
if (handle) {
|
||||
appendedBytes = await runLogStore.append(handle, {
|
||||
stream,
|
||||
chunk: sanitizedChunk,
|
||||
ts,
|
||||
seq: chunkSeq,
|
||||
});
|
||||
persistedLogBytes += appendedBytes;
|
||||
}
|
||||
outputSeq += 1;
|
||||
outputProgressState.pending = {
|
||||
at: new Date(ts),
|
||||
seq: outputSeq,
|
||||
seq: chunkSeq,
|
||||
stream,
|
||||
bytes: persistedLogBytes,
|
||||
};
|
||||
|
|
@ -12359,6 +12361,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
agentId: run.agentId,
|
||||
issueId,
|
||||
ts,
|
||||
seq: chunkSeq,
|
||||
stream,
|
||||
chunk: payloadChunk,
|
||||
truncated: payloadChunk.length !== sanitizedChunk.length,
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ export interface RunLogStore {
|
|||
begin(input: { companyId: string; agentId: string; runId: string }): Promise<RunLogHandle>;
|
||||
append(
|
||||
handle: RunLogHandle,
|
||||
event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string },
|
||||
event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string; seq?: number },
|
||||
): Promise<number>;
|
||||
finalize(handle: RunLogHandle): Promise<RunLogFinalizeSummary>;
|
||||
read(handle: RunLogHandle, opts?: RunLogReadOptions): Promise<RunLogReadResult>;
|
||||
|
|
@ -113,6 +113,10 @@ function createLocalFileRunLogStore(basePath: string): RunLogStore {
|
|||
ts: event.ts,
|
||||
stream: event.stream,
|
||||
chunk: event.chunk,
|
||||
// Monotonic per-run sequence so readers can dedupe and order records
|
||||
// even when several identical chunks share the same millisecond ts
|
||||
// (common for ACP-style token deltas).
|
||||
...(typeof event.seq === "number" && Number.isFinite(event.seq) ? { seq: event.seq } : {}),
|
||||
});
|
||||
const persisted = `${line}\n`;
|
||||
await fs.appendFile(absPath, persisted, "utf8");
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { redactHomePathUserSegments, redactTranscriptEntryPaths } from "@paperclipai/adapter-utils";
|
||||
import type { TranscriptEntry, StdoutLineParser, TranscriptParserSource } from "./types";
|
||||
|
||||
export type RunLogChunk = { ts: string; stream: "stdout" | "stderr" | "system"; chunk: string };
|
||||
export type RunLogChunk = { ts: string; stream: "stdout" | "stderr" | "system"; chunk: string; seq?: number };
|
||||
type TranscriptBuildOptions = { censorUsernameInLogs?: boolean };
|
||||
type RedactionOptions = { enabled: boolean };
|
||||
|
||||
|
|
|
|||
|
|
@ -158,6 +158,46 @@ describe("MarkdownBody", () => {
|
|||
expect(html).toContain("Plain text");
|
||||
});
|
||||
|
||||
it("hides markdown HTML comments instead of rendering placeholder text", () => {
|
||||
const html = renderMarkdown("Before\n\n<!-- -->\n\nAfter");
|
||||
|
||||
expect(html).toContain("Before");
|
||||
expect(html).toContain("After");
|
||||
expect(html).not.toContain("<!--");
|
||||
expect(html).not.toContain("-->");
|
||||
});
|
||||
|
||||
it("hides escaped HTML comment placeholders before attachment images", () => {
|
||||
const html = renderMarkdown("\\<!-- --> ");
|
||||
|
||||
expect(html).toContain('<img src="/api/attachments/57d0805a-1b95-4fa5-abb4-d0c33e2e649c/content" alt=""/>');
|
||||
expect(html).not.toContain("<!--");
|
||||
expect(html).not.toContain("-->");
|
||||
});
|
||||
|
||||
it("hides incomplete streamed HTML comment placeholders before attachment images", () => {
|
||||
const html = renderMarkdown("\\<!-- ");
|
||||
|
||||
expect(html).toContain('<img src="/api/attachments/57d0805a-1b95-4fa5-abb4-d0c33e2e649c/content" alt=""/>');
|
||||
expect(html).not.toContain("<!--");
|
||||
});
|
||||
|
||||
it("hides incomplete encoded HTML comment placeholders", () => {
|
||||
const html = renderMarkdown("<!-- -- ");
|
||||
|
||||
expect(html).toContain('<img src="/api/attachments/57d0805a-1b95-4fa5-abb4-d0c33e2e649c/content" alt=""/>');
|
||||
expect(html).not.toContain("<!--");
|
||||
expect(html).not.toContain("&lt;!--");
|
||||
});
|
||||
|
||||
it("keeps HTML comment markers when they are literal code content", () => {
|
||||
const inlineHtml = renderMarkdown("Use `<!-- -->` as a literal.");
|
||||
const blockHtml = renderMarkdown("```html\n<!-- keep this example -->\n```");
|
||||
|
||||
expect(inlineHtml).toContain("<!-- -->");
|
||||
expect(blockHtml).toContain("<!-- keep this example -->");
|
||||
});
|
||||
|
||||
it("uses soft-break styling by default", () => {
|
||||
const html = renderMarkdown("First line\nSecond line");
|
||||
|
||||
|
|
|
|||
|
|
@ -247,6 +247,30 @@ const tableCellWrapStyle: React.CSSProperties = {
|
|||
wordBreak: "normal",
|
||||
};
|
||||
|
||||
function isHtmlCommentNode(node: MarkdownAstNode) {
|
||||
return node.type === "html" && typeof node.value === "string" && /^<!--[\s\S]*-->$/.test(node.value.trim());
|
||||
}
|
||||
|
||||
function isEscapedHtmlCommentPlaceholder(node: MarkdownAstNode) {
|
||||
if (node.type !== "text" || typeof node.value !== "string") return false;
|
||||
const value = node.value.trim();
|
||||
return /^\\?<!--(?:\s*-{0,2}>?)?$/.test(value) || /^<!--(?:\s*-{0,2}(?:>)?)?$/.test(value);
|
||||
}
|
||||
|
||||
function remarkDropHtmlComments() {
|
||||
return (tree: MarkdownAstNode) => {
|
||||
const visit = (node: MarkdownAstNode) => {
|
||||
const children = node.children;
|
||||
if (!children) return;
|
||||
node.children = children.filter((child) => !isHtmlCommentNode(child) && !isEscapedHtmlCommentPlaceholder(child));
|
||||
for (const child of node.children) {
|
||||
visit(child);
|
||||
}
|
||||
};
|
||||
visit(tree);
|
||||
};
|
||||
}
|
||||
|
||||
function mergeWrapStyle(style?: React.CSSProperties): React.CSSProperties {
|
||||
return {
|
||||
...wrapAnywhereStyle,
|
||||
|
|
@ -726,7 +750,7 @@ function MarkdownBodyImpl({
|
|||
// parent re-renders frequently (see PAP-10767). Memoize both so re-renders
|
||||
// that don't change the inputs are cheap and non-destructive.
|
||||
const remarkPlugins = useMemo<NonNullable<Options["remarkPlugins"]>>(() => {
|
||||
const plugins: NonNullable<Options["remarkPlugins"]> = [remarkGfm];
|
||||
const plugins: NonNullable<Options["remarkPlugins"]> = [remarkGfm, remarkDropHtmlComments];
|
||||
if (enableWikiLinks) {
|
||||
plugins.push(createRemarkWikiLinks({ wikiLinkRoot, resolveWikiLinkHref }));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -286,6 +286,178 @@ describe("useLiveRunTranscripts", () => {
|
|||
container.remove();
|
||||
});
|
||||
|
||||
it("keeps identical same-timestamp log records that carry distinct seq values", async () => {
|
||||
const ts = "2026-04-20T00:00:00.000Z";
|
||||
const tokenRow = (seq: number) =>
|
||||
JSON.stringify({ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq });
|
||||
logMock.mockImplementationOnce(async () => ({
|
||||
runId: "run-1",
|
||||
store: "memory",
|
||||
logRef: "log-1",
|
||||
content: `${tokenRow(1)}\n${tokenRow(2)}\n${tokenRow(3)}\n`,
|
||||
nextOffset: 300,
|
||||
}));
|
||||
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-1", status: "running", adapterType: "gemini_local" }],
|
||||
enableRealtimeUpdates: false,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
const lastCall = buildTranscriptMock.mock.calls.at(-1) as unknown[] | undefined;
|
||||
expect(lastCall?.[0]).toEqual([
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: 1 },
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: 2 },
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: 3 },
|
||||
]);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("keeps repeated unsequenced structured text deltas instead of content-deduping tokens", async () => {
|
||||
const ts = "2026-04-20T00:00:00.000Z";
|
||||
const tokenRow = JSON.stringify({
|
||||
ts,
|
||||
stream: "stdout",
|
||||
chunk: '{"type":"acpx.text_delta","text":" the"}\n',
|
||||
});
|
||||
logMock.mockImplementationOnce(async () => ({
|
||||
runId: "run-1",
|
||||
store: "memory",
|
||||
logRef: "log-1",
|
||||
content: `${tokenRow}\n${tokenRow}\n${tokenRow}\n`,
|
||||
nextOffset: 300,
|
||||
}));
|
||||
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-1", status: "running", adapterType: "gemini_local" }],
|
||||
enableRealtimeUpdates: false,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
const lastCall = buildTranscriptMock.mock.calls.at(-1) as unknown[] | undefined;
|
||||
expect(lastCall?.[0]).toEqual([
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: undefined },
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: undefined },
|
||||
{ ts, stream: "stdout", chunk: '{"type":"acpx.text_delta","text":" the"}\n', seq: undefined },
|
||||
]);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("orders and dedupes sequenced chunks across websocket and persisted-log delivery", async () => {
|
||||
type RunLogResult = { runId: string; store: string; logRef: string; content: string; nextOffset: number };
|
||||
let resolveLog: ((value: RunLogResult) => void) | null = null;
|
||||
logMock.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise<RunLogResult>((resolve) => {
|
||||
resolveLog = resolve;
|
||||
}),
|
||||
);
|
||||
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-1", status: "running", adapterType: "gemini_local" }],
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
const socket = FakeWebSocket.instances[0]!;
|
||||
|
||||
const sendLogEvent = (seq: number, chunk: string) => {
|
||||
socket.onmessage?.(
|
||||
new MessageEvent("message", {
|
||||
data: JSON.stringify({
|
||||
companyId: "company-1",
|
||||
type: "heartbeat.run.log",
|
||||
createdAt: "2026-04-20T00:00:01.000Z",
|
||||
payload: {
|
||||
runId: "run-1",
|
||||
ts: "2026-04-20T00:00:01.000Z",
|
||||
stream: "stdout",
|
||||
chunk,
|
||||
seq,
|
||||
},
|
||||
}),
|
||||
}),
|
||||
);
|
||||
};
|
||||
|
||||
// The websocket races ahead of the poller and its seq-2 chunk arrives
|
||||
// tail-truncated.
|
||||
await act(async () => {
|
||||
sendLogEvent(2, "rld\n");
|
||||
sendLogEvent(3, "!\n");
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
const persistedRow = (seq: number, chunk: string) =>
|
||||
JSON.stringify({ ts: "2026-04-20T00:00:00.500Z", stream: "stdout", chunk, seq });
|
||||
await act(async () => {
|
||||
resolveLog?.({
|
||||
runId: "run-1",
|
||||
store: "memory",
|
||||
logRef: "log-1",
|
||||
content: `${persistedRow(1, "hello\n")}\n${persistedRow(2, "world\n")}\n${persistedRow(3, "!\n")}\n`,
|
||||
nextOffset: 300,
|
||||
});
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
const lastCall = buildTranscriptMock.mock.calls.at(-1) as unknown[] | undefined;
|
||||
expect(lastCall?.[0]).toEqual([
|
||||
{ ts: "2026-04-20T00:00:00.500Z", stream: "stdout", chunk: "hello\n", seq: 1 },
|
||||
{ ts: "2026-04-20T00:00:00.500Z", stream: "stdout", chunk: "world\n", seq: 2 },
|
||||
{ ts: "2026-04-20T00:00:01.000Z", stream: "stdout", chunk: "!\n", seq: 3 },
|
||||
]);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("rebuilds only the transcript for the run that receives live output", async () => {
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
|
|
|
|||
|
|
@ -51,6 +51,14 @@ export function resolveInitialLogOffset(run: RunTranscriptSource, limitBytes: nu
|
|||
return Math.max(0, knownBytes - Math.max(0, limitBytes));
|
||||
}
|
||||
|
||||
function readChunkSeq(value: unknown): number | undefined {
|
||||
return typeof value === "number" && Number.isFinite(value) ? value : undefined;
|
||||
}
|
||||
|
||||
function isStructuredStreamingTextDelta(chunk: string) {
|
||||
return /"type"\s*:\s*"(?:acpx\.text_delta|text)"/.test(chunk);
|
||||
}
|
||||
|
||||
function parsePersistedLogContent(
|
||||
runId: string,
|
||||
content: string,
|
||||
|
|
@ -68,7 +76,7 @@ function parsePersistedLogContent(
|
|||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
try {
|
||||
const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown };
|
||||
const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown; seq?: unknown };
|
||||
const stream = raw.stream === "stderr" || raw.stream === "system" ? raw.stream : "stdout";
|
||||
const chunk = typeof raw.chunk === "string" ? raw.chunk : "";
|
||||
const ts = typeof raw.ts === "string" ? raw.ts : new Date().toISOString();
|
||||
|
|
@ -77,6 +85,7 @@ function parsePersistedLogContent(
|
|||
ts,
|
||||
stream,
|
||||
chunk,
|
||||
seq: readChunkSeq(raw.seq),
|
||||
dedupeKey: `log:${runId}:${ts}:${stream}:${chunk}`,
|
||||
});
|
||||
} catch {
|
||||
|
|
@ -111,6 +120,10 @@ export function useLiveRunTranscripts({
|
|||
const [chunksByRun, setChunksByRun] = useState<Map<string, RunLogChunk[]>>(new Map());
|
||||
const [hydratedRunIds, setHydratedRunIds] = useState<Set<string>>(new Set());
|
||||
const seenChunkKeysRef = useRef(new Set<string>());
|
||||
// Highest sequenced chunk trimmed out of a run's retained window; older
|
||||
// records re-delivered by the other transport are dropped instead of being
|
||||
// re-inserted ahead of newer output.
|
||||
const trimmedSeqFloorByRunRef = useRef(new Map<string, number>());
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
const missingTerminalLogRunIdsRef = useRef(new Set<string>());
|
||||
|
|
@ -149,8 +162,44 @@ export function useLiveRunTranscripts({
|
|||
let changed = false;
|
||||
|
||||
for (const chunk of chunks) {
|
||||
if (seenChunkKeysRef.current.has(chunk.dedupeKey)) continue;
|
||||
seenChunkKeysRef.current.add(chunk.dedupeKey);
|
||||
// Sequenced log chunks (persisted rows and websocket log payloads)
|
||||
// dedupe and order by the server-assigned monotonic seq. Identical
|
||||
// token deltas from ACP-style adapters often share the same
|
||||
// millisecond ts and chunk text, so content-based keys drop real
|
||||
// output; seq keeps every record and restores emit order when the
|
||||
// websocket and the poller interleave.
|
||||
if (typeof chunk.seq === "number") {
|
||||
const seqFloor = trimmedSeqFloorByRunRef.current.get(runId) ?? 0;
|
||||
if (chunk.seq <= seqFloor) continue;
|
||||
const duplicateAt = existing.findIndex((item) => item.seq === chunk.seq);
|
||||
if (duplicateAt !== -1) {
|
||||
// Same record arrived via the other delivery path. Prefer the
|
||||
// longer payload: websocket chunks may be tail-truncated while
|
||||
// the persisted row is complete.
|
||||
if (chunk.chunk.length > existing[duplicateAt]!.chunk.length) {
|
||||
existing[duplicateAt] = { ts: chunk.ts, stream: chunk.stream, chunk: chunk.chunk, seq: chunk.seq };
|
||||
changed = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// Insert in seq order relative to the trailing sequenced chunks so
|
||||
// late-arriving records from the slower delivery path land where
|
||||
// they were emitted. Unsequenced chunks act as an ordering barrier.
|
||||
let insertAt = existing.length;
|
||||
while (insertAt > 0) {
|
||||
const prior = existing[insertAt - 1]!;
|
||||
if (typeof prior.seq !== "number" || prior.seq < chunk.seq) break;
|
||||
insertAt -= 1;
|
||||
}
|
||||
existing.splice(insertAt, 0, { ts: chunk.ts, stream: chunk.stream, chunk: chunk.chunk, seq: chunk.seq });
|
||||
changed = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!isStructuredStreamingTextDelta(chunk.chunk)) {
|
||||
if (seenChunkKeysRef.current.has(chunk.dedupeKey)) continue;
|
||||
seenChunkKeysRef.current.add(chunk.dedupeKey);
|
||||
}
|
||||
existing.push({ ts: chunk.ts, stream: chunk.stream, chunk: chunk.chunk });
|
||||
changed = true;
|
||||
}
|
||||
|
|
@ -159,7 +208,15 @@ export function useLiveRunTranscripts({
|
|||
if (seenChunkKeysRef.current.size > 12000) {
|
||||
seenChunkKeysRef.current.clear();
|
||||
}
|
||||
next.set(runId, existing.slice(-maxChunksPerRun));
|
||||
if (existing.length > maxChunksPerRun) {
|
||||
const trimmed = existing.splice(0, existing.length - maxChunksPerRun);
|
||||
let seqFloor = trimmedSeqFloorByRunRef.current.get(runId) ?? 0;
|
||||
for (const item of trimmed) {
|
||||
if (typeof item.seq === "number" && item.seq > seqFloor) seqFloor = item.seq;
|
||||
}
|
||||
if (seqFloor > 0) trimmedSeqFloorByRunRef.current.set(runId, seqFloor);
|
||||
}
|
||||
next.set(runId, existing);
|
||||
return next;
|
||||
});
|
||||
};
|
||||
|
|
@ -196,6 +253,11 @@ export function useLiveRunTranscripts({
|
|||
logOffsetByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of trimmedSeqFloorByRunRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
trimmedSeqFloorByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of missingTerminalLogRunIdsRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
missingTerminalLogRunIdsRef.current.delete(runId);
|
||||
|
|
@ -316,6 +378,7 @@ export function useLiveRunTranscripts({
|
|||
ts,
|
||||
stream,
|
||||
chunk,
|
||||
seq: readChunkSeq(payload["seq"]),
|
||||
dedupeKey: `log:${runId}:${ts}:${stream}:${chunk}`,
|
||||
}]);
|
||||
return;
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import {
|
|||
buildAssistantPartsFromTranscript,
|
||||
buildIssueChatMessages,
|
||||
isCoTSegmentActive,
|
||||
preserveReadableStreamingRetraction,
|
||||
stabilizeThreadMessages,
|
||||
type IssueChatComment,
|
||||
type IssueChatLinkedRun,
|
||||
|
|
@ -1116,6 +1117,108 @@ describe("buildIssueChatMessages", () => {
|
|||
});
|
||||
|
||||
describe("stabilizeThreadMessages", () => {
|
||||
it("reveals live streamed additions at word boundaries instead of character boundaries", () => {
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"Writing ",
|
||||
"Writing the pla",
|
||||
)).toBe("Writing the ");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"Writing ",
|
||||
"Writing the plan ",
|
||||
)).toBe("Writing the plan ");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"Writing ",
|
||||
"Writing the plan.",
|
||||
)).toBe("Writing the plan.");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"Writing ",
|
||||
"Writing draft",
|
||||
)).toBe("Writing draft");
|
||||
});
|
||||
|
||||
it("holds sliding-window removals until an older paragraph or group boundary drops", () => {
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"First sentence. Second sentence is visible",
|
||||
"irst sentence. Second sentence is visible now ",
|
||||
)).toBe("irst sentence. Second sentence is visible now ");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"First sentence. Second sentence is visible",
|
||||
"Second sentence is visible now ",
|
||||
)).toBe("Second sentence is visible now ");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"Paragraph one.\n\nParagraph two is visible",
|
||||
"Paragraph two is visible now ",
|
||||
)).toBe("Paragraph two is visible now ");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"The answer is 42",
|
||||
"42 is the answer",
|
||||
)).toBe("42 is the answer");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"The quick brown fox jumps over the lazy dog",
|
||||
"quick brown fox jumps over the lazy dog near the river",
|
||||
)).toBe("quick brown fox jumps over the lazy dog near the river");
|
||||
});
|
||||
|
||||
it("keeps live streamed retractions readable until a whole line disappears", () => {
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"First line\nSecond line\nThird line is complete",
|
||||
"First line\nSecond line\nThird line",
|
||||
)).toBe("First line\nSecond line\nThird line is complete");
|
||||
expect(preserveReadableStreamingRetraction(
|
||||
"First line\nSecond line\nThird line is complete",
|
||||
"First line\nSecond line",
|
||||
)).toBe("First line\nSecond line");
|
||||
|
||||
const liveRun: LiveRunForIssue = {
|
||||
id: "run-live-retract",
|
||||
status: "running",
|
||||
invocationSource: "manual",
|
||||
triggerDetail: null,
|
||||
startedAt: "2026-04-06T12:04:00.000Z",
|
||||
finishedAt: null,
|
||||
createdAt: "2026-04-06T12:04:00.000Z",
|
||||
agentId: "agent-1",
|
||||
agentName: "CodexCoder",
|
||||
adapterType: "codex_local",
|
||||
};
|
||||
const buildLiveMessages = (text: string) => buildIssueChatMessages({
|
||||
comments: [],
|
||||
timelineEvents: [],
|
||||
linkedRuns: [],
|
||||
liveRuns: [liveRun],
|
||||
transcriptsByRunId: new Map([
|
||||
["run-live-retract", [{ kind: "assistant", ts: "2026-04-06T12:04:01.000Z", text }]],
|
||||
]),
|
||||
hasOutputForRun: (runId) => runId === "run-live-retract",
|
||||
currentUserId: "user-1",
|
||||
});
|
||||
|
||||
const fullText = "First line\nSecond line\nThird line is complete";
|
||||
const firstStable = stabilizeThreadMessages(buildLiveMessages(fullText), [], new Map());
|
||||
const partialRetractionStable = stabilizeThreadMessages(
|
||||
buildLiveMessages("First line\nSecond line\nThird line"),
|
||||
firstStable.messages,
|
||||
firstStable.cache,
|
||||
);
|
||||
|
||||
expect(partialRetractionStable.messages).toBe(firstStable.messages);
|
||||
expect(partialRetractionStable.messages[0]?.content[0]).toMatchObject({
|
||||
type: "text",
|
||||
text: fullText,
|
||||
});
|
||||
|
||||
const wholeLineRetractionStable = stabilizeThreadMessages(
|
||||
buildLiveMessages("First line\nSecond line"),
|
||||
partialRetractionStable.messages,
|
||||
partialRetractionStable.cache,
|
||||
);
|
||||
|
||||
expect(wholeLineRetractionStable.messages[0]?.content[0]).toMatchObject({
|
||||
type: "text",
|
||||
text: "First line\nSecond line",
|
||||
});
|
||||
});
|
||||
|
||||
it("reuses unchanged message objects across rebuilds", () => {
|
||||
const firstPass = buildIssueChatMessages({
|
||||
comments: [createComment()],
|
||||
|
|
|
|||
|
|
@ -113,6 +113,111 @@ function fingerprintThreadMessage(message: ThreadMessage) {
|
|||
return JSON.stringify(message);
|
||||
}
|
||||
|
||||
function issueChatMessageCustom(message: ThreadMessage): Record<string, unknown> {
|
||||
const custom = message.metadata?.custom;
|
||||
return custom && typeof custom === "object" && !Array.isArray(custom)
|
||||
? custom as Record<string, unknown>
|
||||
: {};
|
||||
}
|
||||
|
||||
function isLiveRunThreadMessage(message: ThreadMessage) {
|
||||
return message.role === "assistant"
|
||||
&& message.status?.type === "running"
|
||||
&& issueChatMessageCustom(message)["kind"] === "live-run";
|
||||
}
|
||||
|
||||
export function preserveReadableStreamingRetraction(previousText: string, nextText: string) {
|
||||
if (!previousText || !nextText) return nextText;
|
||||
|
||||
if (nextText.length >= previousText.length && nextText.startsWith(previousText)) {
|
||||
return revealCompleteStreamingWords(previousText, nextText);
|
||||
}
|
||||
|
||||
const overlapLength = longestSuffixPrefixOverlap(previousText, nextText);
|
||||
if (overlapLength >= 8 && overlapLength < previousText.length) {
|
||||
const removedPrefix = previousText.slice(0, previousText.length - overlapLength);
|
||||
if (isQuietStreamingRemovalBoundary(removedPrefix)) {
|
||||
return nextText;
|
||||
}
|
||||
|
||||
return nextText;
|
||||
}
|
||||
|
||||
if (nextText.length >= previousText.length || !previousText.startsWith(nextText)) {
|
||||
return revealCompleteStreamingWords(previousText, nextText);
|
||||
}
|
||||
|
||||
const nextLength = nextText.length;
|
||||
if (previousText[nextLength] === "\n") return nextText;
|
||||
|
||||
const nextLineBreak = previousText.indexOf("\n", nextLength);
|
||||
if (nextLineBreak === -1) return previousText;
|
||||
return previousText.slice(0, nextLineBreak);
|
||||
}
|
||||
|
||||
function revealCompleteStreamingWords(previousText: string, nextText: string) {
|
||||
if (nextText.length <= previousText.length || !nextText.startsWith(previousText)) {
|
||||
return nextText;
|
||||
}
|
||||
|
||||
const addedText = nextText.slice(previousText.length);
|
||||
if (!addedText) return nextText;
|
||||
|
||||
const boundaryIndex = lastReadableWordBoundary(addedText);
|
||||
if (boundaryIndex === -1) return nextText;
|
||||
return previousText + addedText.slice(0, boundaryIndex + 1);
|
||||
}
|
||||
|
||||
function lastReadableWordBoundary(text: string) {
|
||||
let index = -1;
|
||||
for (let i = 0; i < text.length; i += 1) {
|
||||
const char = text[i];
|
||||
if (/\s/.test(char) || /[.,;:!?)}\]"'`]/.test(char)) {
|
||||
index = i;
|
||||
}
|
||||
}
|
||||
return index;
|
||||
}
|
||||
|
||||
function longestSuffixPrefixOverlap(previousText: string, nextText: string) {
|
||||
const maxLength = Math.min(previousText.length, nextText.length);
|
||||
for (let length = maxLength; length > 0; length -= 1) {
|
||||
if (previousText.endsWith(nextText.slice(0, length))) {
|
||||
return length;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function isQuietStreamingRemovalBoundary(removedPrefix: string) {
|
||||
return /(?:\n\s*\n|\n|[.!?]\s+)$/.test(removedPrefix);
|
||||
}
|
||||
|
||||
function smoothLiveRunRetractions(
|
||||
message: ThreadMessage,
|
||||
previousMessage: ThreadMessage | undefined,
|
||||
): ThreadMessage {
|
||||
if (!previousMessage || !isLiveRunThreadMessage(message) || !isLiveRunThreadMessage(previousMessage)) {
|
||||
return message;
|
||||
}
|
||||
|
||||
let changed = false;
|
||||
const content = message.content.map((part, index) => {
|
||||
if (part.type !== "text" && part.type !== "reasoning") return part;
|
||||
|
||||
const previousPart = previousMessage.content[index];
|
||||
if (previousPart?.type !== part.type) return part;
|
||||
|
||||
const text = preserveReadableStreamingRetraction(previousPart.text, part.text);
|
||||
if (text === part.text) return part;
|
||||
|
||||
changed = true;
|
||||
return { ...part, text };
|
||||
});
|
||||
|
||||
return changed ? ({ ...message, content } as ThreadMessage) : message;
|
||||
}
|
||||
|
||||
export function stabilizeThreadMessages(
|
||||
messages: readonly ThreadMessage[],
|
||||
previousMessages: readonly ThreadMessage[],
|
||||
|
|
@ -122,12 +227,13 @@ export function stabilizeThreadMessages(
|
|||
let sameSequence = previousMessages.length === messages.length;
|
||||
|
||||
const stabilizedMessages = messages.map((message, index) => {
|
||||
const fingerprint = fingerprintThreadMessage(message);
|
||||
const cached = previousById.get(message.id);
|
||||
const displayMessage = smoothLiveRunRetractions(message, cached?.message);
|
||||
const fingerprint = fingerprintThreadMessage(displayMessage);
|
||||
const stableMessage =
|
||||
cached && cached.fingerprint === fingerprint
|
||||
? cached.message
|
||||
: message;
|
||||
: displayMessage;
|
||||
nextById.set(message.id, {
|
||||
fingerprint,
|
||||
message: stableMessage,
|
||||
|
|
|
|||
Loading…
Reference in New Issue