From 3d1d2001ae5dbefae4ae3625e8567b04f0468847 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:29:34 +0800 Subject: [PATCH 01/11] fix(server): enforce owner guard on resume DELETE suspension MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The DELETE /:id/suspensions/:suspensionId route only checked sessionId consistency, not ownership — on hosted tiers anyone knowing the session and suspension ids could delete another session's suspension state. Run resolveSessionParam first, matching its sibling resume routes (H-1). --- apps/server/src/routes/api/resume.ts | 3 +++ apps/server/tests/api/resume.test.ts | 38 ++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/apps/server/src/routes/api/resume.ts b/apps/server/src/routes/api/resume.ts index b796d886..5e9a52c3 100644 --- a/apps/server/src/routes/api/resume.ts +++ b/apps/server/src/routes/api/resume.ts @@ -363,6 +363,9 @@ resumeRoutes.post("/:id/resume", async (c) => { // ── DELETE (abandon) ───────────────────────────────────────────── resumeRoutes.delete("/:id/suspensions/:suspensionId", async (c) => { + const guard = await resolveSessionParam(c); + if (!guard.ok) return guard.response; + const sessionId = c.req.param("id"); const suspensionId = c.req.param("suspensionId"); const store = c.get("store"); diff --git a/apps/server/tests/api/resume.test.ts b/apps/server/tests/api/resume.test.ts index d5457335..3444b5da 100644 --- a/apps/server/tests/api/resume.test.ts +++ b/apps/server/tests/api/resume.test.ts @@ -562,6 +562,44 @@ describe("Resume Routes", () => { expect(res.status).toBe(404); }); + + // Audit 2026-07-16 H-1: this DELETE must enforce the session-owner guard + // on hosted tiers like its sibling routes, not just a sessionId consistency + // check. Anonymous deletion of another session's suspension is a + // cross-tenant destructive write. + it("denies anonymous suspension deletion on a hosted tier (owner guard)", async () => { + const prevTier = process.env.DEPLOYMENT_TIER; + const prevOp = process.env.COVEL_DESKTOP_REST_TOKEN; + process.env.DEPLOYMENT_TIER = "commercial"; + process.env.COVEL_DESKTOP_REST_TOKEN = "operator-secret"; + try { + await createSuspension(store); + const app = createTestApp(makeDefaultDeps(store)); + + const anon = await app.request( + "/api/sessions/sess-1/suspensions/susp-1", + { method: "DELETE" }, + ); + expect(anon.status).toBe(401); + // The denied attempt must not have deleted the suspension. + expect(await store.getSuspension("susp-1")).not.toBeNull(); + + // The operator master credential passes the guard and deletes. + const op = await app.request( + "/api/sessions/sess-1/suspensions/susp-1", + { + method: "DELETE", + headers: { authorization: "Bearer operator-secret" }, + }, + ); + expect(op.status).toBe(200); + } finally { + if (prevTier === undefined) delete process.env.DEPLOYMENT_TIER; + else process.env.DEPLOYMENT_TIER = prevTier; + if (prevOp === undefined) delete process.env.COVEL_DESKTOP_REST_TOKEN; + else process.env.COVEL_DESKTOP_REST_TOKEN = prevOp; + } + }); }); describe("GET /api/sessions/:id/suspensions", () => { From ccda89e5c4413b4c8c83e2e2e9f1650a6351e72a Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:29:42 +0800 Subject: [PATCH 02/11] fix(ai-provider): fail closed on redirects in plugin fetch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetchWithRetry validated only the initial URL and followed redirects, so a plugin hitting an endpoint that returns 302 Location: http://169.254.169.254/… was dialed straight into cloud metadata (SSRF). Force redirect: "manual" (after ...rest so a caller can't override) and throw on any 3xx, matching the core provider HTTP path (H-2). --- packages/ai-provider/src/plugin-utils.ts | 14 +++++++++++++- .../tests/plugin-utils-dns-safety.test.ts | 16 ++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/packages/ai-provider/src/plugin-utils.ts b/packages/ai-provider/src/plugin-utils.ts index 9e438567..7eddc936 100644 --- a/packages/ai-provider/src/plugin-utils.ts +++ b/packages/ai-provider/src/plugin-utils.ts @@ -106,11 +106,23 @@ export async function fetchWithRetry( // following changed DNS to an address that was never policy-checked. const dispatcher = await createPinnedDispatcher(url); try { - return await fetch(input, { + // Only the initial URL is SSRF-checked; a redirect Location is not, and + // undici skips the pinning lookup for an IP-literal host. Follow the core + // provider path: force manual redirect handling and fail closed on a 3xx + // so a `302 Location: http://169.254.169.254/…` can't reach internal + // hosts. `redirect` is placed after `...rest` to override any caller value. + const response = await fetch(input, { ...rest, + redirect: "manual", ...(signal ? { signal } : {}), dispatcher, } as RequestInit); + if (response.status >= 300 && response.status < 400) { + throw new Error( + `baseUrl rejected by SSRF policy: refusing to follow redirect (HTTP ${response.status}) from "${url}".`, + ); + } + return response; } finally { // close() drains active response bodies before releasing the pool, so it // is safe to start shutdown once fetch has returned the response headers. diff --git a/packages/ai-provider/tests/plugin-utils-dns-safety.test.ts b/packages/ai-provider/tests/plugin-utils-dns-safety.test.ts index a4d7e423..6c5aee5a 100644 --- a/packages/ai-provider/tests/plugin-utils-dns-safety.test.ts +++ b/packages/ai-provider/tests/plugin-utils-dns-safety.test.ts @@ -80,6 +80,22 @@ describe("DNS SSRF safety", () => { }); }); + it("refuses to follow a redirect (SSRF via 3xx Location to an IP-literal)", async () => { + lookupMock.mockResolvedValue([{ address: "93.184.216.34", family: 4 }]); + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(null, { + status: 302, + headers: { location: "http://169.254.169.254/latest/meta-data/" }, + }), + ); + + await expect( + fetchWithRetry("https://provider.example.test/v1", { maxRetries: 0 }), + ).rejects.toThrow(/refusing to follow redirect/); + // The initial request must have opted out of automatic redirect following. + expect(fetchMock.mock.calls[0]?.[1]).toMatchObject({ redirect: "manual" }); + }); + it("only allows localhost when every DNS answer is loopback", async () => { lookupMock.mockResolvedValue([{ address: "10.0.0.8", family: 4 }]); const fetchMock = vi.spyOn(globalThis, "fetch"); From 8ba06a73276f06f06eb1c67e67974265585b7cd4 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:29:51 +0800 Subject: [PATCH 03/11] fix(events): detect lost tail frames via transport heartbeat A cross-pod receiver only detected a seq gap when a later frame arrived out of order. A lost *tail* frame (last NOTIFY dropped, no successor) left the receiver silently stale until the next event or reconnect. Each origin now publishes a periodic heartbeat carrying its high-water transportSeq through the same per-session outbox as real frames, so the receiver treats senderSeq >= expected as a hole and resets. seq/epoch are captured at enqueue time so a concurrent emit cannot make the heartbeat overtake a real frame and cause a false reset (M-1). --- packages/events/src/event-bus.ts | 102 ++++++++++- .../events/tests/event-bus-transport.test.ts | 158 ++++++++++++++++++ 2 files changed, 259 insertions(+), 1 deletion(-) diff --git a/packages/events/src/event-bus.ts b/packages/events/src/event-bus.ts index a4fb2437..4f4a6059 100644 --- a/packages/events/src/event-bus.ts +++ b/packages/events/src/event-bus.ts @@ -129,6 +129,12 @@ export const PERSIST_QUEUE_MAX = 1000; export const RECEIVE_PENDING_MAX = 64; /** How long a receiver waits on a transport seq hole before skipping it. */ export const RECEIVE_GAP_FLUSH_MS = 2000; +/** + * How often an origin publishes a liveness heartbeat carrying its current + * high-water transportSeq per active session (audit 2026-07-16 M-1). Bounds how + * long a lost *tail* frame can leave a cross-pod receiver silently stale. + */ +export const TRANSPORT_HEARTBEAT_MS = 15_000; /** FIFO cap on receive-ordering states across transport streams. */ export const RECEIVE_STATES_MAX = 1024; @@ -167,6 +173,17 @@ interface TransportFrame { readonly seq: number; readonly event?: SubscriptionEvent; readonly ref?: { readonly sessionId: string; readonly eventId: string }; + /** + * Liveness frame (audit 2026-07-16 M-1): carries the origin's current + * high-water `transportSeq` so a receiver can detect a lost *tail* frame that + * has no successor to trigger the normal reorder-gap path. Published through + * the same per-session outbox as real frames, so a heartbeat with seq N is + * only ever sent after real frame N — `seq < expected` therefore reliably + * means "caught up", never a false gap. Carries `sessionId` explicitly since + * there is no event/ref to read it from. + */ + readonly heartbeat?: boolean; + readonly sessionId?: string; } /** Per-(origin, session) receive-side ordering state. */ @@ -226,6 +243,13 @@ function parseTransportFrame(payload: string): TransportFrame | undefined { if (typeof frame.seq !== "number" || !Number.isSafeInteger(frame.seq)) { return undefined; } + if (frame.heartbeat === true) { + // Liveness frame: no event/ref, so it must name its session explicitly. + if (typeof frame.sessionId !== "string" || frame.sessionId.length === 0) { + return undefined; + } + return raw as TransportFrame; + } if (frame.event !== undefined) { const e = frame.event as Record | null; if ( @@ -531,6 +555,35 @@ export function createEventBus( rs.pending.clear(); } + /** + * A heartbeat states the origin's high-water transportSeq. Because heartbeats + * ride the same per-session outbox as real frames, one with seq N arrives + * only after real frame N. So `senderSeq < expected` means we are caught up; + * `senderSeq >= expected` means frame(s) in [expected, senderSeq] were lost + * with no successor to trip the normal gap path — surface it as a reset. + */ + function handleHeartbeat(rs: ReceiveState, senderSeq: number): void { + if (senderSeq < rs.expected) return; // caught up (or a stale heartbeat) + if (rs.timer) { + clearTimeout(rs.timer); + rs.timer = undefined; + } + console.warn( + `[EventBus] transport tail gap (expected ${rs.expected}, sender high-water ${senderSeq}) — resetting replay`, + ); + rs.chain = rs.chain + .then(() => invalidateReplay(rs.sessionId)) + .catch((err) => { + console.error("[EventBus] replay invalidation failed:", err); + }); + // Deliver whatever is parked, in order, skipping the hole(s). + const seqs = [...rs.pending.keys()].sort((a, b) => a - b); + for (const seq of seqs) scheduleDeliver(rs, rs.pending.get(seq)!); + rs.pending.clear(); + const maxParked = seqs.length > 0 ? seqs[seqs.length - 1]! : 0; + rs.expected = Math.max(senderSeq, maxParked) + 1; + } + function handleTransportFrame(payload: string): void { const frame = parseTransportFrame(payload); if (!frame) { @@ -538,12 +591,18 @@ export function createEventBus( return; } if (frame.origin === originId) return; // self-echo - const sessionId = frame.event?.sessionId ?? frame.ref!.sessionId; + const sessionId = + frame.event?.sessionId ?? frame.ref?.sessionId ?? frame.sessionId; + if (!sessionId) return; // malformed (no session to route to) const stream = frame.stream ?? "legacy"; const rs = getReceiveState( JSON.stringify([frame.origin, sessionId, stream]), sessionId, ); + if (frame.heartbeat) { + handleHeartbeat(rs, frame.seq); + return; + } if (frame.seq < rs.expected) return; // duplicate/late if (frame.seq === rs.expected) { scheduleDeliver(rs, frame); @@ -569,8 +628,49 @@ export function createEventBus( } } + /** + * Publish a liveness heartbeat per active fanned-out session (audit + * 2026-07-16 M-1). Rides the same per-session outbox as real frames, so the + * seq it carries is a safe high-water mark that never races ahead of an + * in-flight real frame. + */ + function emitTransportHeartbeats(): void { + if (!transport) return; + for (const [sessionId, state] of sessions) { + if (state.transportSeq <= 0) continue; // never fanned out — nothing to signal + // Capture seq + epoch NOW, at enqueue time. Reading them inside the task + // would let an emit() that lands between this tick and the task's + // microtask bump transportSeq first, so the heartbeat would claim that + // frame's seq and (being enqueued earlier) publish before it — a false + // gap + dropped frame on the receiver. Capturing here keeps the exact + // invariant: heartbeat(seq N) is enqueued after frame N and before frame + // N+1, so FIFO publish order never lets it overtake a real frame. + const seq = state.transportSeq; + const epoch = state.epoch; + enqueueOutbox(sessionId, () => { + // Skip if replay was invalidated since enqueue (epoch rolled, seq reset). + if (state.epoch !== epoch) return Promise.resolve(); + const frame = JSON.stringify({ + origin: originId, + stream: epoch, + seq, + heartbeat: true, + sessionId, + } satisfies TransportFrame); + return Promise.resolve(transport.publish(frame)); + }); + } + } + if (transport) { transport.subscribe(handleTransportFrame); + // unref so the heartbeat never keeps the process alive on its own; it dies + // with the process on shutdown (the bus is a process-lifetime singleton). + const heartbeatTimer = setInterval( + emitTransportHeartbeats, + TRANSPORT_HEARTBEAT_MS, + ); + heartbeatTimer.unref?.(); } const bus: EventBus = { diff --git a/packages/events/tests/event-bus-transport.test.ts b/packages/events/tests/event-bus-transport.test.ts index 040ed881..f2e7dfe8 100644 --- a/packages/events/tests/event-bus-transport.test.ts +++ b/packages/events/tests/event-bus-transport.test.ts @@ -9,6 +9,7 @@ import { createEventBus, RECEIVE_GAP_FLUSH_MS, RECEIVE_STATES_MAX, + TRANSPORT_HEARTBEAT_MS, type EventBusTransport, } from "../src/event-bus.js"; @@ -413,3 +414,160 @@ describe("EventBus transport fan-out (audit R-02)", () => { warnSpy.mockRestore(); }); }); + +describe("EventBus transport heartbeat (audit 2026-07-16 M-1)", () => { + const flushMicrotasks = () => new Promise((r) => setTimeout(r, 0)); + + const realFrame = (seq: number, sessionId: string, type: string): string => + JSON.stringify({ + origin: "remote-hb", + stream: "boot-1", + seq, + event: { + id: `remote:${seq}`, + topic: "state", + type, + sessionId, + timestamp: new Date().toISOString(), + payload: {}, + }, + }); + + const heartbeat = (seq: number, sessionId: string): string => + JSON.stringify({ + origin: "remote-hb", + stream: "boot-1", + seq, + heartbeat: true, + sessionId, + }); + + it("resets when a heartbeat reveals a lost tail frame", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const hub = createHub(); + const bus = createEventBus(undefined, { transport: hub.connect() }); + const received: SubscriptionEvent[] = []; + const resets: Array<{ sessionId: string; reason: string }> = []; + bus.onEmit((event) => received.push(event)); + bus.onReset?.((reset) => resets.push(reset)); + + // Frame 1 arrives; the receiver is caught up (expects seq 2 next). + hub.broadcast(realFrame(1, "sess-tail", "t1")); + await flushMicrotasks(); + expect(received.map((e) => e.type)).toEqual(["t1"]); + expect(resets).toEqual([]); + + // Frame 2 is lost. A heartbeat announces the sender's high-water = 2, + // so the receiver must detect the hole and reset (no successor frame + // would otherwise ever trip the gap path). + hub.broadcast(heartbeat(2, "sess-tail")); + await flushMicrotasks(); + expect(resets).toEqual([ + { sessionId: "sess-tail", reason: "transport-gap" }, + ]); + } finally { + warnSpy.mockRestore(); + } + }); + + it("does not reset when a heartbeat matches an already-delivered stream", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const hub = createHub(); + const bus = createEventBus(undefined, { transport: hub.connect() }); + const resets: Array<{ sessionId: string; reason: string }> = []; + bus.onReset?.((reset) => resets.push(reset)); + + hub.broadcast(realFrame(1, "sess-live", "t1")); + hub.broadcast(realFrame(2, "sess-live", "t2")); + await flushMicrotasks(); + + // Caught up: expected is 3, heartbeat high-water is 2 → no gap. + hub.broadcast(heartbeat(2, "sess-live")); + await flushMicrotasks(); + expect(resets).toEqual([]); + } finally { + warnSpy.mockRestore(); + } + }); + + it("does not false-reset when an emit races the heartbeat tick (capture-at-enqueue)", async () => { + vi.useFakeTimers(); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const published: string[] = []; + const hub = createHub(); + const inner = hub.connect(); + const recording: EventBusTransport = { + publish: (payload) => { + published.push(payload); + return inner.publish(payload); + }, + subscribe: inner.subscribe, + }; + const busA = createEventBus(undefined, { transport: recording }); + const busB = createEventBus(undefined, { transport: hub.connect() }); + const receivedB: SubscriptionEvent[] = []; + const resetsB: Array<{ sessionId: string; reason: string }> = []; + busB.onEmit((event) => receivedB.push(event)); + busB.onReset?.((reset) => resetsB.push(reset)); + + // Frame 1 delivered; B is caught up. + busA.emit(makeMessage({ sessionId: "sess-race", topic: "a" })); + await vi.advanceTimersByTimeAsync(0); + expect(receivedB).toHaveLength(1); + + // Heartbeat tick fires (captures seq 1), THEN a second emit lands before + // the outbox microtasks flush — the exact race the fix targets. + vi.advanceTimersByTime(TRANSPORT_HEARTBEAT_MS); + busA.emit(makeMessage({ sessionId: "sess-race", topic: "b" })); + await vi.advanceTimersByTimeAsync(0); + + // The heartbeat must carry seq 1 (not the raced 2), so B sees no gap and + // still receives frame 2. + const hb = published + .map((p) => JSON.parse(p) as Record) + .find((f) => f.heartbeat === true); + expect(hb).toMatchObject({ seq: 1 }); + expect(resetsB).toEqual([]); + expect(receivedB).toHaveLength(2); + } finally { + warnSpy.mockRestore(); + vi.useRealTimers(); + } + }); + + it("periodically publishes a heartbeat carrying the high-water seq", async () => { + vi.useFakeTimers(); + try { + const published: string[] = []; + const hub = createHub(); + const inner = hub.connect(); + const recording: EventBusTransport = { + publish: (payload) => { + published.push(payload); + return inner.publish(payload); + }, + subscribe: inner.subscribe, + }; + const busA = createEventBus(undefined, { transport: recording }); + + busA.emit(makeMessage({ sessionId: "sess-hb" })); + await vi.advanceTimersByTimeAsync(0); // flush the real frame's outbox + published.length = 0; + + await vi.advanceTimersByTimeAsync(TRANSPORT_HEARTBEAT_MS); + const hb = published + .map((p) => JSON.parse(p) as Record) + .find((f) => f.heartbeat === true); + expect(hb).toMatchObject({ + heartbeat: true, + sessionId: "sess-hb", + seq: 1, + }); + } finally { + vi.useRealTimers(); + } + }); +}); From ad28f562ef06593c2c4d81f3d469b431edc8ed1b Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:29:59 +0800 Subject: [PATCH 04/11] fix(server): periodically re-sweep orphaned background jobs The orphaned-job sweep ran only once at boot and skipped jobs younger than the staleness threshold, so a job pending <15min before a crash hung forever. Re-run it on an unref'd interval (drained on shutdown) and anchor freshness on max(startedAt, updatedAt) so a job that writes progress is never force-failed by the now-periodic sweep (M-2). --- apps/server/src/app.ts | 1 + apps/server/src/routes/api/bootstrap.ts | 27 ++++++++++++++++++- apps/server/src/routes/api/plugin-rpc/jobs.ts | 22 ++++++++++++++- 3 files changed, 48 insertions(+), 2 deletions(-) diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 44ea1730..bcc8df6c 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -376,6 +376,7 @@ async function drainPhase( export const drainServerResources = async (): Promise => { await drainPhase("stop world watchers", () => stopWatchers()); + await drainPhase("stop background sweeps", () => api.stopBackgroundSweeps()); await drainPhase("flush event bus", () => api.eventBus.flush()); await drainPhase("close data store", () => store.close()); if (lockSql) { diff --git a/apps/server/src/routes/api/bootstrap.ts b/apps/server/src/routes/api/bootstrap.ts index bfa7d5c2..53ef8bb9 100644 --- a/apps/server/src/routes/api/bootstrap.ts +++ b/apps/server/src/routes/api/bootstrap.ts @@ -59,7 +59,10 @@ import type { MediaStore } from "@covel/store"; import type { MediaStoreBackend, VectorBackend } from "@covel/store"; import { resumeRoutes } from "./resume.js"; import { maybeSweepExpiredSuspensions } from "./suspension-sweep.js"; -import { sweepStalePendingJobs } from "./plugin-rpc/jobs.js"; +import { + sweepStalePendingJobs, + JOB_SWEEP_INTERVAL_MS, +} from "./plugin-rpc/jobs.js"; import { snapshotRoutes } from "./snapshots.js"; import { lorebookRoutes } from "./lorebook.js"; import { runtimeOutputRoutes } from "./runtime-outputs.js"; @@ -191,6 +194,12 @@ export interface ApiBootstrapResult { * the cache for that session). */ readonly prepareToolsForSession: (sessionId: string) => Promise; + /** + * Stop the periodic background sweeps (orphaned-job reaper). Called by the + * graceful-shutdown drain; the timer is also `unref()`'d so it never blocks + * process exit on its own. + */ + readonly stopBackgroundSweeps: () => void; } // ── Bootstrap function ─────────────────────────────────────────── @@ -266,6 +275,21 @@ export async function bootstrapApi( ), ); + // Audit 2026-07-16 M-2: the one-time boot sweep skips jobs younger than the + // staleness threshold and never re-scans, so a job that was pending for + // { + void sweepStalePendingJobs(store).catch((err: unknown) => + console.warn( + "[job-sweep] periodic sweep failed:", + err instanceof Error ? err.message : String(err), + ), + ); + }, JOB_SWEEP_INTERVAL_MS); + jobSweepTimer.unref(); + const { registry, discoveryMap, manifestCache } = await discoverAndRegisterPlugins({ pluginsDir: config.pluginsDir, @@ -627,5 +651,6 @@ export async function bootstrapApi( eventBus, compactorRunner, prepareToolsForSession, + stopBackgroundSweeps: () => clearInterval(jobSweepTimer), }; } diff --git a/apps/server/src/routes/api/plugin-rpc/jobs.ts b/apps/server/src/routes/api/plugin-rpc/jobs.ts index 3c0488a3..c62ecfe9 100644 --- a/apps/server/src/routes/api/plugin-rpc/jobs.ts +++ b/apps/server/src/routes/api/plugin-rpc/jobs.ts @@ -123,6 +123,13 @@ export async function writePluginJob( */ const STALE_PENDING_JOB_MS = 15 * 60_000; +/** + * How often the periodic re-sweep runs (audit 2026-07-16 M-2). Smaller than + * the staleness threshold so a job orphaned just before a crash is reaped soon + * after it crosses `STALE_PENDING_JOB_MS`, instead of only at the next boot. + */ +export const JOB_SWEEP_INTERVAL_MS = 5 * 60_000; + /** * One-shot boot sweep: mark stale `pending` background-job rows as `failed` * so clients polling a job that died with a previous process get a terminal @@ -160,8 +167,21 @@ export async function sweepStalePendingJobs( readonly turnId?: string; }; if (value?.status !== "pending") continue; + // Freshness is the most recent write to the row, not just its start: + // a job that writes progress bumps `updatedAt`, and the sweep now runs + // periodically (not only at boot), so anchoring on `startedAt` alone + // would force-fail a legitimately long-running job. A job that never + // writes progress still ages out at `staleMs` from its start (unchanged + // boot-recovery behavior). const startedAtMs = Date.parse(value.startedAt ?? row.updatedAt); - if (Number.isFinite(startedAtMs) && now - startedAtMs < staleMs) continue; + const updatedAtMs = Date.parse(row.updatedAt); + const lastActivityMs = Math.max( + Number.isFinite(startedAtMs) ? startedAtMs : -Infinity, + Number.isFinite(updatedAtMs) ? updatedAtMs : -Infinity, + ); + if (Number.isFinite(lastActivityMs) && now - lastActivityMs < staleMs) { + continue; + } const startedAt = value.startedAt ?? row.createdAt; const completedAt = new Date(now).toISOString(); From 79bdceb7d83b317d625116430d2f0ca889e3aad9 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:30:06 +0800 Subject: [PATCH 05/11] fix(server): strip owner-token hash and gate event injection Session create/get/patch/list echoed metadata.ownerTokenHash back to the caller; strip it via sanitizeSessionForResponse (L-2). POST /events/emit was gated only on NODE_ENV, so a non-production hosted boot accepted arbitrary cross-session events; add a session-owner check (no-op on self) (L-3). --- apps/server/src/routes/api/events.ts | 7 +++++++ apps/server/src/routes/api/session.ts | 27 +++++++++++++++++++++++---- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/apps/server/src/routes/api/events.ts b/apps/server/src/routes/api/events.ts index 8cd0c3e3..d9d2929a 100644 --- a/apps/server/src/routes/api/events.ts +++ b/apps/server/src/routes/api/events.ts @@ -10,6 +10,7 @@ import type { DataStore } from "@covel/store"; import type { CovelMessage } from "@covel/shared"; import { isEnvTruthy, readRuntimeEnv } from "@covel/shared"; import { errorBody } from "../../api-error.js"; +import { checkSessionOwnerById } from "./session/session-guard.js"; type Env = { Variables: { @@ -41,6 +42,12 @@ eventRoutes.post("/emit", async (c) => { return c.json(errorBody("topic and sessionId are required"), 400); } + // Audit 2026-07-16 L-3: on hosted tiers, gate event injection on the target + // session's owner token so a non-production demo boot can't accept arbitrary + // cross-session events. Strict no-op on self/desktop (unenforced tiers). + const denied = await checkSessionOwnerById(c, c.get("store"), body.sessionId); + if (denied) return denied; + const message: CovelMessage = { id: crypto.randomUUID(), type: "event", diff --git a/apps/server/src/routes/api/session.ts b/apps/server/src/routes/api/session.ts index b46e914e..95468330 100644 --- a/apps/server/src/routes/api/session.ts +++ b/apps/server/src/routes/api/session.ts @@ -84,6 +84,20 @@ type Env = { export const sessionRoutes = new Hono(); +/** + * Strip the persisted owner-token hash before returning a session over the + * wire (audit 2026-07-16 L-2). It is an internal credential check; a caller + * has no use for its own hash and it should not travel in responses. + */ +function sanitizeSessionForResponse< + T extends { readonly metadata?: Record | null }, +>(session: T): T { + const metadata = session.metadata; + if (!metadata || !(SESSION_OWNER_TOKEN_HASH_KEY in metadata)) return session; + const { [SESSION_OWNER_TOKEN_HASH_KEY]: _omit, ...rest } = metadata; + return { ...session, metadata: rest }; +} + /** * Keep persisted active plugins aligned with the in-memory approval gate. * Community server code is never restored implicitly after create/fork or a @@ -177,7 +191,7 @@ sessionRoutes.get("/", async (c) => { // can show RAG status badges without an extra round-trip per row. // listVectorModels is called once and shared across all sessions. const decorated = await decorateSessionList(store, filtered); - return c.json({ items: decorated }); + return c.json({ items: decorated.map(sanitizeSessionForResponse) }); }); // POST /sessions @@ -335,7 +349,10 @@ sessionRoutes.post("/", async (c) => { // `ownerToken` is returned exactly once — it is never readable again // (only its hash is stored). Clients on hosted tiers must persist it. - return c.json({ ...session, ownerToken: owner.token }); + return c.json({ + ...sanitizeSessionForResponse(session), + ownerToken: owner.token, + }); }); // ── Instance endpoints ────────────────────────────────────────── @@ -421,7 +438,9 @@ sessionRoutes.get("/:id", async (c) => { const guard = await resolveSessionParam(c); if (!guard.ok) return guard.response; const session = guard.session; - return c.json(await withEmbeddingMetadata(store, session)); + return c.json( + sanitizeSessionForResponse(await withEmbeddingMetadata(store, session)), + ); }); // PATCH /sessions/:id @@ -458,7 +477,7 @@ sessionRoutes.patch("/:id", async (c) => { } // Return merged result to avoid a second DB read - return c.json({ ...session, ...updates }); + return c.json(sanitizeSessionForResponse({ ...session, ...updates })); }); // DELETE /sessions/:id From c4cdaa537d36af35fae6d810853b15b9fcb5861a Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:30:15 +0800 Subject: [PATCH 06/11] fix(desktop): sender-origin check on remaining IPC channels retry-startup, open-{logs,config,data}-dir, restart-server, pick-data-dir and import:pick-{plugin,world} handled IPC without the isTrustedSender check the secret channels use. Add it as defense-in-depth behind nav-pinning so an untrusted frame can't drive restart / dir / import actions (L-4). --- apps/desktop/src/ipc-handlers.ts | 40 +++++++++++++++++++++++++------- 1 file changed, 32 insertions(+), 8 deletions(-) diff --git a/apps/desktop/src/ipc-handlers.ts b/apps/desktop/src/ipc-handlers.ts index 781c8c98..111f2967 100644 --- a/apps/desktop/src/ipc-handlers.ts +++ b/apps/desktop/src/ipc-handlers.ts @@ -86,28 +86,39 @@ export function registerDesktopIpcHandlers({ }; }); - ipcMain.handle("covel:retry-startup", () => { + // Audit 2026-07-16 L-4: these mutate the app / open OS paths, so gate them on + // the same trusted-sender check the secret channels use — defense in depth + // behind nav-pinning, so an untrusted frame can't drive restart/dir actions. + ipcMain.handle("covel:retry-startup", (event) => { + if (!isTrustedSender(event, "covel:retry-startup")) return; retryStartup(); }); - ipcMain.handle("covel:open-logs-dir", async () => { + ipcMain.handle("covel:open-logs-dir", async (event) => { + if (!isTrustedSender(event, "covel:open-logs-dir")) return; await shell.openPath(paths.logsDir); }); - ipcMain.handle("covel:open-config-dir", async () => { + ipcMain.handle("covel:open-config-dir", async (event) => { + if (!isTrustedSender(event, "covel:open-config-dir")) return; await shell.openPath(paths.covelHome); }); - ipcMain.handle("covel:open-data-dir", async () => { + ipcMain.handle("covel:open-data-dir", async (event) => { + if (!isTrustedSender(event, "covel:open-data-dir")) return; await shell.openPath(paths.dataRoot); }); - ipcMain.handle("covel:restart-server", () => restartServer()); + ipcMain.handle("covel:restart-server", (event) => { + if (!isTrustedSender(event, "covel:restart-server")) return; + return restartServer(); + }); // Pick a directory for the next data_root. Does NOT move data — that's // deliberate per the "drop-old-data" UX contract; app restart starts fresh // in the new location. - ipcMain.handle("covel:pick-data-dir", async () => { + ipcMain.handle("covel:pick-data-dir", async (event) => { + if (!isTrustedSender(event, "covel:pick-data-dir")) return { path: null }; const result = await dialog.showOpenDialog({ title: t("dialog.dataDir.title"), properties: ["openDirectory", "createDirectory"], @@ -265,6 +276,19 @@ export function registerDesktopIpcHandlers({ return handleImport(kind, { sourcePath: picked.filePaths[0] }); } - ipcMain.handle("covel:import:pick-plugin", () => pickAndImport("plugin")); - ipcMain.handle("covel:import:pick-world", () => pickAndImport("world")); + // Audit 2026-07-16 L-4: these drive a native import dialog (installs a + // plugin/world) — a strictly worse action than opening a dir — so gate them + // on the same trusted-sender check as the other dialog-backed channels. + ipcMain.handle("covel:import:pick-plugin", (event) => { + if (!isTrustedSender(event, "covel:import:pick-plugin")) { + return { ok: false, kind: "plugin", message: t("import.cancelled") }; + } + return pickAndImport("plugin"); + }); + ipcMain.handle("covel:import:pick-world", (event) => { + if (!isTrustedSender(event, "covel:import:pick-world")) { + return { ok: false, kind: "world", message: t("import.cancelled") }; + } + return pickAndImport("world"); + }); } From 49beb0b3224b7cb314bb0c2fc25d3141679c9e67 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:30:16 +0800 Subject: [PATCH 07/11] fix(web): preserve streaming text on export and clear stale buffers Resolve getStreamingText(id) at export time so a mid-stream export keeps the partial assistant text instead of a blank row (L-10). Clear the live buffer on any story-runtime completion, including empty content, so stale partial text can't linger (L-11). Remove the scroll listener via a stored handler ref instead of the no-op onscroll=null (L-12). --- apps/web/src/hooks/use-auto-scroll.ts | 14 +++++-- apps/web/src/routes/session.tsx | 9 ++++- .../src/stores/session-store/sse-handler.ts | 38 ++++++++++--------- 3 files changed, 40 insertions(+), 21 deletions(-) diff --git a/apps/web/src/hooks/use-auto-scroll.ts b/apps/web/src/hooks/use-auto-scroll.ts index 48f7f72e..65bdb69c 100644 --- a/apps/web/src/hooks/use-auto-scroll.ts +++ b/apps/web/src/hooks/use-auto-scroll.ts @@ -48,17 +48,25 @@ export function useAutoScroll( [thresholdPx], ); + const scrollHandlerRef = useRef<(() => void) | null>(null); const scrollRef = useCallback( (node: HTMLElement | null) => { const prev = viewportRef.current; - if (prev) prev.onscroll = null; + // `addEventListener` handlers are not removed by clearing `.onscroll`, so + // keep the handler ref and detach it explicitly on node swap (L-12). + if (prev && scrollHandlerRef.current) { + prev.removeEventListener("scroll", scrollHandlerRef.current); + } viewportRef.current = node; + scrollHandlerRef.current = null; if (!node) return; - node.addEventListener("scroll", () => { + const handler = () => { const atBottom = computeIsAtBottom(node); isPinnedRef.current = atBottom; setShowJumpButton((cur) => (cur === !atBottom ? cur : !atBottom)); - }); + }; + scrollHandlerRef.current = handler; + node.addEventListener("scroll", handler); }, [computeIsAtBottom], ); diff --git a/apps/web/src/routes/session.tsx b/apps/web/src/routes/session.tsx index 608d7fec..c57f934f 100644 --- a/apps/web/src/routes/session.tsx +++ b/apps/web/src/routes/session.tsx @@ -5,6 +5,7 @@ import { Loader2, AlertCircle } from "lucide-react"; import { useSession } from "@/stores/session-store.js"; import { getDataService } from "@/services/data-service.js"; import { mergeChatExportMessages } from "@/lib/chat-export.js"; +import { getStreamingText } from "@/stores/streaming-text-store.js"; import { emitToast } from "@/lib/toast-channel.js"; import { useSlotConfig } from "@/hooks/use-slot-config.js"; import { useSettingsDialog } from "@/hooks/use-settings-dialog.js"; @@ -172,7 +173,13 @@ function SessionPage() { } try { const text = msgs - .map((m) => `[${m.role}] ${m.content}`) + // A message still streaming has an empty `content` (live text + // lives in the external store); resolve it so a mid-stream export + // keeps the partial assistant text instead of a blank row (L-10). + .map( + (m) => + `[${m.role}] ${m.content || getStreamingText(m.id) || ""}`, + ) .join("\n\n"); const blob = new Blob([text], { type: "text/plain" }); const url = URL.createObjectURL(blob); diff --git a/apps/web/src/stores/session-store/sse-handler.ts b/apps/web/src/stores/session-store/sse-handler.ts index 43f35aa9..fe46f91e 100644 --- a/apps/web/src/stores/session-store/sse-handler.ts +++ b/apps/web/src/stores/session-store/sse-handler.ts @@ -256,25 +256,29 @@ export function createSseEventHandler( const completedKind = (payload.kind as string) ?? deps.runtimeKindRef.current.get(runtimeId); - if (content && completedKind === "story") { - // Flush a same-frame delta before replacing its placeholder, then drop - // the external live buffer. The completed payload is authoritative. + if (completedKind === "story") { + // A story runtime's stream is finished. Always flush the same-frame + // delta and drop the external live buffer — even on empty content — + // so stale partial text can't linger on screen (audit 2026-07-16 + // L-11). Only publish an authoritative message when content exists. flushNarrativeDeltaBuffer(deps); clearStreamingText(`stream_${turnId ?? "unknown"}_${runtimeId}`); - const msg: StreamMessage = { - id: msgId, - role: "assistant", - content, - timestamp: envelope.timestamp, - turnId, - runtimeId: runtimeId !== "unknown" ? runtimeId : undefined, - }; - deps.dispatch({ - type: "COMPLETE_MESSAGE", - turnId: turnId ?? "unknown", - runtimeId, - message: msg, - }); + if (content) { + const msg: StreamMessage = { + id: msgId, + role: "assistant", + content, + timestamp: envelope.timestamp, + turnId, + runtimeId: runtimeId !== "unknown" ? runtimeId : undefined, + }; + deps.dispatch({ + type: "COMPLETE_MESSAGE", + turnId: turnId ?? "unknown", + runtimeId, + message: msg, + }); + } } if (content) { From 3e6b99beb6cac244cc049f7782e5aef1b1859ee6 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:36:04 +0800 Subject: [PATCH 08/11] fix(server): drain in-flight turns before closing the store on shutdown Graceful shutdown closed the DataStore without waiting for an in-flight turn, so a turn mid-commit could have the store torn out from under it. Add a time-boxed drain phase (awaitActiveTurnsDrained) before store close; the existing 2s drainPhase timeout guarantees a stuck turn can't block shutdown (L-6). --- apps/server/src/app.ts | 6 ++++++ apps/server/src/routes/api/turn-control.ts | 23 ++++++++++++++++++++++ apps/server/tests/api/turn-control.test.ts | 23 ++++++++++++++++++++++ 3 files changed, 52 insertions(+) diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index bcc8df6c..3f268d78 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -26,6 +26,7 @@ import { } from "@covel/runtime"; import { fetchWithRetry, validateBaseUrlForPlugin } from "@covel/ai-provider"; import { bootstrapApi } from "./routes/api/bootstrap.js"; +import { awaitActiveTurnsDrained } from "./routes/api/turn-control.js"; import { createInProcessSessionLock, type SessionLock, @@ -377,6 +378,11 @@ async function drainPhase( export const drainServerResources = async (): Promise => { await drainPhase("stop world watchers", () => stopWatchers()); await drainPhase("stop background sweeps", () => api.stopBackgroundSweeps()); + // Let an in-flight turn finish committing before the store closes under it + // (L-6). Time-boxed by drainPhase, so a stuck turn can't block shutdown. + await drainPhase("await in-flight turns", () => + awaitActiveTurnsDrained(DRAIN_PHASE_TIMEOUT_MS), + ); await drainPhase("flush event bus", () => api.eventBus.flush()); await drainPhase("close data store", () => store.close()); if (lockSql) { diff --git a/apps/server/src/routes/api/turn-control.ts b/apps/server/src/routes/api/turn-control.ts index dab71bee..4c23b76a 100644 --- a/apps/server/src/routes/api/turn-control.ts +++ b/apps/server/src/routes/api/turn-control.ts @@ -92,3 +92,26 @@ export function abortActiveTurn(sessionId: string): { turnId: string } | null { export function hasActiveTurn(sessionId: string): boolean { return activeTurns.has(sessionId); } + +/** Number of sessions with an in-flight turn (for the shutdown drain). */ +export function activeTurnCount(): number { + return activeTurns.size; +} + +/** + * Resolve once no session has an in-flight turn, or after `deadlineMs` + * (whichever comes first). Graceful shutdown calls this before closing the + * store so an in-flight commit is not torn out from under a turn (audit + * 2026-07-16 L-6). The caller (`drainPhase`) also time-boxes it, so a stuck + * turn can never block shutdown indefinitely — worst case the store closes and + * the turn's transaction rolls back, exactly as before. + */ +export async function awaitActiveTurnsDrained( + deadlineMs: number, + pollMs = 50, +): Promise { + const deadline = Date.now() + deadlineMs; + while (activeTurns.size > 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } +} diff --git a/apps/server/tests/api/turn-control.test.ts b/apps/server/tests/api/turn-control.test.ts index 8d13a73d..a9a57cec 100644 --- a/apps/server/tests/api/turn-control.test.ts +++ b/apps/server/tests/api/turn-control.test.ts @@ -8,6 +8,8 @@ import { createMemoryStore } from "@covel/store"; import type { DataStore } from "@covel/store"; import { abortActiveTurn, + activeTurnCount, + awaitActiveTurnsDrained, hasActiveTurn, registerActiveTurn, steerActiveTurn, @@ -45,6 +47,27 @@ describe("turn-control registry", () => { second.release(); expect(hasActiveTurn("sess-2")).toBe(false); }); + + it("awaitActiveTurnsDrained resolves once turns release, and honors the deadline (L-6)", async () => { + // Resolves immediately when nothing is in flight. + await awaitActiveTurnsDrained(1000); + expect(activeTurnCount()).toBe(0); + + const turn = registerActiveTurn("sess-drain", "turn-d"); + expect(activeTurnCount()).toBe(1); + // Release shortly after; the wait must resolve once the count hits 0. + setTimeout(() => turn.release(), 20); + await awaitActiveTurnsDrained(1000, 5); + expect(activeTurnCount()).toBe(0); + + // A turn that never releases must not hang the wait past the deadline. + const stuck = registerActiveTurn("sess-stuck", "turn-s"); + const start = Date.now(); + await awaitActiveTurnsDrained(60, 5); + expect(Date.now() - start).toBeLessThan(1000); + expect(activeTurnCount()).toBe(1); + stuck.release(); + }); }); describe("steer/abort routes", () => { From e294a52cf656df82b312f74f66ff04b5feb92c48 Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:36:04 +0800 Subject: [PATCH 09/11] fix(context): file context.compacted trace under the turn traceId The compactor tagged its trace event with the summaryId, so context.compacted never showed under the turn's traceId in /api/traces. Thread the turn traceId through CompactorRunner.run via the existing options bag and use it for the trace event, falling back to summaryId when absent (L-8). --- .../src/routes/api/bootstrap/compactor.ts | 3 ++- packages/context/src/compactor.ts | 10 ++++++++- packages/context/tests/compactor.test.ts | 21 +++++++++++++++++++ .../src/turn-executor/session-state.ts | 1 + 4 files changed, 33 insertions(+), 2 deletions(-) diff --git a/apps/server/src/routes/api/bootstrap/compactor.ts b/apps/server/src/routes/api/bootstrap/compactor.ts index 381f910c..5419eb70 100644 --- a/apps/server/src/routes/api/bootstrap/compactor.ts +++ b/apps/server/src/routes/api/bootstrap/compactor.ts @@ -48,7 +48,7 @@ export function createBootstrapCompactorRunner({ }; return { - async run(sessionId, systemPromptPreview, messages, locale) { + async run(sessionId, systemPromptPreview, messages, locale, traceId) { // The compactor only ships zh-CN / en-US prompt templates; map the // session locale by prefix (en* → en-US, else the zh-CN default). const compactorLocale = @@ -68,6 +68,7 @@ export function createBootstrapCompactorRunner({ { focusSections, locale: compactorLocale, + ...(traceId ? { traceId } : {}), }, ); }, diff --git a/packages/context/src/compactor.ts b/packages/context/src/compactor.ts index 0975aa87..559e2361 100644 --- a/packages/context/src/compactor.ts +++ b/packages/context/src/compactor.ts @@ -64,6 +64,12 @@ export interface CompactorOptions { /** Deduped list of focus sections from active plugins' summaryFocus fields. */ readonly focusSections?: readonly string[]; readonly locale?: "zh-CN" | "en-US"; + /** + * The current turn's traceId. When set, the `context.compacted` trace event + * is filed under it so it is queryable alongside the rest of the turn instead + * of under a standalone summaryId (audit 2026-07-16 L-8). + */ + readonly traceId?: string; } export interface CompactorResult { @@ -90,6 +96,8 @@ export interface CompactorRunner { * instead of always falling back to zh-CN. */ locale?: string, + /** Current turn's traceId, so the compaction trace joins the turn (L-8). */ + traceId?: string, ): Promise; } @@ -352,7 +360,7 @@ export async function maybeCompact( id: crypto.randomUUID(), sessionId, type: "context.compacted", - traceId: summaryId, + traceId: opts?.traceId ?? summaryId, turnId: turnRangeEnd, payload: { summaryId, diff --git a/packages/context/tests/compactor.test.ts b/packages/context/tests/compactor.test.ts index 1b65cf35..5a361c66 100644 --- a/packages/context/tests/compactor.test.ts +++ b/packages/context/tests/compactor.test.ts @@ -159,6 +159,27 @@ describe("maybeCompact", () => { expect(store.tagTurnMessagesCompacted).toHaveBeenCalledOnce(); }); + it("files the context.compacted trace under the turn traceId when provided (L-8)", async () => { + const messages = makeSimpleHistory(20); + const deps: CompactorDeps = { + store, + estimator, + fastSlotLlm, + contextWindow: 1_000, + }; + + const result = await maybeCompact("sess-1", "", messages, deps, { + threshold: 0.6, + traceId: "turn-trace-123", + }); + + expect(result.compacted).toBe(true); + const traceCall = vi + .mocked(store.addTraceEvent) + .mock.calls.find(([e]) => e.type === "context.compacted"); + expect(traceCall?.[0].traceId).toBe("turn-trace-123"); + }); + it("skips compaction when the fast LLM returns empty/whitespace content", async () => { const messages = makeSimpleHistory(20); const deps: CompactorDeps = { diff --git a/packages/runtime/src/turn-executor/session-state.ts b/packages/runtime/src/turn-executor/session-state.ts index 42f84fc9..6747277f 100644 --- a/packages/runtime/src/turn-executor/session-state.ts +++ b/packages/runtime/src/turn-executor/session-state.ts @@ -107,6 +107,7 @@ export async function loadTurnSessionState(args: { "", freshMessages, input.locale, + deps.emitter?.traceId, ); await runPostCompactionHook(hookOpts, { compacted: result.compacted, From b3b64075b1b12b1c2052ccd4c21051086bdc5ccf Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 15:42:02 +0800 Subject: [PATCH 10/11] refactor(server): simplify job-sweep freshness and drop test-only export MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit updatedAt starts equal to startedAt and only moves forward, so the sweep's freshness is just Date.parse(row.updatedAt) — drop the redundant max() over startedAt. Also drop the test-only activeTurnCount export; the drain test uses hasActiveTurn instead. (ponytail-review shrink, no behavior change.) --- apps/server/src/routes/api/plugin-rpc/jobs.ts | 18 ++++++------------ apps/server/src/routes/api/turn-control.ts | 5 ----- apps/server/tests/api/turn-control.test.ts | 10 ++++------ 3 files changed, 10 insertions(+), 23 deletions(-) diff --git a/apps/server/src/routes/api/plugin-rpc/jobs.ts b/apps/server/src/routes/api/plugin-rpc/jobs.ts index c62ecfe9..61288d61 100644 --- a/apps/server/src/routes/api/plugin-rpc/jobs.ts +++ b/apps/server/src/routes/api/plugin-rpc/jobs.ts @@ -167,18 +167,12 @@ export async function sweepStalePendingJobs( readonly turnId?: string; }; if (value?.status !== "pending") continue; - // Freshness is the most recent write to the row, not just its start: - // a job that writes progress bumps `updatedAt`, and the sweep now runs - // periodically (not only at boot), so anchoring on `startedAt` alone - // would force-fail a legitimately long-running job. A job that never - // writes progress still ages out at `staleMs` from its start (unchanged - // boot-recovery behavior). - const startedAtMs = Date.parse(value.startedAt ?? row.updatedAt); - const updatedAtMs = Date.parse(row.updatedAt); - const lastActivityMs = Math.max( - Number.isFinite(startedAtMs) ? startedAtMs : -Infinity, - Number.isFinite(updatedAtMs) ? updatedAtMs : -Infinity, - ); + // Freshness is the last write to the row (`updatedAt`), not its start: + // the sweep now runs periodically (not only at boot), so a job that + // writes progress must keep itself alive. `updatedAt` starts equal to + // `startedAt` and only moves forward, so a job that never writes progress + // still ages out at `staleMs` from its start (unchanged boot behavior). + const lastActivityMs = Date.parse(row.updatedAt); if (Number.isFinite(lastActivityMs) && now - lastActivityMs < staleMs) { continue; } diff --git a/apps/server/src/routes/api/turn-control.ts b/apps/server/src/routes/api/turn-control.ts index 4c23b76a..9a9ed3a2 100644 --- a/apps/server/src/routes/api/turn-control.ts +++ b/apps/server/src/routes/api/turn-control.ts @@ -93,11 +93,6 @@ export function hasActiveTurn(sessionId: string): boolean { return activeTurns.has(sessionId); } -/** Number of sessions with an in-flight turn (for the shutdown drain). */ -export function activeTurnCount(): number { - return activeTurns.size; -} - /** * Resolve once no session has an in-flight turn, or after `deadlineMs` * (whichever comes first). Graceful shutdown calls this before closing the diff --git a/apps/server/tests/api/turn-control.test.ts b/apps/server/tests/api/turn-control.test.ts index a9a57cec..4190c226 100644 --- a/apps/server/tests/api/turn-control.test.ts +++ b/apps/server/tests/api/turn-control.test.ts @@ -8,7 +8,6 @@ import { createMemoryStore } from "@covel/store"; import type { DataStore } from "@covel/store"; import { abortActiveTurn, - activeTurnCount, awaitActiveTurnsDrained, hasActiveTurn, registerActiveTurn, @@ -51,21 +50,20 @@ describe("turn-control registry", () => { it("awaitActiveTurnsDrained resolves once turns release, and honors the deadline (L-6)", async () => { // Resolves immediately when nothing is in flight. await awaitActiveTurnsDrained(1000); - expect(activeTurnCount()).toBe(0); const turn = registerActiveTurn("sess-drain", "turn-d"); - expect(activeTurnCount()).toBe(1); - // Release shortly after; the wait must resolve once the count hits 0. + expect(hasActiveTurn("sess-drain")).toBe(true); + // Release shortly after; the wait must resolve once the turn is gone. setTimeout(() => turn.release(), 20); await awaitActiveTurnsDrained(1000, 5); - expect(activeTurnCount()).toBe(0); + expect(hasActiveTurn("sess-drain")).toBe(false); // A turn that never releases must not hang the wait past the deadline. const stuck = registerActiveTurn("sess-stuck", "turn-s"); const start = Date.now(); await awaitActiveTurnsDrained(60, 5); expect(Date.now() - start).toBeLessThan(1000); - expect(activeTurnCount()).toBe(1); + expect(hasActiveTurn("sess-stuck")).toBe(true); stuck.release(); }); }); From fd342e9e7586e80f5e7de7718463e1fcceecf83c Mon Sep 17 00:00:00 2001 From: ackness Date: Thu, 16 Jul 2026 17:09:10 +0800 Subject: [PATCH 11/11] revert: back out M-1 heartbeat, M-2 periodic sweep, L-6 drain (ineffective) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #22 review found all three reliability patches unsound; each needs infrastructure beyond a Low/Medium remediation and is reverted to backlog: - L-6 (turn drain): awaited the wrong signal. actions.ts releases the W4 turn-control registry in executeTurn's finally, BEFORE proposal/snapshot commit; plugin-rpc/resume turns never register. So the store could still close mid-commit — false confidence. A correct fix drains the commit pipeline / held session locks, not activeTurns. - M-2 (periodic job sweep): a regression. background-jobs writes updatedAt only at enqueue and completion, never during a run, so a queued/long (>15min) job is force-failed mid-flight and later overwritten to done — clients observe a spurious failed state. Closing the boot blind spot safely needs a real job lease. Reverted to boot-only sweep. - M-1 (transport heartbeat): the sender loop iterates the replay-session LRU (cap 256), so a high-fanout pod evicts sender state and skips heartbeats — weakest exactly when tail-loss is most likely. Needs LRU-independent high-water state; deferred to backlog per the domain review. Keeps the sound fixes (2 High + L-2/L-3/L-4/L-8/L-10/L-11/L-12). --- apps/server/src/app.ts | 7 - apps/server/src/routes/api/bootstrap.ts | 27 +-- apps/server/src/routes/api/plugin-rpc/jobs.ts | 18 +- apps/server/src/routes/api/turn-control.ts | 18 -- apps/server/tests/api/turn-control.test.ts | 21 --- packages/events/src/event-bus.ts | 102 +---------- .../events/tests/event-bus-transport.test.ts | 158 ------------------ 7 files changed, 4 insertions(+), 347 deletions(-) diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 3f268d78..44ea1730 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -26,7 +26,6 @@ import { } from "@covel/runtime"; import { fetchWithRetry, validateBaseUrlForPlugin } from "@covel/ai-provider"; import { bootstrapApi } from "./routes/api/bootstrap.js"; -import { awaitActiveTurnsDrained } from "./routes/api/turn-control.js"; import { createInProcessSessionLock, type SessionLock, @@ -377,12 +376,6 @@ async function drainPhase( export const drainServerResources = async (): Promise => { await drainPhase("stop world watchers", () => stopWatchers()); - await drainPhase("stop background sweeps", () => api.stopBackgroundSweeps()); - // Let an in-flight turn finish committing before the store closes under it - // (L-6). Time-boxed by drainPhase, so a stuck turn can't block shutdown. - await drainPhase("await in-flight turns", () => - awaitActiveTurnsDrained(DRAIN_PHASE_TIMEOUT_MS), - ); await drainPhase("flush event bus", () => api.eventBus.flush()); await drainPhase("close data store", () => store.close()); if (lockSql) { diff --git a/apps/server/src/routes/api/bootstrap.ts b/apps/server/src/routes/api/bootstrap.ts index 53ef8bb9..bfa7d5c2 100644 --- a/apps/server/src/routes/api/bootstrap.ts +++ b/apps/server/src/routes/api/bootstrap.ts @@ -59,10 +59,7 @@ import type { MediaStore } from "@covel/store"; import type { MediaStoreBackend, VectorBackend } from "@covel/store"; import { resumeRoutes } from "./resume.js"; import { maybeSweepExpiredSuspensions } from "./suspension-sweep.js"; -import { - sweepStalePendingJobs, - JOB_SWEEP_INTERVAL_MS, -} from "./plugin-rpc/jobs.js"; +import { sweepStalePendingJobs } from "./plugin-rpc/jobs.js"; import { snapshotRoutes } from "./snapshots.js"; import { lorebookRoutes } from "./lorebook.js"; import { runtimeOutputRoutes } from "./runtime-outputs.js"; @@ -194,12 +191,6 @@ export interface ApiBootstrapResult { * the cache for that session). */ readonly prepareToolsForSession: (sessionId: string) => Promise; - /** - * Stop the periodic background sweeps (orphaned-job reaper). Called by the - * graceful-shutdown drain; the timer is also `unref()`'d so it never blocks - * process exit on its own. - */ - readonly stopBackgroundSweeps: () => void; } // ── Bootstrap function ─────────────────────────────────────────── @@ -275,21 +266,6 @@ export async function bootstrapApi( ), ); - // Audit 2026-07-16 M-2: the one-time boot sweep skips jobs younger than the - // staleness threshold and never re-scans, so a job that was pending for - // { - void sweepStalePendingJobs(store).catch((err: unknown) => - console.warn( - "[job-sweep] periodic sweep failed:", - err instanceof Error ? err.message : String(err), - ), - ); - }, JOB_SWEEP_INTERVAL_MS); - jobSweepTimer.unref(); - const { registry, discoveryMap, manifestCache } = await discoverAndRegisterPlugins({ pluginsDir: config.pluginsDir, @@ -651,6 +627,5 @@ export async function bootstrapApi( eventBus, compactorRunner, prepareToolsForSession, - stopBackgroundSweeps: () => clearInterval(jobSweepTimer), }; } diff --git a/apps/server/src/routes/api/plugin-rpc/jobs.ts b/apps/server/src/routes/api/plugin-rpc/jobs.ts index 61288d61..3c0488a3 100644 --- a/apps/server/src/routes/api/plugin-rpc/jobs.ts +++ b/apps/server/src/routes/api/plugin-rpc/jobs.ts @@ -123,13 +123,6 @@ export async function writePluginJob( */ const STALE_PENDING_JOB_MS = 15 * 60_000; -/** - * How often the periodic re-sweep runs (audit 2026-07-16 M-2). Smaller than - * the staleness threshold so a job orphaned just before a crash is reaped soon - * after it crosses `STALE_PENDING_JOB_MS`, instead of only at the next boot. - */ -export const JOB_SWEEP_INTERVAL_MS = 5 * 60_000; - /** * One-shot boot sweep: mark stale `pending` background-job rows as `failed` * so clients polling a job that died with a previous process get a terminal @@ -167,15 +160,8 @@ export async function sweepStalePendingJobs( readonly turnId?: string; }; if (value?.status !== "pending") continue; - // Freshness is the last write to the row (`updatedAt`), not its start: - // the sweep now runs periodically (not only at boot), so a job that - // writes progress must keep itself alive. `updatedAt` starts equal to - // `startedAt` and only moves forward, so a job that never writes progress - // still ages out at `staleMs` from its start (unchanged boot behavior). - const lastActivityMs = Date.parse(row.updatedAt); - if (Number.isFinite(lastActivityMs) && now - lastActivityMs < staleMs) { - continue; - } + const startedAtMs = Date.parse(value.startedAt ?? row.updatedAt); + if (Number.isFinite(startedAtMs) && now - startedAtMs < staleMs) continue; const startedAt = value.startedAt ?? row.createdAt; const completedAt = new Date(now).toISOString(); diff --git a/apps/server/src/routes/api/turn-control.ts b/apps/server/src/routes/api/turn-control.ts index 9a9ed3a2..dab71bee 100644 --- a/apps/server/src/routes/api/turn-control.ts +++ b/apps/server/src/routes/api/turn-control.ts @@ -92,21 +92,3 @@ export function abortActiveTurn(sessionId: string): { turnId: string } | null { export function hasActiveTurn(sessionId: string): boolean { return activeTurns.has(sessionId); } - -/** - * Resolve once no session has an in-flight turn, or after `deadlineMs` - * (whichever comes first). Graceful shutdown calls this before closing the - * store so an in-flight commit is not torn out from under a turn (audit - * 2026-07-16 L-6). The caller (`drainPhase`) also time-boxes it, so a stuck - * turn can never block shutdown indefinitely — worst case the store closes and - * the turn's transaction rolls back, exactly as before. - */ -export async function awaitActiveTurnsDrained( - deadlineMs: number, - pollMs = 50, -): Promise { - const deadline = Date.now() + deadlineMs; - while (activeTurns.size > 0 && Date.now() < deadline) { - await new Promise((resolve) => setTimeout(resolve, pollMs)); - } -} diff --git a/apps/server/tests/api/turn-control.test.ts b/apps/server/tests/api/turn-control.test.ts index 4190c226..8d13a73d 100644 --- a/apps/server/tests/api/turn-control.test.ts +++ b/apps/server/tests/api/turn-control.test.ts @@ -8,7 +8,6 @@ import { createMemoryStore } from "@covel/store"; import type { DataStore } from "@covel/store"; import { abortActiveTurn, - awaitActiveTurnsDrained, hasActiveTurn, registerActiveTurn, steerActiveTurn, @@ -46,26 +45,6 @@ describe("turn-control registry", () => { second.release(); expect(hasActiveTurn("sess-2")).toBe(false); }); - - it("awaitActiveTurnsDrained resolves once turns release, and honors the deadline (L-6)", async () => { - // Resolves immediately when nothing is in flight. - await awaitActiveTurnsDrained(1000); - - const turn = registerActiveTurn("sess-drain", "turn-d"); - expect(hasActiveTurn("sess-drain")).toBe(true); - // Release shortly after; the wait must resolve once the turn is gone. - setTimeout(() => turn.release(), 20); - await awaitActiveTurnsDrained(1000, 5); - expect(hasActiveTurn("sess-drain")).toBe(false); - - // A turn that never releases must not hang the wait past the deadline. - const stuck = registerActiveTurn("sess-stuck", "turn-s"); - const start = Date.now(); - await awaitActiveTurnsDrained(60, 5); - expect(Date.now() - start).toBeLessThan(1000); - expect(hasActiveTurn("sess-stuck")).toBe(true); - stuck.release(); - }); }); describe("steer/abort routes", () => { diff --git a/packages/events/src/event-bus.ts b/packages/events/src/event-bus.ts index 4f4a6059..a4fb2437 100644 --- a/packages/events/src/event-bus.ts +++ b/packages/events/src/event-bus.ts @@ -129,12 +129,6 @@ export const PERSIST_QUEUE_MAX = 1000; export const RECEIVE_PENDING_MAX = 64; /** How long a receiver waits on a transport seq hole before skipping it. */ export const RECEIVE_GAP_FLUSH_MS = 2000; -/** - * How often an origin publishes a liveness heartbeat carrying its current - * high-water transportSeq per active session (audit 2026-07-16 M-1). Bounds how - * long a lost *tail* frame can leave a cross-pod receiver silently stale. - */ -export const TRANSPORT_HEARTBEAT_MS = 15_000; /** FIFO cap on receive-ordering states across transport streams. */ export const RECEIVE_STATES_MAX = 1024; @@ -173,17 +167,6 @@ interface TransportFrame { readonly seq: number; readonly event?: SubscriptionEvent; readonly ref?: { readonly sessionId: string; readonly eventId: string }; - /** - * Liveness frame (audit 2026-07-16 M-1): carries the origin's current - * high-water `transportSeq` so a receiver can detect a lost *tail* frame that - * has no successor to trigger the normal reorder-gap path. Published through - * the same per-session outbox as real frames, so a heartbeat with seq N is - * only ever sent after real frame N — `seq < expected` therefore reliably - * means "caught up", never a false gap. Carries `sessionId` explicitly since - * there is no event/ref to read it from. - */ - readonly heartbeat?: boolean; - readonly sessionId?: string; } /** Per-(origin, session) receive-side ordering state. */ @@ -243,13 +226,6 @@ function parseTransportFrame(payload: string): TransportFrame | undefined { if (typeof frame.seq !== "number" || !Number.isSafeInteger(frame.seq)) { return undefined; } - if (frame.heartbeat === true) { - // Liveness frame: no event/ref, so it must name its session explicitly. - if (typeof frame.sessionId !== "string" || frame.sessionId.length === 0) { - return undefined; - } - return raw as TransportFrame; - } if (frame.event !== undefined) { const e = frame.event as Record | null; if ( @@ -555,35 +531,6 @@ export function createEventBus( rs.pending.clear(); } - /** - * A heartbeat states the origin's high-water transportSeq. Because heartbeats - * ride the same per-session outbox as real frames, one with seq N arrives - * only after real frame N. So `senderSeq < expected` means we are caught up; - * `senderSeq >= expected` means frame(s) in [expected, senderSeq] were lost - * with no successor to trip the normal gap path — surface it as a reset. - */ - function handleHeartbeat(rs: ReceiveState, senderSeq: number): void { - if (senderSeq < rs.expected) return; // caught up (or a stale heartbeat) - if (rs.timer) { - clearTimeout(rs.timer); - rs.timer = undefined; - } - console.warn( - `[EventBus] transport tail gap (expected ${rs.expected}, sender high-water ${senderSeq}) — resetting replay`, - ); - rs.chain = rs.chain - .then(() => invalidateReplay(rs.sessionId)) - .catch((err) => { - console.error("[EventBus] replay invalidation failed:", err); - }); - // Deliver whatever is parked, in order, skipping the hole(s). - const seqs = [...rs.pending.keys()].sort((a, b) => a - b); - for (const seq of seqs) scheduleDeliver(rs, rs.pending.get(seq)!); - rs.pending.clear(); - const maxParked = seqs.length > 0 ? seqs[seqs.length - 1]! : 0; - rs.expected = Math.max(senderSeq, maxParked) + 1; - } - function handleTransportFrame(payload: string): void { const frame = parseTransportFrame(payload); if (!frame) { @@ -591,18 +538,12 @@ export function createEventBus( return; } if (frame.origin === originId) return; // self-echo - const sessionId = - frame.event?.sessionId ?? frame.ref?.sessionId ?? frame.sessionId; - if (!sessionId) return; // malformed (no session to route to) + const sessionId = frame.event?.sessionId ?? frame.ref!.sessionId; const stream = frame.stream ?? "legacy"; const rs = getReceiveState( JSON.stringify([frame.origin, sessionId, stream]), sessionId, ); - if (frame.heartbeat) { - handleHeartbeat(rs, frame.seq); - return; - } if (frame.seq < rs.expected) return; // duplicate/late if (frame.seq === rs.expected) { scheduleDeliver(rs, frame); @@ -628,49 +569,8 @@ export function createEventBus( } } - /** - * Publish a liveness heartbeat per active fanned-out session (audit - * 2026-07-16 M-1). Rides the same per-session outbox as real frames, so the - * seq it carries is a safe high-water mark that never races ahead of an - * in-flight real frame. - */ - function emitTransportHeartbeats(): void { - if (!transport) return; - for (const [sessionId, state] of sessions) { - if (state.transportSeq <= 0) continue; // never fanned out — nothing to signal - // Capture seq + epoch NOW, at enqueue time. Reading them inside the task - // would let an emit() that lands between this tick and the task's - // microtask bump transportSeq first, so the heartbeat would claim that - // frame's seq and (being enqueued earlier) publish before it — a false - // gap + dropped frame on the receiver. Capturing here keeps the exact - // invariant: heartbeat(seq N) is enqueued after frame N and before frame - // N+1, so FIFO publish order never lets it overtake a real frame. - const seq = state.transportSeq; - const epoch = state.epoch; - enqueueOutbox(sessionId, () => { - // Skip if replay was invalidated since enqueue (epoch rolled, seq reset). - if (state.epoch !== epoch) return Promise.resolve(); - const frame = JSON.stringify({ - origin: originId, - stream: epoch, - seq, - heartbeat: true, - sessionId, - } satisfies TransportFrame); - return Promise.resolve(transport.publish(frame)); - }); - } - } - if (transport) { transport.subscribe(handleTransportFrame); - // unref so the heartbeat never keeps the process alive on its own; it dies - // with the process on shutdown (the bus is a process-lifetime singleton). - const heartbeatTimer = setInterval( - emitTransportHeartbeats, - TRANSPORT_HEARTBEAT_MS, - ); - heartbeatTimer.unref?.(); } const bus: EventBus = { diff --git a/packages/events/tests/event-bus-transport.test.ts b/packages/events/tests/event-bus-transport.test.ts index f2e7dfe8..040ed881 100644 --- a/packages/events/tests/event-bus-transport.test.ts +++ b/packages/events/tests/event-bus-transport.test.ts @@ -9,7 +9,6 @@ import { createEventBus, RECEIVE_GAP_FLUSH_MS, RECEIVE_STATES_MAX, - TRANSPORT_HEARTBEAT_MS, type EventBusTransport, } from "../src/event-bus.js"; @@ -414,160 +413,3 @@ describe("EventBus transport fan-out (audit R-02)", () => { warnSpy.mockRestore(); }); }); - -describe("EventBus transport heartbeat (audit 2026-07-16 M-1)", () => { - const flushMicrotasks = () => new Promise((r) => setTimeout(r, 0)); - - const realFrame = (seq: number, sessionId: string, type: string): string => - JSON.stringify({ - origin: "remote-hb", - stream: "boot-1", - seq, - event: { - id: `remote:${seq}`, - topic: "state", - type, - sessionId, - timestamp: new Date().toISOString(), - payload: {}, - }, - }); - - const heartbeat = (seq: number, sessionId: string): string => - JSON.stringify({ - origin: "remote-hb", - stream: "boot-1", - seq, - heartbeat: true, - sessionId, - }); - - it("resets when a heartbeat reveals a lost tail frame", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - try { - const hub = createHub(); - const bus = createEventBus(undefined, { transport: hub.connect() }); - const received: SubscriptionEvent[] = []; - const resets: Array<{ sessionId: string; reason: string }> = []; - bus.onEmit((event) => received.push(event)); - bus.onReset?.((reset) => resets.push(reset)); - - // Frame 1 arrives; the receiver is caught up (expects seq 2 next). - hub.broadcast(realFrame(1, "sess-tail", "t1")); - await flushMicrotasks(); - expect(received.map((e) => e.type)).toEqual(["t1"]); - expect(resets).toEqual([]); - - // Frame 2 is lost. A heartbeat announces the sender's high-water = 2, - // so the receiver must detect the hole and reset (no successor frame - // would otherwise ever trip the gap path). - hub.broadcast(heartbeat(2, "sess-tail")); - await flushMicrotasks(); - expect(resets).toEqual([ - { sessionId: "sess-tail", reason: "transport-gap" }, - ]); - } finally { - warnSpy.mockRestore(); - } - }); - - it("does not reset when a heartbeat matches an already-delivered stream", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - try { - const hub = createHub(); - const bus = createEventBus(undefined, { transport: hub.connect() }); - const resets: Array<{ sessionId: string; reason: string }> = []; - bus.onReset?.((reset) => resets.push(reset)); - - hub.broadcast(realFrame(1, "sess-live", "t1")); - hub.broadcast(realFrame(2, "sess-live", "t2")); - await flushMicrotasks(); - - // Caught up: expected is 3, heartbeat high-water is 2 → no gap. - hub.broadcast(heartbeat(2, "sess-live")); - await flushMicrotasks(); - expect(resets).toEqual([]); - } finally { - warnSpy.mockRestore(); - } - }); - - it("does not false-reset when an emit races the heartbeat tick (capture-at-enqueue)", async () => { - vi.useFakeTimers(); - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - try { - const published: string[] = []; - const hub = createHub(); - const inner = hub.connect(); - const recording: EventBusTransport = { - publish: (payload) => { - published.push(payload); - return inner.publish(payload); - }, - subscribe: inner.subscribe, - }; - const busA = createEventBus(undefined, { transport: recording }); - const busB = createEventBus(undefined, { transport: hub.connect() }); - const receivedB: SubscriptionEvent[] = []; - const resetsB: Array<{ sessionId: string; reason: string }> = []; - busB.onEmit((event) => receivedB.push(event)); - busB.onReset?.((reset) => resetsB.push(reset)); - - // Frame 1 delivered; B is caught up. - busA.emit(makeMessage({ sessionId: "sess-race", topic: "a" })); - await vi.advanceTimersByTimeAsync(0); - expect(receivedB).toHaveLength(1); - - // Heartbeat tick fires (captures seq 1), THEN a second emit lands before - // the outbox microtasks flush — the exact race the fix targets. - vi.advanceTimersByTime(TRANSPORT_HEARTBEAT_MS); - busA.emit(makeMessage({ sessionId: "sess-race", topic: "b" })); - await vi.advanceTimersByTimeAsync(0); - - // The heartbeat must carry seq 1 (not the raced 2), so B sees no gap and - // still receives frame 2. - const hb = published - .map((p) => JSON.parse(p) as Record) - .find((f) => f.heartbeat === true); - expect(hb).toMatchObject({ seq: 1 }); - expect(resetsB).toEqual([]); - expect(receivedB).toHaveLength(2); - } finally { - warnSpy.mockRestore(); - vi.useRealTimers(); - } - }); - - it("periodically publishes a heartbeat carrying the high-water seq", async () => { - vi.useFakeTimers(); - try { - const published: string[] = []; - const hub = createHub(); - const inner = hub.connect(); - const recording: EventBusTransport = { - publish: (payload) => { - published.push(payload); - return inner.publish(payload); - }, - subscribe: inner.subscribe, - }; - const busA = createEventBus(undefined, { transport: recording }); - - busA.emit(makeMessage({ sessionId: "sess-hb" })); - await vi.advanceTimersByTimeAsync(0); // flush the real frame's outbox - published.length = 0; - - await vi.advanceTimersByTimeAsync(TRANSPORT_HEARTBEAT_MS); - const hb = published - .map((p) => JSON.parse(p) as Record) - .find((f) => f.heartbeat === true); - expect(hb).toMatchObject({ - heartbeat: true, - sessionId: "sess-hb", - seq: 1, - }); - } finally { - vi.useRealTimers(); - } - }); -});