summaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/chunks/index.ts1
-rw-r--r--src/core/chunks/local-echo.test.ts45
-rw-r--r--src/core/chunks/reducer.ts31
-rw-r--r--src/core/chunks/types.ts2
-rw-r--r--src/core/wire/conformance.test.ts7
-rw-r--r--src/core/wire/conformance.ts2
6 files changed, 84 insertions, 4 deletions
diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts
index 3822a79..49381d5 100644
--- a/src/core/chunks/index.ts
+++ b/src/core/chunks/index.ts
@@ -7,6 +7,7 @@ export {
clearGenerating,
foldEvent,
initialState,
+ removeLocalEcho,
} from "./reducer";
export type { ProviderRetryView } from "./retry-banner";
export { formatRetryDelay, viewProviderRetry } from "./retry-banner";
diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts
new file mode 100644
index 0000000..4b5eed4
--- /dev/null
+++ b/src/core/chunks/local-echo.test.ts
@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index";
+
+describe("request-local optimistic echo", () => {
+ it("retains identity through turn-start tags without conflating identical sends", () => {
+ let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+ state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+ state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+ state = removeLocalEcho(state, "r1");
+ expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]);
+ expect(state.provisional.map((p) => p.chunk)).toEqual([
+ { type: "text", text: "same" },
+ { type: "image", url: "data:second" },
+ ]);
+ expect(state.generating).toBe(true);
+ expect(removeLocalEcho(state, "r1")).toBe(state);
+ });
+
+ it("a live user-message confirms only its deduplicated text, not every identical local send", () => {
+ let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+ state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+ state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+ state = foldEvent(state, {
+ type: "user-message",
+ conversationId: "c1",
+ turnId: "t1",
+ text: "same",
+ });
+ state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2");
+ expect(state.provisional).toEqual([
+ { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } },
+ ]);
+ });
+
+ it("does not remove committed rows even if their content is identical", () => {
+ let state = appendUserMessage(initialState(), "same", undefined, "r1");
+ state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]);
+ const committed = state.committed;
+ state = removeLocalEcho(state, "r1");
+ expect(state.committed).toBe(committed);
+ expect(state.provisional).toEqual([]);
+ expect(removeLocalEcho(state, "r1")).toBe(state);
+ expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState());
+ });
+});
diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts
index 0af2ddd..53af9c1 100644
--- a/src/core/chunks/reducer.ts
+++ b/src/core/chunks/reducer.ts
@@ -246,11 +246,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// text) emits no `user-message` and is not de-duped here.
if (event.text.length === 0) return state;
const run = trailingRun(state.provisional, "user");
- const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text);
- if (alreadyEchoed) {
+ const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text);
+ if (echoed !== undefined) {
return {
...state,
- provisional: tagTrailingUserEcho(state.provisional, event.turnId),
+ // Preserve the live text independently of the local send. A delayed
+ // queued ack may remove its images, but cannot erase this server event.
+ provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => {
+ if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p;
+ const { localRequestId: _local, ...confirmed } = p;
+ return confirmed;
+ }),
generating: true,
};
}
@@ -463,6 +469,7 @@ export function appendUserMessage(
state: TranscriptState,
text: string,
images?: readonly ImageInput[],
+ localRequestId?: string,
): TranscriptState {
const provisional = flushAccumulating(state.provisional, state.accumulating);
const userChunks: Chunk[] = [];
@@ -484,7 +491,23 @@ export function appendUserMessage(
}
return {
...state,
- provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)],
+ provisional: [
+ ...provisional,
+ ...userChunks.map(
+ (chunk) =>
+ ({
+ role: "user",
+ chunk,
+ ...(localRequestId !== undefined ? { localRequestId } : {}),
+ }) as const,
+ ),
+ ],
accumulating: null,
};
}
+
+/** Remove only still-local artifacts of one accepted queued send; never history/live text. */
+export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState {
+ const provisional = state.provisional.filter((p) => p.localRequestId !== requestId);
+ return provisional.length === state.provisional.length ? state : { ...state, provisional };
+}
diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts
index 949857a..f7bfe01 100644
--- a/src/core/chunks/types.ts
+++ b/src/core/chunks/types.ts
@@ -10,6 +10,8 @@ export interface AccumulatingChunk {
/** A provisional chunk that has no authoritative seq yet. */
export interface ProvisionalChunk {
+ /** Local send identity only; turn-start tagging does NOT confirm this content. */
+ readonly localRequestId?: string;
readonly role: Role;
readonly chunk: Chunk;
/**
diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts
index 1c5829b..1179fbc 100644
--- a/src/core/wire/conformance.test.ts
+++ b/src/core/wire/conformance.test.ts
@@ -174,6 +174,12 @@ describe("classifies every WsServerMessage type", () => {
event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" },
},
{ type: "chat.error" as const, message: "e" },
+ {
+ type: "chat.queued" as const,
+ conversationId: "conv-1",
+ reason: "compaction" as const,
+ messageId: "queued-1",
+ },
{ type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" },
{
type: "conversation.statusChanged" as const,
@@ -198,6 +204,7 @@ describe("classifies every WsServerMessage type", () => {
"error",
"chat.delta",
"chat.error",
+ "chat.queued",
"conversation.open",
"conversation.statusChanged",
"conversation.compacted",
diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts
index 9847081..78313f9 100644
--- a/src/core/wire/conformance.ts
+++ b/src/core/wire/conformance.ts
@@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string {
return "error";
case "chat.delta":
return "chat.delta";
+ case "chat.queued":
+ return "chat.queued";
case "chat.error":
return "chat.error";
case "conversation.open":