mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: harden resume and SSE replay

Browse Source
This commit is contained in:
User
2026-07-15 00:06:00 +02:00
parent e5366d14a6
commit 2b4797f133
18 changed files with 871 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/bridge/session-bridge.ts
+4 -2
View File
@@ -13,7 +13,7 @@ export type ClientEvent =
| { type: "activity_delta"; text: string }
| { type: "activity_event"; activity: ToolActivity }
| { type: "info"; [k: string]: any }
| { type: "system_event"; [k: string]: any };
| { type: "system_event"; event: string };
export type TurnState = "idle" | "running" | "waiting" | "failed";
@@ -61,7 +61,9 @@ export class SessionBridge {
this.emitToolActivity(m, "end");
// Tool updates and raw payloads remain intentionally dropped.
} else if (m.type === "system_event") {
this.fan(m as ClientEvent);
if (typeof m.event === "string" && m.event.trim() !== "") {
this.fan({ type: "system_event", event: m.event });
}
} else if (m.type === "agent_end") {
if (this.state !== "failed" && !this.pending) this.state = "idle";
this.fan({ type: "system_event", event: "agent_end" });
backend/src/routes/sessions.ts
+25 -5
View File
@@ -8,6 +8,18 @@ import type { ReadinessManager } from "../runtime/readiness-manager.js";
const BOOTSTRAP_FAILURE_MESSAGE =
"Session startup failed. Check configuration and connectivity, then Resume the session.";
const READINESS_FAILURE_MESSAGE =
"Session services are not ready. Check configuration and connectivity, then try again.";
function eventCursor(...values: unknown[]): number {
let cursor = 0;
for (const value of values.flatMap((item) => Array.isArray(item) ? item : [item])) {
if (typeof value !== "string" || !/^\d+$/.test(value)) continue;
const parsed = Number(value);
if (Number.isSafeInteger(parsed)) cursor = Math.max(cursor, parsed);
}
return cursor;
}
export function sessionRoutes(
app: FastifyInstance,
@@ -57,7 +69,7 @@ export function sessionRoutes(
const b = req.body as { question: string; name?: string };
const s = d.getSettings();
const ensure = await d.readiness.ensure(s.workspace ?? "");
if (!ensure.ok) return reply.code(503).send({ error: ensure.error ?? "Ollama/embeddings non disponibili" });
if (!ensure.ok) return reply.code(503).send({ error: READINESS_FAILURE_MESSAGE });
// Settings (global) supply workspace/provider/model/thinking. The new-question
// form sends only the question text. `workspace` selects the tht `-c <config>`.
const { id } = await d.tht.sessionNew({
@@ -115,7 +127,7 @@ export function sessionRoutes(
}
const settings = d.getSettings();
const ensure = await d.readiness.ensure(settings.workspace ?? "");
if (!ensure.ok) return reply.code(503).send({ error: ensure.error ?? "Ollama/embeddings non disponibili" });
if (!ensure.ok) return reply.code(503).send({ error: READINESS_FAILURE_MESSAGE });
const saved = manifest as { provider?: string; model?: string; thinking?: string } | null;
const options = {
provider: saved?.provider,
@@ -131,7 +143,7 @@ export function sessionRoutes(
bindRuntime(id, rt);
info(id, "Resuming session");
bootstrap(id, rt, d.mgr.configure(rt, options), null, () => d.mgr.start(id, rt, options));
return reply.code(200).send({ id });
return reply.code(200).send({ id, alreadyActive: false });
});
app.post("/sessions/:id/close", async (req) => {
const id = (req.params as { id: string }).id;
@@ -146,6 +158,10 @@ export function sessionRoutes(
app.get("/sessions/:id/events", (req, reply) => {
const id = (req.params as any).id;
const rt = d.mgr.get(id);
const afterId = eventCursor(
req.headers["last-event-id"],
(req.query as { lastEventId?: unknown }).lastEventId,
);
// Add CORS headers manually: reply.raw.writeHead bypasses Fastify's onSend hooks
// (where @fastify/cors injects headers), so we must set them explicitly here.
const origin = (req.headers.origin as string | undefined) ?? "*";
@@ -160,8 +176,12 @@ export function sessionRoutes(
// Send the handshake immediately. Without this, Node waits for the first event body and
// proxies/clients cannot establish an idle SSE subscription or inspect its headers.
reply.raw.flushHeaders();
const send = (event: string, data: object) => reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
const off = d.hub.subscribe(id, send, rt?.bridge.pendingWidget() ?? null);
const send = (event: string, data: object, eventId: number) =>
reply.raw.write(`id: ${eventId}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
const off = d.hub.subscribe(id, send, {
afterId,
pending: rt?.bridge.pendingWidget() ?? null,
});
req.raw.on("close", off);
});
app.post("/sessions/:id/rename", async (req, reply) => {
backend/src/sse/sse-hub.ts
+68 -24
View File
@@ -1,48 +1,92 @@
type Send = (event: string, data: object) => void;
type Send = (event: string, data: object, id: number) => void;
/**
* Per-session ring buffer of recent events.
*
* When the browser opens the SSE connection slightly after session creation
* (React re-render, navigation, etc.), events produced by the Pi process in
* that gap would be lost. The buffer replays them to late subscribers.
*
* `system_event` with `event: "agent_end"` acts as the natural sentinel:
* once a subscriber sees it, the buffer for that session is safe to clear.
* Event ids are transport identity: they are monotonically increasing for the lifetime of a
* session id, including across a cold Resume that clears the old buffer and subscribers.
*/
const BUFFER_LIMIT = 200;
interface BufferedEvent { event: string; data: object }
interface BufferedEvent {
id: number;
event: string;
data: object;
}
interface SubscribeOptions {
afterId?: number;
pending?: object | null;
}
function descriptorId(value: unknown): string | null {
if (!value || typeof value !== "object") return null;
const id = (value as { id?: unknown }).id;
return typeof id === "string" && id.trim() !== "" ? id : null;
}
function bufferedGateId(item: BufferedEvent): string | null {
if (item.event !== "ui_request") return null;
const request = (item.data as { ui_request?: unknown }).ui_request;
return descriptorId(request);
}
export class SseHub {
private subs = new Map<string, Set<Send>>();
private buffers = new Map<string, BufferedEvent[]>();
private lastIds = new Map<string, number>();
subscribe(sessionId: string, send: Send, pending?: object | null): () => void {
subscribe(
sessionId: string,
send: Send,
options: SubscribeOptions = {},
): () => void {
if (!this.subs.has(sessionId)) this.subs.set(sessionId, new Set());
this.subs.get(sessionId)!.add(send);
// Replay buffered events so late subscribers don't miss the session lifecycle.
const buf = this.buffers.get(sessionId);
if (buf) for (const { event, data } of buf) send(event, data);
const afterId = Number.isSafeInteger(options.afterId) && (options.afterId ?? 0) >= 0
? options.afterId ?? 0
: 0;
const buf = this.buffers.get(sessionId) ?? [];
for (const item of buf) {
if (item.id > afterId) send(item.event, item.data, item.id);
}
// A pending gate normally already exists in the buffer. If ring-buffer eviction removed it,
// give the fallback a fresh transport id and buffer it for future cursor-based reconnects.
// Matching is by stable descriptor identity only, never by event content.
const pendingId = descriptorId(options.pending);
const pendingBuffered = pendingId !== null && buf.some((item) => bufferedGateId(item) === pendingId);
if (pendingId !== null && !pendingBuffered) {
const data = { type: "ui_request", ui_request: options.pending! };
const item = this.buffer(sessionId, "ui_request", data);
send(item.event, item.data, item.id);
}
// Match the live-event shape (hub.publish sends the full ClientEvent):
// both carry { type, ui_request } so re-emit and live widgets are identical.
if (pending) send("ui_request", { type: "ui_request", ui_request: pending });
return () => this.subs.get(sessionId)?.delete(send);
}
publish(sessionId: string, event: string, data: object): void {
// Buffer the event for late subscribers.
let buf = this.buffers.get(sessionId);
if (!buf) { buf = []; this.buffers.set(sessionId, buf); }
buf.push({ event, data });
if (buf.length > BUFFER_LIMIT) buf.splice(0, buf.length - BUFFER_LIMIT);
for (const s of this.subs.get(sessionId) ?? []) s(event, data);
publish(sessionId: string, event: string, data: object): number {
const item = this.buffer(sessionId, event, data);
for (const send of this.subs.get(sessionId) ?? []) send(event, data, item.id);
return item.id;
}
/** Clear buffer and subscribers for a finished session. */
private buffer(sessionId: string, event: string, data: object): BufferedEvent {
const id = (this.lastIds.get(sessionId) ?? 0) + 1;
this.lastIds.set(sessionId, id);
let buf = this.buffers.get(sessionId);
if (!buf) {
buf = [];
this.buffers.set(sessionId, buf);
}
const item = { id, event, data };
buf.push(item);
if (buf.length > BUFFER_LIMIT) buf.splice(0, buf.length - BUFFER_LIMIT);
return item;
}
/** Clear buffered/runtime bindings while retaining session event-id monotonicity. */
clear(sessionId: string): void {
this.buffers.delete(sessionId);
this.subs.delete(sessionId);