mirror of
https://github.com/cline/cline.git
synced 2026-09-19 02:05:44 +08:00
feat(desktop): reconcile cloud session snapshots (#14080)
* feat(desktop): isolate cloud Hub snapshot reconciliation * fix(desktop): reconcile reflected attachment-only cloud prompts * fix(desktop): preserve unreflected cloud tool and reasoning events * fix(desktop): preserve unfinished aborted cloud output * fix(desktop): retain interrupted output after saved replies * fix(desktop): retain output completed after cloud snapshot * fix(desktop): exclude tool output from empty prompt matches
This commit is contained in:
John Choi
@@ -0,0 +1,505 @@
|
||||
import type { HubEventEnvelope } from "@cline/shared";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { reconcileBufferedCloudEvents } from "./cloud-session-snapshots";
|
||||
|
||||
describe("reconcileBufferedCloudEvents", () => {
|
||||
const event = (
|
||||
name: HubEventEnvelope["event"],
|
||||
id: string,
|
||||
payload: Record<string, unknown> = {},
|
||||
): HubEventEnvelope => ({
|
||||
version: "v1",
|
||||
event: name,
|
||||
eventId: id,
|
||||
timestamp: Date.now(),
|
||||
sessionId: "inner-1",
|
||||
payload,
|
||||
});
|
||||
|
||||
it.each([
|
||||
"Continue",
|
||||
"",
|
||||
])("preserves submitted lifecycle while marking only newly reflected %j prompts", (prompt) => {
|
||||
const submitted = (id: string) =>
|
||||
event("session.pending_prompt_submitted", id, {
|
||||
prompt: {
|
||||
id,
|
||||
prompt,
|
||||
delivery: "queue",
|
||||
...(prompt
|
||||
? {}
|
||||
: {
|
||||
userImages: ["data:image/png;base64,AA=="],
|
||||
attachmentCount: 1,
|
||||
}),
|
||||
},
|
||||
});
|
||||
const first = submitted("q-1");
|
||||
const later = submitted("q-2");
|
||||
const baseline = [
|
||||
{
|
||||
role: "user",
|
||||
content: prompt || [
|
||||
{
|
||||
type: "image",
|
||||
source: { type: "base64", media_type: "image/png", data: "AA==" },
|
||||
},
|
||||
],
|
||||
},
|
||||
];
|
||||
// An older identical user message must not consume a new submission.
|
||||
expect(
|
||||
reconcileBufferedCloudEvents([first], baseline, {
|
||||
baselineMessages: baseline,
|
||||
}),
|
||||
).toEqual([first]);
|
||||
const toolResult = {
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "tool_result", tool_use_id: "call-1", content: "done" },
|
||||
],
|
||||
};
|
||||
expect(
|
||||
reconcileBufferedCloudEvents([first], [...baseline, toolResult], {
|
||||
baselineMessages: baseline,
|
||||
}),
|
||||
).toEqual([first]);
|
||||
const snapshot = [
|
||||
...baseline,
|
||||
toolResult,
|
||||
{
|
||||
...baseline[0],
|
||||
...(prompt
|
||||
? {
|
||||
content: [
|
||||
...toolResult.content,
|
||||
{ type: "text", text: `<user_input>${prompt}</user_input>` },
|
||||
],
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
];
|
||||
const completed = event("run.completed", "done");
|
||||
const running = event("run.started", "next");
|
||||
// Keep the next turn's start between lifecycle events; only its bubble is reflected.
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
[completed, first, running, later],
|
||||
snapshot,
|
||||
{
|
||||
baselineMessages: baseline,
|
||||
},
|
||||
),
|
||||
).toEqual([
|
||||
completed,
|
||||
{ ...first, payload: { ...first.payload, transcriptReflected: true } },
|
||||
running,
|
||||
later,
|
||||
]);
|
||||
// A submission received after the transcript reply cannot be in it.
|
||||
expect(
|
||||
reconcileBufferedCloudEvents([later], snapshot, {
|
||||
baselineMessages: baseline,
|
||||
messagesSnapshotEventCutoff: 0,
|
||||
}),
|
||||
).toEqual([later]);
|
||||
});
|
||||
|
||||
it("replays content the snapshot does NOT contain", () => {
|
||||
const buffered = [
|
||||
event("assistant.delta", "a-1", { text: "unpersisted reply" }),
|
||||
event("run.completed", "done-1"),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{ role: "assistant", content: "a completely different answer" },
|
||||
]).map((item) => item.event),
|
||||
).toEqual(["assistant.delta", "run.completed"]);
|
||||
});
|
||||
|
||||
it("does not let an interior substring claim another run's snapshot", () => {
|
||||
const buffered = [
|
||||
event("assistant.finished", "a-1", { text: "foo" }),
|
||||
event("run.completed", "done-1"),
|
||||
event("assistant.finished", "a-2", { text: "The answer is foobar" }),
|
||||
event("run.completed", "done-2"),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{ role: "assistant", content: "The answer is foobar" },
|
||||
]).map((item) => item.eventId),
|
||||
).toEqual(["a-1", "done-1", "done-2"]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["bar", "foobar", false],
|
||||
["bar", "foobar", true],
|
||||
["Done", "Done", false],
|
||||
] as const)("preserves aborted %j beside saved %j (reverse: %s)", (partial, saved, reverse) => {
|
||||
const aborted = [
|
||||
event("assistant.delta", "partial", { text: partial }),
|
||||
event("run.aborted", "aborted"),
|
||||
];
|
||||
const completed = [
|
||||
event("assistant.delta", "saved-delta", { text: saved }),
|
||||
event("assistant.finished", "saved-finished", { text: saved }),
|
||||
event("run.completed", "completed"),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
reverse ? [...completed, ...aborted] : [...aborted, ...completed],
|
||||
[{ role: "assistant", content: saved }],
|
||||
).map((item) => item.eventId),
|
||||
).toEqual(
|
||||
reverse
|
||||
? ["completed", "partial", "aborted"]
|
||||
: ["partial", "aborted", "completed"],
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
"assistant",
|
||||
"reasoning",
|
||||
] as const)("does not let finished %s hide the other aborted content", (finished) => {
|
||||
const partial = finished === "assistant" ? "reasoning" : "assistant";
|
||||
const buffered = [
|
||||
event(`${finished}.finished`, "saved", {
|
||||
text: "foobar",
|
||||
reasoning: "foobar",
|
||||
}),
|
||||
event(`${partial}.delta`, "partial", { text: "bar" }),
|
||||
event("run.aborted", "aborted"),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "text", text: "foobar" },
|
||||
{ type: "thinking", thinking: "foobar" },
|
||||
],
|
||||
},
|
||||
]).map((item) => item.eventId),
|
||||
).toEqual(["partial", "aborted"]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["assistant", false],
|
||||
["reasoning", false],
|
||||
["reasoning", true],
|
||||
] as const)("preserves unfinished %s after saved output (redacted: %s)", (kind, redacted) => {
|
||||
const saved = [
|
||||
event(`${kind}.delta`, "saved-delta", {
|
||||
text: redacted ? "" : "saved",
|
||||
redacted,
|
||||
}),
|
||||
event(`${kind}.finished`, "saved-finished", {
|
||||
[kind === "assistant" ? "text" : "reasoning"]: redacted
|
||||
? undefined
|
||||
: "saved",
|
||||
}),
|
||||
];
|
||||
const partial = event(`${kind}.delta`, "partial", { text: "unfinished" });
|
||||
const snapshot = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
redacted
|
||||
? { type: "redacted_thinking", data: "opaque" }
|
||||
: kind === "assistant"
|
||||
? { type: "text", text: "saved" }
|
||||
: { type: "thinking", thinking: "saved" },
|
||||
],
|
||||
},
|
||||
];
|
||||
for (const terminal of ["run.aborted", "run.failed"] as const) {
|
||||
const end = event(terminal, "end");
|
||||
const buffered = [...saved, partial, end];
|
||||
expect(reconcileBufferedCloudEvents(buffered, snapshot)).toEqual([
|
||||
partial,
|
||||
end,
|
||||
]);
|
||||
const earlier = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
kind === "assistant"
|
||||
? { type: "text", text: "earlier" }
|
||||
: { type: "thinking", thinking: "earlier" },
|
||||
],
|
||||
},
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
[
|
||||
event(`${kind}.finished`, "earlier", {
|
||||
[kind === "assistant" ? "text" : "reasoning"]: "earlier",
|
||||
}),
|
||||
...buffered,
|
||||
],
|
||||
[...earlier, ...snapshot],
|
||||
{ baselineMessages: earlier },
|
||||
),
|
||||
).toEqual([partial, end]);
|
||||
expect(reconcileBufferedCloudEvents(buffered, [])).toEqual(buffered);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, snapshot, {
|
||||
baselineMessages: snapshot,
|
||||
}),
|
||||
).toEqual(buffered);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, snapshot, {
|
||||
messagesSnapshotEventCutoff: 1,
|
||||
}),
|
||||
).toEqual(
|
||||
terminal === "run.aborted" ? buffered : [saved[1], partial, end],
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
"assistant",
|
||||
"reasoning",
|
||||
] as const)("preserves %s output completed after the snapshot", (kind) => {
|
||||
const key = kind === "assistant" ? "text" : "reasoning";
|
||||
const saved = event(`${kind}.finished`, "saved", { [key]: "saved" });
|
||||
const partial = event(`${kind}.delta`, "partial", { text: "new reply" });
|
||||
const finished = event(`${kind}.finished`, "finished", {
|
||||
[key]: "new reply",
|
||||
});
|
||||
const end = event("run.completed", "end");
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
[saved, partial, finished, end],
|
||||
[
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: kind === "assistant" ? "text" : "thinking",
|
||||
[kind === "assistant" ? "text" : "thinking"]: "saved",
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
{ messagesSnapshotEventCutoff: 2 },
|
||||
),
|
||||
).toEqual([partial, finished, end]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
"run.completed",
|
||||
"run.aborted",
|
||||
"run.failed",
|
||||
] as const)("ignores empty finishes after saved content in %s", (terminal) => {
|
||||
const end = event(terminal, "end");
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
[
|
||||
event("assistant.finished", "saved", { text: "saved" }),
|
||||
event("assistant.finished", "empty", { text: "" }),
|
||||
event("assistant.finished", "missing"),
|
||||
end,
|
||||
],
|
||||
[{ role: "assistant", content: "saved" }],
|
||||
),
|
||||
).toEqual([end]);
|
||||
});
|
||||
|
||||
it("supersedes despite trailing whitespace in the streamed text", () => {
|
||||
const buffered = [
|
||||
event("assistant.finished", "f-1", { text: "the answer \n" }),
|
||||
event("run.completed", "done-1"),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{ role: "assistant", content: "prefix the answer" },
|
||||
]).map((item) => item.event),
|
||||
).toEqual(["run.completed"]);
|
||||
});
|
||||
|
||||
it("drops buffered queue snapshots when a fresh queue snapshot was applied", () => {
|
||||
const buffered = [
|
||||
event("session.pending_prompts", "q-1", { prompts: [] }),
|
||||
event("assistant.delta", "a-1", { text: "live tail" }),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, []).map((item) => item.event),
|
||||
).toEqual(["assistant.delta"]);
|
||||
});
|
||||
|
||||
it("replays the newest buffered queue snapshot when the queue fetch failed", () => {
|
||||
const buffered = [
|
||||
event("session.pending_prompts", "q-1", { prompts: [] }),
|
||||
event("session.pending_prompts", "q-2", {
|
||||
prompts: [{ id: "p-1", prompt: "queued work" }],
|
||||
}),
|
||||
event("assistant.delta", "a-1", { text: "live tail" }),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [], {
|
||||
queueSnapshotApplied: false,
|
||||
}).map((item) => item.eventId),
|
||||
).toEqual(["q-2", "a-1"]);
|
||||
});
|
||||
|
||||
it("replays a queue snapshot received after the queue fetch", () => {
|
||||
const buffered = [
|
||||
event("session.pending_prompts", "q-old", { prompts: [] }),
|
||||
event("assistant.delta", "a-1", { text: "live tail" }),
|
||||
event("session.pending_prompts", "q-new", {
|
||||
prompts: [{ id: "p-1", prompt: "queued work" }],
|
||||
}),
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [], {
|
||||
queueSnapshotEventCutoff: 1,
|
||||
}).map((item) => item.eventId),
|
||||
).toEqual(["a-1", "q-new"]);
|
||||
});
|
||||
|
||||
it("supersedes content independently across two terminal run segments", () => {
|
||||
const buffered = [
|
||||
event("assistant.delta", "a-1", { text: "first tail" }),
|
||||
event("run.completed", "done-1"),
|
||||
event("assistant.delta", "a-2", { text: "second tail" }),
|
||||
event("run.completed", "done-2"),
|
||||
];
|
||||
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{ role: "assistant", content: "prefix first tail" },
|
||||
{ role: "assistant", content: "prefix second tail" },
|
||||
]).map((item) => item.event),
|
||||
).toEqual(["run.completed", "run.completed"]);
|
||||
});
|
||||
|
||||
it("does not let an older identical reply supersede a new buffered turn", () => {
|
||||
const buffered = [
|
||||
event("assistant.delta", "a-2", { text: "Done" }),
|
||||
event("run.completed", "done-2"),
|
||||
];
|
||||
const baseline = [{ role: "assistant", content: "Done" }];
|
||||
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, baseline, {
|
||||
baselineMessages: baseline,
|
||||
}).map((item) => item.event),
|
||||
).toEqual(["assistant.delta", "run.completed"]);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
buffered,
|
||||
[...baseline, { role: "assistant", content: "Done" }],
|
||||
{ baselineMessages: baseline },
|
||||
).map((item) => item.event),
|
||||
).toEqual(["run.completed"]);
|
||||
});
|
||||
|
||||
it("keeps run.failed while suppressing reflected content and dedupes tools by id", () => {
|
||||
const buffered = [
|
||||
event("assistant.delta", "a-1", { text: "partial failure" }),
|
||||
event("tool.started", "tool-1", { toolCallId: "call-1" }),
|
||||
event("run.failed", "failed-1", { error: "boom" }),
|
||||
];
|
||||
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, [
|
||||
{ role: "assistant", content: "saved partial failure" },
|
||||
{
|
||||
role: "assistant",
|
||||
content: [{ type: "tool_use", id: "call-1", name: "read_file" }],
|
||||
},
|
||||
]).map((item) => item.event),
|
||||
).toEqual(["run.failed"]);
|
||||
});
|
||||
|
||||
it("only suppresses tool phases present in the transcript before its cutoff", () => {
|
||||
const started = event("tool.started", "start", { toolCallId: "call-1" });
|
||||
const finished = event("tool.finished", "finish", {
|
||||
toolCallId: "call-1",
|
||||
output: "done",
|
||||
});
|
||||
const snapshot = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [{ type: "tool_use", id: "call-1", name: "read_file" }],
|
||||
},
|
||||
];
|
||||
expect(reconcileBufferedCloudEvents([started, finished], snapshot)).toEqual(
|
||||
[finished],
|
||||
);
|
||||
const completedSnapshot = [
|
||||
...snapshot,
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "tool_result", tool_use_id: "call-1", content: "done" },
|
||||
],
|
||||
},
|
||||
];
|
||||
expect(
|
||||
reconcileBufferedCloudEvents([started, finished], completedSnapshot),
|
||||
).toEqual([]);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents([started, finished], completedSnapshot, {
|
||||
messagesSnapshotEventCutoff: 1,
|
||||
}),
|
||||
).toEqual([finished]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["run.failed", false],
|
||||
["run.aborted", false],
|
||||
["run.aborted", true],
|
||||
] as const)("does not replay persisted thinking for %s (redacted: %s)", (terminal, redacted) => {
|
||||
const buffered = [
|
||||
event("reasoning.delta", "thinking", {
|
||||
text: redacted ? "" : "Checking",
|
||||
redacted,
|
||||
}),
|
||||
event("reasoning.finished", "thought", {
|
||||
reasoning: redacted ? undefined : "Checking",
|
||||
}),
|
||||
event(terminal, "end"),
|
||||
];
|
||||
const snapshot = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
redacted
|
||||
? { type: "redacted_thinking", data: "opaque" }
|
||||
: { type: "thinking", thinking: "Checking" },
|
||||
],
|
||||
},
|
||||
];
|
||||
expect(reconcileBufferedCloudEvents(buffered, snapshot)).toEqual([
|
||||
buffered[2],
|
||||
]);
|
||||
expect(reconcileBufferedCloudEvents(buffered, [])).toEqual(buffered);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, snapshot, {
|
||||
baselineMessages: snapshot,
|
||||
}),
|
||||
).toEqual(buffered);
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(buffered, snapshot, {
|
||||
messagesSnapshotEventCutoff: 0,
|
||||
}),
|
||||
).toEqual(buffered);
|
||||
});
|
||||
|
||||
it("does not treat persisted assistant text as proof that thinking was saved", () => {
|
||||
const thinking = event("reasoning.delta", "thinking", { text: "Checking" });
|
||||
const done = event("run.completed", "done");
|
||||
expect(
|
||||
reconcileBufferedCloudEvents(
|
||||
[
|
||||
thinking,
|
||||
event("assistant.finished", "answer", { text: "Done" }),
|
||||
done,
|
||||
],
|
||||
[{ role: "assistant", content: "Done" }],
|
||||
),
|
||||
).toEqual([thinking, done]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,362 @@
|
||||
import { isUserRunMessage } from "@cline/core";
|
||||
import type { HubEventEnvelope } from "@cline/shared";
|
||||
import type { JsonRecord, PromptInQueue } from "./types";
|
||||
|
||||
export function readSessionRows(
|
||||
payload: Record<string, unknown> | undefined,
|
||||
): JsonRecord[] {
|
||||
return Array.isArray(payload?.sessions)
|
||||
? payload.sessions.filter(
|
||||
(item): item is JsonRecord =>
|
||||
Boolean(item) && typeof item === "object" && !Array.isArray(item),
|
||||
)
|
||||
: [];
|
||||
}
|
||||
|
||||
export function updatedAt(record: JsonRecord): number {
|
||||
const value = record.updatedAt;
|
||||
return typeof value === "number"
|
||||
? value
|
||||
: Date.parse(String(value ?? "")) || 0;
|
||||
}
|
||||
|
||||
export function sessionRowModelId(record: JsonRecord | undefined): string {
|
||||
const metadata =
|
||||
record?.metadata && typeof record.metadata === "object"
|
||||
? (record.metadata as JsonRecord)
|
||||
: undefined;
|
||||
return String(metadata?.model ?? record?.model ?? "").trim();
|
||||
}
|
||||
|
||||
export function isRootSessionRow(record: JsonRecord): boolean {
|
||||
const metadata =
|
||||
record.metadata && typeof record.metadata === "object"
|
||||
? (record.metadata as JsonRecord)
|
||||
: undefined;
|
||||
return !String(
|
||||
metadata?.parentSessionId ?? record.parentSessionId ?? "",
|
||||
).trim();
|
||||
}
|
||||
|
||||
function messageText(
|
||||
message: unknown,
|
||||
kind: "text" | "thinking" = "text",
|
||||
): string {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return "";
|
||||
}
|
||||
const content = (message as JsonRecord).content;
|
||||
if (typeof content === "string") {
|
||||
return kind === "text" ? content.trim() : "";
|
||||
}
|
||||
if (!Array.isArray(content)) {
|
||||
return "";
|
||||
}
|
||||
return content
|
||||
.map((part) =>
|
||||
part && typeof part === "object" && !Array.isArray(part)
|
||||
? kind === "thinking" &&
|
||||
(part as JsonRecord).type === "redacted_thinking"
|
||||
? "[redacted]"
|
||||
: String((part as JsonRecord)[kind] ?? "")
|
||||
: "",
|
||||
)
|
||||
.join("")
|
||||
.trim();
|
||||
}
|
||||
|
||||
function normalizeUserPrompt(text: string): string {
|
||||
const trimmed = text.trim();
|
||||
const match = trimmed.match(/^<user_input\b[^>]*>([\s\S]*)<\/user_input>$/);
|
||||
return (match ? match[1] : trimmed).trim();
|
||||
}
|
||||
|
||||
export function countPromptOccurrences(
|
||||
messages: unknown[],
|
||||
prompts: PromptInQueue[],
|
||||
prompt: string,
|
||||
): number {
|
||||
const expected = normalizeUserPrompt(prompt);
|
||||
return (
|
||||
messages.filter(
|
||||
(message) =>
|
||||
Boolean(message) &&
|
||||
typeof message === "object" &&
|
||||
!Array.isArray(message) &&
|
||||
String((message as JsonRecord).role ?? "").toLowerCase() === "user" &&
|
||||
(expected !== "" || isUserRunMessage(message as JsonRecord)) &&
|
||||
normalizeUserPrompt(messageText(message)) === expected,
|
||||
).length +
|
||||
prompts.filter((item) => normalizeUserPrompt(item.prompt) === expected)
|
||||
.length
|
||||
);
|
||||
}
|
||||
|
||||
export function submittedPromptsFromEvents(
|
||||
events: HubEventEnvelope[],
|
||||
): PromptInQueue[] {
|
||||
return events.flatMap((event) => {
|
||||
if (event.event !== "session.pending_prompt_submitted") return [];
|
||||
const prompt =
|
||||
event.payload?.prompt &&
|
||||
typeof event.payload.prompt === "object" &&
|
||||
!Array.isArray(event.payload.prompt)
|
||||
? (event.payload.prompt as JsonRecord)
|
||||
: undefined;
|
||||
const id = String(prompt?.id ?? "").trim();
|
||||
if (!id) return [];
|
||||
return [
|
||||
{
|
||||
id,
|
||||
prompt: String(prompt?.prompt ?? ""),
|
||||
steer: prompt?.delivery === "steer",
|
||||
attachmentCount:
|
||||
typeof prompt?.attachmentCount === "number"
|
||||
? prompt.attachmentCount
|
||||
: 0,
|
||||
userImages: Array.isArray(prompt?.userImages)
|
||||
? prompt.userImages.filter(
|
||||
(image): image is string => typeof image === "string",
|
||||
)
|
||||
: undefined,
|
||||
},
|
||||
];
|
||||
});
|
||||
}
|
||||
|
||||
const TERMINAL_RUN_EVENTS = new Set([
|
||||
"run.completed",
|
||||
"run.aborted",
|
||||
"run.failed",
|
||||
]);
|
||||
const SUPERSEDABLE_CONTENT_EVENTS = new Set([
|
||||
"assistant.delta",
|
||||
"assistant.finished",
|
||||
"reasoning.delta",
|
||||
"reasoning.finished",
|
||||
]);
|
||||
|
||||
function assistantTexts(
|
||||
messages: unknown[],
|
||||
kind: "text" | "thinking",
|
||||
): string[] {
|
||||
return messages
|
||||
.filter(
|
||||
(message): message is JsonRecord =>
|
||||
Boolean(message) &&
|
||||
typeof message === "object" &&
|
||||
!Array.isArray(message) &&
|
||||
String((message as JsonRecord).role ?? "").toLowerCase() ===
|
||||
"assistant",
|
||||
)
|
||||
.map((message) => messageText(message, kind))
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function newlyPersistedAssistantTexts(
|
||||
snapshotMessages: unknown[],
|
||||
baselineMessages: unknown[],
|
||||
kind: "text" | "thinking" = "text",
|
||||
): string[] {
|
||||
const baselineCounts = new Map<string, number>();
|
||||
for (const text of assistantTexts(baselineMessages, kind)) {
|
||||
baselineCounts.set(text, (baselineCounts.get(text) ?? 0) + 1);
|
||||
}
|
||||
return assistantTexts(snapshotMessages, kind).filter((text) => {
|
||||
const count = baselineCounts.get(text) ?? 0;
|
||||
if (count === 0) return true;
|
||||
if (count === 1) baselineCounts.delete(text);
|
||||
else baselineCounts.set(text, count - 1);
|
||||
return false;
|
||||
});
|
||||
}
|
||||
|
||||
function collectToolCallIds(
|
||||
value: unknown,
|
||||
phase: "tool_use" | "tool_result",
|
||||
result = new Set<string>(),
|
||||
): Set<string> {
|
||||
if (!value || typeof value !== "object") return result;
|
||||
if (Array.isArray(value)) {
|
||||
for (const item of value) collectToolCallIds(item, phase, result);
|
||||
return result;
|
||||
}
|
||||
const record = value as JsonRecord;
|
||||
const id = phase === "tool_use" ? record.id : record.tool_use_id;
|
||||
if (record.type === phase && typeof id === "string") {
|
||||
result.add(id);
|
||||
}
|
||||
for (const child of Object.values(record))
|
||||
collectToolCallIds(child, phase, result);
|
||||
return result;
|
||||
}
|
||||
|
||||
function streamedAssistantText(
|
||||
events: HubEventEnvelope[],
|
||||
kind: "assistant" | "reasoning" = "assistant",
|
||||
): string {
|
||||
let finishedText = "";
|
||||
let deltas = "";
|
||||
for (const event of events) {
|
||||
if (event.event === `${kind}.delta`) {
|
||||
deltas +=
|
||||
kind === "reasoning" && event.payload?.redacted && !event.payload?.text
|
||||
? "[redacted]"
|
||||
: typeof event.payload?.text === "string"
|
||||
? event.payload.text
|
||||
: "";
|
||||
} else if (event.event === `${kind}.finished`) {
|
||||
const text = event.payload?.[kind === "assistant" ? "text" : "reasoning"];
|
||||
const completed = typeof text === "string" && text ? text : deltas;
|
||||
if (completed) finishedText = completed;
|
||||
deltas = "";
|
||||
}
|
||||
}
|
||||
return (finishedText || deltas).trim();
|
||||
}
|
||||
|
||||
/** Reconciles each completed run separately; tools dedupe by stable call id. */
|
||||
export function reconcileBufferedCloudEvents(
|
||||
events: HubEventEnvelope[],
|
||||
snapshotMessages: unknown[],
|
||||
options: {
|
||||
/**
|
||||
* Whether a fresh queue snapshot was fetched and applied during
|
||||
* rehydration. When it was, queue events received before its reply are
|
||||
* stale; later events still win. When the fetch failed, the newest
|
||||
* buffered queue event is the best state available.
|
||||
*/
|
||||
queueSnapshotApplied?: boolean;
|
||||
queueSnapshotEventCutoff?: number;
|
||||
/** Events received after the transcript reply cannot be reflected in it. */
|
||||
messagesSnapshotEventCutoff?: number;
|
||||
baselineMessages?: unknown[];
|
||||
} = {},
|
||||
): HubEventEnvelope[] {
|
||||
const queueSnapshotApplied = options.queueSnapshotApplied !== false;
|
||||
const unclaimedAssistantTexts = newlyPersistedAssistantTexts(
|
||||
snapshotMessages,
|
||||
options.baselineMessages ?? [],
|
||||
);
|
||||
const unclaimedThinking = newlyPersistedAssistantTexts(
|
||||
snapshotMessages,
|
||||
options.baselineMessages ?? [],
|
||||
"thinking",
|
||||
);
|
||||
const snapshotToolCallIds = collectToolCallIds(snapshotMessages, "tool_use");
|
||||
const snapshotToolResultIds = collectToolCallIds(
|
||||
snapshotMessages,
|
||||
"tool_result",
|
||||
);
|
||||
const beforeTranscript = new Set(
|
||||
events.slice(0, options.messagesSnapshotEventCutoff ?? events.length),
|
||||
);
|
||||
const reflectedSubmissions = new Set<HubEventEnvelope>();
|
||||
const unclaimedUserCounts = new Map<string, number>();
|
||||
for (const event of events.slice(
|
||||
0,
|
||||
options.messagesSnapshotEventCutoff ?? events.length,
|
||||
)) {
|
||||
const submitted = submittedPromptsFromEvents([event])[0];
|
||||
if (!submitted) continue;
|
||||
const prompt = normalizeUserPrompt(submitted.prompt);
|
||||
const count =
|
||||
unclaimedUserCounts.get(prompt) ??
|
||||
Math.max(
|
||||
0,
|
||||
countPromptOccurrences(snapshotMessages, [], prompt) -
|
||||
countPromptOccurrences(options.baselineMessages ?? [], [], prompt),
|
||||
);
|
||||
unclaimedUserCounts.set(prompt, Math.max(0, count - 1));
|
||||
if (count > 0) reflectedSubmissions.add(event);
|
||||
}
|
||||
// Queue events are full snapshots, so only the newest one matters.
|
||||
const queueEvents = queueSnapshotApplied
|
||||
? events.slice(options.queueSnapshotEventCutoff ?? events.length)
|
||||
: events;
|
||||
const lastQueueEvent = queueEvents.findLast(
|
||||
(event) => event.event === "session.pending_prompts",
|
||||
);
|
||||
const reconciled: HubEventEnvelope[] = [];
|
||||
let segment: HubEventEnvelope[] = [];
|
||||
|
||||
const flush = (terminal: boolean) => {
|
||||
if (segment.length === 0) return;
|
||||
const snapshotSegment = segment.filter((event) =>
|
||||
beforeTranscript.has(event),
|
||||
);
|
||||
// A buffered run can contain saved replies followed by unsaved output.
|
||||
const contentEnd = (kind: "assistant" | "reasoning") => {
|
||||
if (!terminal) return -1;
|
||||
const finished = snapshotSegment.findLastIndex(
|
||||
(event) => event.event === `${kind}.finished`,
|
||||
);
|
||||
return finished >= 0 || segment.at(-1)?.event === "run.aborted"
|
||||
? finished
|
||||
: snapshotSegment.length - 1;
|
||||
};
|
||||
const assistantEnd = contentEnd("assistant");
|
||||
const reasoningEnd = contentEnd("reasoning");
|
||||
const streamed = streamedAssistantText(
|
||||
snapshotSegment.slice(0, assistantEnd + 1),
|
||||
);
|
||||
const persistedIndex = streamed
|
||||
? unclaimedAssistantTexts.findIndex((text) => text.endsWith(streamed))
|
||||
: -1;
|
||||
const contentPersisted = persistedIndex >= 0;
|
||||
if (contentPersisted) unclaimedAssistantTexts.splice(persistedIndex, 1);
|
||||
const thinking = streamedAssistantText(
|
||||
snapshotSegment.slice(0, reasoningEnd + 1),
|
||||
"reasoning",
|
||||
);
|
||||
const thinkingIndex = thinking
|
||||
? unclaimedThinking.findIndex((text) => text.endsWith(thinking))
|
||||
: -1;
|
||||
if (thinkingIndex >= 0) unclaimedThinking.splice(thinkingIndex, 1);
|
||||
for (const [index, event] of segment.entries()) {
|
||||
// Preserve the turn-start lifecycle; the UI must only skip its user bubble.
|
||||
if (reflectedSubmissions.has(event)) {
|
||||
reconciled.push({
|
||||
...event,
|
||||
payload: { ...event.payload, transcriptReflected: true },
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
beforeTranscript.has(event) &&
|
||||
SUPERSEDABLE_CONTENT_EVENTS.has(event.event) &&
|
||||
(event.event.startsWith("reasoning.")
|
||||
? thinkingIndex >= 0 && index <= reasoningEnd
|
||||
: contentPersisted && index <= assistantEnd)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
event.event === "session.pending_prompts" &&
|
||||
event !== lastQueueEvent
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
// Keep terminal events: run.failed may carry the only error detail.
|
||||
if (beforeTranscript.has(event) && event.event.startsWith("tool.")) {
|
||||
const toolCallId = String(event.payload?.toolCallId ?? "").trim();
|
||||
if (
|
||||
snapshotToolResultIds.has(toolCallId) ||
|
||||
(event.event === "tool.started" &&
|
||||
snapshotToolCallIds.has(toolCallId))
|
||||
)
|
||||
continue;
|
||||
}
|
||||
reconciled.push(event);
|
||||
}
|
||||
segment = [];
|
||||
};
|
||||
|
||||
for (const event of events) {
|
||||
segment.push(event);
|
||||
if (TERMINAL_RUN_EVENTS.has(event.event)) flush(true);
|
||||
}
|
||||
// Never supersede an unterminated tail.
|
||||
flush(false);
|
||||
return reconciled;
|
||||
}
|
||||
Reference in New Issue
Block a user