diff --git a/docs/plans/2026-07-24-overseer-summary-emission.md b/docs/plans/2026-07-24-overseer-summary-emission.md index 7969c3068f..daa614fd13 100644 --- a/docs/plans/2026-07-24-overseer-summary-emission.md +++ b/docs/plans/2026-07-24-overseer-summary-emission.md @@ -1,7 +1,8 @@ # Overseer summary emission (Half B) — 2026-07-24 Status: Piece 1 live; Piece 2 hub text-synth **removed**; Piece 3 in flight -(Claude remote + Codex app-server only; Grok/OpenCode deferred to #89) +(Claude remote + Codex app-server only; Grok/OpenCode deferred to #89). +Option A LLM fallback implemented (default OFF, #90). Owner: feat/overseer-summary-emit (peer of 🔁overseer prep) Scope: FORK-ONLY. Never upstream. The whole overseer feature is fork-private. @@ -133,7 +134,9 @@ ordinary, useful project config, never as surveillance: - Grok / OpenCode first-turn prepend: title + skill-lookup only. Session-summary for those flavors is [#89](https://github.com/heavygee/hapi/issues/89). - kimi + generic ACP / pi: also #89. -- Better LLM / oneshot-agent fallback: [#90](https://github.com/heavygee/hapi/issues/90), default off. +- Better LLM / oneshot-agent fallback: Option A implemented behind + `HAPI_OVERSEER_LLM_FALLBACK` (default off); see § Better fallback / #90. + Option B oneshot agent remains out of scope. ## Better fallback (opt-in — tracked #90) @@ -146,7 +149,7 @@ and is a real cost tax - so it must be **opt-in**, clearly labeled, and rare ### Gate: rarity first, quality never second -Do **not** ship a better fallback until primary emission is good enough that +Do **not** enable a better fallback until primary emission is good enough that fallback is a thin residue - target **well under 5% of turns** (5% is already generous). Measure emit vs missing-line ratio fleet-wide after Piece 3 is live; only then enable LLM fallback. @@ -156,18 +159,34 @@ feed the **full last-turn assistant content** (no input-char truncation that would make the summary worse than the agent would have written). Rarity is the cost control; accuracy is non-negotiable on the rare path. -### Option A — raw OpenAI-compatible completions call +### Option A — raw OpenAI-compatible completions call (implemented) -Hub (or a tiny side worker) POSTs the full last assistant turn text to an -operator-configured base URL (`/v1/chat/completions` or `/v1/responses`) with a -fixed prompt: "emit exactly one AGENT_NOTIFY_SUMMARY JSON line." Local (Ollama / -vLLM / gateway) or remote (OpenAI) - same wire format. +Hub POSTs the full last assistant turn text to an operator-configured base URL +(`/v1/chat/completions` or `/v1/responses`) with a fixed prompt: "emit exactly +one AGENT_NOTIFY_SUMMARY JSON line." Local (Ollama / vLLM / gateway) or remote +(OpenAI) - same wire format. -- Pros: cheap to wire, no session surface, easy to bill/attribute as - `provenance: hub-llm-fallback`. -- Cons: large turns = large prompt tokens (accepted when rare); operator must - provision a key/URL; prefer Chat Completions for local-gateway compatibility, - Responses for OpenAI-native - support both behind one adapter. +**Enable (default OFF — never surprise usage):** + +```bash +export HAPI_OVERSEER_LLM_FALLBACK=1 +export HAPI_OVERSEER_LLM_BASE_URL=http://127.0.0.1:11434/v1 # include /v1 +export HAPI_OVERSEER_LLM_MODEL=llama3.3 +# optional: +export HAPI_OVERSEER_LLM_API_KEY=ollama # Bearer token; empty OK for local +export HAPI_OVERSEER_LLM_API=chat-completions # or: responses +export HAPI_OVERSEER_LLM_TIMEOUT_MS=30000 +``` + +Prefer `chat-completions` for local-gateway compatibility; use `responses` for +OpenAI-native. Failures / non-compliant model output produce **no** Session Log +row (no first-line heuristic). Session-end may still write `completed_fallback` +if the session completes without a later successful notify/LLM row. Events are marked +`provenance: hub-llm-fallback ...` with `payload.synthesis = "llm-fallback"`, +`attentionCandidate = 0` (Session Log only — not inbox / voice). + +**Cost warning:** every missed primary emit becomes a full-turn prompt. Enable +only after the rarity gate, or accept the bill deliberately. ### Option B — out-of-band oneshot agent @@ -180,6 +199,7 @@ in Session Log / inbox so the operator never wonders "wtf usage is this." multi-step retrieval if needed. - Cons: heavier; looks like a phantom session if not carefully labeled; higher cost variance; more moving parts. + **Out of scope for #90** — revisit only if Option A proves insufficient. ### Shared requirements (either option) @@ -192,7 +212,7 @@ in Session Log / inbox so the operator never wonders "wtf usage is this." primary turn lacked a contract - never pretend the primary agent said it. - **Kill-criterion:** if opt-in users report surprise usage, the toggle and provenance labels failed - fix UX before expanding defaults. If fallback - summaries are worse than the heuristic first-line, do not ship. + summaries are worse than a primary `AGENT_NOTIFY_SUMMARY` emit, do not ship. Prefer **Option A** as the first better-fallback ship: smaller blast radius, easier to reason about cost, no phantom sessions. diff --git a/hub/README.md b/hub/README.md index b224b5df78..1f45860b3c 100644 --- a/hub/README.md +++ b/hub/README.md @@ -42,6 +42,19 @@ See `src/configuration.ts` for all options. - `HAPI_RELAY_FORCE_TCP` - Force TCP relay mode (true/1). - `VAPID_SUBJECT` - Contact email/URL for Web Push. +### Optional (Overseer LLM fallback — fork, default OFF) + +Only when primary agents omit `AGENT_NOTIFY_SUMMARY`. Costs a full-turn LLM call +per miss — enable after miss rate is rare (~<5%). See +`docs/plans/2026-07-24-overseer-summary-emission.md`. + +- `HAPI_OVERSEER_LLM_FALLBACK` - `1`/`true` to enable (default: off). +- `HAPI_OVERSEER_LLM_BASE_URL` - OpenAI-compatible base including `/v1` (required when enabled). +- `HAPI_OVERSEER_LLM_MODEL` - Model id (required when enabled). +- `HAPI_OVERSEER_LLM_API_KEY` - Bearer token (optional for local gateways). +- `HAPI_OVERSEER_LLM_API` - `chat-completions` (default) or `responses`. +- `HAPI_OVERSEER_LLM_TIMEOUT_MS` - Request timeout (default: 30000). + ## Running Binary (single executable): diff --git a/hub/src/configuration.ts b/hub/src/configuration.ts index 20779a3a90..dd5f7a5c58 100644 --- a/hub/src/configuration.ts +++ b/hub/src/configuration.ts @@ -21,6 +21,8 @@ * - VAPID_SUBJECT: Contact email or URL for Web Push (defaults to mailto:admin@hapi.run) * - HAPI_HOME: Data directory (default: ~/.hapi) * - DB_PATH: SQLite database path (default: {HAPI_HOME}/hapi.db) + * - HAPI_OVERSEER_LLM_FALLBACK: Opt-in hub LLM summary fallback when AGENT_NOTIFY_SUMMARY is missing (default: off) + * - HAPI_OVERSEER_LLM_BASE_URL / _MODEL / _API_KEY / _API / _TIMEOUT_MS: OpenAI-compatible endpoint for that fallback */ import { existsSync, mkdirSync } from 'node:fs' diff --git a/hub/src/sync/overseerEventRecorder.fallback.test.ts b/hub/src/sync/overseerEventRecorder.fallback.test.ts index 64fc0531b4..520a900064 100644 --- a/hub/src/sync/overseerEventRecorder.fallback.test.ts +++ b/hub/src/sync/overseerEventRecorder.fallback.test.ts @@ -34,12 +34,12 @@ function agentText(message: string) { } describe('OverseerEventRecorder — no hub text synth', () => { - it('does not synthesize from assistant text without AGENT_NOTIFY_SUMMARY', () => { + it('does not synthesize from assistant text without AGENT_NOTIFY_SUMMARY', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('cur', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), 'msg-fb', agentText('Refactored the parser and added tests.\n\nMore detail here.'), @@ -51,12 +51,12 @@ describe('OverseerEventRecorder — no hub text synth', () => { expect(store.inbox.count()).toBe(0) }) - it('still records a real AGENT_NOTIFY_SUMMARY', () => { + it('still records a real AGENT_NOTIFY_SUMMARY', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('cur2', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), 'msg-real', agentText('All done.\nAGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"Review PR","summary":"Shipped"}'), @@ -64,51 +64,17 @@ describe('OverseerEventRecorder — no hub text synth', () => { ) expect(event?.provenance).toBe('AGENT_NOTIFY_SUMMARY') - expect(store.events.list({ eventType: 'progress' })).toHaveLength(0) expect(store.events.count()).toBe(1) }) - it('still records malformed notify as validation_error', () => { - const store = new Store(':memory:') - const recorder = new OverseerEventRecorder(store.events, store.inbox) - const session = store.sessions.getOrCreateSession('cur3', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') - - const event = recorder.onAgentMessage( - toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), - 'msg-bad', - agentText('Working.\nAGENT_NOTIFY_SUMMARY {not valid json'), - Date.now() - ) - - expect(event?.eventType).toBe('validation_error') - expect(store.events.list({ eventType: 'progress' })).toHaveLength(0) - }) - - it('ignores user messages', () => { - const store = new Store(':memory:') - const recorder = new OverseerEventRecorder(store.events, store.inbox) - const session = store.sessions.getOrCreateSession('cur6', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') - - const userContent = { role: 'user', content: { type: 'text', text: 'do the thing' } } - const event = recorder.onAgentMessage( - toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), - 'msg-user', - userContent, - Date.now() - ) - - expect(event).toBeNull() - expect(store.events.count()).toBe(0) - }) - - it('does not synth mid-turn ACP text flushes (multiple messages, no notify)', () => { + it('does not synth mid-turn ACP text flushes', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const live = store.sessions.getOrCreateSession('cur7', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') const snapshot = toSessionSnapshot(makeSession(live.id, 'cursor'), live.tag) - expect(recorder.onAgentMessage(snapshot, 'msg-mid-1', agentText('Pulling the last hour of events.'), Date.now())).toBeNull() - expect(recorder.onAgentMessage(snapshot, 'msg-mid-2', agentText('Found something important.'), Date.now())).toBeNull() + expect(await recorder.onAgentMessage(snapshot, 'msg-mid-1', agentText('Pulling events.'), Date.now(), { thinking: true })).toBeNull() + expect(await recorder.onAgentMessage(snapshot, 'msg-mid-2', agentText('Found something.'), Date.now(), { thinking: true })).toBeNull() expect(store.events.count()).toBe(0) }) }) diff --git a/hub/src/sync/overseerEventRecorder.llmFallback.test.ts b/hub/src/sync/overseerEventRecorder.llmFallback.test.ts new file mode 100644 index 0000000000..bbf9a5bb44 --- /dev/null +++ b/hub/src/sync/overseerEventRecorder.llmFallback.test.ts @@ -0,0 +1,722 @@ +import { describe, expect, it, mock } from 'bun:test' +import type { NotifySummary } from '@hapi/protocol/messages' +import type { Session } from '@hapi/protocol/types' +import { Store } from '../store' +import type { OverseerLlmFallbackClient } from './overseerLlmFallback' +import { OverseerEventRecorder, toSessionSnapshot } from './overseerEventRecorder' + +function makeSession(id: string, flavor: string, overrides?: Partial): Session { + return { + id, + namespace: 'default', + seq: 0, + createdAt: Date.now(), + updatedAt: Date.now(), + active: true, + activeAt: Date.now(), + metadata: { flavor, path: '/tmp', host: 'local' }, + metadataVersion: 1, + agentState: null, + agentStateVersion: 1, + thinking: false, + thinkingAt: 0, + model: null, + modelReasoningEffort: null, + effort: null, + serviceTier: null, + ...overrides + } +} + +function agentText(message: string) { + return { + role: 'agent', + content: { type: 'codex', data: { type: 'message', message } } + } +} + +describe('OverseerEventRecorder LLM fallback', () => { + it('uses hub-llm-fallback provenance and keeps attn=0 when LLM succeeds', async () => { + const store = new Store(':memory:') + const synthesize = mock(async (plainText: string): Promise => { + expect(plainText).toContain('Refactored the parser') + expect(plainText).toContain('More detail here.') + return { + version: 1, + status: 'blocked', + action: 'Unblock CI', + summary: 'LLM distilled summary of the whole turn', + } + }) + const llmFallback: OverseerLlmFallbackClient = { synthesizeNotifySummary: synthesize } + const recorder = new OverseerEventRecorder(store.events, store.inbox, { llmFallback }) + const session = store.sessions.getOrCreateSession('llm1', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-llm', + agentText('Refactored the parser and added tests.\n\nMore detail here.'), + Date.now() + ) + + expect(synthesize).toHaveBeenCalledTimes(1) + expect(event).not.toBeNull() + expect(event?.summary).toBe('LLM distilled summary of the whole turn') + expect(event?.eventType).toBe('blocked') + expect(event?.attentionCandidate).toBe(0) + expect(event?.operatorActionRequired).toBe(0) + expect(event?.provenance).toContain('hub-llm-fallback') + expect(store.inbox.count()).toBe(0) + + const payload = JSON.parse(event!.payloadJson!) as { + synthesized?: boolean + synthesis?: string + notify_summary?: NotifySummary + } + expect(payload.synthesized).toBe(true) + expect(payload.synthesis).toBe('llm-fallback') + expect(payload.notify_summary?.summary).toBe('LLM distilled summary of the whole turn') + }) + + it('records nothing when LLM returns null (no heuristic)', async () => { + const store = new Store(':memory:') + const llmFallback: OverseerLlmFallbackClient = { + synthesizeNotifySummary: mock(async () => null), + } + const recorder = new OverseerEventRecorder(store.events, store.inbox, { llmFallback }) + const session = store.sessions.getOrCreateSession('llm2', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-llm-fail', + agentText('First line wins.\nSecond line.'), + Date.now() + ) + + expect(event).toBeNull() + expect(store.events.count()).toBe(0) + }) + + it('does not call LLM when a real AGENT_NOTIFY_SUMMARY is present', async () => { + const store = new Store(':memory:') + const synthesize = mock(async () => ({ status: 'done', summary: 'should not run' })) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const session = store.sessions.getOrCreateSession('llm3', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-real', + agentText('All done.\nAGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"Review PR","summary":"Shipped"}'), + Date.now() + ) + + expect(synthesize).toHaveBeenCalledTimes(0) + expect(event?.provenance).toBe('AGENT_NOTIFY_SUMMARY') + }) + + it('records nothing when LLM client is not configured', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox) + const session = store.sessions.getOrCreateSession('llm4', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-off', + agentText('No synth path.'), + Date.now() + ) + + expect(event).toBeNull() + expect(store.events.count()).toBe(0) + }) + + it('records nothing when LLM throws (no heuristic)', async () => { + const store = new Store(':memory:') + const llmFallback: OverseerLlmFallbackClient = { + synthesizeNotifySummary: mock(async () => { + throw new Error('network down') + }), + } + const recorder = new OverseerEventRecorder(store.events, store.inbox, { llmFallback }) + const session = store.sessions.getOrCreateSession('llm5', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-throw', + agentText('Should stay quiet.'), + Date.now() + ) + + expect(event).toBeNull() + expect(store.events.count()).toBe(0) + }) + + it('maps stalled LLM status to progress so Session Log All still shows it', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ + status: 'stalled', + summary: 'Agent went quiet mid-turn', + })), + }, + }) + const session = store.sessions.getOrCreateSession('llm-stale', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + + const event = await recorder.onAgentMessage( + toSessionSnapshot(makeSession(session.id, 'cursor'), session.tag), + 'msg-stalled', + agentText('Still working on the rebase.'), + Date.now() + ) + + expect(event?.eventType).toBe('progress') + expect(event?.attentionCandidate).toBe(0) + const payload = JSON.parse(event!.payloadJson!) as { notify_summary?: NotifySummary } + expect(payload.notify_summary?.status).toBe('stalled') + }) + + it('flushes deferred LLM fallback when thinking clears', async () => { + const store = new Store(':memory:') + const synthesize = mock(async () => ({ status: 'done', summary: 'End of turn' })) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const live = store.sessions.getOrCreateSession('llm-think', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(live.id, 'cursor'), live.tag) + + expect(await recorder.onAgentMessage(snapshot, 'msg-mid', agentText('Partial flush.'), Date.now(), { thinking: true })).toBeNull() + expect(synthesize).toHaveBeenCalledTimes(0) + + const flushed = await recorder.flushPendingLlmFallback(snapshot) + expect(synthesize).toHaveBeenCalledTimes(1) + expect(flushed?.summary).toBe('End of turn') + expect(store.events.count()).toBe(1) + }) + + it('flushes pending fallback from onSessionUpdated when thinking is false', async () => { + const store = new Store(':memory:') + const synthesize = mock(async () => ({ status: 'blocked', summary: 'Need a decision' })) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const live = makeSession('sess-alive', 'cursor', { thinking: true }) + const stored = store.sessions.getOrCreateSession('llm-alive', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + await recorder.onAgentMessage(snapshot, 'msg-pending', agentText('No notify yet.'), Date.now(), { thinking: true }) + expect(store.events.count()).toBe(0) + + live.thinking = false + await recorder.onSessionUpdated(live, stored.tag) + + expect(synthesize).toHaveBeenCalledTimes(1) + expect(store.events.list({ eventType: 'blocked' })).toHaveLength(1) + }) + + it('does not insert session-end completed_fallback after a successful LLM flush', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ + status: 'done', + summary: 'LLM caught the last turn', + })), + }, + }) + const live = makeSession('sess-end', 'cursor', { thinking: true }) + const stored = store.sessions.getOrCreateSession('llm-end', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + await recorder.onAgentMessage(snapshot, 'msg-last', agentText('Finishing up.'), Date.now(), { thinking: true }) + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now(), + 'completed', + () => 'Finishing up.' + ) + + expect(event?.provenance).toContain('hub-llm-fallback') + expect(store.events.count()).toBe(1) + expect(store.events.list().some((row) => row.provenance?.includes('session-end'))).toBe(false) + }) + + it('serializes LLM fallbacks on one session so earlier turns keep lower ids', async () => { + const store = new Store(':memory:') + let releaseFirst: ((value: NotifySummary) => void) | undefined + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const synthesize = mock(async (plainText: string): Promise => { + if (plainText.includes('FIRST')) { + firstStarted() + return firstGate + } + return { status: 'done', summary: 'second turn' } + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const stored = store.sessions.getOrCreateSession('llm-ord', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('FIRST turn body'), Date.now()) + await firstStartedP + const second = recorder.onAgentMessage(snapshot, 'msg-b', agentText('SECOND turn body'), Date.now() + 1) + releaseFirst!({ status: 'done', summary: 'first turn' }) + await Promise.all([first, second]) + + const rows = store.events.list().sort((a, b) => a.id - b.id) + expect(rows.map((row) => row.summary)).toEqual(['first turn', 'second turn']) + }) + + it('accumulates ACP thinking segments before fallback', async () => { + const store = new Store(':memory:') + const synthesize = mock(async (plainText: string): Promise => { + expect(plainText).toContain('First chunk') + expect(plainText).toContain('Second chunk') + return { status: 'done', summary: 'both chunks' } + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const stored = store.sessions.getOrCreateSession('llm-acc', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + + const firstTs = Date.now() + const lastTs = firstTs + 120_000 + await recorder.onAgentMessage(snapshot, 'msg-1', agentText('First chunk'), firstTs, { thinking: true }) + await recorder.onAgentMessage(snapshot, 'msg-2', agentText('Second chunk'), lastTs, { thinking: true }) + const flushed = await recorder.flushPendingLlmFallback(snapshot) + + expect(synthesize).toHaveBeenCalledTimes(1) + expect(flushed?.summary).toBe('both chunks') + expect(flushed?.ts).toBe(lastTs) + }) + + it('writes completed_fallback when a later missed turn LLM fails', async () => { + const store = new Store(':memory:') + const synthesize = mock(async (plainText: string): Promise => { + if (plainText.includes('first turn')) { + return { status: 'done', summary: 'caught first' } + } + return null + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const live = makeSession('sess-later', 'cursor', { thinking: true }) + const stored = store.sessions.getOrCreateSession('llm-later', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + await recorder.onAgentMessage(snapshot, 'msg-first', agentText('first turn body'), Date.now(), { thinking: true }) + live.thinking = false + await recorder.onSessionUpdated(live, stored.tag) + expect(store.events.list({ eventType: 'completed' })).toHaveLength(1) + + await recorder.onAgentMessage(snapshot, 'msg-second', agentText('second turn miss'), Date.now() + 1) + expect(synthesize).toHaveBeenCalledTimes(2) + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now() + 2, + 'completed', + () => 'second turn miss' + ) + + expect(event?.provenance).toContain('session-end') + expect(store.events.list().filter((row) => row.provenance?.includes('session-end'))).toHaveLength(1) + expect(store.events.list().filter((row) => row.provenance?.includes('hub-llm-fallback'))).toHaveLength(1) + }) + + it('persists scooped links before awaiting LLM', async () => { + const store = new Store(':memory:') + let release!: (value: NotifySummary) => void + const gate = new Promise((resolve) => { + release = resolve + }) + let started!: () => void + const startedP = new Promise((resolve) => { + started = resolve + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => { + started() + return gate + }), + }, + }) + const stored = store.sessions.getOrCreateSession('llm-scoop', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + + const pending = recorder.onAgentMessage( + snapshot, + 'msg-url', + agentText('See https://example.com/docs for details.'), + Date.now() + ) + await startedP + + expect(store.events.list({ eventType: 'link_seen' })).toHaveLength(1) + expect(store.events.list().filter((row) => row.provenance?.includes('hub-llm-fallback'))).toHaveLength(0) + + release({ status: 'done', summary: 'linked turn' }) + await pending + expect(store.events.list().filter((row) => row.provenance?.includes('hub-llm-fallback'))).toHaveLength(1) + }) + + it('does not append a new turn onto a flush already queued', async () => { + const store = new Store(':memory:') + let releaseFirst!: (value: NotifySummary) => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const synthesize = mock(async (plainText: string): Promise => { + if (plainText.includes('TURN A')) { + firstStarted() + return firstGate + } + return { status: 'done', summary: plainText.includes('TURN C') ? 'turn c' : 'turn b' } + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const stored = store.sessions.getOrCreateSession('llm-detach', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('TURN A body'), Date.now()) + await firstStartedP + await recorder.onAgentMessage(snapshot, 'msg-b', agentText('TURN B body'), Date.now() + 1, { thinking: true }) + const flushedB = recorder.flushPendingLlmFallback(snapshot) + await recorder.onAgentMessage(snapshot, 'msg-c', agentText('TURN C body'), Date.now() + 2, { thinking: true }) + + releaseFirst({ status: 'done', summary: 'turn a' }) + await first + const eventB = await flushedB + expect(eventB?.summary).toBe('turn b') + + const eventC = await recorder.flushPendingLlmFallback(snapshot) + expect(eventC?.summary).toBe('turn c') + }) + + it('queues a later AGENT_NOTIFY_SUMMARY behind an in-flight LLM insert', async () => { + const store = new Store(':memory:') + let releaseFirst!: (value: NotifySummary) => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => { + firstStarted() + return firstGate + }), + }, + }) + const stored = store.sessions.getOrCreateSession('llm-notify-ord', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('TURN A body'), Date.now()) + await firstStartedP + const second = recorder.onAgentMessage( + snapshot, + 'msg-b', + agentText('Done.\nAGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"Review","summary":"real notify"}'), + Date.now() + 1 + ) + releaseFirst({ status: 'done', summary: 'llm first' }) + await Promise.all([first, second]) + + const rows = store.events.list().sort((a, b) => a.id - b.id) + expect(rows.map((row) => row.summary)).toEqual(['llm first', 'real notify']) + }) + + it('does not write completed_fallback for same-turn ACP tool after LLM success', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ + status: 'done', + summary: 'caught this turn', + })), + }, + }) + const live = makeSession('sess-same', 'cursor') + const stored = store.sessions.getOrCreateSession('llm-same', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + await recorder.onAgentMessage(snapshot, 'msg-text', agentText('turn body'), Date.now()) + await recorder.onAgentMessage(snapshot, 'msg-tool', { + role: 'agent', + content: { + type: 'codex', + data: { type: 'tool-call-result', output: { exit_code: 0 } }, + }, + }, Date.now() + 1) + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now() + 2, + 'completed', + () => 'turn body' + ) + expect(event).toBeNull() + expect(store.events.list().filter((row) => row.provenance?.includes('hub-llm-fallback'))).toHaveLength(1) + expect(store.events.list().some((row) => row.provenance?.includes('session-end'))).toBe(false) + }) + + it('writes completed_fallback for a later tool-only turn after a user message', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ + status: 'done', + summary: 'caught earlier turn', + })), + }, + }) + const live = makeSession('sess-tool', 'cursor') + const stored = store.sessions.getOrCreateSession('llm-tool', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + await recorder.onAgentMessage(snapshot, 'msg-text', agentText('earlier turn body'), Date.now()) + await recorder.onAgentMessage(snapshot, 'msg-user', { + role: 'user', + content: { type: 'text', text: 'do the next thing' }, + }, Date.now() + 1) + await recorder.onAgentMessage(snapshot, 'msg-tool', { + role: 'agent', + content: { + type: 'codex', + data: { type: 'tool-call-result', output: { exit_code: 0 } }, + }, + }, Date.now() + 2) + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now() + 3, + 'completed', + () => 'earlier turn body' + ) + expect(event?.provenance).toContain('session-end') + }) + + it('does not let an in-flight earlier LLM cover a later user+tool turn', async () => { + const store = new Store(':memory:') + let releaseFirst!: (value: NotifySummary) => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => { + firstStarted() + return firstGate + }), + }, + }) + const live = makeSession('sess-inflight', 'cursor') + const stored = store.sessions.getOrCreateSession('llm-inflight', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('earlier turn'), Date.now()) + await firstStartedP + await recorder.onAgentMessage(snapshot, 'msg-user', { + role: 'user', + content: { type: 'text', text: 'continue' }, + }, Date.now() + 1) + await recorder.onAgentMessage(snapshot, 'msg-tool', { + role: 'agent', + content: { + type: 'codex', + data: { type: 'tool-call-result', output: { exit_code: 0 } }, + }, + }, Date.now() + 2) + releaseFirst({ status: 'done', summary: 'caught earlier' }) + await first + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now() + 3, + 'completed', + () => 'earlier turn' + ) + expect(event?.provenance).toContain('session-end') + }) + + it('publishes after a successful LLM insert', async () => { + const store = new Store(':memory:') + let published = 0 + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ status: 'done', summary: 'ok' })), + }, + onAsyncSystemEvent: () => { + published += 1 + }, + }) + const stored = store.sessions.getOrCreateSession('llm-pub', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + await recorder.onAgentMessage( + toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag), + 'msg-pub', + agentText('Turn body'), + Date.now() + ) + expect(published).toBe(1) + }) + + it('forgetSession drops deferred LLM state without flushing', async () => { + const store = new Store(':memory:') + const synthesize = mock(async () => ({ status: 'done', summary: 'should not run' })) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { synthesizeNotifySummary: synthesize }, + }) + const stored = store.sessions.getOrCreateSession('llm-forget', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + const snapshot = toSessionSnapshot(makeSession(stored.id, 'cursor'), stored.tag) + await recorder.onAgentMessage(snapshot, 'msg-p', agentText('pending'), Date.now(), { thinking: true }) + recorder.forgetSession(stored.id) + expect(await recorder.flushPendingLlmFallback(snapshot)).toBeNull() + expect(synthesize).toHaveBeenCalledTimes(0) + }) + + it('queues permission requests behind an in-flight LLM insert', async () => { + const store = new Store(':memory:') + let releaseFirst!: (value: NotifySummary) => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => { + firstStarted() + return firstGate + }), + }, + }) + const live = makeSession('sess-perm', 'cursor') + const stored = store.sessions.getOrCreateSession('llm-perm', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('TURN A body'), Date.now()) + await firstStartedP + live.agentState = { + requests: { + req1: { tool: 'Bash', arguments: { command: 'ls' } } + } + } + const perm = recorder.onSessionUpdated(live, stored.tag) + releaseFirst({ status: 'done', summary: 'llm first' }) + await Promise.all([first, perm]) + + const rows = store.events.list().sort((a, b) => a.id - b.id) + expect(rows.map((row) => row.eventType)).toEqual(['completed', 'approval_requested']) + expect(rows[0]?.summary).toBe('llm first') + }) + + it('stamps permission requests at observe time, not queue-drain time', async () => { + const store = new Store(':memory:') + let releaseFirst!: (value: NotifySummary) => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstStartedP = new Promise((resolve) => { + firstStarted = resolve + }) + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => { + firstStarted() + return firstGate + }), + }, + }) + const live = makeSession('sess-perm-ts', 'cursor') + const stored = store.sessions.getOrCreateSession('llm-perm-ts', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + + const first = recorder.onAgentMessage(snapshot, 'msg-a', agentText('TURN A body'), Date.now()) + await firstStartedP + live.agentState = { + requests: { + req1: { tool: 'Bash', arguments: { command: 'ls' } } + } + } + const marked = Date.now() + const perm = recorder.onSessionUpdated(live, stored.tag) + await Bun.sleep(40) + releaseFirst({ status: 'done', summary: 'llm first' }) + await Promise.all([first, perm]) + + const row = store.events.list({ eventType: 'approval_requested' })[0] + expect(row).toBeDefined() + expect(row!.ts).toBeLessThan(marked + 25) + }) + + it('does not bump turn epoch on redelivered user messages', async () => { + const store = new Store(':memory:') + const recorder = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback: { + synthesizeNotifySummary: mock(async () => ({ status: 'done', summary: 'turn' })), + }, + }) + const live = makeSession('sess-redeliver', 'cursor', { thinking: true }) + const stored = store.sessions.getOrCreateSession('llm-redeliver', { flavor: 'cursor', path: '/tmp', host: 'local' }, null, 'default') + live.id = stored.id + const snapshot = toSessionSnapshot(live, stored.tag) + const user = { role: 'user', content: { type: 'text', text: 'go' } } + + await recorder.onAgentMessage(snapshot, 'user-1', user, Date.now()) + await recorder.onAgentMessage(snapshot, 'msg-a', agentText('working'), Date.now() + 1, { thinking: true }) + await recorder.flushPendingLlmFallback(snapshot) + await recorder.onAgentMessage(snapshot, 'user-1', user, Date.now() + 2) + + const event = await recorder.onSessionEnd( + live, + stored.tag, + Date.now() + 3, + 'completed', + () => 'working' + ) + expect(event).toBeNull() + expect(store.events.list().filter((row) => row.provenance?.includes('hub-llm-fallback'))).toHaveLength(1) + expect(store.events.list().some((row) => row.provenance?.includes('session-end'))).toBe(false) + }) +}) diff --git a/hub/src/sync/overseerEventRecorder.test.ts b/hub/src/sync/overseerEventRecorder.test.ts index 636721f28a..c71fe66064 100644 --- a/hub/src/sync/overseerEventRecorder.test.ts +++ b/hub/src/sync/overseerEventRecorder.test.ts @@ -28,7 +28,7 @@ function makeSession(id: string, flavor: string, overrides?: Partial): } describe('OverseerEventRecorder', () => { - it('records AGENT_NOTIFY_SUMMARY from codex assistant text', () => { + it('records AGENT_NOTIFY_SUMMARY from codex assistant text', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('test', { flavor: 'codex', path: '/tmp', host: 'local' }, null, 'default') @@ -44,7 +44,7 @@ describe('OverseerEventRecorder', () => { } } - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'codex'), session.tag), 'msg-1', content, @@ -72,7 +72,7 @@ describe('OverseerEventRecorder', () => { expect(item?.title).toBe('test') }) - it('captures done without action as captured-only', () => { + it('captures done without action as captured-only', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('test2', { flavor: 'claude', path: '/tmp', host: 'local' }, null, 'default') @@ -88,7 +88,7 @@ describe('OverseerEventRecorder', () => { } } - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'claude'), session.tag), 'msg-2', content, @@ -98,12 +98,12 @@ describe('OverseerEventRecorder', () => { expect(event?.attentionCandidate).toBe(0) }) - it('drops sentinel notify actions from suggested_action and inbox', () => { + it('drops sentinel notify actions from suggested_action and inbox', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('test-sentinel', { flavor: 'claude', path: '/tmp', host: 'local' }, null, 'default') - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'claude'), session.tag), 'msg-sentinel', { @@ -145,7 +145,7 @@ describe('OverseerEventRecorder', () => { expect(store.inbox.count()).toBe(0) }) - it('synthesizes approval_requested from permission prompts', () => { + it('synthesizes approval_requested from permission prompts', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('perm', { flavor: 'claude', path: '/tmp', host: 'local' }, null, 'default') @@ -157,7 +157,7 @@ describe('OverseerEventRecorder', () => { } } - recorder.onSessionUpdated(live, session.tag) + await recorder.onSessionUpdated(live, session.tag) const events = store.events.list({ eventType: 'approval_requested' }) expect(events).toHaveLength(1) @@ -171,7 +171,7 @@ describe('OverseerEventRecorder', () => { expect(store.inbox.list()[0]?.title).toBe('perm') }) - it('denormalizes session display name and project into payload.session', () => { + it('denormalizes session display name and project into payload.session', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const stored = store.sessions.getOrCreateSession( @@ -195,7 +195,7 @@ describe('OverseerEventRecorder', () => { } } - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(live, stored.tag), 'msg-meta', content, @@ -212,7 +212,7 @@ describe('OverseerEventRecorder', () => { expect(payload.session.id).toBe(stored.id) }) - it('ignores placeholder notify.project and keeps path-derived project', () => { + it('ignores placeholder notify.project and keeps path-derived project', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const stored = store.sessions.getOrCreateSession( @@ -239,7 +239,7 @@ describe('OverseerEventRecorder', () => { } } - const event = recorder.onAgentMessage( + const event = await recorder.onAgentMessage( toSessionSnapshot(live, stored.tag), 'msg-placeholder', content, @@ -254,7 +254,7 @@ describe('OverseerEventRecorder', () => { expect(event?.tags).not.toContain('agent:') }) - it('titles inbox items from payload.session.name after session delete', () => { + it('titles inbox items from payload.session.name after session delete', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const stored = store.sessions.getOrCreateSession( @@ -264,7 +264,7 @@ describe('OverseerEventRecorder', () => { 'default' ) - recorder.onAgentMessage( + await recorder.onAgentMessage( toSessionSnapshot(makeSession(stored.id, 'codex', { metadata: { flavor: 'codex', path: '/coding/hapi', name: 'meta HAPI triage', host: 'local' } }), stored.tag), @@ -292,7 +292,7 @@ describe('OverseerEventRecorder', () => { expect(itemAfter?.relatedSessionId).toBeNull() }) - it('scoops http(s) URLs into link_seen with artifact_refs kind:url', () => { + it('scoops http(s) URLs into link_seen with artifact_refs kind:url', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('links', { flavor: 'codex', path: '/tmp', host: 'local' }, null, 'default') @@ -312,7 +312,7 @@ describe('OverseerEventRecorder', () => { } } - const notify = recorder.onAgentMessage( + const notify = await recorder.onAgentMessage( toSessionSnapshot(makeSession(session.id, 'codex'), session.tag), 'msg-links', content, @@ -337,7 +337,7 @@ describe('OverseerEventRecorder', () => { expect(payload.session.id).toBe(session.id) }) - it('idempotently scoops the same URL from the same message once', () => { + it('idempotently scoops the same URL from the same message once', async () => { const store = new Store(':memory:') const recorder = new OverseerEventRecorder(store.events, store.inbox) const session = store.sessions.getOrCreateSession('dedupe', { flavor: 'claude', path: '/tmp', host: 'local' }, null, 'default') @@ -352,8 +352,8 @@ describe('OverseerEventRecorder', () => { } } const snapshot = toSessionSnapshot(makeSession(session.id, 'claude'), session.tag) - recorder.onAgentMessage(snapshot, 'msg-dup', content, Date.now()) - recorder.onAgentMessage(snapshot, 'msg-dup', content, Date.now()) + await recorder.onAgentMessage(snapshot, 'msg-dup', content, Date.now()) + await recorder.onAgentMessage(snapshot, 'msg-dup', content, Date.now()) expect(store.events.list({ eventType: 'link_seen' })).toHaveLength(1) }) }) diff --git a/hub/src/sync/overseerEventRecorder.ts b/hub/src/sync/overseerEventRecorder.ts index 11512c4cb7..5a973274f4 100644 --- a/hub/src/sync/overseerEventRecorder.ts +++ b/hub/src/sync/overseerEventRecorder.ts @@ -26,9 +26,32 @@ import { import type { Session } from '@hapi/protocol/types' import type { EventStore, InsertSystemEventInput, StoredSystemEvent } from '../store' import type { InboxStore } from '../store/inboxStore' +import type { OverseerLlmFallbackClient } from './overseerLlmFallback' export type SessionSnapshot = OverseerSessionIdentity +export type OverseerEventRecorderOptions = { + /** Opt-in OpenAI-compatible synthesizer (issue #90). Default: unset / off. */ + llmFallback?: OverseerLlmFallbackClient | null + /** Fired after an async Session Log insert so SSE clients can refetch. */ + onAsyncSystemEvent?: ((sessionId: string) => void) | null +} + +export type OnAgentMessageOptions = { + /** + * When true, defer opt-in LLM fallback until thinking clears so ACP + * mid-turn text flushes do not each trigger a synthesis call. + */ + thinking?: boolean +} + +type PendingLlmFallback = { + messageId: string + plainText: string + ts: number + epoch: number +} + function asRecord(value: unknown): Record | null { return isObject(value) ? value as Record : null } @@ -146,11 +169,27 @@ function buildTags(notify: NotifySummary | null, flavor: string): string | null export class OverseerEventRecorder { private readonly lastAgentMessageAt = new Map() private readonly knownPermissionRequestIds = new Map>() + private readonly llmFallback: OverseerLlmFallbackClient | null + private readonly onAsyncSystemEvent: ((sessionId: string) => void) | null + /** Latest no-notify assistant text awaiting end-of-turn LLM attempt. */ + private readonly pendingLlmFallback = new Map() + private readonly sessionThinking = new Map() + /** Per-session tail so concurrent LLM calls insert in arrival order. */ + private readonly sessionWork = new Map>() + private readonly turnEpoch = new Map() + /** Epoch of the last successful LLM fallback for this session. */ + private readonly llmFallbackSucceededEpoch = new Map() + private readonly seenUserMessageIds = new Map>() + private readonly sessionEnded = new Set() constructor( private readonly events: EventStore, - private readonly inbox?: InboxStore - ) {} + private readonly inbox?: InboxStore, + options?: OverseerEventRecorderOptions + ) { + this.llmFallback = options?.llmFallback ?? null + this.onAsyncSystemEvent = options?.onAsyncSystemEvent ?? null + } list(options: Parameters[0] = {}): StoredSystemEvent[] { return this.events.list(options) @@ -160,12 +199,13 @@ export class OverseerEventRecorder { return this.events.count() } - onAgentMessage( + async onAgentMessage( session: SessionSnapshot, messageId: string, content: unknown, - ts: number - ): StoredSystemEvent | null { + ts: number, + opts: OnAgentMessageOptions = {} + ): Promise { let primary: StoredSystemEvent | null = null if (isAgentMessageContent(content)) { @@ -174,38 +214,53 @@ export class OverseerEventRecorder { const agentBody = unwrapRoleWrappedRecordEnvelope(content) const agentContent = agentBody?.role === 'agent' ? agentBody.content : content + // Scoop URLs before any queued await so a slow/crashed fallback + // cannot drop already-persisted assistant links. + this.scoopLinksFromContent(session, messageId, content, ts) + const plainText = extractAssistantPlainText(agentContent) if (plainText) { if (detectEmptyHapiEventsSentinel(plainText)) { - primary = this.insertSystemEvent(session, { - ts, - sourceKind: 'system', - eventType: 'validation_error', - attentionCandidate: 0, - summary: 'Malformed HAPI_EVENTS sentinel block (empty body)', - relatedSessionId: session.id, - provenance: 'hub-inferred from empty HAPI_EVENTS sentinel pair', - idempotencyKey: `session:${session.id}:message:${messageId}:validation_error:empty_hapi_events`, - payloadFields: { messageId, plainTextPreview: plainText.slice(0, 500) }, - severity: 1 - }) + this.pendingLlmFallback.delete(session.id) + primary = await this.enqueueSessionWork(session.id, () => + Promise.resolve(this.insertInferredEvent(session, { + ts, + sourceKind: 'system', + eventType: 'validation_error', + attentionCandidate: 0, + summary: 'Malformed HAPI_EVENTS sentinel block (empty body)', + relatedSessionId: session.id, + provenance: 'hub-inferred from empty HAPI_EVENTS sentinel pair', + idempotencyKey: `session:${session.id}:message:${messageId}:validation_error:empty_hapi_events`, + payloadFields: { messageId, plainTextPreview: plainText.slice(0, 500) }, + severity: 1 + })) + ) } else if (detectMalformedNotifySummaryLine(plainText)) { - primary = this.insertSystemEvent(session, { - ts, - sourceKind: 'system', - eventType: 'validation_error', - attentionCandidate: 0, - summary: 'Malformed AGENT_NOTIFY_SUMMARY line on last turn', - relatedSessionId: session.id, - provenance: 'hub-inferred from malformed AGENT_NOTIFY_SUMMARY JSON', - idempotencyKey: `session:${session.id}:message:${messageId}:validation_error:malformed_notify`, - payloadFields: { messageId }, - severity: 1 - }) + this.pendingLlmFallback.delete(session.id) + primary = await this.enqueueSessionWork(session.id, () => + Promise.resolve(this.insertInferredEvent(session, { + ts, + sourceKind: 'system', + eventType: 'validation_error', + attentionCandidate: 0, + summary: 'Malformed AGENT_NOTIFY_SUMMARY line on last turn', + relatedSessionId: session.id, + provenance: 'hub-inferred from malformed AGENT_NOTIFY_SUMMARY JSON', + idempotencyKey: `session:${session.id}:message:${messageId}:validation_error:malformed_notify`, + payloadFields: { messageId }, + severity: 1 + })) + ) } else { const notify = extractNotifySummary(plainText) if (notify) { - primary = this.recordNotifySummary(session, messageId, notify, ts) + this.pendingLlmFallback.delete(session.id) + primary = await this.enqueueSessionWork(session.id, () => { + const stored = this.recordNotifySummary(session, messageId, notify, ts) + if (stored) this.onAsyncSystemEvent?.(session.id) + return Promise.resolve(stored) + }) } } } @@ -213,75 +268,224 @@ export class OverseerEventRecorder { if (!primary) { const toolFailure = extractToolFailureSummary(agentContent) if (toolFailure) { - primary = this.insertSystemEvent(session, { - ts, - sourceKind: 'system', - sourceRef: session.id, - eventType: 'failed', - attentionCandidate: 1, - operatorActionRequired: 1, - summary: toolFailure, - relatedSessionId: session.id, - provenance: 'hub-inferred from tool-call-result exit code', - idempotencyKey: `session:${session.id}:message:${messageId}:tool_failed`, - payloadFields: { messageId }, - severity: deriveSeverity('failed'), - tags: buildTags(null, session.flavor) - }) + primary = await this.enqueueSessionWork(session.id, () => + Promise.resolve(this.insertInferredEvent(session, { + ts, + sourceKind: 'system', + sourceRef: session.id, + eventType: 'failed', + attentionCandidate: 1, + operatorActionRequired: 1, + summary: toolFailure, + relatedSessionId: session.id, + provenance: 'hub-inferred from tool-call-result exit code', + idempotencyKey: `session:${session.id}:message:${messageId}:tool_failed`, + payloadFields: { messageId }, + severity: deriveSeverity('failed'), + tags: buildTags(null, session.flavor) + })) + ) } } - // No hub-synthesized "first line of assistant text" events. Session - // Log is fed by AGENT_NOTIFY_SUMMARY (agent wrapup) only. Opt-in LLM - // fallback (#90 / HAPI_OVERSEER_LLM_FALLBACK) is a separate layer. - // Per-tool / mid-turn narrative belongs in session-flow experiments, - // not here. + // Opt-in LLM only (#90). No first-line heuristic. Defer while + // thinking so ACP mid-turn flushes do not each hit the LLM. + if (!primary && plainText && this.llmFallback) { + if (opts.thinking) { + this.rememberPendingLlmFallback(session.id, messageId, plainText, ts) + } else { + this.pendingLlmFallback.delete(session.id) + this.bumpTurnEpoch(session.id) + const epoch = this.currentTurnEpoch(session.id) + primary = await this.enqueueSessionWork(session.id, () => + this.tryLlmFallback(session, messageId, plainText, ts, epoch) + ) + } + } + return primary } - // Always scoop URLs from any ingestible message text (agent or user). + const seen = this.seenUserMessageIds.get(session.id) ?? new Set() + if (!seen.has(messageId)) { + seen.add(messageId) + this.seenUserMessageIds.set(session.id, seen) + this.bumpTurnEpoch(session.id) + } this.scoopLinksFromContent(session, messageId, content, ts) return primary } - onSessionUpdated(session: Session, tag?: string | null): void { - this.syncPermissionRequests(session, tag ?? null) + async onSessionUpdated(session: Session, tag?: string | null): Promise { + const wasThinking = this.sessionThinking.get(session.id) === true + this.sessionThinking.set(session.id, session.thinking) + if (session.thinking && !wasThinking) { + this.bumpTurnEpoch(session.id) + } + await this.syncPermissionRequests(session, tag ?? null) + // Flush whenever thinking is clear and a deferred turn is waiting — + // not only on a true→false edge. Keepalives often never sent the + // thinking=true update through this recorder. + if (!session.thinking) { + await this.flushPendingLlmFallback(toSessionSnapshot(session, tag ?? null)) + } } - onSessionEnd( + async onSessionEnd( session: Session, tag: string | null, ts: number, reason: string | undefined, getLastAgentPlainText: () => string | null - ): StoredSystemEvent | null { + ): Promise { + this.sessionEnded.add(session.id) this.knownPermissionRequestIds.delete(session.id) + this.sessionThinking.delete(session.id) + const snapshot = toSessionSnapshot(session, tag) + const pending = this.takePendingLlmFallback(session.id) + + return this.enqueueSessionWork(session.id, async () => { + try { + const llmEvent = pending + ? await this.runPendingLlmFallback(snapshot, pending) + : null + if (reason !== 'completed') { + return llmEvent + } - if (reason !== 'completed') { - return null - } + const lastText = getLastAgentPlainText() + if (lastText && extractNotifySummary(lastText)) { + return llmEvent + } + // Successful LLM row already captured this turn (epoch), including + // same-turn ACP tool/usage messages after the text flush. + if (llmEvent || this.llmFallbackSucceededEpoch.get(session.id) === this.currentTurnEpoch(session.id)) { + return llmEvent + } - const lastText = getLastAgentPlainText() - if (lastText && extractNotifySummary(lastText)) { - return null - } + const stored = this.insertSystemEvent(snapshot, { + ts, + sourceKind: 'system', + sourceRef: session.id, + eventType: 'completed', + attentionCandidate: 0, + summary: 'Session ended without AGENT_NOTIFY_SUMMARY; hub inferred completion', + relatedSessionId: session.id, + provenance: 'hub-inferred from session-end completed signal', + idempotencyKey: `session:${session.id}:session_end:${ts}:completed_fallback`, + payloadFields: { reason }, + severity: deriveSeverity('completed'), + tags: buildTags(null, snapshot.flavor) + }) + if (stored) this.onAsyncSystemEvent?.(session.id) + return stored + } finally { + this.clearTurnState(session.id) + } + }) + } - // Session ended without a self-report — still a rare hub signal (not - // per-message text synth). Keeps teardown visible when agents bail. - const snapshot = toSessionSnapshot(session, tag) - return this.insertSystemEvent(snapshot, { + async flushPendingLlmFallback(session: SessionSnapshot): Promise { + const pending = this.takePendingLlmFallback(session.id) + if (!pending) return null + return this.enqueueSessionWork(session.id, () => this.runPendingLlmFallback(session, pending)) + } + + private takePendingLlmFallback(sessionId: string): PendingLlmFallback | undefined { + const pending = this.pendingLlmFallback.get(sessionId) + if (pending) this.pendingLlmFallback.delete(sessionId) + return pending + } + + private rememberPendingLlmFallback( + sessionId: string, + messageId: string, + plainText: string, + ts: number + ): void { + const prev = this.pendingLlmFallback.get(sessionId) + // New ACP text segment in the same thinking turn — keep the whole turn. + // Same messageId is a redelivery; replace rather than duplicate. + const combined = prev && prev.messageId !== messageId + ? `${prev.plainText}\n${plainText}` + : plainText + this.pendingLlmFallback.set(sessionId, { + messageId, + plainText: combined, ts, - sourceKind: 'system', - sourceRef: session.id, - eventType: 'completed', - attentionCandidate: 0, - summary: 'Session ended without AGENT_NOTIFY_SUMMARY; hub inferred completion', - relatedSessionId: session.id, - provenance: 'hub-inferred from session-end completed signal', - idempotencyKey: `session:${session.id}:session_end:${ts}:completed_fallback`, - payloadFields: { reason }, - severity: deriveSeverity('completed'), - tags: buildTags(null, snapshot.flavor) + epoch: prev?.epoch ?? this.currentTurnEpoch(sessionId) + }) + } + + private async runPendingLlmFallback( + session: SessionSnapshot, + pending: PendingLlmFallback + ): Promise { + if (extractNotifySummary(pending.plainText)) return null + return this.tryLlmFallback(session, pending.messageId, pending.plainText, pending.ts, pending.epoch) + } + + private enqueueSessionWork(sessionId: string, work: () => Promise): Promise { + const previous = this.sessionWork.get(sessionId) ?? Promise.resolve() + const run = previous.then(work, work) + const tail: Promise = run.then(() => undefined, () => undefined) + this.sessionWork.set(sessionId, tail) + void tail.then(() => { + if (this.sessionWork.get(sessionId) === tail) { + this.sessionWork.delete(sessionId) + } }) + return run + } + + /** + * Opt-in LLM synthesis only. Failures return null — never invent a + * first-line heuristic Session Log row. + */ + private async tryLlmFallback( + session: SessionSnapshot, + messageId: string, + plainText: string, + ts: number, + epoch: number + ): Promise { + if (!this.llmFallback) return null + try { + const notify = await this.llmFallback.synthesizeNotifySummary(plainText) + if (!notify) return null + // Session Log All hides `stale` (ambient silence). Keep LLM + // fallbacks visible as captured-only progress. + const mapped = mapNotifyStatusToEventType(notify.status) + const eventType = mapped === 'stale' ? 'progress' : mapped + const stored = this.insertSystemEvent(session, { + ts, + sourceKind: 'system', + sourceRef: session.id, + eventType, + attentionCandidate: 0, + operatorActionRequired: 0, + summary: buildEventSummaryFromNotify(notify), + relatedSessionId: session.id, + provenance: 'hub-llm-fallback (no AGENT_NOTIFY_SUMMARY from primary agent)', + idempotencyKey: `session:${session.id}:message:${messageId}:turn_fallback`, + payloadFields: { + messageId, + synthesized: true, + synthesis: 'llm-fallback', + notify_summary: notify, + suggested_action: notify.action ?? null, + }, + notifyProject: notify.project ?? null, + severity: deriveSeverity(eventType), + tags: buildTags(notify, session.flavor), + }) + if (stored) { + this.llmFallbackSucceededEpoch.set(session.id, epoch) + this.onAsyncSystemEvent?.(session.id) + } + return stored + } catch { + return null + } } /** @@ -303,6 +507,31 @@ export class OverseerEventRecorder { this.lastAgentMessageAt.set(sessionId, ts) } + /** Drop deferred LLM state when a session is deleted (do not flush). */ + forgetSession(sessionId: string): void { + this.pendingLlmFallback.delete(sessionId) + this.sessionWork.delete(sessionId) + this.clearTurnState(sessionId) + this.lastAgentMessageAt.delete(sessionId) + this.knownPermissionRequestIds.delete(sessionId) + this.sessionThinking.delete(sessionId) + this.sessionEnded.delete(sessionId) + } + + private clearTurnState(sessionId: string): void { + this.turnEpoch.delete(sessionId) + this.llmFallbackSucceededEpoch.delete(sessionId) + this.seenUserMessageIds.delete(sessionId) + } + + private currentTurnEpoch(sessionId: string): number { + return this.turnEpoch.get(sessionId) ?? 0 + } + + private bumpTurnEpoch(sessionId: string): void { + this.turnEpoch.set(sessionId, this.currentTurnEpoch(sessionId) + 1) + } + private scoopLinksFromContent( session: SessionSnapshot, messageId: string, @@ -381,7 +610,8 @@ export class OverseerEventRecorder { }) } - private syncPermissionRequests(session: Session, tag: string | null): void { + private async syncPermissionRequests(session: Session, tag: string | null): Promise { + if (this.sessionEnded.has(session.id)) return const requests = session.agentState?.requests ?? null if (!requests) { this.knownPermissionRequestIds.delete(session.id) @@ -394,27 +624,47 @@ export class OverseerEventRecorder { for (const requestId of currentIds) { if (known.has(requestId)) continue + known.add(requestId) + this.knownPermissionRequestIds.set(session.id, new Set(known)) const request = asRecord(requests[requestId]) const toolName = typeof request?.tool === 'string' ? request.tool : 'tool' const summary = `Permission requested: ${toolName}` - this.insertSystemEvent(snapshot, { - ts: Date.now(), - sourceKind: 'system', - sourceRef: session.id, - eventType: 'approval_requested', - attentionCandidate: 1, - operatorActionRequired: 1, - summary, - relatedSessionId: session.id, - provenance: 'hub-inferred from permission prompt', - idempotencyKey: `session:${session.id}:permission:${requestId}`, - payloadFields: { requestId, request }, - severity: deriveSeverity('approval_requested'), - tags: buildTags(null, snapshot.flavor) - }) + const ts = Date.now() + await this.enqueueSessionWork(session.id, () => + Promise.resolve(this.insertInferredEvent(snapshot, { + ts, + sourceKind: 'system', + sourceRef: session.id, + eventType: 'approval_requested', + attentionCandidate: 1, + operatorActionRequired: 1, + summary, + relatedSessionId: session.id, + provenance: 'hub-inferred from permission prompt', + idempotencyKey: `session:${session.id}:permission:${requestId}`, + payloadFields: { requestId, request }, + severity: deriveSeverity('approval_requested'), + tags: buildTags(null, snapshot.flavor) + })) + ) } - this.knownPermissionRequestIds.set(session.id, currentIds) + if (!this.sessionEnded.has(session.id)) { + this.knownPermissionRequestIds.set(session.id, currentIds) + } + } + + private insertInferredEvent( + session: SessionSnapshot, + input: Omit & { + riskDetected?: 0 | 1 + payloadFields?: Record + notifyProject?: string | null + } + ): StoredSystemEvent | null { + const stored = this.insertSystemEvent(session, input) + if (stored) this.onAsyncSystemEvent?.(session.id) + return stored } private insertSystemEvent( diff --git a/hub/src/sync/overseerLlmFallback.test.ts b/hub/src/sync/overseerLlmFallback.test.ts new file mode 100644 index 0000000000..ed6fb52568 --- /dev/null +++ b/hub/src/sync/overseerLlmFallback.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, it, mock } from 'bun:test' +import { + OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT, + createOverseerLlmFallbackClient, + extractTextFromChatCompletionsBody, + extractTextFromResponsesBody, + parseNotifySummaryFromLlmText, + type OverseerLlmFetch, +} from './overseerLlmFallback' +import type { OverseerLlmFallbackEnabledConfig } from './overseerLlmFallbackConfig' + +const baseConfig: OverseerLlmFallbackEnabledConfig = { + enabled: true, + baseUrl: 'http://llm.test/v1', + apiKey: 'test-key', + model: 'test-model', + api: 'chat-completions', + timeoutMs: 5_000, +} + +describe('parseNotifySummaryFromLlmText', () => { + it('parses a bare AGENT_NOTIFY_SUMMARY line', () => { + const text = 'AGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"Review PR","summary":"Shipped fix"}' + const notify = parseNotifySummaryFromLlmText(text) + expect(notify?.status).toBe('done') + expect(notify?.summary).toBe('Shipped fix') + expect(notify?.action).toBe('Review PR') + }) + + it('strips markdown fences before parse', () => { + const text = '```\nAGENT_NOTIFY_SUMMARY {"status":"blocked","summary":"Waiting on review"}\n```' + expect(parseNotifySummaryFromLlmText(text)?.summary).toBe('Waiting on review') + }) + + it('returns null for empty or non-compliant text', () => { + expect(parseNotifySummaryFromLlmText('')).toBeNull() + expect(parseNotifySummaryFromLlmText('just a paragraph')).toBeNull() + expect(parseNotifySummaryFromLlmText('AGENT_NOTIFY_SUMMARY {}')).toBeNull() + expect(parseNotifySummaryFromLlmText('AGENT_NOTIFY_SUMMARY {"status":"done"}')).toBeNull() + expect(parseNotifySummaryFromLlmText('AGENT_NOTIFY_SUMMARY {"summary":"no status"}')).toBeNull() + expect(parseNotifySummaryFromLlmText('AGENT_NOTIFY_SUMMARY {"status":"nope","summary":"bad status"}')).toBeNull() + }) +}) + +describe('response body extractors', () => { + it('reads chat completions choices[0].message.content', () => { + expect(extractTextFromChatCompletionsBody({ + choices: [{ message: { content: 'AGENT_NOTIFY_SUMMARY {"status":"done","summary":"ok"}' } }], + })).toContain('AGENT_NOTIFY_SUMMARY') + }) + + it('reads responses output_text when present', () => { + expect(extractTextFromResponsesBody({ + output_text: 'AGENT_NOTIFY_SUMMARY {"status":"failed","summary":"boom"}', + })).toContain('failed') + }) + + it('aggregates responses output message content text parts', () => { + expect(extractTextFromResponsesBody({ + output: [{ + type: 'message', + content: [{ type: 'output_text', text: 'AGENT_NOTIFY_SUMMARY {"status":"stalled","summary":"idle"}' }], + }], + })).toContain('stalled') + }) +}) + +describe('createOverseerLlmFallbackClient', () => { + it('POSTs chat completions with full turn text and parses notify', async () => { + const fetchMock = mock(async (input: string, init?: RequestInit) => { + expect(input).toBe('http://llm.test/v1/chat/completions') + expect(init?.method).toBe('POST') + const headers = init?.headers as Record + expect(headers.Authorization).toBe('Bearer test-key') + const body = JSON.parse(String(init?.body)) as { + model: string + messages: Array<{ role: string; content: string }> + } + expect(body.model).toBe('test-model') + expect(body.messages[0]?.role).toBe('system') + expect(body.messages[0]?.content).toBe(OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT) + expect(body.messages[1]?.content).toContain('FULL TURN BODY THAT IS LONG') + return new Response(JSON.stringify({ + choices: [{ + message: { + content: 'AGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"Merge","summary":"Turn complete"}', + }, + }], + }), { status: 200, headers: { 'content-type': 'application/json' } }) + }) + + const client = createOverseerLlmFallbackClient(baseConfig, { + fetchImpl: fetchMock as unknown as OverseerLlmFetch, + }) + const notify = await client.synthesizeNotifySummary('FULL TURN BODY THAT IS LONG\nline two') + expect(notify?.summary).toBe('Turn complete') + expect(notify?.status).toBe('done') + expect(fetchMock).toHaveBeenCalledTimes(1) + }) + + it('POSTs /responses when api=responses and uses store:false', async () => { + const fetchMock = mock(async (input: string, init?: RequestInit) => { + expect(input).toBe('http://llm.test/v1/responses') + const body = JSON.parse(String(init?.body)) as { + model: string + store: boolean + instructions: string + input: string + } + expect(body.store).toBe(false) + expect(body.instructions).toBe(OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT) + expect(body.input).toContain('assistant turn text') + expect(body.model).toBe('test-model') + return new Response(JSON.stringify({ + output_text: 'AGENT_NOTIFY_SUMMARY {"status":"needs_review","summary":"Please look"}', + }), { status: 200 }) + }) + + const client = createOverseerLlmFallbackClient( + { ...baseConfig, api: 'responses' }, + { fetchImpl: fetchMock as unknown as OverseerLlmFetch }, + ) + const notify = await client.synthesizeNotifySummary('assistant turn text') + expect(notify?.status).toBe('needs_review') + }) + + it('returns null on HTTP error so caller can fall through', async () => { + const fetchMock = mock(async () => new Response('nope', { status: 500 })) + const client = createOverseerLlmFallbackClient(baseConfig, { + fetchImpl: fetchMock as unknown as OverseerLlmFetch, + }) + expect(await client.synthesizeNotifySummary('text')).toBeNull() + }) + + it('returns null when model output is not a notify line', async () => { + const fetchMock = mock(async () => new Response(JSON.stringify({ + choices: [{ message: { content: 'Sorry, I cannot help with that.' } }], + }), { status: 200 })) + const client = createOverseerLlmFallbackClient(baseConfig, { + fetchImpl: fetchMock as unknown as OverseerLlmFetch, + }) + expect(await client.synthesizeNotifySummary('text')).toBeNull() + }) + + it('returns null for empty turn text without calling fetch', async () => { + const fetchMock = mock(async () => new Response('{}', { status: 200 })) + const client = createOverseerLlmFallbackClient(baseConfig, { + fetchImpl: fetchMock as unknown as OverseerLlmFetch, + }) + expect(await client.synthesizeNotifySummary(' \n ')).toBeNull() + expect(fetchMock).toHaveBeenCalledTimes(0) + }) +}) diff --git a/hub/src/sync/overseerLlmFallback.ts b/hub/src/sync/overseerLlmFallback.ts new file mode 100644 index 0000000000..c4dd1f6cf6 --- /dev/null +++ b/hub/src/sync/overseerLlmFallback.ts @@ -0,0 +1,182 @@ +import { NOTIFY_SUMMARY_STATUSES } from '@hapi/protocol' +import { extractNotifySummary, type NotifySummary } from '@hapi/protocol/messages' +import type { OverseerLlmFallbackEnabledConfig } from './overseerLlmFallbackConfig' + +/** + * Fixed system prompt for Option A hub LLM fallback. + * Asks for exactly one AGENT_NOTIFY_SUMMARY line — same contract as primary agents. + */ +export const OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT = [ + 'You summarize an AI coding agent turn for session tracking.', + 'Reply with exactly one line and nothing else (no markdown fences, no prose):', + 'AGENT_NOTIFY_SUMMARY {"version":1,"status":"done|blocked|needs_review|needs_decision|failed|stalled","action":"<=12 words","summary":"one-line triage"}', + 'Use status blocked if unsure. action must be concrete when status is done and follow-up remains.', +].join('\n') + +export type OverseerLlmFallbackClient = { + synthesizeNotifySummary(plainText: string): Promise +} + +export type OverseerLlmFetch = ( + input: string, + init?: RequestInit +) => Promise + +export type OverseerLlmFallbackClientOptions = { + fetchImpl?: OverseerLlmFetch +} + +function isObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +/** Strip common markdown fences so local models that wrap output still parse. */ +function stripMarkdownFences(text: string): string { + const trimmed = text.trim() + const fenced = trimmed.match(/^```(?:\w+)?\s*\n?([\s\S]*?)\n?```$/u) + if (fenced?.[1]) return fenced[1].trim() + return trimmed +} + +/** + * Parse an AGENT_NOTIFY_SUMMARY from LLM output. + * Uses the same end-anchored extractor as primary agent turns. + */ +const LLM_NOTIFY_STATUSES = new Set(NOTIFY_SUMMARY_STATUSES) + +function isCompliantLlmNotify(notify: NotifySummary): boolean { + const summary = notify.summary?.trim() + if (!summary) return false + if (!notify.status || !LLM_NOTIFY_STATUSES.has(notify.status)) return false + return true +} + +export function parseNotifySummaryFromLlmText(text: string): NotifySummary | null { + if (typeof text !== 'string' || text.trim().length === 0) return null + const parsed = extractNotifySummary(stripMarkdownFences(text)) + if (!parsed || !isCompliantLlmNotify(parsed)) return null + return parsed +} + +export function extractTextFromChatCompletionsBody(body: unknown): string | null { + if (!isObject(body)) return null + const choices = body.choices + if (!Array.isArray(choices) || choices.length === 0) return null + const first = choices[0] + if (!isObject(first)) return null + const message = first.message + if (!isObject(message)) return null + const content = message.content + if (typeof content === 'string' && content.trim().length > 0) return content + if (Array.isArray(content)) { + const parts: string[] = [] + for (const part of content) { + if (typeof part === 'string') parts.push(part) + else if (isObject(part) && typeof part.text === 'string') parts.push(part.text) + } + const joined = parts.join('\n').trim() + return joined.length > 0 ? joined : null + } + return null +} + +export function extractTextFromResponsesBody(body: unknown): string | null { + if (!isObject(body)) return null + if (typeof body.output_text === 'string' && body.output_text.trim().length > 0) { + return body.output_text + } + const output = body.output + if (!Array.isArray(output)) return null + const parts: string[] = [] + for (const item of output) { + if (!isObject(item)) continue + if (item.type !== 'message') continue + const content = item.content + if (!Array.isArray(content)) continue + for (const part of content) { + if (!isObject(part)) continue + if ((part.type === 'output_text' || part.type === 'text') && typeof part.text === 'string') { + parts.push(part.text) + } + } + } + const joined = parts.join('\n').trim() + return joined.length > 0 ? joined : null +} + +function joinUrl(baseUrl: string, path: string): string { + return `${baseUrl.replace(/\/+$/, '')}/${path.replace(/^\/+/, '')}` +} + +export function createOverseerLlmFallbackClient( + config: OverseerLlmFallbackEnabledConfig, + options: OverseerLlmFallbackClientOptions = {} +): OverseerLlmFallbackClient { + const fetchImpl: OverseerLlmFetch = options.fetchImpl + ?? ((input, init) => fetch(input, init)) + + return { + async synthesizeNotifySummary(plainText: string): Promise { + const turn = plainText.trim() + if (!turn) return null + + const controller = new AbortController() + const timer = setTimeout(() => controller.abort(), config.timeoutMs) + + try { + const headers: Record = { + 'Content-Type': 'application/json', + } + if (config.apiKey) { + headers.Authorization = `Bearer ${config.apiKey}` + } + + let url: string + let body: Record + if (config.api === 'responses') { + url = joinUrl(config.baseUrl, 'responses') + body = { + model: config.model, + instructions: OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT, + input: turn, + store: false, + } + } else { + url = joinUrl(config.baseUrl, 'chat/completions') + body = { + model: config.model, + messages: [ + { role: 'system', content: OVERSEER_LLM_FALLBACK_SYSTEM_PROMPT }, + { role: 'user', content: turn }, + ], + } + } + + const response = await fetchImpl(url, { + method: 'POST', + headers, + body: JSON.stringify(body), + signal: controller.signal, + }) + if (!response.ok) return null + + let parsed: unknown + try { + parsed = await response.json() + } catch { + return null + } + + const text = config.api === 'responses' + ? extractTextFromResponsesBody(parsed) + : extractTextFromChatCompletionsBody(parsed) + if (!text) return null + return parseNotifySummaryFromLlmText(text) + } catch { + return null + } finally { + clearTimeout(timer) + } + }, + } +} diff --git a/hub/src/sync/overseerLlmFallbackConfig.test.ts b/hub/src/sync/overseerLlmFallbackConfig.test.ts new file mode 100644 index 0000000000..5b9b846915 --- /dev/null +++ b/hub/src/sync/overseerLlmFallbackConfig.test.ts @@ -0,0 +1,163 @@ +import { afterEach, describe, expect, it } from 'bun:test' +import { loadOverseerLlmFallbackConfig, redactOverseerLlmBaseUrlForLog } from './overseerLlmFallbackConfig' + +const ENV_KEYS = [ + 'HAPI_OVERSEER_LLM_FALLBACK', + 'HAPI_OVERSEER_LLM_BASE_URL', + 'HAPI_OVERSEER_LLM_API_KEY', + 'HAPI_OVERSEER_LLM_MODEL', + 'HAPI_OVERSEER_LLM_API', + 'HAPI_OVERSEER_LLM_TIMEOUT_MS', +] as const + +const saved: Partial> = {} + +function stashEnv(): void { + for (const key of ENV_KEYS) { + saved[key] = process.env[key] + delete process.env[key] + } +} + +function restoreEnv(): void { + for (const key of ENV_KEYS) { + const value = saved[key] + if (value === undefined) delete process.env[key] + else process.env[key] = value + } +} + +afterEach(() => { + restoreEnv() +}) + +describe('loadOverseerLlmFallbackConfig', () => { + it('defaults to disabled when env is unset', () => { + stashEnv() + const config = loadOverseerLlmFallbackConfig() + expect(config.enabled).toBe(false) + if (config.enabled) throw new Error('expected disabled') + expect(config.reasonDisabled).toBe('flag_off') + }) + + it('stays disabled when flag is on but base URL or model missing', () => { + stashEnv() + process.env.HAPI_OVERSEER_LLM_FALLBACK = '1' + process.env.HAPI_OVERSEER_LLM_MODEL = 'llama3.3' + const config = loadOverseerLlmFallbackConfig() + expect(config.enabled).toBe(false) + if (config.enabled) throw new Error('expected disabled') + expect(config.reasonDisabled).toBe('incomplete_config') + }) + + it('enables with chat-completions defaults when flag + url + model set', () => { + stashEnv() + process.env.HAPI_OVERSEER_LLM_FALLBACK = 'true' + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'http://127.0.0.1:11434/v1' + process.env.HAPI_OVERSEER_LLM_MODEL = 'llama3.3' + const config = loadOverseerLlmFallbackConfig() + expect(config.enabled).toBe(true) + if (!config.enabled) throw new Error('expected enabled') + expect(config.baseUrl).toBe('http://127.0.0.1:11434/v1') + expect(config.model).toBe('llama3.3') + expect(config.api).toBe('chat-completions') + expect(config.apiKey).toBe('') + expect(config.timeoutMs).toBe(30_000) + }) + + it('accepts responses api mode and custom timeout/key', () => { + stashEnv() + process.env.HAPI_OVERSEER_LLM_FALLBACK = '1' + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://api.openai.com/v1/' + process.env.HAPI_OVERSEER_LLM_MODEL = 'gpt-4.1-mini' + process.env.HAPI_OVERSEER_LLM_API = 'responses' + process.env.HAPI_OVERSEER_LLM_API_KEY = 'sk-test' + process.env.HAPI_OVERSEER_LLM_TIMEOUT_MS = '12000' + const config = loadOverseerLlmFallbackConfig() + expect(config.enabled).toBe(true) + if (!config.enabled) throw new Error('expected enabled') + expect(config.baseUrl).toBe('https://api.openai.com/v1') + expect(config.api).toBe('responses') + expect(config.apiKey).toBe('sk-test') + expect(config.timeoutMs).toBe(12_000) + }) + + it('rejects partially numeric timeout values', () => { + stashEnv() + process.env.HAPI_OVERSEER_LLM_FALLBACK = '1' + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'http://127.0.0.1:11434/v1' + process.env.HAPI_OVERSEER_LLM_MODEL = 'llama3.3' + process.env.HAPI_OVERSEER_LLM_TIMEOUT_MS = '30s' + const suffix = loadOverseerLlmFallbackConfig() + expect(suffix.enabled).toBe(false) + if (suffix.enabled) throw new Error('expected disabled') + expect(suffix.reasonDisabled).toBe('invalid_timeout') + + process.env.HAPI_OVERSEER_LLM_TIMEOUT_MS = '1e3' + const scientific = loadOverseerLlmFallbackConfig() + expect(scientific.enabled).toBe(false) + if (scientific.enabled) throw new Error('expected disabled') + expect(scientific.reasonDisabled).toBe('invalid_timeout') + + process.env.HAPI_OVERSEER_LLM_TIMEOUT_MS = '2147483648' + const overflow = loadOverseerLlmFallbackConfig() + expect(overflow.enabled).toBe(false) + if (overflow.enabled) throw new Error('expected disabled') + expect(overflow.reasonDisabled).toBe('invalid_timeout') + }) + + it('rejects malformed fallback base URLs', () => { + stashEnv() + process.env.HAPI_OVERSEER_LLM_FALLBACK = '1' + process.env.HAPI_OVERSEER_LLM_MODEL = 'llama3.3' + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'localhost:11434/v1' + const missingScheme = loadOverseerLlmFallbackConfig() + expect(missingScheme.enabled).toBe(false) + if (missingScheme.enabled) throw new Error('expected disabled') + expect(missingScheme.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = '/' + const slash = loadOverseerLlmFallbackConfig() + expect(slash.enabled).toBe(false) + if (slash.enabled) throw new Error('expected disabled') + expect(slash.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://host/v1?tenant=x' + const query = loadOverseerLlmFallbackConfig() + expect(query.enabled).toBe(false) + if (query.enabled) throw new Error('expected disabled') + expect(query.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://host/v1#frag' + const hash = loadOverseerLlmFallbackConfig() + expect(hash.enabled).toBe(false) + if (hash.enabled) throw new Error('expected disabled') + expect(hash.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://user:secret@gateway/v1' + const userinfo = loadOverseerLlmFallbackConfig() + expect(userinfo.enabled).toBe(false) + if (userinfo.enabled) throw new Error('expected disabled') + expect(userinfo.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://host/v1?' + const emptyQuery = loadOverseerLlmFallbackConfig() + expect(emptyQuery.enabled).toBe(false) + if (emptyQuery.enabled) throw new Error('expected disabled') + expect(emptyQuery.reasonDisabled).toBe('invalid_base_url') + + process.env.HAPI_OVERSEER_LLM_BASE_URL = 'https://host/v1#' + const emptyHash = loadOverseerLlmFallbackConfig() + expect(emptyHash.enabled).toBe(false) + if (emptyHash.enabled) throw new Error('expected disabled') + expect(emptyHash.reasonDisabled).toBe('invalid_base_url') + }) + + it('redacts URL userinfo for startup logs', () => { + expect(redactOverseerLlmBaseUrlForLog('https://user:secret@gateway/v1')).toBe( + 'https://REDACTED:REDACTED@gateway/v1' + ) + expect(redactOverseerLlmBaseUrlForLog('https://gateway/v1')).toBe('https://gateway/v1') + expect(redactOverseerLlmBaseUrlForLog('not a url')).toBe('[invalid-url]') + }) +}) diff --git a/hub/src/sync/overseerLlmFallbackConfig.ts b/hub/src/sync/overseerLlmFallbackConfig.ts new file mode 100644 index 0000000000..dd11f32fdf --- /dev/null +++ b/hub/src/sync/overseerLlmFallbackConfig.ts @@ -0,0 +1,131 @@ +/** + * Opt-in hub LLM fallback for missing AGENT_NOTIFY_SUMMARY (fork issue #90). + * + * Default OFF. Enable only after primary emission miss rate is rare (~<5%). + * Env-only for v1 — never surprise usage. + * + * HAPI_OVERSEER_LLM_FALLBACK=1 + * HAPI_OVERSEER_LLM_BASE_URL=http://127.0.0.1:11434/v1 + * HAPI_OVERSEER_LLM_MODEL=llama3.3 + * HAPI_OVERSEER_LLM_API_KEY= # optional for local gateways + * HAPI_OVERSEER_LLM_API=chat-completions|responses # default chat-completions + * HAPI_OVERSEER_LLM_TIMEOUT_MS=30000 + */ + +export type OverseerLlmApiMode = 'chat-completions' | 'responses' + +export type OverseerLlmFallbackEnabledConfig = { + enabled: true + baseUrl: string + apiKey: string + model: string + api: OverseerLlmApiMode + timeoutMs: number +} + +export type OverseerLlmFallbackDisabledConfig = { + enabled: false + reasonDisabled: 'flag_off' | 'incomplete_config' | 'invalid_api' | 'invalid_timeout' | 'invalid_base_url' +} + +export type OverseerLlmFallbackConfig = + | OverseerLlmFallbackEnabledConfig + | OverseerLlmFallbackDisabledConfig + +const DEFAULT_TIMEOUT_MS = 30_000 +/** Node/Bun setTimeout clamps delays above 2^31-1 ms to 1 ms. */ +const MAX_TIMEOUT_MS = 2_147_483_647 + +function envTruthy(value: string | undefined): boolean { + if (!value) return false + const normalized = value.trim().toLowerCase() + return normalized === '1' || normalized === 'true' || normalized === 'yes' || normalized === 'on' +} + +function normalizeBaseUrl(raw: string): string { + return raw.trim().replace(/\/+$/, '') +} + +export function redactOverseerLlmBaseUrlForLog(url: string): string { + try { + const parsed = new URL(url) + if (parsed.username !== '' || parsed.password !== '') { + parsed.username = parsed.username !== '' ? 'REDACTED' : '' + parsed.password = parsed.password !== '' ? 'REDACTED' : '' + } + return parsed.toString().replace(/\/$/, '') + } catch { + return '[invalid-url]' + } +} + +function isAbsoluteHttpUrl(value: string): boolean { + try { + const parsed = new URL(value) + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') return false + // URL() strips empty `?` / `#`; joinUrl would then glue the path into query/hash. + if (value.includes('?') || value.includes('#')) return false + if (parsed.username !== '' || parsed.password !== '') return false + return true + } catch { + return false + } +} + +function parseApiMode(raw: string | undefined): OverseerLlmApiMode | null { + if (!raw || raw.trim() === '') return 'chat-completions' + const normalized = raw.trim().toLowerCase() + if (normalized === 'chat-completions' || normalized === 'chat_completions' || normalized === 'chat') { + return 'chat-completions' + } + if (normalized === 'responses' || normalized === 'response') { + return 'responses' + } + return null +} + +export function loadOverseerLlmFallbackConfig( + env: NodeJS.ProcessEnv = process.env +): OverseerLlmFallbackConfig { + if (!envTruthy(env.HAPI_OVERSEER_LLM_FALLBACK)) { + return { enabled: false, reasonDisabled: 'flag_off' } + } + + const baseUrlRaw = env.HAPI_OVERSEER_LLM_BASE_URL?.trim() ?? '' + const model = env.HAPI_OVERSEER_LLM_MODEL?.trim() ?? '' + if (!baseUrlRaw || !model) { + return { enabled: false, reasonDisabled: 'incomplete_config' } + } + + const api = parseApiMode(env.HAPI_OVERSEER_LLM_API) + if (!api) { + return { enabled: false, reasonDisabled: 'invalid_api' } + } + + let timeoutMs = DEFAULT_TIMEOUT_MS + const timeoutRaw = env.HAPI_OVERSEER_LLM_TIMEOUT_MS?.trim() + if (timeoutRaw) { + if (!/^\d+$/.test(timeoutRaw)) { + return { enabled: false, reasonDisabled: 'invalid_timeout' } + } + const parsed = Number.parseInt(timeoutRaw, 10) + if (!Number.isFinite(parsed) || parsed <= 0 || parsed > MAX_TIMEOUT_MS) { + return { enabled: false, reasonDisabled: 'invalid_timeout' } + } + timeoutMs = parsed + } + + const baseUrl = normalizeBaseUrl(baseUrlRaw) + if (!isAbsoluteHttpUrl(baseUrl)) { + return { enabled: false, reasonDisabled: 'invalid_base_url' } + } + + return { + enabled: true, + baseUrl, + apiKey: env.HAPI_OVERSEER_LLM_API_KEY?.trim() ?? '', + model, + api, + timeoutMs, + } +} diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index f4bde7e656..b9996e3520 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -43,6 +43,8 @@ import { } from './rpcGateway' import { SessionCache } from './sessionCache' import { OverseerEventRecorder, toSessionSnapshot } from './overseerEventRecorder' +import { createOverseerLlmFallbackClient } from './overseerLlmFallback' +import { loadOverseerLlmFallbackConfig, redactOverseerLlmBaseUrlForLog } from './overseerLlmFallbackConfig' import { OverseerEntity } from './overseerEntity' import { extractAssistantPlainText } from '@hapi/protocol/messages' import type { InboxOperatorAction } from '@hapi/protocol' @@ -165,7 +167,23 @@ export class SyncEngine { (sessionId, updatedAt) => this.recordSessionActivity(sessionId, updatedAt) ) this.rpcGateway = new RpcGateway(io, rpcRegistry) - this.overseerEvents = new OverseerEventRecorder(store.events, store.inbox) + const llmFallbackConfig = loadOverseerLlmFallbackConfig() + const llmFallback = llmFallbackConfig.enabled + ? createOverseerLlmFallbackClient(llmFallbackConfig) + : null + if (llmFallbackConfig.enabled) { + console.log( + `[overseer] LLM summary fallback ENABLED (api=${llmFallbackConfig.api}, model=${llmFallbackConfig.model}, base=${redactOverseerLlmBaseUrlForLog(llmFallbackConfig.baseUrl)})` + ) + } else if (llmFallbackConfig.reasonDisabled !== 'flag_off') { + console.warn(`[overseer] LLM summary fallback disabled (${llmFallbackConfig.reasonDisabled})`) + } + this.overseerEvents = new OverseerEventRecorder(store.events, store.inbox, { + llmFallback, + onAsyncSystemEvent: (sessionId) => { + this.eventPublisher.emit({ type: 'session-updated', sessionId }) + } + }) this.overseer = new OverseerEntity({ events: store.events, inbox: store.inbox, @@ -296,10 +314,12 @@ export class SyncEngine { this.sessionCache.refreshSession(event.sessionId) const after = this.sessionCache.getSession(event.sessionId) if (after) { - this.overseerEvents.onSessionUpdated( + void this.overseerEvents.onSessionUpdated( after, this.store.sessions.getSession(after.id)?.tag ?? null - ) + ).catch((error) => { + console.error('[overseer] onSessionUpdated failed', error) + }) } if (after?.metadata && !this.hasSameAgentSessionIds(before?.metadata ?? null, after.metadata)) { if (!this.canRunCursorDedup(after)) { @@ -328,8 +348,11 @@ export class SyncEngine { toSessionSnapshot(session, storedSession?.tag ?? null), event.message.id, event.message.content, - event.message.createdAt - ) + event.message.createdAt, + { thinking: session.thinking } + ).catch((error) => { + console.error('[overseer] onAgentMessage failed', error) + }) } } @@ -388,6 +411,17 @@ export class SyncEngine { }): void { this.sessionCache.handleSessionAlive(payload) this.triggerDedupIfNeeded(payload.sid) + const session = this.getSession(payload.sid) + if (session) { + // thinking=true→false usually arrives on session-alive, not + // session-updated. Flush deferred LLM fallback on that path. + void this.overseerEvents.onSessionUpdated( + session, + this.store.sessions.getSession(session.id)?.tag ?? null + ).catch((error) => { + console.error('[overseer] onSessionUpdated from session-alive failed', error) + }) + } } handleSessionReady(payload: { sid: string; time: number }): void { @@ -407,27 +441,32 @@ export class SyncEngine { this.sessionCache.handleSessionEnd(payload) const session = this.getSession(payload.sid) - if (session) { - this.overseerEvents.onSessionEnd( - session, - this.store.sessions.getSession(session.id)?.tag ?? null, - payload.time, - payload.reason, - () => this.getLastAgentPlainText(session.id) - ) - } this.eventPublisher.emit({ type: 'session-ended', sessionId: payload.sid, reason: payload.reason }) - // Retry dedup now that this session is inactive — a prior dedup may have - // skipped it because it was still active at the time. Cursor ACP rows that - // never reached session-ready must not dedup-merge the original on failure. - if (shouldRetryDedup) { - this.triggerDedupIfNeeded(payload.sid) - } - this.sessionReadyIds.delete(payload.sid) + // Await recorder work before dedup so a queued LLM/completed_fallback + // insert still has a live relatedSessionId. + void (async () => { + try { + if (session) { + await this.overseerEvents.onSessionEnd( + session, + this.store.sessions.getSession(session.id)?.tag ?? null, + payload.time, + payload.reason, + () => this.getLastAgentPlainText(session.id) + ) + } + } catch (error) { + console.error('[overseer] onSessionEnd failed', error) + } + if (shouldRetryDedup) { + this.triggerDedupIfNeeded(payload.sid) + } + this.sessionReadyIds.delete(payload.sid) + })() } handleBackgroundTaskDelta(sessionId: string, delta: { started: number; completed: number }): void { @@ -444,15 +483,29 @@ export class SyncEngine { private expireInactive(): void { const expired = this.sessionCache.expireInactive() - // Sort by most recent first so dedup keeps the newest session when multiple - // duplicates for the same agent thread expire in the same sweep. - const sorted = expired - .map((id) => this.sessionCache.getSession(id)) - .filter((s): s is NonNullable => s != null) - .sort((a, b) => (b.activeAt - a.activeAt) || (b.updatedAt - a.updatedAt)) - for (const session of sorted) { - this.triggerDedupIfNeeded(session.id) - } + void (async () => { + for (const sessionId of expired) { + const session = this.sessionCache.getSession(sessionId) + if (!session) continue + try { + await this.overseerEvents.onSessionUpdated( + session, + this.store.sessions.getSession(sessionId)?.tag ?? null + ) + } catch (error) { + console.error('[overseer] onSessionUpdated from expireInactive failed', error) + } + } + // Sort by most recent first so dedup keeps the newest session when multiple + // duplicates for the same agent thread expire in the same sweep. + const sorted = expired + .map((id) => this.sessionCache.getSession(id)) + .filter((s): s is NonNullable => s != null) + .sort((a, b) => (b.activeAt - a.activeAt) || (b.updatedAt - a.updatedAt)) + for (const session of sorted) { + this.triggerDedupIfNeeded(session.id) + } + })() this.machineCache.expireInactive() this.overseerEvents.checkStaleSessions(this.sessionCache.getSessions()) // Piggybacked on the inactivity tick; not a logical part of expireInactive @@ -748,6 +801,7 @@ export class SyncEngine { async deleteSession(sessionId: string): Promise { await this.sessionCache.deleteSession(sessionId) + this.overseerEvents.forgetSession(sessionId) } async applySessionConfig( diff --git a/web/src/hooks/useSSE.ts b/web/src/hooks/useSSE.ts index fe462d82af..6258dfad77 100644 --- a/web/src/hooks/useSSE.ts +++ b/web/src/hooks/useSSE.ts @@ -487,6 +487,10 @@ export function useSSE(options: { ingestIncomingMessages(event.sessionId, [event.message]) } + if (event.type === 'session-updated' || event.type === 'message-received' || event.type === 'session-ended') { + void queryClient.invalidateQueries({ queryKey: ['session-system-events', event.sessionId] }) + } + if (event.type === 'session-added' || event.type === 'session-updated' || event.type === 'session-removed') { if (event.type === 'session-removed') { removeSessionSummary(event.sessionId) diff --git a/web/src/routes/settings/index.test.tsx b/web/src/routes/settings/index.test.tsx index b87b94fffa..55907f4bcb 100644 --- a/web/src/routes/settings/index.test.tsx +++ b/web/src/routes/settings/index.test.tsx @@ -23,12 +23,17 @@ vi.mock('@tanstack/react-router', () => ({ useNavigate: () => navigate, })) -vi.mock('@hapi/protocol', () => ({ PROTOCOL_VERSION: 1 })) - vi.mock('@/lib/app-context', () => ({ - useAppContext: () => ({ api: null, token: '', baseUrl: '' }), + useAppContext: () => ({ + api: { + fetchSystemEvents: vi.fn(async () => ({ total: 0, events: [] })), + fetchInboxItems: vi.fn(async () => ({ total: 0, items: [] })), + }, + }), })) +vi.mock('@hapi/protocol', () => ({ PROTOCOL_VERSION: 1 })) + vi.mock('@/hooks/useTheme', () => ({ useAppearance: () => ({ appearance: 'system', setAppearance }), getAppearanceOptions: () => [