summaryrefslogtreecommitdiffhomepage
path: root/src/adapters
diff options
context:
space:
mode:
Diffstat (limited to 'src/adapters')
-rw-r--r--src/adapters/ws/index.test.ts37
-rw-r--r--src/adapters/ws/index.ts20
-rw-r--r--src/adapters/ws/logic.test.ts29
-rw-r--r--src/adapters/ws/logic.ts14
4 files changed, 97 insertions, 3 deletions
diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts
index 6461493..e4e4ba4 100644
--- a/src/adapters/ws/index.test.ts
+++ b/src/adapters/ws/index.test.ts
@@ -499,3 +499,40 @@ it("routes chat.queued only through the typed chat callback, never surface/error
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 25e9911..b1e17cf 100644
--- a/src/adapters/ws/index.ts
+++ b/src/adapters/ws/index.ts
@@ -2,6 +2,7 @@ import type {
ChatDeltaMessage,
ChatErrorMessage,
ChatQueuedMessage,
+ ChatResyncRequiredMessage,
ConversationCompactedMessage,
ConversationOpenMessage,
ConversationStatusChangedMessage,
@@ -25,7 +26,9 @@ export interface WebSocketLike {
export interface SurfaceSocketOptions {
url: string;
onMessage: (msg: SurfaceServerMessage) => void;
- onChat?: (msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage) => void;
+ onChat?: (
+ msg: 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). */
@@ -101,8 +104,19 @@ 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" || msg.type === "chat.queued") {
- opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage);
+ if (
+ msg.type === "chat.delta" ||
+ msg.type === "chat.error" ||
+ msg.type === "chat.queued" ||
+ msg.type === "chat.resync-required"
+ ) {
+ opts.onChat?.(
+ msg as
+ | ChatDeltaMessage
+ | ChatErrorMessage
+ | ChatQueuedMessage
+ | ChatResyncRequiredMessage,
+ );
} else if (msg.type === "conversation.open") {
opts.onConversationOpen?.(msg as ConversationOpenMessage);
} else if (msg.type === "conversation.statusChanged") {
diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts
index db6444a..4bc2aad 100644
--- a/src/adapters/ws/logic.test.ts
+++ b/src/adapters/ws/logic.test.ts
@@ -609,3 +609,32 @@ describe("chat.queued admission acknowledgement", () => {
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 0efcd90..abc0464 100644
--- a/src/adapters/ws/logic.ts
+++ b/src/adapters/ws/logic.ts
@@ -2,6 +2,7 @@ import type {
ChatDeltaMessage,
ChatErrorMessage,
ChatQueuedMessage,
+ ChatResyncRequiredMessage,
ConversationCompactedMessage,
ConversationOpenMessage,
ConversationStatusChangedMessage,
@@ -27,6 +28,7 @@ const VALID_SERVER_TYPES = new Set([
"chat.delta",
"chat.error",
"chat.queued",
+ "chat.resync-required",
"conversation.open",
"conversation.statusChanged",
"conversation.compacted",
@@ -113,6 +115,18 @@ export function parseServerMessage(data: string): WsServerMessage | 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;