buzzert/Sybil-2
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

make PWA chat resume idempotent

Browse Source
This commit is contained in:
James Magahern
2026-07-23 17:12:11 -07:00
parent abc1124d27
commit c5217b2710
10 changed files with 326 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
dist/default.conf vendored
View File

@@ -12,17 +12,43 @@ server {
location /api/ {
proxy_pass http://server:8787/;
proxy_http_version 1.1;
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 3600s;
proxy_send_timeout 3600s;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Connection "";
}
location = /sw.js {
add_header Cache-Control "no-store, no-cache, must-revalidate" always;
expires -1;
try_files $uri =404;
}
location = /manifest.webmanifest {
default_type application/manifest+json;
add_header Cache-Control "no-store, no-cache, must-revalidate" always;
expires -1;
try_files $uri =404;
}
location = /index.html {
add_header Cache-Control "no-store, no-cache, must-revalidate" always;
expires -1;
try_files $uri =404;
}
location /assets/ {
add_header Cache-Control "public, max-age=31536000, immutable" always;
try_files $uri =404;
}
location / {
add_header Cache-Control "no-store, no-cache, must-revalidate" always;
expires -1;
try_files $uri $uri/ /index.html;
}
}

10
docs/api/rest.md
View File

@@ -323,6 +323,16 @@ Behavior notes:
- `CHAT_SHELL_EXEC_TIMEOUT_MS=120000` (optional)
- When a tool call is executed, backend stores a chat `Message` with `role: "tool"` and tool metadata (`metadata.kind = "tool_call"`). Streaming requests emit an initiated SSE `tool_call` event before execution, then persist each completed or failed tool call as its terminal SSE `tool_call` event is emitted, then store the assistant output when the completion finishes.
## Streaming Chat
### `POST /v1/chat-completions/stream`
- The request accepts the chat-completion fields above plus optional `persist` and `clientRequestId` fields.
- `clientRequestId` is only valid for a persisted request with a `chatId`, may be up to 128 characters, and should be a stable unique value generated once per user submission.
- Retrying with the same `chatId` and `clientRequestId` replays the matching active or completed stream rather than starting a duplicate provider call.
- The server persists the ID in `metadata.clientRequestId` on the submitted user message and completed assistant message.
- The complete request, SSE event, persistence, retry, and attach contracts are defined in `docs/api/streaming-chat.md`.
## Searches
### `GET /v1/searches`

6
docs/api/streaming-chat.md
View File

