mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-06 18:39:47 +00:00
perf(hub,web): emit structured patches for session todos/teamState/metadata/agentState writes (closes #895, second half of #884) (#897)
* perf(hub,web): emit structured patches for session todos/teamState/metadata/agentState writes (#895, closes second half of #884) Today the four CLI handlers in `sessionHandlers.ts` that write session-scoped state (TodoWrite messages -> setSessionTodos; team-state deltas -> setSessionTeamState; update-metadata RPC; update-state RPC) emit `session-updated` events with no `data` payload. `syncEngine.handleRealtimeEvent` intercepts each one, re-reads the row from SQLite, and broadcasts the entire ~5KB Session via SSE. That works (the web client's `isSessionRecord` shortcut keeps the cache patched), but it costs a DB read and a full-payload SSE fan-out per write, and any failure mode that drops the broadcast data falls through to `useSSE.ts:509-512` and triggers per-session REST refetches - the storm vector documented in #884. This is the architectural follow-up to #885. #885 added `staleTime` on the detail query (eliminates focus / mount refetches inside a 30s window). This PR removes the structural reason these four writes touch the REST path at all. `SessionPatchSchema` learns four optional structured fields: - `todos` (array) - `teamState` (object) - `metadata` (versioned `{ version, value }` wrapper) - `agentState` (versioned `{ version, value }` wrapper) `.strict()` preserved so unknown keys still throw. The versioned wrappers mirror the existing socket.io `update-session` broadcast at lines 211 / 259 so metadata and agentState always travel as an atomic (version, value) pair - caches need the version to reject stale patches. Each of the four emit-sites now carries a structured `data` payload with the delta it just wrote. `syncEngine.handleRealtimeEvent` for `session-updated` events with non-empty patch data: applies the patch to the in-memory Session in place via the new `sessionCache.applySessionPatch`, then forwards the event as-is. Empty patches, no-data events, and patches against uncached sessions all fall back to the legacy `refreshSession` path so behavior for other emitters (e.g. `cursor/codexDesktop.ts`) is unchanged. Dedup hook against agent-session-id changes preserved on the fast path. `patchSessionDetail` is no longer a blanket spread - it enumerates each field explicitly so the versioned metadata / agentState patches can be unwrapped into the Session's flat (metadata, metadataVersion) and (agentState, agentStateVersion) pairs. Spreading the patch wholesale would have written a `{ version, value }` object into `session.metadata` and corrupted the cache. `patchSessionSummary` recomputes the touched derivations - `todoProgress` from todos, `pendingRequestsCount` / `pendingRequestKinds` from agentState, SessionSummaryMetadata from metadata - via three new pure helpers exposed from `shared/src/sessionSummary.ts` (`computeTodoProgress`, `computePendingRequestKinds`, `toSessionSummaryMetadata`). `toSessionSummary` is refactored to use these helpers - identical output, single source of truth. - shared: `SessionPatchSchema` parses each new patch shape, stays strict, rejects empty metadata without `version`, rejects full Session payloads (those go through `isSessionRecord`). - shared: summary derivation helpers covered against bare AgentState / Metadata inputs (the shape the SSE patch path provides). - hub: each emit-site asserted to carry the expected structured payload. - hub: `applySessionPatch` unit-tests cover todos / metadata / agentState application, empty-patch rejection (forces caller back to refreshSession), cross-namespace guard, and missing-session fallback. Empirical wire round-trip verifies each patch shape survives `JSON.stringify` intact and routes the web client through `getSessionPatch` (non-empty result) instead of the REST invalidation fallback. Per #884 expectation: with this fix on top of #885, idle GET /api/sessions/<id> rate is expected to drop to near-zero on the reporter's 100+ session install. Operator (heavygee) will attach the live-measured before / after to the PR post-merge. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(hub): snapshot metadata reference before applySessionPatch mutation so dedup-on-id-change fires The structured-patch fast path added in 05147a6a (closes #884 second half) broke the dedup-on-metadata-change trigger in handleRealtimeEvent. Root cause: applySessionPatch MUTATES the cached Session in place (reassigns session.metadata = patch.metadata.value). The dedup check compares before vs after agent session IDs, but `before = getSession(id)` and `after = getSession(id)` returned the SAME object reference, so before.metadata had already been overwritten by the time the check ran. hasSameAgentSessionIds always returned true and dedup silently never fired on the fast path. The legacy refreshSession path got dedup for free because it REPLACES the cache map entry with a new Session object, leaving the pre-refresh reference intact for the comparator. Fix: capture beforeMetadata before applySessionPatch runs; use it for both branches so the comparison contract is identical. Adds syncEngineHandleRealtimeEvent.test.ts with three regression guards: - structured metadata patch with changed cursorSessionId fires dedup - todos-only patch does NOT fire dedup (no false positives) - legacy refresh path (no patch data) still fires dedup Co-authored-by: Cursor <cursoragent@cursor.com> * fix(schemas): reorder SessionPatchSchema fields so soup-merge with codex-usage layer conflicts cleanly Pure reorder (no semantic change). feat/codex-usage-indicator-rebased adds a flat `metadata: MetadataSchema.nullable().optional()` + `metadataVersion` to SessionPatchSchema in the same line range upstream/main has the model/ modelReasoningEffort fields. My branch added the versioned `metadata` field at the END of the object, so git 3-way merge silently auto-merged both, producing an invalid object literal with duplicate `metadata` keys. By placing my `metadata` / `agentState` / `todos` / `teamState` insertions in the SAME line range codex inserts (between updatedAt and model), git now raises an explicit CONFLICT during the soup merge, which can be resolved correctly once and replayed by rerere. No behavior change on a clean upstream/main merge. Pure cosmetic; no test or runtime impact. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(sse): propagate TeamDelete clear through structured patch path Closes PR #897 Major review (HAPI Bot, 2026-06-13): TeamDelete events drove `applyTeamStateDelta` to return `null`, but the emit-site coalesced that to `undefined`. JSON serialization then dropped the key, the hub cache skipped its assignment branch (`patch.teamState !== undefined` was false), and the web client saw an empty patch and fell back to REST invalidation — exactly the storm path this PR was supposed to close. Sidebar / NotificationHub / dedup all served stale team state until the next full refresh. Fix in four coordinated places (wire ↔ cache contract): - shared/src/schemas.ts: `teamState: TeamStateSchema.nullable().optional()` so `null` is a valid wire shape meaning "cleared". Comment documents the discriminator contract for consumers. - hub/src/socket/handlers/cli/sessionHandlers.ts: drop the `?? undefined` coalesce so `null` survives JSON serialization. - hub/src/sync/sessionCache.ts (applySessionPatch): use `Object.prototype.hasOwnProperty.call(patch, 'teamState')` to discriminate "field absent" from "field is null", then map null → undefined to match the cached `Session.teamState` type. - web/src/hooks/useSSE.ts (patchSessionDetail): same hasOwnProperty discriminator + null → undefined mapping. Regression tests: - schemas.sessionPatch.test.ts: `{ teamState: null }` parses successfully (locks the wire contract). - sessionCache.applySessionPatch.test.ts: TeamDelete clears cached teamState; todos-only patch leaves teamState untouched (guards the hasOwnProperty branch against a regression back to `!== undefined`). Co-authored-by: Cursor <cursoragent@cursor.com> * fix(sse): version-gate metadata/agentState patches against cache regression Closes PR #897 follow-up Major review (HAPI Bot, 2026-06-16): the new structured SSE patch path unwraps versioned metadata/agentState fields without checking the cached metadataVersion/agentStateVersion. SSE reconnects + the existing per-query invalidation can leave a detail cache repopulated by a fresh REST refetch BEFORE a buffered older patch replays. Without the gate the older patch overwrites the newer cache, regressing resume / session-id / pending-requests state. Mirrors the hub-side CLI room handler contract (`incoming.version > currentVersion`, `web/src/hooks/useSSE.ts`): - `patchSessionDetail`: gate metadata/agentState assignment behind `isNewerVersionedPatch(patch.version, nextSession.<field>Version)`. The pre-patch version is captured by `{ ...previous.session }` so the comparison is against the cache-at-write-time. - `patchSessionSummary`: read the detail cache (via queryClient) for the canonical metadataVersion / agentStateVersion. Use `>=` (not `>`) because the callsite runs `patchSessionDetail` first — when detail accepts a newer patch the cache already holds the new version, so matching `>=` keeps summary aligned with detail's acceptance; when detail rejects, `>=` aligns summary with detail's rejection. - Exported `isNewerVersionedPatch(patchVersion, currentVersion)` as a pure helper so the rule is unit-testable in isolation. - Test: `useSSE.test.ts` pins the 4 cases (newer ✓ / older ✗ / same-version ✗ / first-write currentVersion=0 ✓). Hub-side `applySessionPatch` does NOT need the same gate: in-process events from `handleUpdateMetadata` / `handleUpdateState` are emitted only AFTER the optimistic-concurrency check at the store layer succeeds, and `syncEngine.handleRealtimeEvent` consumes them synchronously in order. The vulnerability is the SSE reconnect/replay window on the web client. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(sse): include updatedAt in structured patches + pendingRequests summary Closes PR #897 post-rebase bot review (HAPI Bot, 2026-06-18): Major — structured patches dropped session.updatedAt. TodoWrite, teamState, metadata, and agentState DB writes all touch sessions.updated_at, but the fast path forwarded only field deltas. Hub/web caches and session list ordering stayed stale until a full refresh. All four emit-sites in sessionHandlers now reload the stored row after a successful write and include updatedAt in the SSE patch payload (applySessionPatch already applies it via Math.max). Minor — agentState summary patches updated pendingRequestsCount/kinds but left pendingRequests stale, so SessionAttentionIndicator tooltips showed old request tools after an SSE patch. patchSessionSummary now uses computePendingRequestsCount + computePendingRequests alongside the existing kinds helper. Tests: sessionHandlers.test.ts asserts updatedAt on todos/metadata/agentState patches. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web,hub): apply serviceTier in structured session patch path Closes PR #897 bot Minor (2026-06-18): field-by-field patchSessionDetail stopped copying serviceTier after the spread refactor, so Codex Fast/ Standard could show stale tier until a full refetch. Mirror nullable hasOwnProperty handling in patchSessionDetail and hub applySessionPatch. Co-authored-by: Cursor <cursoragent@cursor.com> * test(hub): allow same-ms updatedAt on structured patch emit asserts Date.now() resolution makes create+update land on the same millisecond in unit tests; the store still touches updated_at. Use >= so CI is not flaky. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): refuse versioned summary SSE patches without detail version source When session detail is not cached, defaulting metadata/agentState versions to 0 let stale buffered patches overwrite a freshly refetched list and suppress list invalidation. Bail out so the list refetches instead. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): keep updatedAt monotonic when applying SSE session patches Stale versioned metadata/agentState replays can still carry an older updatedAt. Use Math.max on detail and summary paths so rejected replays cannot rewind list/detail clocks while patched=true suppresses invalidation. Co-authored-by: Cursor <cursoragent@cursor.com> * chore: retrigger Codex PR review after infra stream failure Prior pr-review run died on reconnect (stream closed before response.completed); no code findings. Empty commit to re-fire pull_request_target. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): compare all summary metadata fields in keep-alive skip isRenderIrrelevantPatch omitted path/machineId/flavor/worktree, so a same-ms metadata patch could be dropped while summaryPatched stayed true and list invalidation never repaired grouping/icon/path. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(sse): version-wrap todos/teamState patches for dual-SSE races Global + session EventSources can deliver out of order. Carry store todos_updated_at / team_state_updated_at as patch versions, gate web applies, and tighten keep-alive skip compares (metadata + request tool/kind). Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): put SSE version watermarks on SessionSummary Requiring a detail query to apply versioned list patches forced O(N) /sessions invalidation on every global SSE write. Gate against summary watermarks instead; skip no-op detail clones on duplicate deliveries. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(hub): ratchet todosUpdatedAt on rewind rebuild replaceSessionTodos was stamping the remaining TodoWrite's older createdAt, so a lagged pre-rewind structured SSE patch could resurrect deleted todos. Advance the watermark on force-replace instead. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): apply copilotAgentMode in structured detail SSE patches Field-by-field detail mapper dropped the new Copilot keep-alive field, so detailPatched suppressed invalidation and SessionChat kept a stale mode. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Debian <heavygee@oos-linux.in.lockhouse>
This commit is contained in:
HeavyGee
co-authored by
Cursor
Debian
Cursor
Debian
parent
3c83fe58c9
commit
8e34e7599b
@@ -84,6 +84,160 @@ describe('cli session handlers', () => {
|
||||
expect(webEvents).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('emits a structured todos patch when a TodoWrite message lands (closes second half of #884)', () => {
|
||||
const store = new Store(':memory:')
|
||||
const session = store.sessions.getOrCreateSession('todos-session', { path: '/tmp', host: 'h' }, null, 'default')
|
||||
const socket = new FakeSocket()
|
||||
const webEvents: SyncEvent[] = []
|
||||
|
||||
registerSessionHandlers(socket as unknown as CliSocketWithData, {
|
||||
store,
|
||||
resolveSessionAccess: () => ({ ok: true, value: session as StoredSession }),
|
||||
emitAccessError: () => {
|
||||
throw new Error('unexpected access error')
|
||||
},
|
||||
onWebappEvent: (event) => {
|
||||
webEvents.push(event)
|
||||
}
|
||||
})
|
||||
|
||||
socket.trigger('message', {
|
||||
sid: session.id,
|
||||
message: {
|
||||
role: 'agent',
|
||||
content: {
|
||||
type: 'output',
|
||||
data: {
|
||||
type: 'assistant',
|
||||
message: {
|
||||
content: [
|
||||
{
|
||||
type: 'tool_use',
|
||||
name: 'TodoWrite',
|
||||
input: {
|
||||
todos: [
|
||||
{ content: 'pending thing', status: 'pending' },
|
||||
{ content: 'done thing', status: 'completed' }
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
const sessionUpdated = webEvents.find((e) => e.type === 'session-updated')
|
||||
expect(sessionUpdated).toBeDefined()
|
||||
if (!sessionUpdated || sessionUpdated.type !== 'session-updated') return
|
||||
expect(sessionUpdated.data).toMatchObject({
|
||||
todos: {
|
||||
version: expect.any(Number),
|
||||
value: [
|
||||
{ content: 'pending thing', status: 'pending' },
|
||||
{ content: 'done thing', status: 'completed' }
|
||||
]
|
||||
}
|
||||
})
|
||||
expect(typeof (sessionUpdated.data as { updatedAt?: number }).updatedAt).toBe('number')
|
||||
expect((sessionUpdated.data as { updatedAt?: number }).updatedAt).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('emits a structured metadata patch on update-metadata RPC (closes second half of #884)', () => {
|
||||
const store = new Store(':memory:')
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
'metadata-patch-session',
|
||||
{ path: '/tmp/project', host: 'example' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
const socket = new FakeSocket()
|
||||
const webEvents: SyncEvent[] = []
|
||||
|
||||
registerSessionHandlers(socket as unknown as CliSocketWithData, {
|
||||
store,
|
||||
resolveSessionAccess: () => ({ ok: true, value: session as StoredSession }),
|
||||
emitAccessError: () => {
|
||||
throw new Error('unexpected access error')
|
||||
},
|
||||
onWebappEvent: (event) => {
|
||||
webEvents.push(event)
|
||||
}
|
||||
})
|
||||
|
||||
socket.trigger(
|
||||
'update-metadata',
|
||||
{
|
||||
sid: session.id,
|
||||
expectedVersion: session.metadataVersion,
|
||||
metadata: { lifecycleState: 'archived' }
|
||||
},
|
||||
() => {}
|
||||
)
|
||||
|
||||
const sessionUpdated = webEvents.find((e) => e.type === 'session-updated')
|
||||
expect(sessionUpdated).toBeDefined()
|
||||
if (!sessionUpdated || sessionUpdated.type !== 'session-updated') return
|
||||
const data = sessionUpdated.data as { metadata?: { version: number; value: Record<string, unknown> }; updatedAt?: number } | undefined
|
||||
expect(data?.metadata?.version).toBe(session.metadataVersion + 1)
|
||||
// Merged value: original path/host preserved + new lifecycleState applied.
|
||||
expect(data?.metadata?.value).toMatchObject({
|
||||
path: '/tmp/project',
|
||||
host: 'example',
|
||||
lifecycleState: 'archived'
|
||||
})
|
||||
expect(typeof data?.updatedAt).toBe('number')
|
||||
// Same-ms create+update is common in unit tests; store still touches updated_at.
|
||||
expect(data?.updatedAt).toBeGreaterThanOrEqual(session.updatedAt)
|
||||
})
|
||||
|
||||
it('emits a structured agentState patch on update-state RPC (closes second half of #884)', () => {
|
||||
const store = new Store(':memory:')
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
'agent-state-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
const socket = new FakeSocket()
|
||||
const webEvents: SyncEvent[] = []
|
||||
|
||||
registerSessionHandlers(socket as unknown as CliSocketWithData, {
|
||||
store,
|
||||
resolveSessionAccess: () => ({ ok: true, value: session as StoredSession }),
|
||||
emitAccessError: () => {
|
||||
throw new Error('unexpected access error')
|
||||
},
|
||||
onWebappEvent: (event) => {
|
||||
webEvents.push(event)
|
||||
}
|
||||
})
|
||||
|
||||
socket.trigger(
|
||||
'update-state',
|
||||
{
|
||||
sid: session.id,
|
||||
expectedVersion: session.agentStateVersion,
|
||||
agentState: { controlledByUser: true }
|
||||
},
|
||||
() => {}
|
||||
)
|
||||
|
||||
const sessionUpdated = webEvents.find((e) => e.type === 'session-updated')
|
||||
expect(sessionUpdated).toBeDefined()
|
||||
if (!sessionUpdated || sessionUpdated.type !== 'session-updated') return
|
||||
const data = sessionUpdated.data as {
|
||||
agentState?: { version: number; value: { controlledByUser?: boolean } }
|
||||
updatedAt?: number
|
||||
} | undefined
|
||||
expect(data?.agentState?.version).toBe(session.agentStateVersion + 1)
|
||||
expect(data?.agentState?.value).toMatchObject({ controlledByUser: true })
|
||||
expect(typeof data?.updatedAt).toBe('number')
|
||||
// Same-ms create+update is common in unit tests; store still touches updated_at.
|
||||
expect(data?.updatedAt).toBeGreaterThanOrEqual(session.updatedAt)
|
||||
})
|
||||
|
||||
it('update-metadata broadcasts the merged value, not the pre-merge payload', () => {
|
||||
const store = new Store(':memory:')
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import type { ClientToServerEvents } from '@hapi/protocol'
|
||||
import { z } from 'zod'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { CodexCollaborationMode, PermissionMode } from '@hapi/protocol/types'
|
||||
import type { CopilotAgentMode } from '@hapi/protocol'
|
||||
import type { AgentState, CodexCollaborationMode, Metadata, PermissionMode } from '@hapi/protocol/types'
|
||||
import { isRedundantGoalStatusEventContent } from '@hapi/protocol/messages'
|
||||
import type { Store, StoredSession } from '../../../store'
|
||||
import type { SyncEvent } from '../../../sync/syncEngine'
|
||||
@@ -142,7 +142,18 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
if (todos) {
|
||||
const updated = store.sessions.setSessionTodos(sid, todos, msg.createdAt, session.namespace)
|
||||
if (updated) {
|
||||
onWebappEvent?.({ type: 'session-updated', sessionId: sid })
|
||||
const stored = store.sessions.getSession(sid)
|
||||
onWebappEvent?.({
|
||||
type: 'session-updated',
|
||||
sessionId: sid,
|
||||
data: {
|
||||
todos: {
|
||||
version: stored?.todosUpdatedAt ?? msg.createdAt,
|
||||
value: todos
|
||||
},
|
||||
updatedAt: stored?.updatedAt ?? msg.createdAt
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -153,7 +164,21 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
const newTeamState = applyTeamStateDelta(existingTeamState ?? null, teamDelta)
|
||||
const updated = store.sessions.setSessionTeamState(sid, newTeamState, msg.createdAt, session.namespace)
|
||||
if (updated) {
|
||||
onWebappEvent?.({ type: 'session-updated', sessionId: sid })
|
||||
const stored = store.sessions.getSession(sid)
|
||||
// Versioned clear: value null = TeamDelete. Consumers gate on
|
||||
// version (store team_state_updated_at) so dual SSE cannot
|
||||
// resurrect a deleted team from a lagged older event.
|
||||
onWebappEvent?.({
|
||||
type: 'session-updated',
|
||||
sessionId: sid,
|
||||
data: {
|
||||
teamState: {
|
||||
version: stored?.teamStateUpdatedAt ?? msg.createdAt,
|
||||
value: newTeamState
|
||||
},
|
||||
updatedAt: stored?.updatedAt ?? msg.createdAt
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -223,6 +248,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
}
|
||||
|
||||
if (result.result === 'success') {
|
||||
const stored = store.sessions.getSession(sid)
|
||||
const update = {
|
||||
id: randomUUID(),
|
||||
seq: Date.now(),
|
||||
@@ -241,7 +267,19 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
}
|
||||
}
|
||||
socket.to(`session:${sid}`).emit('update', update)
|
||||
onWebappEvent?.({ type: 'session-updated', sessionId: sid })
|
||||
onWebappEvent?.({
|
||||
type: 'session-updated',
|
||||
sessionId: sid,
|
||||
// The unknown-cast here mirrors the schema's MetadataSchema.nullable()
|
||||
// shape: the store returns raw JSON, the wire schema parses it on
|
||||
// both ends. Keeping the broadcast shape identical to the socket.io
|
||||
// `update-session` body (line ~213) lets the same patch travel
|
||||
// through both fan-out channels without divergence.
|
||||
data: {
|
||||
metadata: { version: result.version, value: result.value as Metadata | null },
|
||||
updatedAt: stored?.updatedAt ?? Date.now()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -276,6 +314,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
}
|
||||
|
||||
if (result.result === 'success') {
|
||||
const stored = store.sessions.getSession(sid)
|
||||
const update = {
|
||||
id: randomUUID(),
|
||||
seq: Date.now(),
|
||||
@@ -288,7 +327,14 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
|
||||
}
|
||||
}
|
||||
socket.to(`session:${sid}`).emit('update', update)
|
||||
onWebappEvent?.({ type: 'session-updated', sessionId: sid })
|
||||
onWebappEvent?.({
|
||||
type: 'session-updated',
|
||||
sessionId: sid,
|
||||
data: {
|
||||
agentState: { version: result.version, value: agentState as AgentState | null },
|
||||
updatedAt: stored?.updatedAt ?? Date.now()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -67,10 +67,9 @@ export class SessionStore {
|
||||
replaceSessionTodos(
|
||||
id: string,
|
||||
todos: unknown,
|
||||
todosUpdatedAt: number | null,
|
||||
namespace: string
|
||||
): boolean {
|
||||
return replaceSessionTodos(this.db, id, todos, todosUpdatedAt, namespace)
|
||||
return replaceSessionTodos(this.db, id, todos, namespace)
|
||||
}
|
||||
|
||||
setSessionTeamState(id: string, teamState: unknown, updatedAt: number, namespace: string): boolean {
|
||||
|
||||
@@ -841,3 +841,82 @@ describe('updateSessionMetadata: protocol resume token preservation', () => {
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('replaceSessionTodos: watermark ratchet (PR #897 rewind race)', () => {
|
||||
it('advances todosUpdatedAt past the prior write even when rebuilding older content', () => {
|
||||
const store = makeStore()
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
'rewind-todos-watermark',
|
||||
{ path: '/tmp/project' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
const priorAt = 1_700_000_000_000
|
||||
const lateTodos = [{ content: 'late', status: 'pending', activeForm: 'doing late' }]
|
||||
const earlyTodos = [{ content: 'early', status: 'pending', activeForm: 'doing early' }]
|
||||
|
||||
expect(store.sessions.setSessionTodos(session.id, lateTodos, priorAt, 'default')).toBe(true)
|
||||
expect(store.sessions.getSession(session.id)?.todosUpdatedAt).toBe(priorAt)
|
||||
|
||||
// Rewind would otherwise stamp the remaining TodoWrite's older createdAt.
|
||||
expect(store.sessions.replaceSessionTodos(session.id, earlyTodos, 'default')).toBe(true)
|
||||
|
||||
const after = store.sessions.getSession(session.id)
|
||||
expect(after?.todosUpdatedAt).toBe(priorAt + 1)
|
||||
expect(after?.todos).toEqual(earlyTodos)
|
||||
|
||||
// A lagged pre-rewind structured patch using priorAt must lose the
|
||||
// store-side monotonic write too (defense in depth vs SSE gate).
|
||||
expect(store.sessions.setSessionTodos(session.id, lateTodos, priorAt, 'default')).toBe(false)
|
||||
expect(store.sessions.getSession(session.id)?.todos).toEqual(earlyTodos)
|
||||
|
||||
store.close()
|
||||
})
|
||||
|
||||
it('stamps Date.now() when replacing into a null watermark', () => {
|
||||
const store = makeStore()
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
'rewind-todos-null-watermark',
|
||||
{ path: '/tmp/project' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
const before = Date.now()
|
||||
expect(store.sessions.replaceSessionTodos(
|
||||
session.id,
|
||||
[{ content: 'only', status: 'pending', activeForm: 'doing' }],
|
||||
'default'
|
||||
)).toBe(true)
|
||||
const after = store.sessions.getSession(session.id)
|
||||
expect(after?.todosUpdatedAt).toBeGreaterThanOrEqual(before)
|
||||
expect(after?.todosUpdatedAt).toBeLessThanOrEqual(Date.now())
|
||||
|
||||
store.close()
|
||||
})
|
||||
|
||||
it('clears todos while still ratcheting the watermark', () => {
|
||||
const store = makeStore()
|
||||
const session = store.sessions.getOrCreateSession(
|
||||
'rewind-todos-clear',
|
||||
{ path: '/tmp/project' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
expect(store.sessions.setSessionTodos(
|
||||
session.id,
|
||||
[{ content: 'gone', status: 'pending', activeForm: 'going' }],
|
||||
50,
|
||||
'default'
|
||||
)).toBe(true)
|
||||
|
||||
expect(store.sessions.replaceSessionTodos(session.id, null, 'default')).toBe(true)
|
||||
const after = store.sessions.getSession(session.id)
|
||||
expect(after?.todos).toBeNull()
|
||||
expect(after?.todosUpdatedAt).toBe(51)
|
||||
|
||||
store.close()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -377,12 +377,22 @@ export function setSessionTodos(
|
||||
}
|
||||
}
|
||||
|
||||
/** Force-replace todos after rewind/fork (ignores monotonic timestamp guard). */
|
||||
/**
|
||||
* Force-replace todos after rewind/fork (ignores the normal
|
||||
* `todos_updated_at < candidate` guard so an older remaining TodoWrite can
|
||||
* land). The watermark itself MUST still advance: SSE clients gate structured
|
||||
* todos patches on `todosUpdatedAt`, and dual EventSources can deliver a
|
||||
* buffered pre-rewind patch after the post-rewind Session. Writing the
|
||||
* remaining message's older `createdAt` here would let that stale patch win
|
||||
* and resurrect deleted todos (PR #897 HAPI Bot 2026-08-03 Major).
|
||||
*
|
||||
* Ratchet: `null → now`, else `previous + 1`. Always strictly greater than any
|
||||
* prior write on this row, so lagged pre-rewind versions are rejected.
|
||||
*/
|
||||
export function replaceSessionTodos(
|
||||
db: Database,
|
||||
id: string,
|
||||
todos: unknown,
|
||||
todosUpdatedAt: number | null,
|
||||
namespace: string
|
||||
): boolean {
|
||||
try {
|
||||
@@ -391,16 +401,18 @@ export function replaceSessionTodos(
|
||||
const result = db.prepare(`
|
||||
UPDATE sessions
|
||||
SET todos = @todos,
|
||||
todos_updated_at = @todos_updated_at,
|
||||
updated_at = CASE WHEN updated_at > @updated_at THEN updated_at ELSE @updated_at END,
|
||||
todos_updated_at = CASE
|
||||
WHEN todos_updated_at IS NULL THEN @now
|
||||
ELSE todos_updated_at + 1
|
||||
END,
|
||||
updated_at = CASE WHEN updated_at > @now THEN updated_at ELSE @now END,
|
||||
seq = seq + 1
|
||||
WHERE id = @id
|
||||
AND namespace = @namespace
|
||||
`).run({
|
||||
id,
|
||||
todos: json,
|
||||
todos_updated_at: todosUpdatedAt,
|
||||
updated_at: now,
|
||||
now,
|
||||
namespace
|
||||
})
|
||||
return result.changes === 1
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
import { describe, expect, it } from 'bun:test'
|
||||
import type { SyncEvent } from '@hapi/protocol/types'
|
||||
import { Store } from '../store'
|
||||
import type { EventPublisher } from './eventPublisher'
|
||||
import { SessionCache } from './sessionCache'
|
||||
|
||||
function createPublisher(events: SyncEvent[]): EventPublisher {
|
||||
return {
|
||||
emit: (event: SyncEvent) => {
|
||||
events.push(event)
|
||||
}
|
||||
} as unknown as EventPublisher
|
||||
}
|
||||
|
||||
// Companion guard for syncEngine.handleRealtimeEvent's new "forward structured
|
||||
// patches without DB refresh" branch (closes the second half of #884). The
|
||||
// hub-side fast-path is only safe if applySessionPatch keeps the in-memory
|
||||
// cache consistent with what just landed in the DB — otherwise subsequent
|
||||
// callers like NotificationHub.getSession would see stale data and the
|
||||
// cache-vs-DB divergence would manifest as ghost notifications, stale
|
||||
// pendingRequestsCount, or wrong todos progress in the session list.
|
||||
describe('SessionCache.applySessionPatch', () => {
|
||||
it('applies a todos patch in place when the session is cached', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'todos-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
const todos = [
|
||||
{ content: 'one', status: 'pending' as const, priority: 'medium' as const, id: '1' }
|
||||
]
|
||||
const applied = cache.applySessionPatch(created.id, {
|
||||
todos: { version: 100, value: todos }
|
||||
})
|
||||
|
||||
expect(applied).toBe(true)
|
||||
expect(cache.getSession(created.id)?.todos).toEqual(todos)
|
||||
expect(cache.getSession(created.id)?.todosUpdatedAt).toBe(100)
|
||||
})
|
||||
|
||||
it('applies a versioned metadata patch by unwrapping value + version', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'meta-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
const nextVersion = created.metadataVersion + 1
|
||||
const applied = cache.applySessionPatch(created.id, {
|
||||
metadata: {
|
||||
version: nextVersion,
|
||||
value: { path: '/tmp', host: 'h', lifecycleState: 'archived' }
|
||||
}
|
||||
})
|
||||
|
||||
expect(applied).toBe(true)
|
||||
const after = cache.getSession(created.id)
|
||||
expect(after?.metadata?.lifecycleState).toBe('archived')
|
||||
expect(after?.metadataVersion).toBe(nextVersion)
|
||||
})
|
||||
|
||||
it('applies a versioned agentState patch with null value', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'agent-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
{ controlledByUser: true },
|
||||
'default'
|
||||
)
|
||||
expect(created.agentState).not.toBeNull()
|
||||
|
||||
const nextVersion = created.agentStateVersion + 1
|
||||
const applied = cache.applySessionPatch(created.id, {
|
||||
agentState: { version: nextVersion, value: null }
|
||||
})
|
||||
|
||||
expect(applied).toBe(true)
|
||||
const after = cache.getSession(created.id)
|
||||
expect(after?.agentState).toBeNull()
|
||||
expect(after?.agentStateVersion).toBe(nextVersion)
|
||||
})
|
||||
|
||||
it('returns false (caller falls back to refresh) when the session is not cached', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const applied = cache.applySessionPatch('does-not-exist', { todos: { version: 1, value: [] } })
|
||||
expect(applied).toBe(false)
|
||||
})
|
||||
|
||||
it('returns false when patch data fails SessionPatchSchema', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'bad-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
// Bogus shape: { metadata: { value: ... } } is missing the required version.
|
||||
const applied = cache.applySessionPatch(created.id, {
|
||||
metadata: { value: { path: '/x', host: 'y' } }
|
||||
})
|
||||
expect(applied).toBe(false)
|
||||
})
|
||||
|
||||
it('refuses an empty patch ({}) so the caller falls back to refreshSession', () => {
|
||||
// Web-side getSessionPatch rejects empty payloads (Object.keys length 0)
|
||||
// and would fall through to REST invalidation — exactly the storm we
|
||||
// are closing. The empty-patch guard keeps the syncEngine on the safe
|
||||
// legacy refresh path for these no-op events.
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'empty-patch-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
expect(cache.applySessionPatch(created.id, {})).toBe(false)
|
||||
})
|
||||
|
||||
it('clears cached teamState when a null teamState patch lands (TeamDelete)', () => {
|
||||
// PR #897 review (HAPI Bot, 2026-06-13 Major): TeamDelete events
|
||||
// drive applyTeamStateDelta to return null; the emit-site sends
|
||||
// { teamState: null } as the explicit clear signal. Without
|
||||
// hasOwnProperty-discrimination, `if (patch.teamState !== undefined)`
|
||||
// skipped the clear path and left the hub cache holding stale
|
||||
// pre-delete TeamState — sidebar / NotificationHub / dedup all
|
||||
// would serve stale data until the next full refresh.
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'teamstate-clear-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
// Seed cached teamState (the pre-delete state).
|
||||
const seedApplied = cache.applySessionPatch(created.id, {
|
||||
teamState: { version: 10, value: { teamName: 'crew', members: [{ name: 'a' }] } }
|
||||
})
|
||||
expect(seedApplied).toBe(true)
|
||||
expect(cache.getSession(created.id)?.teamState?.teamName).toBe('crew')
|
||||
|
||||
// TeamDelete: null teamState value must clear the cache.
|
||||
const cleared = cache.applySessionPatch(created.id, { teamState: { version: 11, value: null } })
|
||||
expect(cleared).toBe(true)
|
||||
expect(cache.getSession(created.id)?.teamState).toBeUndefined()
|
||||
})
|
||||
|
||||
it('leaves teamState untouched when the patch does not carry the key', () => {
|
||||
// Guard the hasOwnProperty discriminator against a refactor back to
|
||||
// `if (patch.teamState !== undefined)` — a todos-only patch must
|
||||
// NOT clear teamState, which a naive `?? undefined` assignment on
|
||||
// the unconditional branch would do.
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'teamstate-untouched-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
cache.applySessionPatch(created.id, {
|
||||
teamState: { version: 10, value: { teamName: 'crew', members: [{ name: 'a' }] } }
|
||||
})
|
||||
expect(cache.getSession(created.id)?.teamState?.teamName).toBe('crew')
|
||||
|
||||
const todosOnly = cache.applySessionPatch(created.id, {
|
||||
todos: {
|
||||
version: 20,
|
||||
value: [{ content: 'one', status: 'pending' as const, priority: 'medium' as const, id: '1' }]
|
||||
}
|
||||
})
|
||||
expect(todosOnly).toBe(true)
|
||||
expect(cache.getSession(created.id)?.teamState?.teamName).toBe('crew')
|
||||
})
|
||||
|
||||
it('refuses cross-namespace patches even if the session exists', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
const cache = new SessionCache(store, createPublisher(events))
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'ns-guard-session',
|
||||
{ path: '/tmp', host: 'h' },
|
||||
null,
|
||||
'tenant-a'
|
||||
)
|
||||
|
||||
const applied = cache.applySessionPatch(created.id, { todos: { version: 1, value: [] } }, 'tenant-b')
|
||||
expect(applied).toBe(false)
|
||||
})
|
||||
})
|
||||
@@ -1,4 +1,4 @@
|
||||
import { AgentStateSchema, MetadataSchema, TeamStateSchema } from '@hapi/protocol/schemas'
|
||||
import { AgentStateSchema, MetadataSchema, SessionPatchSchema, TeamStateSchema } from '@hapi/protocol/schemas'
|
||||
import type { CodexCollaborationMode, CopilotAgentMode, PermissionMode, Session, SessionPatch } from '@hapi/protocol/types'
|
||||
import type { Store } from '../store'
|
||||
import { clampAliveTime } from './aliveTime'
|
||||
@@ -94,7 +94,9 @@ export class SessionCache {
|
||||
/**
|
||||
* After fork hydrate / rewind truncate, re-scan the transcript for the
|
||||
* latest TodoWrite (or clear todos). Bypasses the one-shot backfill flag
|
||||
* and the monotonic todosUpdatedAt guard.
|
||||
* and the normal `setSessionTodos` monotonic guard. Watermark still
|
||||
* ratchets inside `replaceSessionTodos` — do not pass the remaining
|
||||
* message's older `createdAt` as the SSE version.
|
||||
*/
|
||||
rebuildTodosFromTranscript(sessionId: string): void {
|
||||
const stored = this.store.sessions.getSession(sessionId)
|
||||
@@ -102,20 +104,19 @@ export class SessionCache {
|
||||
|
||||
this.todoBackfillAttemptedSessionIds.delete(sessionId)
|
||||
const messages = this.store.messages.getAllMessages(sessionId)
|
||||
let found: { todos: unknown; at: number } | null = null
|
||||
let foundTodos: unknown | null = null
|
||||
for (let i = messages.length - 1; i >= 0; i -= 1) {
|
||||
const message = messages[i]
|
||||
if (!message) continue
|
||||
const todos = extractTodoWriteTodosFromMessageContent(message.content)
|
||||
if (todos) {
|
||||
found = { todos, at: message.createdAt }
|
||||
foundTodos = todos
|
||||
break
|
||||
}
|
||||
}
|
||||
this.store.sessions.replaceSessionTodos(
|
||||
sessionId,
|
||||
found?.todos ?? null,
|
||||
found?.at ?? null,
|
||||
foundTodos,
|
||||
stored.namespace
|
||||
)
|
||||
this.todoBackfillAttemptedSessionIds.add(sessionId)
|
||||
@@ -197,6 +198,8 @@ export class SessionCache {
|
||||
backgroundTaskCount: existing?.backgroundTaskCount ?? 0,
|
||||
todos,
|
||||
teamState,
|
||||
todosUpdatedAt: stored.todosUpdatedAt ?? 0,
|
||||
teamStateUpdatedAt: stored.teamStateUpdatedAt ?? 0,
|
||||
model: stored.model,
|
||||
modelReasoningEffort: stored.modelReasoningEffort,
|
||||
effort: stored.effort,
|
||||
@@ -244,6 +247,86 @@ export class SessionCache {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a structured patch to the cached Session in place.
|
||||
*
|
||||
* Returns `true` if the patch parsed, carried at least one field, and a
|
||||
* Session was present to update. Returns `false` when:
|
||||
* - the patch data fails SessionPatchSchema (caller falls back to
|
||||
* refreshSession),
|
||||
* - the patch is the empty object `{}` — the web client's
|
||||
* `getSessionPatch` rejects empty payloads and would fall through to
|
||||
* REST invalidation, so we route empty events through the legacy
|
||||
* refresh path instead (caller falls back to refreshSession),
|
||||
* - the session is not in the cache (caller falls back to refreshSession
|
||||
* so the DB read can hydrate it),
|
||||
* - the patch's namespace hint disagrees with the cached session
|
||||
* namespace (cross-namespace event, caller skips).
|
||||
*
|
||||
* Companion to syncEngine.handleRealtimeEvent. Closes the second half of
|
||||
* #884 by giving the four no-data emit-sites in cli/sessionHandlers.ts a
|
||||
* way to propagate their delta straight through to SSE without a DB
|
||||
* re-read or full-Session broadcast.
|
||||
*/
|
||||
applySessionPatch(sessionId: string, data: unknown, namespace?: string): boolean {
|
||||
const parsed = SessionPatchSchema.safeParse(data)
|
||||
if (!parsed.success) {
|
||||
return false
|
||||
}
|
||||
|
||||
// Empty patch ({}): forward would hit the web-side fallback that
|
||||
// triggers a REST refetch. Let the caller fall back to refreshSession
|
||||
// so the existing full-Session broadcast path keeps the cache
|
||||
// coherent.
|
||||
if (Object.keys(parsed.data).length === 0) {
|
||||
return false
|
||||
}
|
||||
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (namespace && session.namespace !== namespace) {
|
||||
return false
|
||||
}
|
||||
|
||||
const patch = parsed.data
|
||||
|
||||
if (patch.active !== undefined) session.active = patch.active
|
||||
if (patch.thinking !== undefined) session.thinking = patch.thinking
|
||||
if (patch.activeAt !== undefined) session.activeAt = patch.activeAt
|
||||
if (patch.updatedAt !== undefined) session.updatedAt = Math.max(session.updatedAt, patch.updatedAt)
|
||||
if (patch.model !== undefined) session.model = patch.model
|
||||
if (patch.modelReasoningEffort !== undefined) session.modelReasoningEffort = patch.modelReasoningEffort
|
||||
if (patch.effort !== undefined) session.effort = patch.effort
|
||||
if (Object.prototype.hasOwnProperty.call(patch, 'serviceTier')) {
|
||||
session.serviceTier = patch.serviceTier ?? null
|
||||
}
|
||||
if (patch.permissionMode !== undefined) session.permissionMode = patch.permissionMode
|
||||
if (patch.collaborationMode !== undefined) session.collaborationMode = patch.collaborationMode
|
||||
if (patch.backgroundTaskCount !== undefined) session.backgroundTaskCount = patch.backgroundTaskCount
|
||||
if (patch.todos !== undefined) {
|
||||
session.todos = patch.todos.value
|
||||
session.todosUpdatedAt = patch.todos.version
|
||||
}
|
||||
// Versioned teamState: key present + value null = TeamDelete clear.
|
||||
if (patch.teamState !== undefined) {
|
||||
session.teamState = patch.teamState.value ?? undefined
|
||||
session.teamStateUpdatedAt = patch.teamState.version
|
||||
}
|
||||
if (patch.metadata !== undefined) {
|
||||
session.metadata = patch.metadata.value
|
||||
session.metadataVersion = patch.metadata.version
|
||||
}
|
||||
if (patch.agentState !== undefined) {
|
||||
session.agentState = patch.agentState.value
|
||||
session.agentStateVersion = patch.agentState.version
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
handleSessionAlive(payload: {
|
||||
sid: string
|
||||
time: number
|
||||
|
||||
@@ -386,12 +386,51 @@ export class SyncEngine {
|
||||
|
||||
handleRealtimeEvent(event: SyncEvent): void {
|
||||
if (event.type === 'session-updated' && event.sessionId) {
|
||||
// Snapshot agent session IDs before refresh — safe because JS is single-threaded
|
||||
// and refreshSession replaces the Map entry with a new object.
|
||||
// Closes the second half of #884: when a CLI handler emits a
|
||||
// structured patch (todos / teamState / metadata / agentState),
|
||||
// apply it in place and forward the patch as-is. This skips both
|
||||
// the DB re-read AND the full-Session SSE broadcast that the
|
||||
// legacy no-data path went through. Web clients hit
|
||||
// getSessionPatch's truthy path and patch the cache instead of
|
||||
// falling through to the per-session REST invalidation that drove
|
||||
// the refetch storm.
|
||||
//
|
||||
// `applySessionPatch` MUTATES the cached Session in place (it
|
||||
// reassigns `session.metadata = patch.metadata.value`), so we
|
||||
// MUST snapshot the metadata reference BEFORE calling it.
|
||||
// Reading `before?.metadata` after the mutation would see the
|
||||
// new value and `hasSameAgentSessionIds` would always return
|
||||
// true — breaking the dedup-on-metadata-id-change trigger that
|
||||
// the legacy `refreshSession` path got for free (refresh
|
||||
// REPLACES the cache entry, leaving the old object reference
|
||||
// intact for the caller). Use the snapshot for BOTH branches
|
||||
// so the comparison contract is identical.
|
||||
const before = this.sessionCache.getSession(event.sessionId)
|
||||
const beforeMetadata = before?.metadata ?? null
|
||||
const patchApplied = event.data
|
||||
? this.sessionCache.applySessionPatch(event.sessionId, event.data, event.namespace)
|
||||
: false
|
||||
|
||||
if (patchApplied) {
|
||||
this.eventPublisher.emit(event)
|
||||
const after = this.sessionCache.getSession(event.sessionId)
|
||||
if (after?.metadata && !this.hasSameAgentSessionIds(beforeMetadata, after.metadata)) {
|
||||
if (!this.canRunCursorDedup(after)) {
|
||||
return
|
||||
}
|
||||
void this.sessionCache.deduplicateByAgentSessionId(event.sessionId).catch(() => {
|
||||
// best-effort: dedup failure is harmless, web-side safety net hides remaining duplicates
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// No-data event (or data we can't apply directly, e.g. full
|
||||
// Session payload from a different emitter): fall back to the
|
||||
// legacy refresh-from-DB-and-broadcast path.
|
||||
this.sessionCache.refreshSession(event.sessionId)
|
||||
const after = this.sessionCache.getSession(event.sessionId)
|
||||
if (after?.metadata && !this.hasSameAgentSessionIds(before?.metadata ?? null, after.metadata)) {
|
||||
if (after?.metadata && !this.hasSameAgentSessionIds(beforeMetadata, after.metadata)) {
|
||||
if (!this.canRunCursorDedup(after)) {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
import { describe, expect, it } from 'bun:test'
|
||||
import type { SessionPatch, SyncEvent } from '@hapi/protocol/types'
|
||||
import { RpcRegistry } from '../socket/rpcRegistry'
|
||||
import { Store } from '../store'
|
||||
import { SessionCache } from './sessionCache'
|
||||
import { SyncEngine } from './syncEngine'
|
||||
|
||||
/**
|
||||
* Regression guard for the in-place-mutation interaction between
|
||||
* `SessionCache.applySessionPatch` and `SyncEngine.handleRealtimeEvent`'s
|
||||
* dedup trigger.
|
||||
*
|
||||
* `applySessionPatch` MUTATES the cached Session in place (it reassigns
|
||||
* `session.metadata = patch.metadata.value`). The dedup-on-metadata-change
|
||||
* branch in `handleRealtimeEvent` needs to compare BEFORE and AFTER agent
|
||||
* session IDs to decide whether to trigger `deduplicateByAgentSessionId`.
|
||||
* Without snapshotting the metadata reference before the mutation, `before`
|
||||
* and `after` resolve to the SAME object reference, `hasSameAgentSessionIds`
|
||||
* is always true, and dedup silently never fires on the fast path. The
|
||||
* legacy `refreshSession` path got dedup for free because it REPLACED the
|
||||
* cache entry with a new Session object, leaving the pre-refresh reference
|
||||
* intact for the comparator.
|
||||
*
|
||||
* This was a real regression introduced by the refetch-storm fix
|
||||
* (#884 second half) — the fast-path replaced the refresh-then-broadcast
|
||||
* path for the four CLI emit-sites including the `update-metadata` RPC
|
||||
* handler, which is exactly where a Cursor session id change would land
|
||||
* (CLI resume re-stamps `metadata.cursorSessionId`).
|
||||
*/
|
||||
describe('SyncEngine.handleRealtimeEvent dedup-on-metadata-change', () => {
|
||||
function makeEngine(): { engine: SyncEngine; cache: SessionCache; dedupCalls: string[] } {
|
||||
const store = new Store(':memory:')
|
||||
const engine = new SyncEngine(
|
||||
store,
|
||||
{} as never,
|
||||
new RpcRegistry(),
|
||||
{ broadcast() {} } as never
|
||||
)
|
||||
const cache = (engine as unknown as { sessionCache: SessionCache }).sessionCache
|
||||
const dedupCalls: string[] = []
|
||||
const originalDedup = cache.deduplicateByAgentSessionId.bind(cache)
|
||||
cache.deduplicateByAgentSessionId = async (sessionId: string) => {
|
||||
dedupCalls.push(sessionId)
|
||||
// Do not actually run dedup; we only need to assert the trigger
|
||||
// fired. Running the real merge logic would require additional
|
||||
// store fixtures and is covered by sessionCache tests.
|
||||
void originalDedup
|
||||
}
|
||||
return { engine, cache, dedupCalls }
|
||||
}
|
||||
|
||||
it('triggers dedup when a structured metadata patch changes the agent session id', () => {
|
||||
const { engine, cache, dedupCalls } = makeEngine()
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'cursor-session-fast-path',
|
||||
{ path: '/tmp', host: 'h', flavor: 'cursor', cursorSessionId: 'cursor-old' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
const patch: SessionPatch = {
|
||||
metadata: {
|
||||
version: created.metadataVersion + 1,
|
||||
value: {
|
||||
path: '/tmp',
|
||||
host: 'h',
|
||||
flavor: 'cursor',
|
||||
cursorSessionId: 'cursor-new'
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const event: SyncEvent = {
|
||||
type: 'session-updated',
|
||||
sessionId: created.id,
|
||||
data: patch
|
||||
}
|
||||
engine.handleRealtimeEvent(event)
|
||||
|
||||
expect(dedupCalls).toEqual([created.id])
|
||||
expect(cache.getSession(created.id)?.metadata?.cursorSessionId).toBe('cursor-new')
|
||||
})
|
||||
|
||||
it('does not trigger dedup when the patch leaves agent session ids unchanged', () => {
|
||||
const { engine, cache, dedupCalls } = makeEngine()
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'cursor-session-fast-path-noop',
|
||||
{ path: '/tmp', host: 'h', flavor: 'cursor', cursorSessionId: 'cursor-stable' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
// A todos patch carries no metadata and must NOT trigger dedup.
|
||||
const event: SyncEvent = {
|
||||
type: 'session-updated',
|
||||
sessionId: created.id,
|
||||
data: { todos: { version: 1, value: [] } } satisfies SessionPatch
|
||||
}
|
||||
engine.handleRealtimeEvent(event)
|
||||
|
||||
expect(dedupCalls).toEqual([])
|
||||
})
|
||||
|
||||
it('triggers dedup on the legacy refresh fallback path (no patch data)', () => {
|
||||
// Tighten the contract for the legacy refresh-from-DB branch:
|
||||
// a no-`data` session-updated event still needs to fire dedup
|
||||
// when the DB read surfaces a new agent session id. The shared
|
||||
// `beforeMetadata` snapshot covers both branches, this guards
|
||||
// against a refactor breaking the legacy path.
|
||||
const { engine, cache, dedupCalls } = makeEngine()
|
||||
|
||||
const created = cache.getOrCreateSession(
|
||||
'cursor-session-legacy-path',
|
||||
{ path: '/tmp', host: 'h', flavor: 'cursor', cursorSessionId: 'cursor-legacy-old' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
|
||||
// Persist a metadata change to the DB without going through the
|
||||
// cache mutation path, so refreshSession's DB read picks up the
|
||||
// new value when handleRealtimeEvent fires.
|
||||
const updateResult = (engine as unknown as { store: Store }).store.sessions.updateSessionMetadata(
|
||||
created.id,
|
||||
{ path: '/tmp', host: 'h', flavor: 'cursor', cursorSessionId: 'cursor-legacy-new' },
|
||||
created.metadataVersion,
|
||||
'default',
|
||||
{ touchUpdatedAt: false }
|
||||
)
|
||||
expect(updateResult.result).toBe('success')
|
||||
|
||||
const event: SyncEvent = {
|
||||
type: 'session-updated',
|
||||
sessionId: created.id
|
||||
}
|
||||
engine.handleRealtimeEvent(event)
|
||||
|
||||
expect(dedupCalls).toEqual([created.id])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,104 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { SessionPatchSchema } from './schemas';
|
||||
|
||||
// Guard the contract for the second-half-of-#884 fix. The web client routes
|
||||
// `session-updated` events to the structured-patch path only when the event's
|
||||
// `data` parses as a SessionPatch — these tests pin the schema shape so a
|
||||
// future refactor that drops `.strict()`, the versioned (version, value)
|
||||
// wrappers, or any of the new optional fields breaks the build instead of
|
||||
// silently re-introducing the refetch storm.
|
||||
describe('SessionPatchSchema structured patches (closes #884 follow-up)', () => {
|
||||
it('parses a versioned todos patch', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
todos: {
|
||||
version: 10,
|
||||
value: [{ content: 'thing', status: 'pending' }]
|
||||
}
|
||||
});
|
||||
expect(parsed.success).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects bare todos without a version (dual-SSE watermark required)', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
todos: [{ content: 'thing', status: 'pending' }]
|
||||
});
|
||||
expect(parsed.success).toBe(false);
|
||||
});
|
||||
|
||||
it('parses a versioned teamState patch', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
teamState: {
|
||||
version: 11,
|
||||
value: {
|
||||
teamName: 'crew',
|
||||
members: [{ name: 'one' }]
|
||||
}
|
||||
}
|
||||
});
|
||||
expect(parsed.success).toBe(true);
|
||||
});
|
||||
|
||||
it('parses a teamState clear patch (null value = TeamDelete)', () => {
|
||||
// PR #897 review (HAPI Bot, 2026-06-13 Major + 2026-07-30 Major):
|
||||
// teamState travels as { version, value }; value null clears the
|
||||
// cached row. Version gates dual-SSE reordering.
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
teamState: { version: 12, value: null }
|
||||
});
|
||||
expect(parsed.success).toBe(true);
|
||||
if (parsed.success) {
|
||||
expect(parsed.data.teamState?.value).toBeNull();
|
||||
}
|
||||
});
|
||||
|
||||
it('parses a versioned metadata patch', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
metadata: {
|
||||
version: 7,
|
||||
value: { path: '/tmp', host: 'h' }
|
||||
}
|
||||
});
|
||||
expect(parsed.success).toBe(true);
|
||||
});
|
||||
|
||||
it('parses a versioned agentState patch with null value', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
agentState: { version: 3, value: null }
|
||||
});
|
||||
expect(parsed.success).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects metadata without a version (must stay versioned for cache safety)', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
metadata: { value: { path: '/tmp', host: 'h' } }
|
||||
});
|
||||
expect(parsed.success).toBe(false);
|
||||
});
|
||||
|
||||
it('stays strict and rejects unknown keys (catches silent .strict() removal)', () => {
|
||||
const parsed = SessionPatchSchema.safeParse({
|
||||
todos: { version: 1, value: [] },
|
||||
notARealField: true
|
||||
});
|
||||
expect(parsed.success).toBe(false);
|
||||
});
|
||||
|
||||
it('rejects a full Session payload (full-session SSE goes through isSessionRecord instead)', () => {
|
||||
const fullSession = {
|
||||
id: 's1',
|
||||
namespace: 'default',
|
||||
seq: 1,
|
||||
createdAt: 1,
|
||||
updatedAt: 1,
|
||||
active: true,
|
||||
activeAt: 1,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentState: null,
|
||||
agentStateVersion: 0,
|
||||
thinking: false,
|
||||
thinkingAt: 0
|
||||
};
|
||||
expect(SessionPatchSchema.safeParse(fullSession).success).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -295,6 +295,12 @@ export const SessionSchema = z.object({
|
||||
backgroundTaskCount: z.number().optional(),
|
||||
todos: TodosSchema.optional(),
|
||||
teamState: TeamStateSchema.optional(),
|
||||
// Watermarks for structured SSE patches (PR #897). Dual EventSource
|
||||
// connections can deliver todos/teamState out of order; caches reject
|
||||
// stale patches with version <= these fields. Optional so older
|
||||
// full-session payloads and hand-built Session literals stay valid.
|
||||
todosUpdatedAt: z.number().optional(),
|
||||
teamStateUpdatedAt: z.number().optional(),
|
||||
model: z.string().nullable().optional().default(null),
|
||||
modelReasoningEffort: z.string().nullable().optional().default(null),
|
||||
effort: z.string().nullable().optional().default(null),
|
||||
@@ -306,11 +312,52 @@ export const SessionSchema = z.object({
|
||||
|
||||
export type Session = z.infer<typeof SessionSchema>
|
||||
|
||||
// Versioned wrappers mirror the socket.io `update-session` broadcast shape so
|
||||
// metadata/agentState always travel as an atomic (version, value) pair — the
|
||||
// version is the only safe way for downstream caches to reject stale patches.
|
||||
const VersionedMetadataPatchSchema = z.object({
|
||||
version: z.number(),
|
||||
value: MetadataSchema.nullable()
|
||||
})
|
||||
|
||||
const VersionedAgentStatePatchSchema = z.object({
|
||||
version: z.number(),
|
||||
value: AgentStateSchema.nullable()
|
||||
})
|
||||
|
||||
// Same dual-SSE race as metadata/agentState: global + session EventSources
|
||||
// have no shared order. Version = store `todos_updated_at` /
|
||||
// `team_state_updated_at`. Normal TodoWrite / team writes stamp message
|
||||
// `createdAt`; rewind/fork `replaceSessionTodos` ratchets the watermark
|
||||
// so a lagged pre-rewind patch cannot resurrect deleted todos.
|
||||
const VersionedTodosPatchSchema = z.object({
|
||||
version: z.number(),
|
||||
value: TodosSchema
|
||||
})
|
||||
|
||||
const VersionedTeamStatePatchSchema = z.object({
|
||||
version: z.number(),
|
||||
// `null` value = TeamDelete clear. Discriminator remains "key present".
|
||||
value: TeamStateSchema.nullable()
|
||||
})
|
||||
|
||||
export const SessionPatchSchema = z.object({
|
||||
active: z.boolean().optional(),
|
||||
thinking: z.boolean().optional(),
|
||||
activeAt: z.number().optional(),
|
||||
updatedAt: z.number().optional(),
|
||||
// Structured-patch fields for the second half of #884. Letting the four
|
||||
// hub-side emit-sites in cli/sessionHandlers.ts (todos, teamState,
|
||||
// metadata, agentState writes) carry their delta means the web client's
|
||||
// SSE handler can patch the cache in place instead of falling through to
|
||||
// the invalidation fallback that triggers per-session REST refetches.
|
||||
// Versioned wrappers for metadata/agentState mirror the socket.io
|
||||
// `update-session` broadcast shape — the version field is the only safe
|
||||
// way for downstream caches to reject stale patches.
|
||||
metadata: VersionedMetadataPatchSchema.optional(),
|
||||
agentState: VersionedAgentStatePatchSchema.optional(),
|
||||
todos: VersionedTodosPatchSchema.optional(),
|
||||
teamState: VersionedTeamStatePatchSchema.optional(),
|
||||
model: z.string().nullable().optional(),
|
||||
modelReasoningEffort: z.string().nullable().optional(),
|
||||
effort: z.string().nullable().optional(),
|
||||
|
||||
@@ -2,9 +2,13 @@ import { describe, expect, it } from 'bun:test'
|
||||
import type { Session } from './schemas'
|
||||
import {
|
||||
PENDING_REQUEST_SUMMARY_CAP,
|
||||
computePendingRequestKinds,
|
||||
computePendingRequestsCount,
|
||||
computeTodoProgress,
|
||||
getPendingRequestKinds,
|
||||
getPendingRequests,
|
||||
toSessionSummary
|
||||
toSessionSummary,
|
||||
toSessionSummaryMetadata
|
||||
} from './sessionSummary'
|
||||
|
||||
function makeSession(overrides: Partial<Session> = {}): Session {
|
||||
@@ -259,3 +263,52 @@ describe('getPendingRequestKinds', () => {
|
||||
expect(kinds).toEqual(['permission', 'input'])
|
||||
})
|
||||
})
|
||||
|
||||
// The SSE patch path (useSSE.ts patchSessionSummary) calls these directly
|
||||
// against the patch payload — no full Session needed — to keep the session
|
||||
// list summary consistent with structured todos/teamState/metadata/agentState
|
||||
// patches landing for the second half of #884.
|
||||
describe('summary derivation helpers', () => {
|
||||
it('computeTodoProgress returns null for empty / undefined todos', () => {
|
||||
expect(computeTodoProgress(undefined)).toBeNull()
|
||||
expect(computeTodoProgress([])).toBeNull()
|
||||
})
|
||||
|
||||
it('computeTodoProgress counts completed vs total', () => {
|
||||
const progress = computeTodoProgress([
|
||||
{ content: 'a', status: 'pending', priority: 'medium', id: '1' },
|
||||
{ content: 'b', status: 'completed', priority: 'medium', id: '2' },
|
||||
{ content: 'c', status: 'completed', priority: 'medium', id: '3' }
|
||||
])
|
||||
expect(progress).toEqual({ completed: 2, total: 3 })
|
||||
})
|
||||
|
||||
it('computePendingRequestKinds works on a bare AgentState without a Session', () => {
|
||||
const kinds = computePendingRequestKinds({
|
||||
requests: {
|
||||
req1: { tool: 'Bash', arguments: {} },
|
||||
req2: { tool: 'AskUserQuestion', arguments: {} }
|
||||
}
|
||||
})
|
||||
expect(kinds).toEqual(['permission', 'input'])
|
||||
})
|
||||
|
||||
it('computePendingRequestsCount handles null agentState', () => {
|
||||
expect(computePendingRequestsCount(null)).toBe(0)
|
||||
expect(computePendingRequestsCount(undefined)).toBe(0)
|
||||
})
|
||||
|
||||
it('toSessionSummaryMetadata returns null for null metadata', () => {
|
||||
expect(toSessionSummaryMetadata(null)).toBeNull()
|
||||
expect(toSessionSummaryMetadata(undefined)).toBeNull()
|
||||
})
|
||||
|
||||
it('toSessionSummaryMetadata derives agentSessionId from the first non-null source id', () => {
|
||||
const summary = toSessionSummaryMetadata({
|
||||
path: '/p',
|
||||
host: 'h',
|
||||
cursorSessionId: 'cursor-xyz'
|
||||
})
|
||||
expect(summary?.agentSessionId).toBe('cursor-xyz')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Metadata, Session, WorktreeMetadata } from './schemas'
|
||||
import type { AgentState, Metadata, Session, TodoItem, WorktreeMetadata } from './schemas'
|
||||
import { isKnownFlavor } from './flavors'
|
||||
import type { AgentFlavor } from './modes'
|
||||
|
||||
@@ -22,8 +22,9 @@ export type PendingRequest = {
|
||||
id: string
|
||||
kind: PendingRequestKind
|
||||
tool: string
|
||||
/** Epoch ms when the request was raised; falls back to `session.updatedAt`
|
||||
* for older requests that were stored without `createdAt`. */
|
||||
/** Epoch ms when the request was raised; falls back to the caller-supplied
|
||||
* `fallbackSince` (typically `session.updatedAt`) for older requests
|
||||
* stored without `createdAt`. */
|
||||
since: number
|
||||
}
|
||||
|
||||
@@ -49,6 +50,12 @@ export type SessionSummary = {
|
||||
activeAt: number
|
||||
updatedAt: number
|
||||
metadata: SessionSummaryMetadata | null
|
||||
/** Watermarks for structured SSE patches (PR #897). List cache must gate
|
||||
* without requiring a detail query — otherwise global SSE forces O(N)
|
||||
* /sessions invalidation for every versioned write. */
|
||||
metadataVersion: number
|
||||
agentStateVersion: number
|
||||
todosUpdatedAt: number
|
||||
todoProgress: { completed: number; total: number } | null
|
||||
pendingRequestsCount: number
|
||||
pendingRequestKinds: PendingRequestKind[]
|
||||
@@ -65,35 +72,11 @@ export type SessionSummary = {
|
||||
effort: string | null
|
||||
}
|
||||
|
||||
export function getPendingRequests(
|
||||
session: Session,
|
||||
cap: number = PENDING_REQUEST_SUMMARY_CAP
|
||||
): PendingRequest[] {
|
||||
const requests = session.agentState?.requests
|
||||
if (!requests) {
|
||||
return []
|
||||
}
|
||||
|
||||
const items: PendingRequest[] = []
|
||||
for (const [id, request] of Object.entries(requests)) {
|
||||
items.push({
|
||||
id,
|
||||
kind: classifyKind(request.tool),
|
||||
tool: request.tool,
|
||||
since: typeof request.createdAt === 'number' ? request.createdAt : session.updatedAt
|
||||
})
|
||||
}
|
||||
|
||||
items.sort((a, b) => {
|
||||
if (a.since !== b.since) return a.since - b.since
|
||||
return a.id < b.id ? -1 : a.id > b.id ? 1 : 0
|
||||
})
|
||||
|
||||
return cap >= items.length ? items : items.slice(0, cap)
|
||||
}
|
||||
|
||||
export function getPendingRequestKinds(session: Session): PendingRequestKind[] {
|
||||
const requests = session.agentState?.requests
|
||||
// Re-exported as a standalone derivation so SSE patch handlers can recompute
|
||||
// summary fields from a structured `agentState` patch without needing the
|
||||
// full Session in hand.
|
||||
export function computePendingRequestKinds(agentState: AgentState | null | undefined): PendingRequestKind[] {
|
||||
const requests = agentState?.requests
|
||||
if (!requests) {
|
||||
return []
|
||||
}
|
||||
@@ -108,6 +91,59 @@ export function getPendingRequestKinds(session: Session): PendingRequestKind[] {
|
||||
: Array.from(kinds)
|
||||
}
|
||||
|
||||
export function computePendingRequests(
|
||||
agentState: AgentState | null | undefined,
|
||||
fallbackSince: number,
|
||||
cap: number = PENDING_REQUEST_SUMMARY_CAP
|
||||
): PendingRequest[] {
|
||||
const requests = agentState?.requests
|
||||
if (!requests) {
|
||||
return []
|
||||
}
|
||||
|
||||
const items: PendingRequest[] = []
|
||||
for (const [id, request] of Object.entries(requests)) {
|
||||
items.push({
|
||||
id,
|
||||
kind: classifyKind(request.tool),
|
||||
tool: request.tool,
|
||||
since: typeof request.createdAt === 'number' ? request.createdAt : fallbackSince
|
||||
})
|
||||
}
|
||||
|
||||
items.sort((a, b) => {
|
||||
if (a.since !== b.since) return a.since - b.since
|
||||
return a.id < b.id ? -1 : a.id > b.id ? 1 : 0
|
||||
})
|
||||
|
||||
return cap >= items.length ? items : items.slice(0, cap)
|
||||
}
|
||||
|
||||
export function getPendingRequestKinds(session: Session): PendingRequestKind[] {
|
||||
return computePendingRequestKinds(session.agentState)
|
||||
}
|
||||
|
||||
export function getPendingRequests(
|
||||
session: Session,
|
||||
cap: number = PENDING_REQUEST_SUMMARY_CAP
|
||||
): PendingRequest[] {
|
||||
return computePendingRequests(session.agentState, session.updatedAt, cap)
|
||||
}
|
||||
|
||||
export function computePendingRequestsCount(agentState: AgentState | null | undefined): number {
|
||||
return agentState?.requests ? Object.keys(agentState.requests).length : 0
|
||||
}
|
||||
|
||||
export function computeTodoProgress(todos: TodoItem[] | undefined): SessionSummary['todoProgress'] {
|
||||
if (!todos?.length) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
completed: todos.filter((todo) => todo.status === 'completed').length,
|
||||
total: todos.length
|
||||
}
|
||||
}
|
||||
|
||||
const AGENT_SESSION_ID_FIELD_BY_FLAVOR = {
|
||||
claude: 'claudeSessionId',
|
||||
codex: 'codexSessionId',
|
||||
@@ -144,36 +180,37 @@ function getSummaryAgentSessionId(metadata: Metadata): string | undefined {
|
||||
?? undefined
|
||||
}
|
||||
|
||||
export function toSessionSummaryMetadata(metadata: Metadata | null | undefined): SessionSummaryMetadata | null {
|
||||
if (!metadata) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
name: metadata.name,
|
||||
path: metadata.path,
|
||||
machineId: metadata.machineId ?? undefined,
|
||||
summary: metadata.summary ? { text: metadata.summary.text } : undefined,
|
||||
flavor: metadata.flavor ?? null,
|
||||
worktree: metadata.worktree,
|
||||
agentSessionId: getSummaryAgentSessionId(metadata),
|
||||
lifecycleState: metadata.lifecycleState
|
||||
}
|
||||
}
|
||||
|
||||
export function toSessionSummary(session: Session): SessionSummary {
|
||||
const pendingRequestsCount = session.agentState?.requests ? Object.keys(session.agentState.requests).length : 0
|
||||
|
||||
const metadata: SessionSummaryMetadata | null = session.metadata ? {
|
||||
name: session.metadata.name,
|
||||
path: session.metadata.path,
|
||||
machineId: session.metadata.machineId ?? undefined,
|
||||
summary: session.metadata.summary ? { text: session.metadata.summary.text } : undefined,
|
||||
flavor: session.metadata.flavor ?? null,
|
||||
worktree: session.metadata.worktree,
|
||||
agentSessionId: getSummaryAgentSessionId(session.metadata),
|
||||
lifecycleState: session.metadata.lifecycleState
|
||||
} : null
|
||||
|
||||
const todoProgress = session.todos?.length ? {
|
||||
completed: session.todos.filter(t => t.status === 'completed').length,
|
||||
total: session.todos.length
|
||||
} : null
|
||||
|
||||
return {
|
||||
id: session.id,
|
||||
active: session.active,
|
||||
thinking: session.thinking,
|
||||
activeAt: session.activeAt,
|
||||
updatedAt: session.updatedAt,
|
||||
metadata,
|
||||
todoProgress,
|
||||
pendingRequestsCount,
|
||||
pendingRequestKinds: getPendingRequestKinds(session),
|
||||
pendingRequests: getPendingRequests(session),
|
||||
metadata: toSessionSummaryMetadata(session.metadata),
|
||||
metadataVersion: session.metadataVersion,
|
||||
agentStateVersion: session.agentStateVersion,
|
||||
todosUpdatedAt: session.todosUpdatedAt ?? 0,
|
||||
todoProgress: computeTodoProgress(session.todos),
|
||||
pendingRequestsCount: computePendingRequestsCount(session.agentState),
|
||||
pendingRequestKinds: computePendingRequestKinds(session.agentState),
|
||||
pendingRequests: computePendingRequests(session.agentState, session.updatedAt),
|
||||
backgroundTaskCount: session.backgroundTaskCount ?? 0,
|
||||
futureScheduledMessageCount: 0,
|
||||
nextScheduledAt: null,
|
||||
|
||||
@@ -19,6 +19,9 @@ function makeSummary(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
@@ -20,6 +20,9 @@ function makeSession(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
@@ -16,6 +16,9 @@ function makeSession(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
@@ -28,6 +28,9 @@ function makeSession(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
@@ -1,7 +1,15 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { SessionSummary } from '@/types/api'
|
||||
import type { Session } from '@/types/api'
|
||||
import { isGlobalScopedMessageStreamEvent, isRenderIrrelevantPatch, isRenderIrrelevantSessionPatch, shouldInvalidateSessionListForEvent } from './useSSE'
|
||||
import {
|
||||
applySessionDetailPatch,
|
||||
canApplyVersionedSummaryPatch,
|
||||
isGlobalScopedMessageStreamEvent,
|
||||
isNewerVersionedPatch,
|
||||
isRenderIrrelevantPatch,
|
||||
isRenderIrrelevantSessionPatch,
|
||||
shouldInvalidateSessionListForEvent
|
||||
} from './useSSE'
|
||||
|
||||
function makeSummary(overrides: Partial<SessionSummary> = {}): SessionSummary {
|
||||
return {
|
||||
@@ -11,6 +19,9 @@ function makeSummary(overrides: Partial<SessionSummary> = {}): SessionSummary {
|
||||
activeAt: 1_000,
|
||||
updatedAt: 2_000,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
@@ -24,6 +35,81 @@ function makeSummary(overrides: Partial<SessionSummary> = {}): SessionSummary {
|
||||
} as SessionSummary
|
||||
}
|
||||
|
||||
describe('canApplyVersionedSummaryPatch (PR #897 review, HAPI Bot 2026-07-23 Major)', () => {
|
||||
it('allows non-versioned patches without a detail cache', () => {
|
||||
expect(canApplyVersionedSummaryPatch({}, false)).toBe(true)
|
||||
expect(canApplyVersionedSummaryPatch({ metadata: undefined, agentState: undefined }, false)).toBe(true)
|
||||
})
|
||||
|
||||
it('refuses metadata/agentState/todos summary patches when detail version source is missing', () => {
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ metadata: { version: 1, value: null } },
|
||||
false
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ agentState: { version: 1, value: null } },
|
||||
false
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ todos: { version: 1, value: [] } },
|
||||
false
|
||||
)
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('allows teamState-only patches without detail (summary no-op)', () => {
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ teamState: { version: 1, value: null } },
|
||||
false
|
||||
)
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('allows versioned summary patches when detail is present', () => {
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ metadata: { version: 2, value: null } },
|
||||
true
|
||||
)
|
||||
).toBe(true)
|
||||
expect(
|
||||
canApplyVersionedSummaryPatch(
|
||||
{ todos: { version: 2, value: [] } },
|
||||
true
|
||||
)
|
||||
).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('isNewerVersionedPatch (PR #897 review, HAPI Bot 2026-06-16 Major)', () => {
|
||||
// Pin the version-monotonicity contract for structured metadata /
|
||||
// agentState patches. Without this gate, an SSE reconnect that replays
|
||||
// a buffered older patch after a fresh REST refetch would regress the
|
||||
// cache (e.g. drop a newer resume id / pending request). Mirrors the
|
||||
// hub's CLI room handler check (`incoming.version > currentVersion`).
|
||||
it('accepts a strictly newer patch', () => {
|
||||
expect(isNewerVersionedPatch(5, 4)).toBe(true)
|
||||
})
|
||||
|
||||
it('rejects an older patch (the bug case: stale buffered patch on reconnect)', () => {
|
||||
expect(isNewerVersionedPatch(4, 5)).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects a same-version patch (idempotent / duplicate replay)', () => {
|
||||
expect(isNewerVersionedPatch(5, 5)).toBe(false)
|
||||
})
|
||||
|
||||
it('accepts the first write into a freshly-cached session (currentVersion=0)', () => {
|
||||
expect(isNewerVersionedPatch(1, 0)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('useSSE scope handling', () => {
|
||||
it('invalidates the global session list when message ownership changes', () => {
|
||||
expect(shouldInvalidateSessionListForEvent('global', 'messages-invalidated')).toBe(true)
|
||||
@@ -68,7 +154,21 @@ describe('isRenderIrrelevantPatch', () => {
|
||||
['model', { model: 'opus' }],
|
||||
['modelReasoningEffort', { modelReasoningEffort: 'high' }],
|
||||
['effort', { effort: 'medium' }],
|
||||
['pendingRequestsCount', { pendingRequestsCount: 2 }]
|
||||
['pendingRequestsCount', { pendingRequestsCount: 2 }],
|
||||
['metadata.path', { metadata: { path: '/other', name: undefined } }],
|
||||
['metadata.flavor', { metadata: { path: '/tmp', flavor: 'claude' as const } }],
|
||||
['metadata.machineId', { metadata: { path: '/tmp', machineId: 'Teemo' } }],
|
||||
['metadata.worktree.branch', {
|
||||
metadata: {
|
||||
path: '/tmp',
|
||||
worktree: {
|
||||
basePath: '/tmp',
|
||||
branch: 'feat/x',
|
||||
name: 'x',
|
||||
worktreePath: '/tmp/x'
|
||||
}
|
||||
}
|
||||
}]
|
||||
] as Array<[string, Partial<SessionSummary>]>)('reports %s changes as relevant', (_field, change) => {
|
||||
const current = makeSummary()
|
||||
const next = makeSummary({ ...change, activeAt: 11_000 })
|
||||
@@ -138,3 +238,47 @@ describe('isRenderIrrelevantSessionPatch', () => {
|
||||
expect(isRenderIrrelevantSessionPatch(session, {})).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('applySessionDetailPatch (PR #897 review, Copilot keep-alive)', () => {
|
||||
const session = {
|
||||
id: 'session-1',
|
||||
active: true,
|
||||
thinking: false,
|
||||
activeAt: 1_000,
|
||||
updatedAt: 2_000,
|
||||
model: 'gpt-5',
|
||||
effort: null,
|
||||
permissionMode: 'default',
|
||||
collaborationMode: undefined,
|
||||
copilotAgentMode: 'interactive',
|
||||
serviceTier: null,
|
||||
metadataVersion: 1,
|
||||
agentStateVersion: 1,
|
||||
todosUpdatedAt: 0,
|
||||
teamStateUpdatedAt: 0
|
||||
} as unknown as Session
|
||||
|
||||
it('applies a copilotAgentMode keep-alive change to the detail session', () => {
|
||||
// Hub emits copilotAgentMode from markSessionActive keep-alives. The
|
||||
// field-by-field mapper must copy it — otherwise detailPatched=true
|
||||
// suppresses invalidation and SessionChat keeps the stale mode.
|
||||
const next = applySessionDetailPatch(session, {
|
||||
active: true,
|
||||
thinking: false,
|
||||
activeAt: 11_000,
|
||||
copilotAgentMode: 'plan'
|
||||
})
|
||||
expect(next).not.toBeNull()
|
||||
expect(next?.copilotAgentMode).toBe('plan')
|
||||
expect(next?.activeAt).toBe(11_000)
|
||||
})
|
||||
|
||||
it('returns null for a keep-alive that only repeats the current Copilot mode', () => {
|
||||
expect(applySessionDetailPatch(session, {
|
||||
active: true,
|
||||
thinking: false,
|
||||
activeAt: 11_000,
|
||||
copilotAgentMode: 'interactive'
|
||||
})).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
+172
-7
@@ -1,6 +1,14 @@
|
||||
import { useEffect, useMemo, useRef, useState } from 'react'
|
||||
import { useQueryClient } from '@tanstack/react-query'
|
||||
import { isObject, toSessionSummary } from '@hapi/protocol'
|
||||
import {
|
||||
computePendingRequestKinds,
|
||||
computePendingRequests,
|
||||
computePendingRequestsCount,
|
||||
computeTodoProgress,
|
||||
isObject,
|
||||
toSessionSummary,
|
||||
toSessionSummaryMetadata
|
||||
} from '@hapi/protocol'
|
||||
import { MachinePatchSchema, MachineSchema, SessionPatchSchema, SessionSchema } from '@hapi/protocol/schemas'
|
||||
import type {
|
||||
Machine,
|
||||
@@ -38,6 +46,36 @@ export function shouldInvalidateSessionListForEvent(scope: SSEScope, eventType:
|
||||
return scope === 'global' && eventType === 'messages-invalidated'
|
||||
}
|
||||
|
||||
// Version-monotonicity gate for structured patches carrying metadata or
|
||||
// agentState. SSE reconnects + per-query invalidation can leave the cache
|
||||
// holding state that's NEWER than a buffered older patch about to replay;
|
||||
// applying that older patch would regress resume / session-id / pending-
|
||||
// requests state. Mirrors the CLI room handler contract: strictly newer
|
||||
// only. Exported so the rule is unit-testable in isolation from the hook.
|
||||
export function isNewerVersionedPatch(patchVersion: number, currentVersion: number): boolean {
|
||||
return patchVersion > currentVersion
|
||||
}
|
||||
|
||||
/**
|
||||
* @deprecated Prefer gating against SessionSummary watermarks. Kept for unit
|
||||
* tests of the old "detail required" rule; summary path no longer uses this
|
||||
* for metadata/agentState/todos (teamState is a summary no-op).
|
||||
*/
|
||||
export function canApplyVersionedSummaryPatch(
|
||||
patch: Pick<SessionPatch, 'metadata' | 'agentState' | 'todos' | 'teamState'>,
|
||||
detailPresent: boolean
|
||||
): boolean {
|
||||
// teamState is not rendered on SessionSummary — never block list patches.
|
||||
if (
|
||||
patch.metadata === undefined
|
||||
&& patch.agentState === undefined
|
||||
&& patch.todos === undefined
|
||||
) {
|
||||
return true
|
||||
}
|
||||
return detailPresent
|
||||
}
|
||||
|
||||
type VisibilityState = 'visible' | 'hidden'
|
||||
|
||||
type ToastEvent = Extract<SyncEvent, { type: 'toast' }>
|
||||
@@ -89,6 +127,69 @@ export function isRenderIrrelevantSessionPatch(session: Session, patch: SessionP
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a validated SessionPatch onto a detail Session. Returns null when
|
||||
* nothing render-relevant changed (keep prior object identity). Field-by-field
|
||||
* only — never wholesale-spread versioned `{ version, value }` wrappers.
|
||||
* Exported for unit tests (Copilot mode keep-alive must not be dropped).
|
||||
*/
|
||||
export function applySessionDetailPatch(session: Session, patch: SessionPatch): Session | null {
|
||||
if (isRenderIrrelevantSessionPatch(session, patch)) {
|
||||
return null
|
||||
}
|
||||
let changed = false
|
||||
const nextSession: Session = { ...session }
|
||||
const assign = <K extends keyof Session>(key: K, value: Session[K]) => {
|
||||
if (nextSession[key] !== value) {
|
||||
nextSession[key] = value
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if (patch.active !== undefined) assign('active', patch.active)
|
||||
if (patch.thinking !== undefined) assign('thinking', patch.thinking)
|
||||
if (patch.activeAt !== undefined) assign('activeAt', patch.activeAt)
|
||||
// Monotonic with hub applySessionPatch: a rejected stale
|
||||
// metadata/agentState replay must not rewind updatedAt.
|
||||
if (patch.updatedAt !== undefined) {
|
||||
const nextUpdatedAt = Math.max(nextSession.updatedAt, patch.updatedAt)
|
||||
assign('updatedAt', nextUpdatedAt)
|
||||
}
|
||||
if (patch.model !== undefined) assign('model', patch.model)
|
||||
if (patch.modelReasoningEffort !== undefined) assign('modelReasoningEffort', patch.modelReasoningEffort)
|
||||
if (patch.effort !== undefined) assign('effort', patch.effort)
|
||||
if (Object.prototype.hasOwnProperty.call(patch, 'serviceTier')) {
|
||||
assign('serviceTier', patch.serviceTier ?? null)
|
||||
}
|
||||
if (patch.permissionMode !== undefined) assign('permissionMode', patch.permissionMode)
|
||||
if (patch.collaborationMode !== undefined) assign('collaborationMode', patch.collaborationMode)
|
||||
if (patch.copilotAgentMode !== undefined) assign('copilotAgentMode', patch.copilotAgentMode)
|
||||
if (patch.backgroundTaskCount !== undefined) assign('backgroundTaskCount', patch.backgroundTaskCount)
|
||||
// Version gates: dual SSE can deliver duplicates out of order.
|
||||
// Only mark changed when a strictly newer version lands —
|
||||
// otherwise keep previous object identity (no redundant render).
|
||||
if (patch.todos !== undefined && isNewerVersionedPatch(patch.todos.version, nextSession.todosUpdatedAt ?? 0)) {
|
||||
nextSession.todos = patch.todos.value
|
||||
nextSession.todosUpdatedAt = patch.todos.version
|
||||
changed = true
|
||||
}
|
||||
if (patch.teamState !== undefined && isNewerVersionedPatch(patch.teamState.version, nextSession.teamStateUpdatedAt ?? 0)) {
|
||||
nextSession.teamState = patch.teamState.value ?? undefined
|
||||
nextSession.teamStateUpdatedAt = patch.teamState.version
|
||||
changed = true
|
||||
}
|
||||
if (patch.metadata !== undefined && isNewerVersionedPatch(patch.metadata.version, nextSession.metadataVersion)) {
|
||||
nextSession.metadata = patch.metadata.value
|
||||
nextSession.metadataVersion = patch.metadata.version
|
||||
changed = true
|
||||
}
|
||||
if (patch.agentState !== undefined && isNewerVersionedPatch(patch.agentState.version, nextSession.agentStateVersion)) {
|
||||
nextSession.agentState = patch.agentState.value
|
||||
nextSession.agentStateVersion = patch.agentState.version
|
||||
changed = true
|
||||
}
|
||||
return changed ? nextSession : null
|
||||
}
|
||||
|
||||
function isSessionRecord(value: unknown): value is Session {
|
||||
return SessionSchema.safeParse(value).success
|
||||
}
|
||||
@@ -102,6 +203,30 @@ function isSessionRecord(value: unknown): value is Session {
|
||||
* `updatedAt` and sorts on active/pendingRequestsCount/updatedAt - so storing
|
||||
* it costs a new object identity and a full list re-render for nothing.
|
||||
*/
|
||||
export function sameSessionSummaryMetadata(
|
||||
current: SessionSummary['metadata'],
|
||||
next: SessionSummary['metadata']
|
||||
): boolean {
|
||||
if (current === next) {
|
||||
return true
|
||||
}
|
||||
if (current == null || next == null) {
|
||||
return current == null && next == null
|
||||
}
|
||||
return current.name === next.name
|
||||
&& current.path === next.path
|
||||
&& current.machineId === next.machineId
|
||||
&& current.summary?.text === next.summary?.text
|
||||
&& current.flavor === next.flavor
|
||||
&& current.agentSessionId === next.agentSessionId
|
||||
&& current.lifecycleState === next.lifecycleState
|
||||
&& current.worktree?.basePath === next.worktree?.basePath
|
||||
&& current.worktree?.branch === next.worktree?.branch
|
||||
&& current.worktree?.name === next.worktree?.name
|
||||
&& current.worktree?.worktreePath === next.worktree?.worktreePath
|
||||
&& current.worktree?.createdAt === next.worktree?.createdAt
|
||||
}
|
||||
|
||||
export function isRenderIrrelevantPatch(current: SessionSummary, next: SessionSummary): boolean {
|
||||
return current.active === next.active
|
||||
&& current.thinking === next.thinking
|
||||
@@ -111,6 +236,25 @@ export function isRenderIrrelevantPatch(current: SessionSummary, next: SessionSu
|
||||
&& current.modelReasoningEffort === next.modelReasoningEffort
|
||||
&& current.effort === next.effort
|
||||
&& current.pendingRequestsCount === next.pendingRequestsCount
|
||||
// Structured SSE patches (#897) can move these without touching the
|
||||
// keep-alive fields above; omit them and a todos/metadata/agentState
|
||||
// patch would be dropped as "activeAt-only" churn.
|
||||
&& current.todoProgress?.completed === next.todoProgress?.completed
|
||||
&& current.todoProgress?.total === next.todoProgress?.total
|
||||
&& (current.todoProgress == null) === (next.todoProgress == null)
|
||||
&& current.pendingRequestKinds.length === next.pendingRequestKinds.length
|
||||
&& current.pendingRequestKinds.every((kind, i) => kind === next.pendingRequestKinds[i])
|
||||
&& current.pendingRequests.length === next.pendingRequests.length
|
||||
&& current.pendingRequests.every((req, i) => {
|
||||
const nextReq = next.pendingRequests[i]
|
||||
return req.id === nextReq?.id
|
||||
&& req.kind === nextReq.kind
|
||||
&& req.tool === nextReq.tool
|
||||
})
|
||||
&& sameSessionSummaryMetadata(current.metadata, next.metadata)
|
||||
&& current.metadataVersion === next.metadataVersion
|
||||
&& current.agentStateVersion === next.agentStateVersion
|
||||
&& current.todosUpdatedAt === next.todosUpdatedAt
|
||||
}
|
||||
|
||||
function getSessionPatch(value: unknown): SessionPatch | null {
|
||||
@@ -412,7 +556,11 @@ export function useSSE(options: {
|
||||
active: patch.active ?? current.active,
|
||||
thinking: patch.thinking ?? current.thinking,
|
||||
activeAt: patch.activeAt ?? current.activeAt,
|
||||
updatedAt: patch.updatedAt ?? current.updatedAt,
|
||||
// Monotonic: stale versioned-patch replays can carry an
|
||||
// older updatedAt; never move the list clock backward.
|
||||
updatedAt: patch.updatedAt !== undefined
|
||||
? Math.max(current.updatedAt, patch.updatedAt)
|
||||
: current.updatedAt,
|
||||
backgroundTaskCount: Object.prototype.hasOwnProperty.call(patch, 'backgroundTaskCount')
|
||||
? patch.backgroundTaskCount ?? 0
|
||||
: current.backgroundTaskCount,
|
||||
@@ -423,6 +571,25 @@ export function useSSE(options: {
|
||||
effort: Object.prototype.hasOwnProperty.call(patch, 'effort') ? patch.effort ?? null : current.effort
|
||||
}
|
||||
|
||||
// Gate versioned fields against THIS summary's watermarks —
|
||||
// not the detail query. Global SSE covers every session;
|
||||
// requiring detail would force O(N) /sessions invalidation.
|
||||
// teamState is a summary no-op (not rendered on the list).
|
||||
if (patch.todos !== undefined && patch.todos.version >= current.todosUpdatedAt) {
|
||||
nextSummary.todoProgress = computeTodoProgress(patch.todos.value)
|
||||
nextSummary.todosUpdatedAt = patch.todos.version
|
||||
}
|
||||
if (patch.agentState !== undefined && patch.agentState.version >= current.agentStateVersion) {
|
||||
nextSummary.pendingRequestsCount = computePendingRequestsCount(patch.agentState.value)
|
||||
nextSummary.pendingRequestKinds = computePendingRequestKinds(patch.agentState.value)
|
||||
nextSummary.pendingRequests = computePendingRequests(patch.agentState.value, nextSummary.updatedAt)
|
||||
nextSummary.agentStateVersion = patch.agentState.version
|
||||
}
|
||||
if (patch.metadata !== undefined && patch.metadata.version >= current.metadataVersion) {
|
||||
nextSummary.metadata = toSessionSummaryMetadata(patch.metadata.value)
|
||||
nextSummary.metadataVersion = patch.metadata.version
|
||||
}
|
||||
|
||||
patched = true
|
||||
// The keep-alive patch repeats every field every ~10s per active
|
||||
// session, and `activeAt` is the only one that actually moves.
|
||||
@@ -445,15 +612,13 @@ export function useSSE(options: {
|
||||
return previous
|
||||
}
|
||||
patched = true
|
||||
if (isRenderIrrelevantSessionPatch(previous.session, patch)) {
|
||||
const nextSession = applySessionDetailPatch(previous.session, patch)
|
||||
if (!nextSession) {
|
||||
return previous
|
||||
}
|
||||
return {
|
||||
...previous,
|
||||
session: {
|
||||
...previous.session,
|
||||
...patch
|
||||
}
|
||||
session: nextSession
|
||||
}
|
||||
})
|
||||
return patched
|
||||
|
||||
@@ -9,6 +9,9 @@ function makeSummary(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 1000,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
@@ -16,6 +16,9 @@ function makeSession(overrides: Partial<SessionSummary> & { id: string }): Sessi
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: null,
|
||||
metadataVersion: 0,
|
||||
agentStateVersion: 0,
|
||||
todosUpdatedAt: 0,
|
||||
todoProgress: null,
|
||||
pendingRequestsCount: 0,
|
||||
pendingRequestKinds: [],
|
||||
|
||||
Reference in New Issue
Block a user