officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chat: durable Postgres event queue + cursor replay on reconnect (Phase 2+3)

Browse Source 
Completes the turn/session decoupling so nothing is lost across disconnects.

Phase 2 (durability):
- New chat_session_events table (global monotonic id = cursor) + queries
  appendChatEvent / getChatEventsSince / pruneChatEventsOlderThan.
- Every durable outbound ServerMessage now goes through emitToSession: appended
  to the queue (even while the client is disconnected) and delivered live with
  its seq. Streaming deltas stay ephemeral (live-only, never persisted).

Phase 3 (resilient transport):
- New 'resume-cursor' client message → handleResumeCursor re-binds the socket to
  the (still-live) session (cancels idle-GC via attachWs) and replays every event
  since the client's cursor.
- useChatWebSocket already auto-reconnects; added an onOpen hook. useChat tracks
  the max seq and, on every (re)connect with an established session, sends
  resume-cursor — so a dropped connection self-heals with no manual navigate
  away/back, and background task notifications that landed while offline replay.

Verified end-to-end: disconnect after a turn's result but before a background
task finishes, reconnect with the cursor → the missed task:notification is
replayed from Postgres, no duplicates.

Note: the DB is managed via drizzle push/direct DDL (no __drizzle_migrations
table), so 0001 was applied directly; the generated migration is committed for
the record.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-07-27 00:12:46 +00:00
co-authored by Claude Opus 4.8
parent 449f28b1e5
commit 6b3eb247a3
11 changed files with 1792 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/chat/websocket.ts
+59 -16
View File
@@ -8,7 +8,7 @@ import { ensureGeneralChatSessionsCwd } from './claude-sessions';
import * as sidecar from '@@/sidecar-registry';
import { join } from 'path';
import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path';
import { getUserSettings, getEmailAccounts } from 'officerdb';
import { getUserSettings, getEmailAccounts, appendChatEvent, getChatEventsSince } from 'officerdb';
import { mkdirSync } from 'node:fs';
import { logger } from './logger';
@@ -90,12 +90,31 @@ const wsToSessionMap = new WeakMap<any, string>();
const pingTimers = new WeakMap<ServerWebSocket<WSData>, ReturnType<typeof setInterval>>();
const PING_INTERVAL_MS = 25_000;
function sendToClient(ws: ServerWebSocket<WSData> | null, msg: ServerMessage): void {
function sendToClient(ws: ServerWebSocket<WSData> | null, msg: ServerMessage, seq?: number): void {
if (ws?.readyState === 1) {
ws.send(JSON.stringify(msg));
ws.send(JSON.stringify(seq === undefined ? msg : { ...msg, seq }));
}
}
// Persist a durable session event to the Postgres queue (for replay across reconnects) and deliver it
// live to the attached socket with its cursor `seq`. Events are queued even while the client is
// DISCONNECTED (ws null) — that's what lets a reconnecting client replay what it missed (e.g. a
// background task:notification). Streaming deltas are ephemeral: delivered live, never persisted.
async function emitToSession(sessionId: string, msg: ServerMessage): Promise<void> {
const ws = (sessionManager.getSession(sessionId)?.ws ?? null) as ServerWebSocket<WSData> | null;
if (msg.type === 'assistant:delta') {
sendToClient(ws, msg);
return;
}
let seq: number | undefined;
try {
seq = await appendChatEvent(sessionId, msg);
} catch (err) {
logger.error('Failed to persist chat event', { sessionId, error: String(err) });
}
sendToClient(ws, msg, seq);
}
export async function open(ws: ServerWebSocket<WSData>): Promise<void> {
const timer = setInterval(() => {
try {
@@ -122,6 +141,8 @@ export function message(ws: ServerWebSocket<WSData>, raw: string | Buffer): void
await handleStop(ws);
} else if (clientMsg.type === 'disconnect') {
await handleDisconnect(ws);
} else if (clientMsg.type === 'resume-cursor') {
await handleResumeCursor(ws, clientMsg);
}
} catch (err) {
logger.error('Error handling WebSocket message', { email: ws.data.email, error: String(err) });
@@ -149,11 +170,10 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
const session = sessionManager.getSession(sessionId);
if (!session) return;
const ws = session.ws as ServerWebSocket<WSData> | null;
// All durable sends go through emitToSession (persist + deliver with cursor seq). Deltas stay live-only.
switch (event.type) {
case 'delta': {
sendToClient(ws, { type: 'assistant:delta', text: event.text });
await emitToSession(sessionId, { type: 'assistant:delta', text: event.text });
session.streamBuffer += event.text;
break;
}
@@ -162,7 +182,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
// Flush streaming buffer as complete text
const text = event.text || session.streamBuffer;
if (text) {
sendToClient(ws, { type: 'assistant:text', text });
await emitToSession(sessionId, { type: 'assistant:text', text });
const assistantMsg: Message = {
id: randomUUID(),
@@ -181,7 +201,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
case 'tool:start': {
// Flush any pending streaming text first
if (session.streamBuffer) {
sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer });
await emitToSession(sessionId, { type: 'assistant:text', text: session.streamBuffer });
const assistantMsg: Message = {
id: randomUUID(),
@@ -195,7 +215,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
session.streamBuffer = '';
}
sendToClient(ws, {
await emitToSession(sessionId, {
type: 'tool:start',
toolCallId: event.toolCallId,
toolName: event.toolName,
@@ -216,7 +236,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
}
case 'tool:result': {
sendToClient(ws, {
await emitToSession(sessionId, {
type: 'tool:result',
toolCallId: event.toolCallId,
output: event.output,
@@ -238,7 +258,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
case 'result': {
// Flush any remaining streaming buffer
if (session.streamBuffer) {
sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer });
await emitToSession(sessionId, { type: 'assistant:text', text: session.streamBuffer });
const assistantMsg: Message = {
id: randomUUID(),
@@ -253,7 +273,7 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
session.streamBuffer = '';
}
sendToClient(ws, { type: 'result', sessionId, cost: event.cost });
await emitToSession(sessionId, { type: 'result', sessionId, cost: event.cost });
session.isGenerating = false;
session.meta.cost.inputTokens += event.cost.inputTokens;
@@ -265,26 +285,26 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
}
case 'error': {
sendToClient(ws, { type: 'error', message: event.message });
await emitToSession(sessionId, { type: 'error', message: event.message });
session.isGenerating = false;
break;
}
case 'stopped': {
sendToClient(ws, { type: 'stopped' });
await emitToSession(sessionId, { type: 'stopped' });
session.isGenerating = false;
break;
}
case 'task:started': {
// Background task launched (run_in_background / Monitor). Independent of turn state.
sendToClient(ws, { type: 'task:started', taskId: event.taskId, description: event.description, taskType: event.taskType });
await emitToSession(sessionId, { type: 'task:started', taskId: event.taskId, description: event.description, taskType: event.taskType });
break;
}
case 'task:notification': {
// Background task finished — arrives AFTER the turn's 'result'; the whole point of the fix.
sendToClient(ws, { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary });
await emitToSession(sessionId, { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary });
break;
}
}
@@ -585,6 +605,29 @@ async function handleDisconnect(ws: ServerWebSocket<WSData>): Promise<void> {
sendToClient(ws, { type: 'disconnected' });
}
// Reconnect: re-bind this socket to the (possibly still-live) session and replay every durable event
// queued since the client's cursor — so a brief disconnect never loses turn output or a background
// task:notification. attachWs cancels the pending idle-GC. If the in-memory session was already
// GC'd, we still replay history from Postgres (new turns will respawn the session).
async function handleResumeCursor(
ws: ServerWebSocket<WSData>,
msg: { sessionId: string; cursor: number },
): Promise<void> {
const { sessionId, cursor } = msg;
if (sessionManager.getSession(sessionId)) {
sessionManager.attachWs(sessionId, ws);
wsToSessionMap.set(ws as any, sessionId);
}
try {
const events = await getChatEventsSince(sessionId, cursor ?? 0);
for (const { id, event } of events) {
sendToClient(ws, event as ServerMessage, id);
}
} catch (err) {
logger.error('Failed to replay chat events on resume-cursor', { sessionId, error: String(err) });
}
}
export const chatWebsocket = {
open,
message,