summaryrefslogtreecommitdiffhomepage
path: root/src/adapters
diff options
context:
space:
mode:
Diffstat (limited to 'src/adapters')
-rw-r--r--src/adapters/idb/index.test.ts15
-rw-r--r--src/adapters/idb/index.ts23
-rw-r--r--src/adapters/ws/index.test.ts137
-rw-r--r--src/adapters/ws/index.ts50
-rw-r--r--src/adapters/ws/logic.test.ts76
-rw-r--r--src/adapters/ws/logic.ts59
6 files changed, 351 insertions, 9 deletions
diff --git a/src/adapters/idb/index.test.ts b/src/adapters/idb/index.test.ts
index 04e2177..aa56f15 100644
--- a/src/adapters/idb/index.test.ts
+++ b/src/adapters/idb/index.test.ts
@@ -128,3 +128,18 @@ describe("createIdbChunkStore", () => {
expect(loaded2[0]?.chunk).toEqual(textChunk("b1"));
});
});
+
+it("persists epoch across adapter recreation and append, including empty histories", async () => {
+ const indexedDB = new IDBFactory();
+ const store = createIdbChunkStore({ indexedDB });
+ await store.append("c", [makeChunk(1, "legacy")]);
+ expect(await store.historyEpoch("c")).toBeUndefined();
+ await store.reset("c", 2);
+ const reopened = createIdbChunkStore({ indexedDB });
+ expect(await reopened.load("c")).toEqual([]);
+ expect(await reopened.historyEpoch("c")).toBe(2);
+ await reopened.append("c", [makeChunk(101, "new")]);
+ expect(await reopened.historyEpoch("c")).toBe(2);
+ await reopened.delete("c");
+ expect(await reopened.historyEpoch("c")).toBeUndefined();
+});
diff --git a/src/adapters/idb/index.ts b/src/adapters/idb/index.ts
index 3dffef9..8b1848a 100644
--- a/src/adapters/idb/index.ts
+++ b/src/adapters/idb/index.ts
@@ -20,6 +20,7 @@ interface ChunkRecord {
interface MetaRecord {
conversationId: string;
lastAccess: number;
+ historyEpoch?: number;
}
export interface CreateIdbChunkStoreOptions {
@@ -92,6 +93,26 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa
}
return {
+ async historyEpoch(conversationId) {
+ const db = await getDb();
+ const tx = db.transaction(META_STORE, "readonly");
+ const meta = await requestToPromise<MetaRecord | undefined>(
+ tx.objectStore(META_STORE).get(conversationId),
+ );
+ await txComplete(tx);
+ return meta?.historyEpoch;
+ },
+ async reset(conversationId, historyEpoch) {
+ const db = await getDb();
+ const tx = db.transaction([CHUNKS_STORE, META_STORE], "readwrite");
+ tx.objectStore(CHUNKS_STORE).delete(keyRangeFor(conversationId));
+ tx.objectStore(META_STORE).put({
+ conversationId,
+ historyEpoch,
+ lastAccess: Date.now(),
+ } satisfies MetaRecord);
+ await txComplete(tx);
+ },
async load(conversationId: string): Promise<readonly StoredChunk[]> {
const db = await getDb();
const tx = db.transaction(CHUNKS_STORE, "readonly");
@@ -122,7 +143,9 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa
} satisfies ChunkRecord);
}
+ const meta = await requestToPromise<MetaRecord | undefined>(metaStore.get(conversationId));
metaStore.put({
+ ...meta,
conversationId,
lastAccess: Date.now(),
} satisfies MetaRecord);
diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts
index 9d821e2..e4e4ba4 100644
--- a/src/adapters/ws/index.test.ts
+++ b/src/adapters/ws/index.test.ts
@@ -6,7 +6,7 @@ interface FakeSocket extends WebSocketLike {
sent: string[];
resolveOpen(): void;
invokeMessage(data: string): void;
- invokeClose(): void;
+ invokeClose(code?: number): void;
}
function fakeSocket(): FakeSocket {
@@ -44,8 +44,8 @@ function fakeSocket(): FakeSocket {
invokeMessage(data: string) {
onmessage?.({ data });
},
- invokeClose() {
- onclose?.({ code: 1000, reason: "" });
+ invokeClose(code = 1000) {
+ onclose?.({ code, reason: "" });
},
sent,
};
@@ -405,3 +405,134 @@ describe("createSurfaceSocket", () => {
expect(JSON.parse(ws.sent[0] ?? "")).toEqual({ type: "chat.send", message: "queued" });
});
});
+
+describe("reconnect backoff after open", () => {
+ it.each([1013, 1006, 1011])("grows and caps delays across repeated early %s closes", (code) => {
+ vi.useFakeTimers();
+ const random = vi.spyOn(Math, "random").mockReturnValue(0.5);
+ const sockets: FakeSocket[] = [];
+ const handle = createSurfaceSocket({
+ url: "ws://test",
+ onMessage: vi.fn(),
+ socketFactory: () => {
+ const ws = fakeSocket();
+ sockets.push(ws);
+ return ws;
+ },
+ });
+ try {
+ for (const delay of [500, 1000, 2000, 4000, 8000, 16000, 30000, 30000]) {
+ const count = sockets.length;
+ sockets.at(-1)?.resolveOpen();
+ vi.advanceTimersByTime(100);
+ sockets.at(-1)?.invokeClose(code);
+ vi.advanceTimersByTime(delay - 1);
+ expect(sockets).toHaveLength(count);
+ vi.advanceTimersByTime(1);
+ expect(sockets).toHaveLength(count + 1);
+ }
+ } finally {
+ handle.close();
+ random.mockRestore();
+ vi.useRealTimers();
+ }
+ });
+
+ it.each([
+ "healthy",
+ "normal",
+ "going away",
+ ])("resets accumulated backoff after a %s connection", (mode) => {
+ vi.useFakeTimers();
+ const random = vi.spyOn(Math, "random").mockReturnValue(0.5);
+ const sockets: FakeSocket[] = [];
+ const handle = createSurfaceSocket({
+ url: "ws://test",
+ onMessage: vi.fn(),
+ socketFactory: () => {
+ const ws = fakeSocket();
+ sockets.push(ws);
+ return ws;
+ },
+ });
+ try {
+ sockets[0]?.resolveOpen();
+ sockets[0]?.invokeClose(1013);
+ vi.advanceTimersByTime(500);
+ sockets[1]?.resolveOpen();
+ if (mode === "healthy") vi.advanceTimersByTime(30000);
+ sockets[1]?.invokeClose(mode === "normal" ? 1000 : mode === "going away" ? 1001 : 1013);
+ vi.advanceTimersByTime(499);
+ expect(sockets).toHaveLength(2);
+ vi.advanceTimersByTime(1);
+ expect(sockets).toHaveLength(3);
+ handle.close();
+ vi.advanceTimersByTime(60000);
+ expect(sockets).toHaveLength(3);
+ } finally {
+ handle.close();
+ random.mockRestore();
+ vi.useRealTimers();
+ }
+ });
+});
+
+it("routes chat.queued only through the typed chat callback, never surface/error handling", () => {
+ const ws = fakeSocket();
+ const onChat = vi.fn();
+ const onMessage = vi.fn();
+ const handle = createSurfaceSocket({
+ url: "ws://test",
+ socketFactory: () => ws,
+ onChat,
+ onMessage,
+ });
+ const ack = {
+ type: "chat.queued",
+ conversationId: "c1",
+ reason: "compaction",
+ messageId: "q1",
+ requestId: "r1",
+ };
+ ws.invokeMessage(JSON.stringify(ack));
+ expect(onChat).toHaveBeenCalledExactlyOnceWith(ack);
+ expect(onMessage).not.toHaveBeenCalled();
+ handle.close();
+});
+
+it("routes resync notices to chat without reconnect/resend or surface errors", () => {
+ vi.useFakeTimers();
+ const ws = fakeSocket();
+ const onChat = vi.fn();
+ const onMessage = vi.fn();
+ const onReopen = vi.fn();
+ const factory = vi.fn(() => ws);
+ const handle = createSurfaceSocket({
+ url: "ws://test",
+ socketFactory: factory,
+ onChat,
+ onMessage,
+ onReopen,
+ });
+ try {
+ ws.resolveOpen();
+ const notice = {
+ type: "chat.resync-required",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "replay-limit",
+ };
+ ws.invokeMessage(JSON.stringify(notice));
+ ws.invokeMessage(JSON.stringify(notice));
+ vi.advanceTimersByTime(1000);
+ expect(onChat).toHaveBeenCalledTimes(2);
+ expect(onChat).toHaveBeenCalledWith(notice);
+ expect(onMessage).not.toHaveBeenCalled();
+ expect(onReopen).not.toHaveBeenCalled();
+ expect(factory).toHaveBeenCalledTimes(1);
+ expect(ws.sent).toEqual([]);
+ } finally {
+ handle.close();
+ vi.useRealTimers();
+ }
+});
diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts
index 76d9e75..e1d7d22 100644
--- a/src/adapters/ws/index.ts
+++ b/src/adapters/ws/index.ts
@@ -1,6 +1,10 @@
import type {
+ ChatAcceptedMessage,
ChatDeltaMessage,
ChatErrorMessage,
+ ChatQueuedMessage,
+ ChatResyncRequiredMessage,
+ ChatSubscriptionRejectedMessage,
ConversationCompactedMessage,
ConversationOpenMessage,
ConversationStatusChangedMessage,
@@ -24,7 +28,15 @@ export interface WebSocketLike {
export interface SurfaceSocketOptions {
url: string;
onMessage: (msg: SurfaceServerMessage) => void;
- onChat?: (msg: ChatDeltaMessage | ChatErrorMessage) => void;
+ onChat?: (
+ msg:
+ | ChatAcceptedMessage
+ | ChatSubscriptionRejectedMessage
+ | ChatDeltaMessage
+ | ChatErrorMessage
+ | ChatQueuedMessage
+ | ChatResyncRequiredMessage,
+ ) => void;
/** Broadcast when a conversation is "opened" (e.g. CLI `--open` flag). */
onConversationOpen?: (msg: ConversationOpenMessage) => void;
/** Broadcast when a conversation's lifecycle status changes (active/idle/closed). */
@@ -67,7 +79,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
let disposed = false;
let reconnectAttempt = 0;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+ let healthyTimer: ReturnType<typeof setTimeout> | null = null;
let isOpen = false;
+ function clearHealthyTimer(): void {
+ if (healthyTimer !== null) clearTimeout(healthyTimer);
+ healthyTimer = null;
+ }
const queue: string[] = [];
function connect(isReconnect: boolean): void {
@@ -77,7 +94,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
socket.onopen = () => {
if (disposed) return;
isOpen = true;
- reconnectAttempt = 0;
+ // Opening is not proof of recovery: replay may immediately overflow the
+ // server's slow-consumer bound. Only a sustained open resets the streak.
+ healthyTimer = setTimeout(() => {
+ healthyTimer = null;
+ reconnectAttempt = 0;
+ }, 30_000);
for (const raw of queue.splice(0)) {
socket?.send(raw);
}
@@ -90,8 +112,23 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
if (disposed) return;
const msg = parseServerMessage(ev.data);
if (msg !== null) {
- if (msg.type === "chat.delta" || msg.type === "chat.error") {
- opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage);
+ if (
+ msg.type === "chat.accepted" ||
+ msg.type === "chat.subscription-rejected" ||
+ msg.type === "chat.delta" ||
+ msg.type === "chat.error" ||
+ msg.type === "chat.queued" ||
+ msg.type === "chat.resync-required"
+ ) {
+ opts.onChat?.(
+ msg as
+ | ChatAcceptedMessage
+ | ChatSubscriptionRejectedMessage
+ | ChatDeltaMessage
+ | ChatErrorMessage
+ | ChatQueuedMessage
+ | ChatResyncRequiredMessage,
+ );
} else if (msg.type === "conversation.open") {
opts.onConversationOpen?.(msg as ConversationOpenMessage);
} else if (msg.type === "conversation.statusChanged") {
@@ -112,8 +149,10 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
}
};
- socket.onclose = () => {
+ socket.onclose = (ev) => {
+ clearHealthyTimer();
if (disposed) return;
+ if (isOpen && (ev.code === 1000 || ev.code === 1001)) reconnectAttempt = 0;
isOpen = false;
scheduleReconnect();
};
@@ -143,6 +182,7 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
},
close(): void {
disposed = true;
+ clearHealthyTimer();
if (reconnectTimer !== null) {
clearTimeout(reconnectTimer);
reconnectTimer = null;
diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts
index 2179ad0..4bc2aad 100644
--- a/src/adapters/ws/logic.test.ts
+++ b/src/adapters/ws/logic.test.ts
@@ -562,3 +562,79 @@ describe("parseServerMessage — subagent tab groups", () => {
).toBeNull();
});
});
+
+it("preserves a compaction epoch and rejects absent or invalid epochs", () => {
+ const msg = {
+ type: "conversation.compacted",
+ conversationId: "c",
+ newConversationId: "archive",
+ messagesSummarized: 2,
+ messagesKept: 1,
+ historyEpoch: 7,
+ };
+ expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg);
+ for (const historyEpoch of [undefined, -1, 1.5, "7"]) {
+ expect(parseServerMessage(JSON.stringify({ ...msg, historyEpoch }))).toBeNull();
+ }
+});
+
+describe("chat.queued admission acknowledgement", () => {
+ const ack = {
+ type: "chat.queued",
+ conversationId: "c1",
+ reason: "compaction",
+ messageId: "q1",
+ };
+
+ it.each([undefined, "request-1"])("preserves optional requestId %s", (requestId) => {
+ const msg = { ...ack, ...(requestId !== undefined ? { requestId } : {}) };
+ expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg);
+ });
+
+ it.each([
+ { conversationId: undefined },
+ { conversationId: 12 },
+ { messageId: undefined },
+ { messageId: null },
+ { reason: undefined },
+ { reason: "other" },
+ { requestId: null },
+ { requestId: 12 },
+ ])("rejects malformed acknowledgement %j", (bad) => {
+ expect(parseServerMessage(JSON.stringify({ ...ack, ...bad }))).toBeNull();
+ });
+
+ it("serializes a correlated send", () => {
+ const msg = { type: "chat.send" as const, message: "hello", requestId: "r1" };
+ expect(JSON.parse(serialize(msg))).toEqual(msg);
+ });
+});
+
+describe("chat.resync-required", () => {
+ it.each(["replay-limit", "payload-limit"])("parses %s with optional turn identity", (reason) => {
+ const msg = { type: "chat.resync-required", conversationId: "c1", reason };
+ expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg);
+ expect(parseServerMessage(JSON.stringify({ ...msg, turnId: "t1" }))).toEqual({
+ ...msg,
+ turnId: "t1",
+ });
+ });
+ it.each([
+ {},
+ { conversationId: null },
+ { conversationId: 1 },
+ { reason: "unknown" },
+ { reason: null },
+ { turnId: null },
+ { turnId: 42 },
+ ])("rejects malformed notice %j", (bad) => {
+ const msg = {
+ type: "chat.resync-required",
+ conversationId: "c1",
+ reason: "replay-limit",
+ ...bad,
+ };
+ if (Object.keys(bad).length === 0) delete (msg as { conversationId?: unknown }).conversationId;
+ expect(parseServerMessage(JSON.stringify(msg))).toBeNull();
+ });
+});
diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts
index cecc66c..d167669 100644
--- a/src/adapters/ws/logic.ts
+++ b/src/adapters/ws/logic.ts
@@ -1,6 +1,8 @@
import type {
ChatDeltaMessage,
ChatErrorMessage,
+ ChatQueuedMessage,
+ ChatResyncRequiredMessage,
ConversationCompactedMessage,
ConversationOpenMessage,
ConversationStatusChangedMessage,
@@ -23,8 +25,12 @@ const VALID_SERVER_TYPES = new Set([
"surface",
"update",
"error",
+ "chat.accepted",
+ "chat.subscription-rejected",
"chat.delta",
"chat.error",
+ "chat.queued",
+ "chat.resync-required",
"conversation.open",
"conversation.statusChanged",
"conversation.compacted",
@@ -105,12 +111,56 @@ export function parseServerMessage(data: string): WsServerMessage | null {
: { type: "error", message: parsed.message };
return msg;
}
+ case "chat.accepted": {
+ if (typeof parsed.conversationId !== "string" || typeof parsed.requestId !== "string")
+ return null;
+ return {
+ type: "chat.accepted",
+ conversationId: parsed.conversationId,
+ requestId: parsed.requestId,
+ };
+ }
+ case "chat.subscription-rejected": {
+ if (typeof parsed.conversationId !== "string" || parsed.reason !== "subscription-limit")
+ return null;
+ return {
+ type: "chat.subscription-rejected",
+ conversationId: parsed.conversationId,
+ reason: parsed.reason,
+ };
+ }
case "chat.delta": {
const event = parsed.event;
if (!isRecord(event)) return null;
if (typeof event.type !== "string") return null;
return { type: "chat.delta", event: event as unknown as ChatDeltaMessage["event"] };
}
+ case "chat.resync-required": {
+ if (typeof parsed.conversationId !== "string") return null;
+ if (parsed.turnId !== undefined && typeof parsed.turnId !== "string") return null;
+ if (parsed.reason !== "replay-limit" && parsed.reason !== "payload-limit") return null;
+ const msg: ChatResyncRequiredMessage = {
+ type: "chat.resync-required",
+ conversationId: parsed.conversationId,
+ reason: parsed.reason,
+ ...(parsed.turnId !== undefined ? { turnId: parsed.turnId } : {}),
+ };
+ return msg;
+ }
+ case "chat.queued": {
+ if (typeof parsed.conversationId !== "string") return null;
+ if (typeof parsed.messageId !== "string") return null;
+ if (parsed.reason !== "compaction") return null;
+ if (parsed.requestId !== undefined && typeof parsed.requestId !== "string") return null;
+ const msg: ChatQueuedMessage = {
+ type: "chat.queued",
+ conversationId: parsed.conversationId,
+ messageId: parsed.messageId,
+ reason: "compaction",
+ ...(parsed.requestId !== undefined ? { requestId: parsed.requestId } : {}),
+ };
+ return msg;
+ }
case "chat.error": {
if (typeof parsed.message !== "string") return null;
const conversationId = parsed.conversationId;
@@ -119,7 +169,11 @@ export function parseServerMessage(data: string): WsServerMessage | null {
conversationId !== undefined
? { type: "chat.error", conversationId, message: parsed.message }
: { type: "chat.error", message: parsed.message };
- return msg;
+ if (parsed.requestId !== undefined && typeof parsed.requestId !== "string") return null;
+ return {
+ ...msg,
+ ...(typeof parsed.requestId === "string" ? { requestId: parsed.requestId } : {}),
+ };
}
case "conversation.open": {
if (typeof parsed.conversationId !== "string") return null;
@@ -180,7 +234,10 @@ export function parseServerMessage(data: string): WsServerMessage | null {
if (typeof parsed.newConversationId !== "string") return null;
if (typeof parsed.messagesSummarized !== "number") return null;
if (typeof parsed.messagesKept !== "number") return null;
+ if (!Number.isSafeInteger(parsed.historyEpoch) || (parsed.historyEpoch as number) < 0)
+ return null;
const msg: ConversationCompactedMessage = {
+ historyEpoch: parsed.historyEpoch as number,
type: "conversation.compacted",
conversationId: parsed.conversationId,
newConversationId: parsed.newConversationId,
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-26 18:28:11 +0000