@@ -21,6 +21,7 @@ Authentication:
{
"chatId": "optional-chat-id",
"persist": true,
"clientRequestId": "optional-client-generated-id",
"provider": "openai|anthropic|xai|gemini|hermes-agent",
"model": "string",
"messages": [
@@ -61,6 +62,9 @@ Notes:
- If `persist` is `true` and `chatId` is omitted, backend creates a new chat.
- If `chatId` is provided, backend validates it exists.
- If `persist` is `false`, `chatId` must be omitted. Backend does not create a chat and does not persist input messages, tool-call messages, assistant output, or `LlmCall` metadata.
- `clientRequestId` is optional and is only valid for a persisted stream with a `chatId`. Clients should generate one stable, unique value per user submission and reuse it when retrying a disconnected request.
- A retry with the same `chatId` and `clientRequestId` attaches to and replays the matching active stream. If that submission already completed, the endpoint replays `meta` and `done` without invoking the provider again. This makes retrying the initial streaming `POST` idempotent.
- `clientRequestId` values may be up to 128 characters. The server stores the value in `metadata.clientRequestId` on the submitted user message and completed assistant message.
- For persisted streams, backend stores only new non-assistant input history rows to avoid duplicates.
- `additionalSystemPrompt`, when present directly or loaded from stored chat settings, is prepended to the provider request as a `system` message and is not inserted into the persisted chat transcript by this endpoint.
- `enabledTools` limits Sybil-managed tools for this request. When omitted for a saved chat, the stored chat setting is used; otherwise all available tools are enabled by default. An empty array disables Sybil-managed tools.
@@ -70,7 +74,7 @@ Notes:
Persisted chat streams with a `chatId` are backend-owned active runs:
- Once started, the backend keeps the stream running even if the HTTP client disconnects or refreshes.
- While running, `GET /v1/active-runs` includes the `chatId`.
- Starting a second persisted stream for the same active `chatId` returns `409`.
- Starting a second persisted stream for the same active `chatId` returns `409`, unless its `clientRequestId` matches the active submission, in which case the existing stream is replayed.
- Clients can reattach with `POST /v1/chats/:chatId/stream/attach`.
## Attach Endpoint

7
server/src/llm/streaming.ts
View File

@@ -119,7 +119,12 @@ export async function* runMultiplexStream(req: MultiplexRequest): AsyncGenerator
if (shouldPersist && chatId && call) {
await prisma.$transaction(async (tx) => {
await tx.message.create({
data: { chatId, role: "assistant" as any, content: text },
data: {
chatId,
role: "assistant" as any,
content: text,
metadata: req.clientRequestId ? ({ clientRequestId: req.clientRequestId } as any) : undefined,
},
});
await tx.llmCall.update({
where: { id: call.id },

1
server/src/llm/types.ts
View File

@@ -33,6 +33,7 @@ export type ChatMessage = {
export type MultiplexRequest = {
chatId?: string;
persist?: boolean;
clientRequestId?: string;
provider: Provider;
model: string;
messages: ChatMessage[];

128
server/src/routes.ts
View File

@@ -88,7 +88,7 @@ function withRequestUserLocation<T extends { userLocation?: string }>(body: T, r
return body.userLocation ? body : { ...body, userLocation: inferRequestUserLocation(req) };
}
async function storeNonAssistantMessages(chatId: string, messages: IncomingChatMessage[]) {
async function storeNonAssistantMessages(chatId: string, messages: IncomingChatMessage[], clientRequestId?: string) {
const incoming = messages.filter((m) => m.role !== "assistant");
if (!incoming.length) return;
@@ -109,14 +109,21 @@ async function storeNonAssistantMessages(chatId: string, messages: IncomingChatM
const toInsert = sharedPrefix === existingNonAssistant.length ? incoming.slice(existingNonAssistant.length) : incoming;
if (!toInsert.length) return;
const finalUserMessageIndex = toInsert.map((message) => message.role).lastIndexOf("user");
await prisma.message.createMany({
data: toInsert.map((m) => ({
chatId,
role: m.role as any,
content: m.content,
name: m.name,
metadata: m.attachments?.length ? ({ attachments: m.attachments } as any) : undefined,
})),
data: toInsert.map((m, index) => {
const metadata = {
...(m.attachments?.length ? { attachments: m.attachments } : {}),
...(clientRequestId && index === finalUserMessageIndex ? { clientRequestId } : {}),
};
return {
chatId,
role: m.role as any,
content: m.content,
name: m.name,
metadata: Object.keys(metadata).length ? (metadata as any) : undefined,
};
}),
});
}
@@ -169,6 +176,7 @@ const CompletionStreamBody = z
.object({
chatId: z.string().optional(),
persist: z.boolean().optional(),
clientRequestId: z.string().trim().min(1).max(128).optional(),
provider: ProviderSchema,
model: z.string().min(1),
messages: z.array(CompletionMessageSchema),
@@ -186,6 +194,13 @@ const CompletionStreamBody = z
path: ["chatId"],
});
}
if (value.clientRequestId && (value.persist === false || !value.chatId)) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: "clientRequestId requires a persisted stream with chatId",
path: ["clientRequestId"],
});
}
});
function mergeAttachmentsIntoMetadata(metadata: unknown, attachments?: ChatAttachment[]) {
@@ -399,6 +414,7 @@ function buildSseHeaders(originHeader: string | undefined) {
type SearchRunRequest = z.infer<typeof SearchRunBody>;
const activeChatStreams = new Map<string, ActiveSseStream>();
const activeChatStreamRequestIds = new Map<string, string>();
const activeSearchStreams = new Map<string, ActiveSseStream>();
const STARRED_PROJECT_ID = "starred";
@@ -554,6 +570,7 @@ function writeSseEvent(reply: FastifyReply, event: SseStreamEvent) {
}
async function streamActiveRun(req: FastifyRequest, reply: FastifyReply, stream: ActiveSseStream) {
if (reply.raw.destroyed || reply.raw.writableEnded) return reply;
reply.raw.writeHead(200, buildSseHeaders(typeof req.headers.origin === "string" ? req.headers.origin : undefined));
reply.raw.flushHeaders?.();
@@ -588,10 +605,24 @@ function mapChatStreamEvent(ev: StreamEvent): SseStreamEvent {
return { event: ev.type, data: ev };
}
function startActiveChatStream(chatId: string, body: z.infer<typeof CompletionStreamBody>) {
function registerActiveChatStream(chatId: string, clientRequestId?: string) {
const stream = new ActiveSseStream();
activeChatStreams.set(chatId, stream);
if (clientRequestId) {
activeChatStreamRequestIds.set(chatId, clientRequestId);
} else {
activeChatStreamRequestIds.delete(chatId);
}
return stream;
}
function clearActiveChatStream(chatId: string, stream: ActiveSseStream) {
if (activeChatStreams.get(chatId) !== stream) return;
activeChatStreams.delete(chatId);
activeChatStreamRequestIds.delete(chatId);
}
function executeActiveChatStream(chatId: string, body: z.infer<typeof CompletionStreamBody>, stream: ActiveSseStream) {
void (async () => {
let sawTerminalEvent = false;
try {
@@ -611,13 +642,54 @@ function startActiveChatStream(chatId: string, body: z.infer<typeof CompletionSt
} catch (err) {
stream.complete({ event: "error", data: { message: getErrorMessage(err) } });
} finally {
activeChatStreams.delete(chatId);
clearActiveChatStream(chatId, stream);
}
})();
}
function startActiveChatStream(chatId: string, body: z.infer<typeof CompletionStreamBody>) {
const stream = registerActiveChatStream(chatId, body.clientRequestId);
executeActiveChatStream(chatId, body, stream);
return stream;
}
function getMetadataClientRequestId(metadata: unknown) {
if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return null;
const clientRequestId = (metadata as Record<string, unknown>).clientRequestId;
return typeof clientRequestId === "string" ? clientRequestId : null;
}
async function findCompletedChatSubmission(chatId: string, clientRequestId: string) {
const assistantMessages = await prisma.message.findMany({
where: { chatId, role: "assistant" as any },
orderBy: { createdAt: "desc" },
select: { content: true, metadata: true },
});
return assistantMessages.find((message) => getMetadataClientRequestId(message.metadata) === clientRequestId) ?? null;
}
function completeChatSubmissionStream(
stream: ActiveSseStream,
chatId: string,
body: z.infer<typeof CompletionStreamBody>,
assistantText: string
) {
stream.emit("meta", {
type: "meta",
chatId,
callId: null,
provider: body.provider,
model: body.model,
});
stream.complete({
event: "done",
data: {
type: "done",
text: assistantText,
},
});
}
async function executeSearchRunStream(searchId: string, body: SearchRunRequest, stream: ActiveSseStream) {
const startedAt = performance.now();
const query = body.query?.trim();
@@ -1353,15 +1425,39 @@ export async function registerRoutes(app: FastifyInstance) {
if (!exists) return app.httpErrors.notFound("chat not found");
}
// Store only new non-assistant messages to avoid duplicate history entries.
if (body.persist !== false && body.chatId) {
await storeNonAssistantMessages(body.chatId, body.messages);
}
if (body.persist !== false && body.chatId) {
if (activeChatStreams.has(body.chatId)) {
const activeStream = activeChatStreams.get(body.chatId);
if (activeStream) {
if (body.clientRequestId && activeChatStreamRequestIds.get(body.chatId) === body.clientRequestId) {
return streamActiveRun(req, reply, activeStream);
}
return app.httpErrors.conflict("chat completion already running");
}
if (body.clientRequestId) {
const reservedStream = registerActiveChatStream(body.chatId, body.clientRequestId);
try {
const completedSubmission = await findCompletedChatSubmission(body.chatId, body.clientRequestId);
if (completedSubmission) {
completeChatSubmissionStream(reservedStream, body.chatId, body, completedSubmission.content);
clearActiveChatStream(body.chatId, reservedStream);
return streamActiveRun(req, reply, reservedStream);
}
// Store only new non-assistant messages to avoid duplicate history entries.
await storeNonAssistantMessages(body.chatId, body.messages, body.clientRequestId);
const configuredBody = await applyStoredChatSettings(body);
executeActiveChatStream(body.chatId, configuredBody, reservedStream);
return streamActiveRun(req, reply, reservedStream);
} catch (err) {
reservedStream.complete({ event: "error", data: { message: getErrorMessage(err) } });
clearActiveChatStream(body.chatId, reservedStream);
throw err;
}
}
// Legacy requests without an idempotency key retain the original behavior.
await storeNonAssistantMessages(body.chatId, body.messages);
const stream = startActiveChatStream(body.chatId, await applyStoredChatSettings(body));
return streamActiveRun(req, reply, stream);
}

2
web/public/sw.js
View File

@@ -8,5 +8,5 @@ self.addEventListener("activate", (event) => {
self.addEventListener("fetch", (event) => {
if (event.request.mode !== "navigate") return;
event.respondWith(fetch(event.request));
event.respondWith(fetch(new Request(event.request, { cache: "no-store" })));
});

270
web/src/App.tsx
View File

@@ -206,6 +206,13 @@ function getActiveRunsAfterResume() {
return retryAfterAppResume(getActiveRuns);
}
function createClientRequestId() {
if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
return crypto.randomUUID();
}
return `web-${Date.now()}-${Math.random().toString(36).slice(2)}`;
}
function isRecoverableStreamDisconnect(error: unknown) {
if (error instanceof TypeError) return true;
const message = (error instanceof Error ? error.message : String(error)).toLowerCase();
@@ -1085,7 +1092,7 @@ export default function App() {
resetWorkspaceState();
};
const refreshCollections = async (preferredSelection?: SidebarSelection) => {
const refreshCollections = async (preferredSelection?: SidebarSelection, reportTransientError = true) => {
setIsLoadingCollections(true);
try {
const nextWorkspaceItems = await listWorkspaceItems();
@@ -1113,7 +1120,7 @@ export default function App() {
const message = err instanceof Error ? err.message : String(err);
if (message.includes("bearer token")) {
handleAuthFailure(message);
} else {
} else if (reportTransientError || !isRecoverableStreamDisconnect(err)) {
setError(message);
}
} finally {
@@ -2204,141 +2211,166 @@ export default function App() {
});
}
let streamErrorMessage: string | null = null;
const abortController = new AbortController();
chatStreamAbortRefs.current.set(chatId, abortController);
const target: SidebarSelection = { kind: "chat", id: chatId };
const clientRequestId = createClientRequestId();
try {
await runCompletionStream(
{
chatId,
provider,
model: selectedModel,
messages: requestMessages,
},
{
onMeta: (payload) => {
if (payload.chatId !== chatId) return;
while (true) {
let streamErrorMessage: string | null = null;
const abortController = new AbortController();
chatStreamAbortRefs.current.set(chatId, abortController);
try {
await runCompletionStream(
{
chatId,
clientRequestId,
provider,
model: selectedModel,
messages: requestMessages,
},
onToolCall: (payload) => {
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
return {
...current,
[chatId]: {
messages: upsertOptimisticToolMessage(pendingState.messages, payload, "temp-assistant-"),
},
};
});
},
onDelta: (payload) => {
if (!payload.text) return;
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
let updated = false;
const nextMessages = pendingState.messages.map((message, index, all) => {
const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-");
if (!isTarget) return message;
updated = true;
return { ...message, content: message.content + payload.text };
{
onMeta: (payload) => {
if (payload.chatId !== chatId) return;
},
onToolCall: (payload) => {
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
return {
...current,
[chatId]: {
messages: upsertOptimisticToolMessage(pendingState.messages, payload, "temp-assistant-"),
},
};
});
return updated ? { ...current, [chatId]: { messages: nextMessages } } : current;
});
},
onDone: (payload) => {
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
let updated = false;
const nextMessages = pendingState.messages.map((message, index, all) => {
const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-");
if (!isTarget) return message;
updated = true;
return { ...message, content: payload.text };
},
onDelta: (payload) => {
if (!payload.text) return;
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
let updated = false;
const nextMessages = pendingState.messages.map((message, index, all) => {
const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-");
if (!isTarget) return message;
updated = true;
return { ...message, content: message.content + payload.text };
});
return updated ? { ...current, [chatId]: { messages: nextMessages } } : current;
});
return updated ? { ...current, [chatId]: { messages: nextMessages } } : current;
});
},
onDone: (payload) => {
setPendingChatStates((current) => {
const pendingState = current[chatId];
if (!pendingState) return current;
let updated = false;
const nextMessages = pendingState.messages.map((message, index, all) => {
const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-");
if (!isTarget) return message;
updated = true;
return { ...message, content: payload.text };
});
return updated ? { ...current, [chatId]: { messages: nextMessages } } : current;
});
},
onError: (payload) => {
streamErrorMessage = payload.message;
},
},
onError: (payload) => {
streamErrorMessage = payload.message;
},
},
{ signal: abortController.signal }
);
{ signal: abortController.signal }
);
if (streamErrorMessage) {
throw new Error(streamErrorMessage);
}
if (streamErrorMessage) {
throw new Error(streamErrorMessage);
}
backgroundSuspendedChatStreamsRef.current.delete(chatId);
await refreshCollections();
const currentSelection = selectedItemRef.current;
if (currentSelection?.kind === "chat" && currentSelection.id === chatId) {
await refreshChat(chatId);
}
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
if (currentSelection?.kind === "chat" && currentSelection.id === chatId) {
requestSettleTranscriptTailSpacer();
}
return target;
} catch (err) {
const wasBackgroundSuspended =
abortController.signal.aborted && backgroundSuspendedChatStreamsRef.current.delete(chatId);
const shouldResumeStream =
wasBackgroundSuspended ||
(!abortController.signal.aborted && streamErrorMessage === null && isRecoverableStreamDisconnect(err));
backgroundSuspendedChatStreamsRef.current.delete(chatId);
const persistedChat = await retryAfterAppResume(() => getChat(chatId));
await refreshCollections(target, false);
if (isCurrentSelection(target)) {
setSelectedChat(persistedChat);
setSelectedSearch(null);
}
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
if (isCurrentSelection(target)) {
requestSettleTranscriptTailSpacer();
}
return target;
} catch (caughtError) {
let err: unknown = caughtError;
const wasBackgroundSuspended =
abortController.signal.aborted && backgroundSuspendedChatStreamsRef.current.delete(chatId);
const shouldResumeStream =
wasBackgroundSuspended ||
(!abortController.signal.aborted && streamErrorMessage === null && isRecoverableStreamDisconnect(err));
if (shouldResumeStream) {
if (chatStreamAbortRefs.current.get(chatId) === abortController) {
chatStreamAbortRefs.current.delete(chatId);
}
try {
const resumedRuns = await getActiveRunsAfterResume();
setActiveRuns(buildActiveRunsState(resumedRuns));
if (shouldResumeStream) {
try {
const persistedChat = await retryAfterAppResume(() => getChat(chatId));
const userMessageWasAccepted = hasNewPersistedUserMessage(
persistedChat,
previousMessageIds,
content,
attachments
);
setError(null);
if (isCurrentSelection(target)) {
setSelectedChat(persistedChat);
setSelectedSearch(null);
}
setPendingChatStates((current) => ({
...current,
[chatId]: {
messages: persistedChat.messages.concat(
...(userMessageWasAccepted ? [] : [optimisticUserMessage]),
{
...optimisticAssistantMessage,
id: `temp-assistant-resume-${Date.now()}`,
content: "",
}
),
},
}));
if (resumedRuns.chats.includes(chatId)) {
removePendingChatState(chatId);
await attachToActiveChatStream(chatId, true);
return target;
// Retrying the same idempotent request either starts an unaccepted
// submission or replays the existing backend-owned stream.
continue;
} catch (resumeError) {
err = resumeError;
}
const persistedChat = await retryAfterAppResume(() => getChat(chatId));
await refreshCollections(target);
if (isCurrentSelection(target)) {
setSelectedChat(persistedChat);
setSelectedSearch(null);
} else if (streamErrorMessage) {
try {
const persistedChat = await retryAfterAppResume(() => getChat(chatId));
if (hasNewPersistedUserMessage(persistedChat, previousMessageIds, content, attachments)) {
if (isCurrentSelection(target)) {
setSelectedChat(persistedChat);
setSelectedSearch(null);
setError(streamErrorMessage);
}
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
requestSettleTranscriptTailSpacer();
return target;
}
} catch (reconcileError) {
err = reconcileError;
}
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
if (hasNewPersistedUserMessage(persistedChat, previousMessageIds, content, attachments)) {
requestSettleTranscriptTailSpacer();
return target;
}
err = new Error(
wasBackgroundSuspended
? "The app was backgrounded before the message reached the server. Please send it again."
: "The connection dropped before the message reached the server. Please send it again."
);
} catch (resumeError) {
err = resumeError;
}
}
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
pendingReplyScrollRef.current = false;
setTranscriptTailSpacer(TRANSCRIPT_BOTTOM_GAP);
throw err;
} finally {
if (chatStreamAbortRefs.current.get(chatId) === abortController) {
chatStreamAbortRefs.current.delete(chatId);
removePendingChatState(chatId);
removeActiveRun("chat", chatId);
pendingReplyScrollRef.current = false;
setTranscriptTailSpacer(TRANSCRIPT_BOTTOM_GAP);
throw err;
} finally {
if (chatStreamAbortRefs.current.get(chatId) === abortController) {
chatStreamAbortRefs.current.delete(chatId);
}
}
}
};

1
web/src/lib/api.ts
View File

@@ -589,6 +589,7 @@ export async function runCompletionStream(
body: {
chatId?: string | null;
persist?: boolean;
clientRequestId?: string;
provider: Provider;
model: string;
messages: CompletionRequestMessage[];

15
web/src/pwa.ts
View File

@@ -2,8 +2,19 @@ export function registerServiceWorker() {
if (!import.meta.env.PROD || !("serviceWorker" in navigator)) return;
window.addEventListener("load", () => {
void navigator.serviceWorker.register("/sw.js").catch((error: unknown) => {
console.warn("Sybil service worker registration failed", error);
const hadController = Boolean(navigator.serviceWorker.controller);
let isReloadingForUpdate = false;
navigator.serviceWorker.addEventListener("controllerchange", () => {
if (!hadController || isReloadingForUpdate) return;
isReloadingForUpdate = true;
window.location.reload();
});
void navigator.serviceWorker
.register("/sw.js", { updateViaCache: "none" })
.then((registration) => registration.update())
.catch((error: unknown) => {
console.warn("Sybil service worker registration failed", error);
});
});
}