From 2513e2139043b3091ec8d61b60dcfef502c4af27 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:35:03 -0700 Subject: [PATCH] fix(native-chat): publish structured session status from the host so the sidebar never goes stale (#18776) * fix(native-chat): publish structured session status from the host The sidebar learned whether a structured chat was mid-turn by replaying the session journal in the renderer, through a reader whose lifetime was tied to the chat pane. Hiding the pane stopped the reader before the turn's settlement arrived, so the row stayed on "working" until the chat was reopened. The same coupling meant a tab never opened this session showed no status at all, and a reloaded renderer lost every settled row. The host owns the journal, so it now projects each session's status once per journal publication and fans the changes out on one stream per client (`agentSession.subscribeStatus`). The projection survives eviction of an idle session's provider child and is republished when readable sessions are restored. The renderer bridge subscribes to that feed per runtime target and never opens a transcript reader; the observation hook is gone. Additive wire surface behind the existing structured capability; old hosts reject the method and the renderer retries, showing no status. * fix(native-chat): negotiate the status feed and stop losing a change on subscribe The status stream is additive to a surface that already shipped, so a host advertising agent-session.structured.v1 can still answer subscribeStatus with method_not_found. Every renderer error path reconnected, so a remote host one release behind got a relay round-trip every 5s and no sidebar status at all. Give the method its own capability and probe it before subscribing; a failed probe still retries, an absent capability does not. Re-projecting on subscribe also wrote straight into the shared cache, so a second client could pin the first to a stale summary. Route those diffs through publish() before the arriving subscriber is registered. * fix(native-chat): bound the status prompt, merge snapshots, and prove the unread path One status frame carries every retained session and a send admits 256 KB per prompt, so ~16 large-prompt sessions could push the snapshot past the 4 MB outbound guard and into the retry loop. Bound latestPrompt to the same 200-char single-line preview every other agent-status row already carries. A snapshot also replaced the cached map wholesale, so the empty first frame from a restarting host retracted every row before restore republished them. Merge instead; the tab map, not this feed, decides which sessions are listed. Tests: the hidden-pane claim now sits at the host, where a journal with no transcript subscriber is driven from running to idle; the RPC test reads a real projection instead of its own stub. * fix(native-chat): merge the duplicated status-event type import * test(native-chat): pin the restart status publication, and log the unsupported host Startup restore indexes a readable session and publishes its status, which is what puts a never-reopened tab back in the sidebar. Only an Electron screenshot covered that wiring; a sitting status subscriber now pins it directly. The terminal "host too old" branch was silent, so a mixed-version report showed an empty sidebar with nothing in the log to explain it. --------- Co-authored-by: Merge Sim --- ...structured-agent-session-history-result.ts | 4 +- .../structured-agent-session-host-lifetime.ts | 23 ++ .../structured-agent-session-host.ts | 43 ++-- ...session-restart-status-publication.test.ts | 149 +++++++++++ ...ructured-agent-session-status-feed.test.ts | 242 ++++++++++++++++++ .../structured-agent-session-status-feed.ts | 130 ++++++++++ ...ructured-agent-session-subscribers.test.ts | 92 +++++++ .../structured-agent-session-subscribers.ts | 11 + .../structured-agent-session-status-stream.ts | 62 +++++ ...tructured-agent-session-subscription-id.ts | 29 +++ .../methods/structured-agent-session.test.ts | 79 +++++- .../rpc/methods/structured-agent-session.ts | 52 ++-- ...tructuredAgentSessionStatusBridge.test.tsx | 226 +++++++++------- .../StructuredAgentSessionStatusBridge.tsx | 73 +++--- ...use-structured-agent-session-read.test.tsx | 31 +-- .../use-structured-agent-session-read.ts | 7 - .../structured-agent-session-client.ts | 50 +++- ...ructured-agent-session-status-feed.test.ts | 140 ++++++++++ .../structured-agent-session-status-feed.ts | 211 +++++++++++++++ src/shared/agent-session-wire.ts | 30 ++- src/shared/protocol-version.ts | 5 + ...tructured-agent-session-projection.test.ts | 48 ++++ .../structured-agent-session-projection.ts | 29 +++ ...ss-version-agent-session-wire.unit.test.ts | 24 +- 24 files changed, 1556 insertions(+), 234 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-restart-status-publication.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-status-stream.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts create mode 100644 src/renderer/src/runtime/structured-agent-session-status-feed.test.ts create mode 100644 src/renderer/src/runtime/structured-agent-session-status-feed.ts diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts index 70fc9a43ed2..b8e198c9b6b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts @@ -8,7 +8,7 @@ import type { import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { readAgentSessionHistory } from './agent-session-history-page' -function providerSessionMetadata( +export function structuredAgentSessionProviderSessionMetadata( record: AgentSessionRecord | null ): AgentProviderSessionMetadata | undefined { const head = record ? agentSessionProviderHandleChainHead(record.providerHandleChain) : null @@ -27,7 +27,7 @@ export function readStructuredAgentSessionHistoryResult(input: { }): AgentSessionHistoryResult { const result = readAgentSessionHistory(input.journal, input.request) const fence = input.record?.lease.runtimeFence - const providerSession = providerSessionMetadata(input.record) + const providerSession = structuredAgentSessionProviderSessionMetadata(input.record) if (fence === undefined) { return providerSession ? { ...result, providerSession } : result } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts index 2afd94ba128..ba62db1704e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts @@ -19,6 +19,8 @@ import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' import { releaseStoredStructuredAgentSessionOwner } from './structured-agent-session-lease-release' +import { resumeHeldStructuredAgentSession } from './structured-agent-session-hold-resume' +import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire' export type StructuredAgentSessionLifetimeContext = { deps: StructuredAgentSessionHostDeps @@ -67,6 +69,27 @@ export async function evictHeldStructuredAgentSession( ) } +/** The first hold on a childless session: reconcile the lease, settle recovery, then attach. */ +export async function resumeStructuredAgentSessionForHold( + context: StructuredAgentSessionLifetimeContext & { + reconcileLeases: (sessionId: string) => Promise + }, + sessionId: string, + attach: Parameters[0]['attach'] +): Promise { + const unreconciled = await context.reconcileLeases(sessionId) + if (unreconciled) { + throw new Error(unreconciled.code) + } + await context.runtimeState.resolveRecovery(sessionId) + await resumeHeldStructuredAgentSession({ + sessionId, + deps: context.deps, + now: context.now, + attach + }) +} + export function createStructuredAgentSessionHolds( context: StructuredAgentSessionLifetimeContext, input: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 8989f5e4d72..e4c7d191067 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -33,13 +33,13 @@ import { attachStructuredAgentSession } from './structured-agent-session-attach- import { createStructuredAgentSessionHolds, evictHeldStructuredAgentSession, + resumeStructuredAgentSessionForHold, type StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' import type { StructuredAgentSessionHolds, StructuredAgentSessionHoldOptions } from './structured-agent-session-holds' -import { resumeHeldStructuredAgentSession } from './structured-agent-session-hold-resume' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' import { listStructuredAgentSessionTabs } from './structured-agent-session-host-tabs' import { @@ -57,6 +57,7 @@ import type { StructuredAgentSessionHostDeps, StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' import { StructuredAgentSessionEventRecovery } from './structured-agent-session-event-recovery' import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel' import { withTimeout } from '../../../shared/promise-timeout-fallback' @@ -66,7 +67,14 @@ const HANDOFF_DRAIN_TIMEOUT_MS = 5_000 export class StructuredAgentSessionHost { private readonly sessions = new Map() - private readonly subscribers = new AgentSessionSubscribers() + private readonly statusFeed = new StructuredAgentSessionStatusFeed({ + sessions: this.sessions, + getRecord: (sessionId) => this.deps.store.getRecord(sessionId), + now: () => this.now() + }) + private readonly subscribers = new AgentSessionSubscribers({ + onJournalPublished: (sessionId, journal) => this.statusFeed.publish(sessionId, journal) + }) private readonly tasks = new StructuredAgentSessionTaskQueue() private readonly runtimeState: StructuredAgentSessionHostRuntimeState private readonly reconcileLeases: (sessionId: string) => Promise @@ -112,7 +120,12 @@ export class StructuredAgentSessionHost { now: this.now }) this.holds = createStructuredAgentSessionHolds(this.lifetimeContext(), { - resume: (sessionId) => this.resumeForHold(sessionId), + resume: (sessionId) => + resumeStructuredAgentSessionForHold( + { ...this.lifetimeContext(), reconcileLeases: this.reconcileLeases }, + sessionId, + (params) => this.attach({ callerKey: 'trusted-local:surface-hold' }, params) + ), evict: (sessionId) => this.close(sessionId) }) this.readableRestorer = new StructuredAgentSessionReadableRestorer({ @@ -125,7 +138,10 @@ export class StructuredAgentSessionHost { hasSession: (sessionId) => this.sessions.has(sessionId), // Site 10: cannot overwrite a live entry — the restorer returns early on // `hasSession` inside the same serialized step as this `set`. - onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored), + onReadable: (sessionId, restored) => { + this.sessions.set(sessionId, restored) + this.statusFeed.publish(sessionId) + }, restoreHandoff: (sessionId) => this.handoffs.restore(sessionId) }) this.eventRecovery = new StructuredAgentSessionEventRecovery({ @@ -160,20 +176,6 @@ export class StructuredAgentSessionHost { /** That surface is gone. The child outlives it by the release grace, and by any running turn. */ release = (sessionId: string, holderId: string): void => this.holds.release(sessionId, holderId) - private async resumeForHold(sessionId: string): Promise { - const unreconciled = await this.reconcileLeases(sessionId) - if (unreconciled) { - throw new Error(unreconciled.code) - } - await this.runtimeState.resolveRecovery(sessionId) - await resumeHeldStructuredAgentSession({ - sessionId, - deps: this.deps, - now: () => this.now(), - attach: (params) => this.attach({ callerKey: 'trusted-local:surface-hold' }, params) - }) - } - handleAdapterEvent = (event: Parameters[0]) => this.eventRecovery.handle(event) @@ -342,6 +344,11 @@ export class StructuredAgentSessionHost { ) => this.backgroundTasks.publish(sessionId, state) unsubscribe = (sessionId: string, id: string): void => this.subscribers.close(sessionId, id) + /** Every session's projected status for session lists; unlike `subscribe`, retains nothing. */ + subscribeStatus = ( + subscriber: Parameters[0] + ): (() => void) => this.statusFeed.subscribe(subscriber) + private requireSession(sessionId: string): StructuredAgentSessionHostSession { const session = this.sessions.get(sessionId) if (!session) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-status-publication.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-status-publication.test.ts new file mode 100644 index 00000000000..884fba78a4e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-status-publication.test.ts @@ -0,0 +1,149 @@ +// Startup restore has to publish status, not just index the session. +// +// A tab nobody reopens after a restart still owes the sidebar a row. The host restores such a +// session read-only, without a provider child, so the only thing that can surface its state is +// the status publication the restore wiring makes. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import type { + AgentSessionMutationEnvelope, + AgentSessionStatusEvent +} from '../../../shared/agent-session-wire' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestAttachParams, + hostTestMessage, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' + +const CALLER = { callerKey: 'client-1' } + +const hosts: StructuredAgentSessionHost[] = [] +let root = '' + +function adapter(): StructuredAgentSessionAdapter { + return { + acquire: async ({ fence, spawnToken }) => ({ + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex', threadId: THREAD }, + origin: 'created', + mintedAtFence: fence, + observedAt: NOW + } + }), + dispatch: async () => ({ + state: 'accepted', + providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 1 } + }), + cancelTurn: async () => ({ cancelled: true }), + answerPrompt: async () => undefined, + setOption: async () => undefined + } +} + +function createHost(store: AgentSessionRecordStore): StructuredAgentSessionHost { + const host = new StructuredAgentSessionHost({ + store, + adapter: adapter(), + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-a', + probeOwner: async () => ({ + outcome: 'indeterminate', + reason: 'read does not need ownership' + }), + now: () => NOW + }) + hosts.push(host) + return host +} + +function sendEnvelope( + store: AgentSessionRecordStore, + fields: Record +): AgentSessionMutationEnvelope { + return { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields + }) + } +} + +/** Persists one turn, then hands back a restarted host over the same directories. */ +async function restartWithPersistedTurn(): Promise { + root = await mkdtemp(join(tmpdir(), 'orca-restart-status-')) + resetHostTestOperationIds() + const directory = join(root, 'store') + const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) + const host = createHost(store) + expect(await host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) + const body = hostTestMessage('persisted conversation') + await host.send(CALLER, { envelope: sendEnvelope(store, { body }), body }) + await host.flushAllStreamedEvents() + return createHost(await AgentSessionRecordStore.open({ directory, hostId: 'local' })) +} + +afterEach(async () => { + await Promise.all(hosts.splice(0).map((host) => host.flushAllStreamedEvents())) + await rm(root, { recursive: true, force: true }) + root = '' +}) + +describe('structured session restart status publication', () => { + // Served by the subscribe-time re-projection rather than the restore's own publish, so this + // covers what a restored journal projects — not the restore wiring. The test below pins that. + it('projects the persisted turn of a session restored without a provider', async () => { + const restarted = await restartWithPersistedTurn() + + await restarted.restoreReadableSessions() + const events: AgentSessionStatusEvent[] = [] + restarted.subscribeStatus({ id: 'session-list', emit: (event) => events.push(event) }) + + expect(events).toEqual([ + { + type: 'snapshot', + sessions: [ + expect.objectContaining({ + sessionId: SESSION, + workspaceId: 'workspace-1', + agent: 'codex', + status: 'idle', + latestPrompt: 'persisted conversation' + }) + ] + } + ]) + }) + + it('publishes a restored session to a list already sitting on the stream', async () => { + const restarted = await restartWithPersistedTurn() + const events: AgentSessionStatusEvent[] = [] + restarted.subscribeStatus({ id: 'session-list', emit: (event) => events.push(event) }) + expect(events).toEqual([{ type: 'snapshot', sessions: [] }]) + + await restarted.restoreReadableSessions() + + // The restore wiring publishes; without it this list never hears about the session at all. + expect(events.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ sessionId: SESSION, status: 'idle' }) + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts new file mode 100644 index 00000000000..7efd147c420 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts @@ -0,0 +1,242 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { AgentSessionStatusEvent } from '../../../shared/agent-session-wire' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' + +const SESSION = 'status-session' +const TURN_IDENTITY = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-1', + ordinal: 0 +} as const +const USER_IDENTITY = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-1', + ordinal: 1 +} as const + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-agent-status-feed-')) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +async function openJournal(sessionId = SESSION) { + return journals.open({ + identity: { + sessionId, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + }, + journalDir: join(root, sessionId) + }) +} + +function indexed(session: { journal: Awaited> }) { + return { + journal: session.journal, + params: { location: { workspaceId: 'workspace-1' }, provider: 'codex' as const } + } +} + +function feedFor(sessions: Map> }>) { + let now = 1_000 + const feed = new StructuredAgentSessionStatusFeed({ + sessions: { + get: (sessionId: string) => { + const session = sessions.get(sessionId) + return session ? indexed(session) : undefined + }, + [Symbol.iterator]: function* () { + for (const [sessionId, session] of sessions) { + yield [sessionId, indexed(session)] as const + } + } + } as unknown as ReadonlyMap>, + getRecord: () => null, + now: () => (now += 1) + }) + const events: AgentSessionStatusEvent[] = [] + const dispose = feed.subscribe({ id: 'list-1', emit: (event) => events.push(event) }) + return { feed, events, dispose } +} + +describe('StructuredAgentSessionStatusFeed', () => { + it('opens with every readable session and reports no status before a persisted turn', async () => { + const journal = await openJournal() + const { events } = feedFor(new Map([[SESSION, { journal }]])) + + expect(events).toEqual([ + { + type: 'snapshot', + sessions: [ + { + sessionId: SESSION, + workspaceId: 'workspace-1', + agent: 'codex', + status: null, + latestPrompt: '', + updatedAt: expect.any(Number) + } + ] + } + ]) + }) + + it('publishes working, then idle once the running marker is tombstoned, and never a repeat', async () => { + const journal = await openJournal() + const { feed, events } = feedFor(new Map([[SESSION, { journal }]])) + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'write a poem' }] }, + { fence: 1 } + ) + await journal.appendItem( + TURN_IDENTITY, + { kind: 'status', text: 'Working', turnLifecycle: { turnId: 'turn-1', state: 'running' } }, + { fence: 1 } + ) + + feed.publish(SESSION) + feed.publish(SESSION) + expect(events.slice(1)).toEqual([ + { + type: 'status', + session: expect.objectContaining({ + sessionId: SESSION, + status: 'working', + latestPrompt: 'write a poem' + }) + } + ]) + + await journal.appendTombstone(TURN_IDENTITY, { fence: 1 }) + feed.publish(SESSION) + expect(events.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ sessionId: SESSION, status: 'idle' }) + }) + expect(events).toHaveLength(3) + }) + + it('reports a pending approval as attention', async () => { + const journal = await openJournal() + const { feed, events } = feedFor(new Map([[SESSION, { journal }]])) + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'run it' }] }, + { fence: 1 } + ) + await journal.appendItem( + TURN_IDENTITY, + { + kind: 'approval', + title: 'Run command?', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 1 } + ) + + feed.publish(SESSION) + expect(events.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'attention' }) + }) + }) + + it('keeps the last projection for an evicted session and serves it to a new subscriber', async () => { + const journal = await openJournal() + const sessions = new Map([[SESSION, { journal }]]) + const { feed, events } = feedFor(sessions) + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] }, + { fence: 1 } + ) + feed.publish(SESSION) + expect(events.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'idle', latestPrompt: 'hello' }) + }) + + // Eviction drops the host's index entry; the projection it already made stays true. + sessions.delete(SESSION) + feed.publish(SESSION) + const late: AgentSessionStatusEvent[] = [] + feed.subscribe({ id: 'list-late', emit: (event) => late.push(event) }) + + expect(events).toHaveLength(2) + expect(late).toEqual([ + { + type: 'snapshot', + sessions: [expect.objectContaining({ sessionId: SESSION, status: 'idle' })] + } + ]) + }) + + it('tells the sitting subscribers about a change a new subscriber re-projected', async () => { + const journal = await openJournal() + const { feed, events } = feedFor(new Map([[SESSION, { journal }]])) + // Journal appends and the feed's publish are separate queue submissions, so the journal + // can already hold the turn when a second client connects and re-projects it. + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] }, + { fence: 1 } + ) + + const late: AgentSessionStatusEvent[] = [] + feed.subscribe({ id: 'list-late', emit: (event) => late.push(event) }) + + expect(events.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'idle', latestPrompt: 'hello' }) + }) + // The arriving subscriber reads that same state once, from its snapshot. + expect(late).toEqual([ + { + type: 'snapshot', + sessions: [expect.objectContaining({ status: 'idle', latestPrompt: 'hello' })] + } + ]) + // The cache is not left holding a value nobody was told about. + feed.publish(SESSION) + expect(events).toHaveLength(2) + }) + + it('ends a closed subscriber and keeps publishing to the rest', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + const others: AgentSessionStatusEvent[] = [] + feed.subscribe({ id: 'list-2', emit: (event) => others.push(event) }) + + dispose() + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] }, + { fence: 1 } + ) + feed.publish(SESSION) + + expect(events.at(-1)).toEqual({ type: 'end' }) + expect(others.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'idle' }) + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts new file mode 100644 index 00000000000..5ad494cf830 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -0,0 +1,130 @@ +// The host's answer to "what is every structured session doing", fanned out to session lists. +// +// A client used to learn whether a turn was running by replaying the journal through its own +// reducer, which tied the answer to whichever surface happened to hold a reader open: hide the +// chat and the sidebar froze on the last thing it had heard. The host always has the journal, so +// it projects the status once per journal publication and sends only the changes. +// +// The last projection is kept after the session's provider child is evicted: an idle session is +// still idle without a process, and a renderer that reloads must not lose every settled row until +// each chat is reopened. Restart is the one boundary that forgets, and restoring readable sessions +// republishes them. + +import { agentProviderSessionsEqual } from '../../../shared/agent-session-resume' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../shared/agent-session-wire' +import { projectStructuredAgentSessionStatusSummary } from '../../../shared/structured-agent-session-projection' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { structuredAgentSessionProviderSessionMetadata } from './structured-agent-session-history-result' + +export type StructuredAgentSessionStatusSubscriber = { + id: string + emit: (event: AgentSessionStatusEvent) => void +} + +type StatusFeedSession = { + journal: AgentSessionJournal + params: { location: { workspaceId: string }; provider: AgentSessionRecord['provider'] } +} + +export type StructuredAgentSessionStatusFeedDeps = { + sessions: ReadonlyMap + getRecord: (sessionId: string) => AgentSessionRecord | null + now: () => number +} + +function summariesEqual(a: AgentSessionStatusSummary, b: AgentSessionStatusSummary): boolean { + return ( + a.workspaceId === b.workspaceId && + a.agent === b.agent && + a.status === b.status && + a.latestPrompt === b.latestPrompt && + agentProviderSessionsEqual(undefined, a.providerSession, b.providerSession) + ) +} + +export class StructuredAgentSessionStatusFeed { + private readonly subscribers = new Map() + private readonly published = new Map() + + constructor(private readonly deps: StructuredAgentSessionStatusFeedDeps) {} + + /** Opens with every session this host has projected, live ones re-read, then only changes. */ + subscribe(subscriber: StructuredAgentSessionStatusSubscriber): () => void { + // Re-project before registering: a change found here has to reach the subscribers that + // already read the old value, and the arriving one carries it in its snapshot instead. + for (const [sessionId] of this.deps.sessions) { + this.publish(sessionId) + } + this.subscribers.set(subscriber.id, subscriber) + this.emit(subscriber, { type: 'snapshot', sessions: [...this.published.values()] }) + return () => this.unsubscribe(subscriber.id) + } + + unsubscribe(id: string): void { + const subscriber = this.subscribers.get(id) + if (!subscriber) { + return + } + this.subscribers.delete(id) + try { + subscriber.emit({ type: 'end' }) + } catch { + // The transport is already gone; teardown must remain idempotent. + } + } + + /** Re-projects one session after its journal changed; equal projections are not re-sent. */ + publish(sessionId: string, journal?: AgentSessionJournal): void { + const session = this.deps.sessions.get(sessionId) + if (!session) { + return + } + const summary = this.summaryFor(sessionId, session, journal ?? session.journal) + const previous = this.published.get(sessionId) + if (previous && summariesEqual(previous, summary)) { + return + } + this.published.set(sessionId, summary) + this.broadcast({ type: 'status', session: summary }) + } + + private summaryFor( + sessionId: string, + session: StatusFeedSession, + journal: AgentSessionJournal + ): AgentSessionStatusSummary { + // An unreadable journal projects as "no turn": the chat itself shows the reset. + const items = journal.isReadOnly ? [] : journal.snapshot().items + const providerSession = structuredAgentSessionProviderSessionMetadata( + this.deps.getRecord(sessionId) + ) + return { + sessionId, + workspaceId: session.params.location.workspaceId, + agent: session.params.provider, + ...projectStructuredAgentSessionStatusSummary(items), + ...(providerSession ? { providerSession } : {}), + updatedAt: this.deps.now() + } + } + + private broadcast(event: AgentSessionStatusEvent): void { + // A Map skips entries deleted mid-iteration, so a failing subscriber can drop itself here. + for (const subscriber of this.subscribers.values()) { + this.emit(subscriber, event) + } + } + + /** A dead transport must not poison every later publication. */ + private emit(subscriber: StructuredAgentSessionStatusSubscriber, event: AgentSessionStatusEvent) { + try { + subscriber.emit(event) + } catch { + this.subscribers.delete(subscriber.id) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts index 6e156d91532..81bcfa82b40 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' import type { AgentSessionHandoffStatus, + AgentSessionStatusEvent, AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' import { @@ -16,6 +17,7 @@ import { journalDatabaseFile } from '../agent-session-journal/journal-paths' import { insertJournalRow } from '../agent-session-journal/journal-row-table' import type { JournalRow } from '../agent-session-journal/journal-row-schema' import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' import { AgentSessionSubscribers } from './structured-agent-session-subscribers' const SESSION = 'subscriber-session' @@ -70,6 +72,96 @@ describe('AgentSessionSubscribers', () => { ]) }) + it('reports every content publication to the journal hook, subscribed or not', async () => { + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + }, + journalDir: join(root, 'hook-journal') + }) + const published: string[] = [] + const subscribers = new AgentSessionSubscribers({ + onJournalPublished: (sessionId, published_journal) => { + expect(published_journal).toBe(journal) + published.push(sessionId) + } + }) + + subscribers.publish(SESSION, journal) + subscribers.reset(SESSION, journal, 'epoch_changed', 1) + subscribers.snapshot(SESSION, journal, 1) + subscribers.handoff(SESSION, 1, { + owner: 'native', + direction: null, + phase: 'idle', + stage: null, + operationId: null + }) + + expect(published).toEqual([SESSION, SESSION, SESSION]) + }) + + it('settles a session nobody is reading, from running to idle', async () => { + // The defect this whole feed exists for: status used to come from a transcript reader, so a + // session with no open pane had no reader and froze on whatever it last said. Nothing here + // ever calls `subscribers.open`. + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + }, + journalDir: join(root, 'unread-journal') + }) + const statusFeed = new StructuredAgentSessionStatusFeed({ + sessions: new Map([ + [ + SESSION, + { journal, params: { location: { workspaceId: 'workspace-1' }, provider: 'codex' } } + ] + ]), + getRecord: () => null, + now: () => 1_000 + }) + const subscribers = new AgentSessionSubscribers({ + onJournalPublished: (sessionId, published) => statusFeed.publish(sessionId, published) + }) + const statuses: AgentSessionStatusEvent[] = [] + statusFeed.subscribe({ id: 'session-list', emit: (event) => statuses.push(event) }) + const turn = { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 } as const + + await journal.appendItem( + { ...turn, ordinal: 1 }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'write a poem' }] }, + { fence: 1 } + ) + await journal.appendItem( + turn, + { kind: 'status', text: 'Working', turnLifecycle: { turnId: 'turn-1', state: 'running' } }, + { fence: 1 } + ) + subscribers.publish(SESSION, journal) + + expect(statuses.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'working', latestPrompt: 'write a poem' }) + }) + + await journal.appendTombstone(turn, { fence: 1 }) + subscribers.publish(SESSION, journal) + + expect(statuses.at(-1)).toEqual({ + type: 'status', + session: expect.objectContaining({ status: 'idle' }) + }) + }) + it('publishes handoff-only changes without serializing a transcript snapshot', async () => { const journal = await journals.open({ identity: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index d3131505384..29dffa6a687 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -36,9 +36,17 @@ type Subscriber = { fence: number } +export type AgentSessionSubscribersHooks = { + /** Fires after any publication that can change journal content, whether or not anyone + * is subscribed to the transcript: session lists project status from this same edge. */ + onJournalPublished?: (sessionId: string, journal: AgentSessionJournal) => void +} + export class AgentSessionSubscribers { private readonly bySession = new Map>() + constructor(private readonly hooks: AgentSessionSubscribersHooks = {}) {} + /** Opens the stream with a bounded tail page or, when the client's cursor * still resolves, with the rows it missed. Returns the disposer. */ open(input: { @@ -99,6 +107,7 @@ export class AgentSessionSubscribers { for (const subscriber of this.subscribers(sessionId)) { this.deliver(subscriber, journal) } + this.hooks.onJournalPublished?.(sessionId, journal) } /** Force every subscriber back to a bounded tail page — recovery, epoch @@ -123,6 +132,7 @@ export class AgentSessionSubscribers { subscriber.cursor = page.liveCursor ?? page.window.nextCursor subscriber.fence = fence } + this.hooks.onJournalPublished?.(sessionId, journal) } snapshot( @@ -143,6 +153,7 @@ export class AgentSessionSubscribers { subscriber.cursor = page.liveCursor ?? page.window.nextCursor subscriber.fence = fence } + this.hooks.onJournalPublished?.(sessionId, journal) } handoff(sessionId: string, fence: number, handoff: AgentSessionHandoffStatus): void { diff --git a/src/main/runtime/rpc/methods/structured-agent-session-status-stream.ts b/src/main/runtime/rpc/methods/structured-agent-session-status-stream.ts new file mode 100644 index 00000000000..8637089c254 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-status-stream.ts @@ -0,0 +1,62 @@ +// `agentSession.subscribeStatus` — every structured session's projected status on one stream. +// +// Session lists read turn state from here instead of replaying transcripts: one stream per client +// covers every session, and unlike a transcript subscription it retains none of them. + +import { defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core' +import { requireStructuredHost as requireHost } from './structured-agent-session-gate' +import { structuredAgentSessionStatusSubscriptionId } from './structured-agent-session-subscription-id' + +/** Ties a stream to both ends that can close it — the runtime's subscription registry and the + * transport abort — so either one runs `onClose` exactly once. */ +export function bindStructuredAgentSessionStream( + ctx: RpcContext, + subscriptionId: string, + onClose: () => void +): { isClosed: () => boolean } { + let closed = false + let releaseTransportSubscription = (): void => {} + const onTransportAbort = (): void => releaseTransportSubscription() + const cleanup = (): void => { + closed = true + ctx.signal?.removeEventListener('abort', onTransportAbort) + onClose() + } + let registration: { releaseIfCurrent: () => void } + if (typeof ctx.runtime.registerOwnedSubscriptionCleanup === 'function') { + registration = ctx.runtime.registerOwnedSubscriptionCleanup( + subscriptionId, + cleanup, + ctx.connectionId + ) + } else { + ctx.runtime.registerSubscriptionCleanup(subscriptionId, cleanup, ctx.connectionId) + registration = { releaseIfCurrent: () => ctx.runtime.cleanupSubscription(subscriptionId) } + } + releaseTransportSubscription = registration.releaseIfCurrent + ctx.signal?.addEventListener('abort', onTransportAbort, { once: true }) + if (ctx.signal?.aborted) { + onTransportAbort() + } + return { isClosed: () => closed } +} + +export const STRUCTURED_AGENT_SESSION_STATUS_METHODS: RpcAnyMethod[] = [ + defineStreamingMethod({ + name: 'agentSession.subscribeStatus', + params: null, + handler: async (_params, ctx, emit) => { + const host = requireHost(ctx) + const subscriptionId = structuredAgentSessionStatusSubscriptionId(ctx) + let dispose = (): void => {} + const stream = bindStructuredAgentSessionStream(ctx, subscriptionId, () => dispose()) + if (stream.isClosed()) { + return + } + dispose = host.subscribeStatus({ id: subscriptionId, emit }) + if (stream.isClosed()) { + dispose() + } + } + }) +] diff --git a/src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts b/src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts new file mode 100644 index 00000000000..f3d00232359 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts @@ -0,0 +1,29 @@ +// Subscription ids for the streaming `agentSession.*` methods. +// +// Shared control multiplexes several streams over one socket, so the frame id keeps one +// subscriber from evicting another. It is appended only when present: collapsing a missing +// frame id to a constant is the collision the rule exists to prevent. + +import type { RpcContext } from '../core' + +const SUBSCRIPTION_PREFIX = 'agentSession' + +function withFrameId(ctx: RpcContext, base: string): string { + return ctx.requestId ? `${base}:${ctx.requestId}` : base +} + +/** The id a session's streams share before the frame id. `unsubscribe` addresses this + * directly and sweeps `${base}:` to reach every frame under it. */ +export function structuredAgentSessionSubscriptionBase(ctx: RpcContext, sessionId: string): string { + return `${SUBSCRIPTION_PREFIX}:${ctx.connectionId ?? 'local'}:${sessionId}` +} + +/** One session's transcript stream. */ +export function structuredAgentSessionSubscriptionId(ctx: RpcContext, sessionId: string): string { + return withFrameId(ctx, structuredAgentSessionSubscriptionBase(ctx, sessionId)) +} + +/** The status feed, which is per connection rather than per session. */ +export function structuredAgentSessionStatusSubscriptionId(ctx: RpcContext): string { + return withFrameId(ctx, `${SUBSCRIPTION_PREFIX}.status:${ctx.connectionId ?? 'local'}`) +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index e4888e4a9df..69b04711960 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -2,8 +2,14 @@ // accepts once they can. import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' +import type { AgentSessionJournal } from '../../../native-chat/agent-session-journal/journal-store' import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host' import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' +import { + StructuredAgentSessionStatusFeed, + type StructuredAgentSessionStatusSubscriber +} from '../../../native-chat/agent-session-wire/structured-agent-session-status-feed' import { RUNTIME_CAPABILITIES, RUNTIME_PROTOCOL_VERSION, @@ -64,6 +70,44 @@ function request(method: string, params: unknown): RpcRequest { let hostCalls: Record> let runtimeCalls: Record> +const STATUS_SESSION = 'session-status' +const STATUS_ITEMS: AgentJournalRenderItem[] = [ + { + itemId: 'user-1', + sequence: 1, + revision: 1, + observedAt: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'write a poem' }] } + }, + { + itemId: 'turn-1', + sequence: 2, + revision: 1, + observedAt: 2, + body: { kind: 'status', text: 'Working', turnLifecycle: { turnId: 'turn-1', state: 'running' } } + } +] + +/** One indexed session over a journal that reads back fixed items; the projection is real. */ +function statusFeed(): StructuredAgentSessionStatusFeed { + return new StructuredAgentSessionStatusFeed({ + sessions: new Map([ + [ + STATUS_SESSION, + { + journal: { + isReadOnly: false, + snapshot: () => ({ items: STATUS_ITEMS }) + } as unknown as AgentSessionJournal, + params: { location: { workspaceId: 'workspace-1' }, provider: 'codex' as const } + } + ] + ]), + getRecord: () => null, + now: () => 1_000 + }) +} + function hostStub(): StructuredAgentSessionHost { hostCalls = { attach: vi.fn(async () => ({ @@ -122,6 +166,11 @@ function hostStub(): StructuredAgentSessionHost { })), history: vi.fn(() => ({ ok: true, page: { items: [] } })), subscribe: vi.fn(() => () => undefined), + // A real feed, so the snapshot this method hands back is a genuine projection rather + // than a shape the stub restated. + subscribeStatus: vi.fn((subscriber: StructuredAgentSessionStatusSubscriber) => + statusFeed().subscribe(subscriber) + ), unsubscribe: vi.fn() } return hostCalls as unknown as StructuredAgentSessionHost @@ -240,7 +289,7 @@ describe('capability gating', () => { } // Bump deliberately: the whole agentSession.* surface is behind the structured capability, // so an additive method is invisible to old clients and needs no protocol bump. - expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(17) + expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(18) }) it('hides the surface from a declared client that did not advertise it', async () => { @@ -583,3 +632,31 @@ describe('parameter validation', () => { expect(response).toMatchObject({ ok: true }) }) }) + +describe('agentSession.subscribeStatus', () => { + it('is invisible to a client without the structured capability', async () => { + const reply = await call('agentSession.subscribeStatus', null, { clientKind: 'runtime' }) + expect(reply.ok).toBe(false) + expect(hostCalls.subscribeStatus).not.toHaveBeenCalled() + }) + + it('opens the host status feed with a projected snapshot as its first reply', async () => { + const reply = await call('agentSession.subscribeStatus', null, STRUCTURED_CLIENT) + expect(reply).toMatchObject({ + ok: true, + result: { + type: 'snapshot', + sessions: [ + { + sessionId: STATUS_SESSION, + workspaceId: 'workspace-1', + agent: 'codex', + status: 'working', + latestPrompt: 'write a poem' + } + ] + } + }) + expect(hostCalls.subscribeStatus).toHaveBeenCalledOnce() + }) +}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index b69ff6fd628..61b0f4dcf4f 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -21,6 +21,14 @@ import { import type { AgentSessionAttachParams } from '../../../native-chat/agent-session-wire/structured-agent-session-attach' import { STRUCTURED_AGENT_SESSION_HOLD_METHODS } from './structured-agent-session-hold' import { resolveUncommittedStructuredCreate } from './structured-agent-session-precommit-refusal' +import { + bindStructuredAgentSessionStream, + STRUCTURED_AGENT_SESSION_STATUS_METHODS +} from './structured-agent-session-status-stream' +import { + structuredAgentSessionSubscriptionBase as subscriptionBaseFor, + structuredAgentSessionSubscriptionId as subscriptionIdFor +} from './structured-agent-session-subscription-id' import { AttachParams, CancelParams, @@ -37,15 +45,6 @@ import { UnsubscribeParams } from './structured-agent-session-schemas' -const SUBSCRIPTION_PREFIX = 'agentSession' - -function subscriptionIdFor(ctx: RpcContext, sessionId: string): string { - const base = `${SUBSCRIPTION_PREFIX}:${ctx.connectionId ?? 'local'}:${sessionId}` - // Shared control multiplexes several streams over one socket; the frame id - // keeps one subscriber from evicting another on the same session. - return ctx.requestId ? `${base}:${ctx.requestId}` : base -} - /** * The attach-shaped entries take the location from the client instead of resolving it from a * worktree, so they never reach the worktree-resolving create-support check. Ask the executing @@ -245,33 +244,12 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ // Retain-only: reading history must never be what starts a provider process. Current clients // explicitly hold every open surface before subscribing. const streamHolder = `subscription:${subscriptionId}` - let closed = false let dispose = (): void => {} - let releaseTransportSubscription = (): void => {} - const onTransportAbort = (): void => releaseTransportSubscription() - const cleanup = () => { - closed = true - ctx.signal?.removeEventListener('abort', onTransportAbort) + const stream = bindStructuredAgentSessionStream(ctx, subscriptionId, () => { dispose() host.release(params.sessionId, streamHolder) - } - let registration: { releaseIfCurrent: () => void } - if (typeof ctx.runtime.registerOwnedSubscriptionCleanup === 'function') { - registration = ctx.runtime.registerOwnedSubscriptionCleanup( - subscriptionId, - cleanup, - ctx.connectionId - ) - } else { - ctx.runtime.registerSubscriptionCleanup(subscriptionId, cleanup, ctx.connectionId) - registration = { releaseIfCurrent: () => ctx.runtime.cleanupSubscription(subscriptionId) } - } - releaseTransportSubscription = registration.releaseIfCurrent - ctx.signal?.addEventListener('abort', onTransportAbort, { once: true }) - if (ctx.signal?.aborted) { - onTransportAbort() - } - if (closed) { + }) + if (stream.isClosed()) { return } // The host emits the opening snapshot (or the missed batch) synchronously @@ -282,7 +260,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ emit, ...(params.cursor ? { cursor: params.cursor } : {}) }) - if (closed) { + if (stream.isClosed()) { dispose() } else { // Fire-and-forget, but never unhandled: a resume that refuses leaves the stream holding a @@ -300,8 +278,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ params: UnsubscribeParams, handler: async (params, ctx) => { requireHost(ctx) - const connection = ctx.connectionId ?? 'local' - const base = `${SUBSCRIPTION_PREFIX}:${connection}:${params.sessionId}` + const base = subscriptionBaseFor(ctx, params.sessionId) if (params.subscriptionId) { ctx.runtime.cleanupSubscription(`${base}:${params.subscriptionId}`) return { unsubscribed: true } @@ -311,5 +288,6 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ return { unsubscribed: true } } }), - ...STRUCTURED_AGENT_SESSION_HOLD_METHODS + ...STRUCTURED_AGENT_SESSION_HOLD_METHODS, + ...STRUCTURED_AGENT_SESSION_STATUS_METHODS ] diff --git a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx index 8a61399d669..a8eb22ae7a5 100644 --- a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx +++ b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx @@ -2,17 +2,23 @@ import { act, cleanup, render, waitFor } from '@testing-library/react' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../../shared/agent-session-wire' import type { Tab } from '../../../../shared/tab-types' +import type * as RuntimeRpcClientModule from '@/runtime/runtime-rpc-client' const mocks = vi.hoisted(() => ({ - call: vi.fn(), removeAgentStatus: vi.fn(), setAgentStatus: vi.fn(), store: null as null | { getState: () => Record setState: (state: Record) => void }, - subscribe: vi.fn(), + subscribeStatus: vi.fn(), + subscribeTranscript: vi.fn(), + supportsCapability: vi.fn(), unsubscribe: vi.fn() })) @@ -73,17 +79,22 @@ vi.mock('@/lib/worktree-runtime-owner', () => ({ state.testRuntimeOwner ?? null })) +vi.mock('@/runtime/runtime-rpc-client', async (importOriginal) => ({ + ...(await importOriginal()), + runtimeEnvironmentSupportsCapability: mocks.supportsCapability +})) + vi.mock('@/runtime/structured-agent-session-client', () => ({ - callStructuredAgentSession: mocks.call, - subscribeStructuredAgentSession: mocks.subscribe + callStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSession: mocks.subscribeTranscript, + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus })) import { getStructuredAgentSessionTabs, StructuredAgentSessionStatusBridge } from './StructuredAgentSessionStatusBridge' -import { resetStructuredAgentSessionReadOwnersForTests } from './structured-agent-session-read-owner' -import { useStructuredAgentSessionRead } from './use-structured-agent-session-read' +import { resetStructuredAgentSessionStatusFeedsForTests } from '@/runtime/structured-agent-session-status-feed' const structuredTab = { id: 'structured-tab-1', @@ -100,60 +111,40 @@ const structuredTab = { agentSessionAgent: 'codex' } satisfies Tab -const userItem = { - itemId: 'item-1', - revision: 1, - sequence: 1, - observedAt: 1, - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] } -} as const +const providerSession = { key: 'session_id', id: '01a002e9-9a1c-7d42-a642-e481f64446f1' } as const -const historyResult = { - ok: true, - providerSession: { key: 'session_id', id: '01a002e9-9a1c-7d42-a642-e481f64446f1' }, - page: { +function summary(overrides: Partial = {}): AgentSessionStatusSummary { + return { sessionId: 'session-1', - epoch: 'epoch-1', - fence: 1, - direction: 'tail', - items: [userItem], - removedItemIds: [], - submissions: [], - window: { - oldest: { epoch: 'epoch-1', sequence: 1 }, - newest: { epoch: 'epoch-1', sequence: 1 }, - nextCursor: { epoch: 'epoch-1', sequence: 1 } - }, - liveCursor: { epoch: 'epoch-1', sequence: 1 }, - hasOlder: false, - hasNewer: false + workspaceId: 'wt-1', + agent: 'codex', + status: 'working', + latestPrompt: 'hello', + providerSession, + updatedAt: 1, + ...overrides } } -function ActiveSessionRead(): null { - useStructuredAgentSessionRead({ - sessionId: structuredTab.entityId, - target: { kind: 'local' }, - isVisible: true - }) - return null +function statuses(): Record[] { + return Object.values(mocks.store?.getState().agentStatusByPaneKey ?? {}) } -function ActiveComposition(): React.JSX.Element { - return ( - <> - - - - ) +/** The host side of the most recent status subscription. */ +function feed(index = 0): { target: unknown; emit: (event: AgentSessionStatusEvent) => void } { + const call = mocks.subscribeStatus.mock.calls[index] + if (!call) { + throw new Error('status feed not subscribed') + } + return { target: call[0], emit: call[1] as (event: AgentSessionStatusEvent) => void } } describe('StructuredAgentSessionStatusBridge', () => { beforeEach(() => { vi.clearAllMocks() - resetStructuredAgentSessionReadOwnersForTests() - mocks.call.mockResolvedValue(historyResult) - mocks.subscribe.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + resetStructuredAgentSessionStatusFeedsForTests() + mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + mocks.supportsCapability.mockResolvedValue(true) mocks.store?.setState({ agentStatusByPaneKey: {}, testRuntimeOwner: null, @@ -163,7 +154,7 @@ describe('StructuredAgentSessionStatusBridge', () => { afterEach(() => { cleanup() - resetStructuredAgentSessionReadOwnersForTests() + resetStructuredAgentSessionStatusFeedsForTests() }) it('reuses the structured-tab projection for an unchanged tab map', () => { @@ -194,65 +185,111 @@ describe('StructuredAgentSessionStatusBridge', () => { ]) }) - it('keeps restored inactive tabs transport-neutral', async () => { + it('projects the host status feed without opening a transcript reader', async () => { render() - await act(() => Promise.resolve()) + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + expect(feed().target).toEqual({ kind: 'local' }) + expect(mocks.subscribeTranscript).not.toHaveBeenCalled() - expect(mocks.call).not.toHaveBeenCalled() - expect(mocks.subscribe).not.toHaveBeenCalled() + act(() => feed().emit({ type: 'snapshot', sessions: [summary()] })) + + expect(statuses()).toEqual([ + expect.objectContaining({ + state: 'working', + prompt: 'hello', + agentType: 'codex', + sessionBoundary: false, + tabId: structuredTab.id, + worktreeId: 'wt-1', + terminalTitle: 'Codex Chat', + terminalResumeEligible: false, + providerSession + }) + ]) + }) + + // Hiddenness is the host's side of this: see structured-agent-session-subscribers.test.ts, + // which drives an unsubscribed journal through the feed. Here the transport is a mock, so + // only the summary-to-store mapping is under test. + it('maps each host status onto the sidebar agent state', async () => { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + act(() => feed().emit({ type: 'snapshot', sessions: [summary()] })) + expect(statuses()).toEqual([expect.objectContaining({ state: 'working' })]) + + act(() => feed().emit({ type: 'status', session: summary({ status: 'idle', updatedAt: 2 }) })) + expect(statuses()).toEqual([expect.objectContaining({ state: 'done', sessionBoundary: true })]) + + act(() => + feed().emit({ type: 'status', session: summary({ status: 'attention', updatedAt: 3 }) }) + ) + expect(statuses()).toEqual([expect.objectContaining({ state: 'blocked' })]) + }) + + it('shows no status before a persisted turn', async () => { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + + act(() => feed().emit({ type: 'snapshot', sessions: [summary({ status: null })] })) expect(mocks.setAgentStatus).not.toHaveBeenCalled() + + act(() => feed().emit({ type: 'status', session: summary({ updatedAt: 2 }) })) + expect(statuses()).toEqual([expect.objectContaining({ state: 'working' })]) }) - it('shares the visible pane subscriber with status projection', async () => { - render() - - await waitFor(() => expect(mocks.setAgentStatus).toHaveBeenCalledOnce()) - expect(mocks.call).toHaveBeenCalledOnce() - expect(mocks.subscribe).toHaveBeenCalledOnce() - expect(mocks.setAgentStatus.mock.calls[0]?.[5]).toEqual({ - providerSession: historyResult.providerSession, - terminalResumeEligible: false - }) - }) - - it('keeps the status map reference stable for coalesced assistant deltas', async () => { - render() - await waitFor(() => expect(mocks.setAgentStatus).toHaveBeenCalledOnce()) + it('keeps the status map reference stable for repeated equal summaries', async () => { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + act(() => feed().emit({ type: 'snapshot', sessions: [summary()] })) const before = mocks.store?.getState().agentStatusByPaneKey - const onEvent = mocks.subscribe.mock.calls[0]?.[2] as (event: unknown) => void act(() => { - for (let sequence = 2; sequence <= 12; sequence += 1) { - onEvent({ - type: 'batch', - sessionId: 'session-1', - batch: { - cursor: { epoch: 'epoch-1', sequence }, - items: [ - { - itemId: 'assistant-1', - revision: sequence, - sequence, - observedAt: sequence, - body: { - kind: 'message', - role: 'assistant', - blocks: [{ type: 'text', text: `delta-${sequence}` }] - } - } - ], - removedItemIds: [], - submissions: [] - } - }) + for (let updatedAt = 2; updatedAt <= 12; updatedAt += 1) { + feed().emit({ type: 'status', session: summary({ updatedAt }) }) } }) - await act(async () => new Promise((resolve) => setTimeout(resolve, 60))) expect(mocks.setAgentStatus).toHaveBeenCalledOnce() expect(mocks.store?.getState().agentStatusByPaneKey).toBe(before) }) + it('drops the status and the feed when the last structured tab closes', async () => { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + act(() => feed().emit({ type: 'snapshot', sessions: [summary()] })) + expect(statuses()).toHaveLength(1) + + act(() => mocks.store?.setState({ unifiedTabsByWorktree: { 'wt-1': [] } })) + + expect(statuses()).toEqual([]) + await waitFor(() => expect(mocks.unsubscribe).toHaveBeenCalledOnce()) + }) + + it('reconnects after the host ends the stream', async () => { + vi.useFakeTimers() + try { + render() + await act(() => Promise.resolve()) + expect(mocks.subscribeStatus).toHaveBeenCalledOnce() + + act(() => feed().emit({ type: 'end' })) + await act(() => vi.advanceTimersByTimeAsync(300)) + + expect(mocks.unsubscribe).toHaveBeenCalledOnce() + expect(mocks.subscribeStatus).toHaveBeenCalledTimes(2) + } finally { + vi.useRealTimers() + } + }) + + it('keys the feed by the worktree runtime environment', async () => { + mocks.store?.setState({ testRuntimeOwner: 'env-1' }) + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + + expect(feed().target).toEqual({ kind: 'environment', environmentId: 'env-1' }) + }) + it('does not project an unknown provider as Codex', async () => { mocks.store?.setState({ unifiedTabsByWorktree: { @@ -262,8 +299,7 @@ describe('StructuredAgentSessionStatusBridge', () => { render() await act(() => Promise.resolve()) - expect(mocks.call).not.toHaveBeenCalled() - expect(mocks.subscribe).not.toHaveBeenCalled() + expect(mocks.subscribeStatus).not.toHaveBeenCalled() expect(mocks.setAgentStatus).not.toHaveBeenCalled() }) }) diff --git a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx index 8409858dd17..d4cb74ab93b 100644 --- a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx +++ b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx @@ -1,19 +1,14 @@ -import { useEffect, useMemo } from 'react' +import { useEffect, useMemo, useSyncExternalStore } from 'react' import { useShallow } from 'zustand/react/shallow' -import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' import { agentProviderSessionsEqual } from '../../../../shared/agent-session-resume' -import { - hasPersistedStructuredAgentSessionTurn, - projectStructuredAgentSessionStatus, - structuredAgentSessionPaneKey -} from '../../../../shared/structured-agent-session-projection' -import type { StructuredAgentSessionState } from '../../../../shared/structured-agent-session-reducer' +import type { AgentSessionStatusSummary } from '../../../../shared/agent-session-wire' +import { structuredAgentSessionPaneKey } from '../../../../shared/structured-agent-session-projection' import type { Tab } from '../../../../shared/tab-types' import { isAgentSessionHandleProvider } from '../../../../shared/agent-session-provider-handle' import { getRuntimeEnvironmentIdForWorktree } from '@/lib/worktree-runtime-owner' import { useAppStore } from '@/store' -import { getActiveRuntimeTarget } from '@/runtime/runtime-rpc-client' -import { useStructuredAgentSessionReadObservation } from './use-structured-agent-session-read' +import { getActiveRuntimeTarget, type RuntimeClientTarget } from '@/runtime/runtime-rpc-client' +import { getStructuredAgentSessionStatusFeed } from '@/runtime/structured-agent-session-status-feed' type StructuredTab = Tab & { contentType: 'agent-session' } @@ -47,35 +42,40 @@ export function getStructuredAgentSessionTabs( return tabs } -function latestPrompt(state: StructuredAgentSessionState): string { - for (let index = state.items.length - 1; index >= 0; index -= 1) { - const body = state.items[index]?.body - if (body?.kind === 'message' && body.role === 'user') { - return body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n') - } - } - return '' +/** The host's projected status for one session, live while the caller is mounted. */ +function useStructuredAgentSessionStatusSummary( + sessionId: string, + target: RuntimeClientTarget +): AgentSessionStatusSummary | null { + const feed = useMemo(() => getStructuredAgentSessionStatusFeed(target), [target]) + useEffect(() => feed.activate(), [feed]) + return useSyncExternalStore( + feed.subscribe, + () => feed.getSnapshot().get(sessionId) ?? null, + () => null + ) } -function projectStatus( - tab: StructuredTab, - state: StructuredAgentSessionState, - providerSession: AgentProviderSessionMetadata | undefined -): void { +function projectStatus(tab: StructuredTab, summary: AgentSessionStatusSummary | null): void { const paneKey = structuredAgentSessionPaneKey(tab.id, tab.entityId) const store = useAppStore.getState() - if (!hasPersistedStructuredAgentSessionTurn(state.items)) { + // No persisted turn yet (or nothing known): the row shows no agent status at all. + if (!summary?.status) { if (store.agentStatusByPaneKey?.[paneKey]) { store.removeAgentStatus(paneKey) } return } - const projection = projectStructuredAgentSessionStatus(state.items) const desired = { - state: projection === 'working' ? 'working' : projection === 'attention' ? 'blocked' : 'done', - prompt: latestPrompt(state), + state: + summary.status === 'working' + ? 'working' + : summary.status === 'attention' + ? 'blocked' + : 'done', + prompt: summary.latestPrompt, agentType: tab.agentSessionAgent, - sessionBoundary: projection === 'idle' + sessionBoundary: summary.status === 'idle' } as const const current = store.agentStatusByPaneKey?.[paneKey] if ( @@ -87,7 +87,11 @@ function projectStatus( current.tabId === tab.id && current.worktreeId === tab.worktreeId && current.terminalResumeEligible === false && - agentProviderSessionsEqual(tab.agentSessionAgent, current.providerSession, providerSession) + agentProviderSessionsEqual( + tab.agentSessionAgent, + current.providerSession, + summary.providerSession + ) ) { return } @@ -98,7 +102,7 @@ function projectStatus( undefined, { tabId: tab.id, worktreeId: tab.worktreeId }, { - ...(providerSession ? { providerSession } : {}), + ...(summary.providerSession ? { providerSession: summary.providerSession } : {}), terminalResumeEligible: false } ) @@ -112,13 +116,10 @@ function StructuredAgentSessionStatusProjection({ tab }: { tab: StructuredTab }) () => getActiveRuntimeTarget({ activeRuntimeEnvironmentId: environmentId }), [environmentId] ) - const { providerSession, state } = useStructuredAgentSessionReadObservation({ - sessionId: tab.entityId, - target - }) + const summary = useStructuredAgentSessionStatusSummary(tab.entityId, target) useEffect(() => { - projectStatus(tab, state, providerSession) - }, [providerSession, state, tab]) + projectStatus(tab, summary) + }, [summary, tab]) useEffect( () => () => useAppStore.getState().removeAgentStatus(structuredAgentSessionPaneKey(tab.id, tab.entityId)), diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx index 2d320694c9c..9e1ce8067b7 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx @@ -19,10 +19,7 @@ vi.mock('@/runtime/structured-agent-session-client', () => ({ subscribeStructuredAgentSession: mocks.subscribe })) -import { - useStructuredAgentSessionRead, - useStructuredAgentSessionReadObservation -} from './use-structured-agent-session-read' +import { useStructuredAgentSessionRead } from './use-structured-agent-session-read' import { resetStructuredAgentSessionReadOwnersForTests } from './structured-agent-session-read-owner' const LOCAL_TARGET = { kind: 'local' } as const @@ -357,32 +354,6 @@ describe('useStructuredAgentSessionRead history window', () => { second.unmount() }) - it('shares one subscriber when pane and projection observe the same visible session', async () => { - const unsubscribe = vi.fn() - mocks.call.mockResolvedValue({ ok: true, page: page('tail', [], false) }) - mocks.subscribe.mockResolvedValue({ unsubscribe }) - - const view = renderHook(() => { - const pane = useStructuredAgentSessionRead({ - sessionId: 'session-shared', - target: LOCAL_TARGET, - isVisible: true - }) - const projection = useStructuredAgentSessionReadObservation({ - sessionId: 'session-shared', - target: LOCAL_TARGET - }) - return { pane, projection } - }) - - await waitFor(() => expect(mocks.subscribe).toHaveBeenCalledOnce()) - expect(mocks.call).toHaveBeenCalledOnce() - expect(view.result.current.pane.state).toBe(view.result.current.projection.state) - - view.unmount() - expect(unsubscribe).toHaveBeenCalledOnce() - }) - it('preserves cached state while switching away and refreshes once on re-entry', async () => { const unsubscribe = vi.fn() mocks.call.mockImplementation((_target, _method, params) => { diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts index 894730f5e65..834d11d3fa1 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts @@ -20,13 +20,6 @@ function useReadOwnerSnapshot( return { owner, snapshot } } -export function useStructuredAgentSessionReadObservation(args: { - sessionId: string - target: RuntimeClientTarget -}): StructuredAgentSessionReadSnapshot { - return useReadOwnerSnapshot(args.sessionId, args.target).snapshot -} - export function useStructuredAgentSessionRead(args: { sessionId: string target: RuntimeClientTarget diff --git a/src/renderer/src/runtime/structured-agent-session-client.ts b/src/renderer/src/runtime/structured-agent-session-client.ts index 0e8d2ce16f2..71be3f3449d 100644 --- a/src/renderer/src/runtime/structured-agent-session-client.ts +++ b/src/renderer/src/runtime/structured-agent-session-client.ts @@ -1,5 +1,8 @@ import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope' -import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import type { + AgentSessionStatusEvent, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' import { getRuntimeEnvironmentRevision } from './runtime-environment-revision' import { callRuntimeRpc, type RuntimeClientTarget } from './runtime-rpc-client' @@ -11,10 +14,11 @@ export function callStructuredAgentSession( return callRuntimeRpc(target, method, params) } -export async function subscribeStructuredAgentSession( +async function subscribeStructuredAgentSessionMethod( target: RuntimeClientTarget, + method: string, params: unknown, - onEvent: (event: AgentSessionSubscribeEvent) => void, + onEvent: (event: TEvent) => void, onError: (error: unknown) => void, onClose: () => void ): Promise<{ unsubscribe: () => void }> { @@ -23,15 +27,15 @@ export async function subscribeStructuredAgentSession( onError(response.error) return } - onEvent(response.result as AgentSessionSubscribeEvent) + onEvent(response.result as TEvent) } if (target.kind === 'local') { - return window.api.runtime.subscribe({ method: 'agentSession.subscribe', params }, onResponse) + return window.api.runtime.subscribe({ method, params }, onResponse) } return window.api.runtimeEnvironments.subscribe( { selector: target.environmentId, - method: 'agentSession.subscribe', + method, params, timeoutMs: 15_000, expectedEnvironmentPairingRevision: getRuntimeEnvironmentRevision(target.environmentId) @@ -39,3 +43,37 @@ export async function subscribeStructuredAgentSession( { onResponse, onError, onClose } ) } + +export function subscribeStructuredAgentSession( + target: RuntimeClientTarget, + params: unknown, + onEvent: (event: AgentSessionSubscribeEvent) => void, + onError: (error: unknown) => void, + onClose: () => void +): Promise<{ unsubscribe: () => void }> { + return subscribeStructuredAgentSessionMethod( + target, + 'agentSession.subscribe', + params, + onEvent, + onError, + onClose + ) +} + +/** Every structured session's projected status on one runtime, as the host publishes it. */ +export function subscribeStructuredAgentSessionStatus( + target: RuntimeClientTarget, + onEvent: (event: AgentSessionStatusEvent) => void, + onError: (error: unknown) => void, + onClose: () => void +): Promise<{ unsubscribe: () => void }> { + return subscribeStructuredAgentSessionMethod( + target, + 'agentSession.subscribeStatus', + {}, + onEvent, + onError, + onClose + ) +} diff --git a/src/renderer/src/runtime/structured-agent-session-status-feed.test.ts b/src/renderer/src/runtime/structured-agent-session-status-feed.test.ts new file mode 100644 index 00000000000..d9cb4dd3d72 --- /dev/null +++ b/src/renderer/src/runtime/structured-agent-session-status-feed.test.ts @@ -0,0 +1,140 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../shared/agent-session-wire' +import { AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' + +const mocks = vi.hoisted(() => ({ + subscribeStatus: vi.fn(), + supportsCapability: vi.fn(), + unsubscribe: vi.fn() +})) + +vi.mock('./structured-agent-session-client', () => ({ + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus +})) + +vi.mock('./runtime-rpc-client', () => ({ + runtimeEnvironmentSupportsCapability: mocks.supportsCapability +})) + +import { + getStructuredAgentSessionStatusFeed, + resetStructuredAgentSessionStatusFeedsForTests +} from './structured-agent-session-status-feed' + +const REMOTE = { kind: 'environment', environmentId: 'env-1' } as const +const LOCAL = { kind: 'local' } as const + +function summary( + sessionId: string, + status: AgentSessionStatusSummary['status'] = 'idle' +): AgentSessionStatusSummary { + return { + sessionId, + workspaceId: 'wt-1', + agent: 'codex', + status, + latestPrompt: 'hello', + updatedAt: 1 + } +} + +/** The event callback the feed handed to the most recent subscription. */ +function hostEmit(index = 0): (event: AgentSessionStatusEvent) => void { + const call = mocks.subscribeStatus.mock.calls[index] + if (!call) { + throw new Error('status feed not subscribed') + } + return call[1] as (event: AgentSessionStatusEvent) => void +} + +describe('structured agent session status feed', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.clearAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + mocks.supportsCapability.mockResolvedValue(true) + }) + + afterEach(() => { + resetStructuredAgentSessionStatusFeedsForTests() + vi.useRealTimers() + }) + + it('never subscribes, and never retries, against a host without the status feed', async () => { + mocks.supportsCapability.mockResolvedValue(false) + getStructuredAgentSessionStatusFeed(REMOTE).activate() + + await vi.advanceTimersByTimeAsync(0) + expect(mocks.supportsCapability).toHaveBeenCalledWith( + 'env-1', + AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY + ) + expect(mocks.subscribeStatus).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + + await vi.advanceTimersByTimeAsync(60_000) + expect(mocks.subscribeStatus).not.toHaveBeenCalled() + expect(mocks.supportsCapability).toHaveBeenCalledOnce() + }) + + it('subscribes once the remote host advertises the status feed', async () => { + getStructuredAgentSessionStatusFeed(REMOTE).activate() + + await vi.advanceTimersByTimeAsync(0) + expect(mocks.subscribeStatus).toHaveBeenCalledOnce() + expect(mocks.subscribeStatus.mock.calls[0]?.[0]).toEqual(REMOTE) + }) + + it('reconnects when the capability probe fails, which is not an answer', async () => { + mocks.supportsCapability.mockRejectedValue(new Error('relay unreachable')) + getStructuredAgentSessionStatusFeed(REMOTE).activate() + + await vi.advanceTimersByTimeAsync(0) + expect(mocks.supportsCapability).toHaveBeenCalledOnce() + + await vi.advanceTimersByTimeAsync(300) + expect(mocks.supportsCapability).toHaveBeenCalledTimes(2) + expect(mocks.subscribeStatus).not.toHaveBeenCalled() + }) + + it('probes nothing for a local host, which is this build', async () => { + getStructuredAgentSessionStatusFeed(LOCAL).activate() + + await vi.advanceTimersByTimeAsync(0) + expect(mocks.supportsCapability).not.toHaveBeenCalled() + expect(mocks.subscribeStatus).toHaveBeenCalledOnce() + }) + + it('merges a snapshot over the cached rows instead of retracting them', async () => { + const feed = getStructuredAgentSessionStatusFeed(LOCAL) + feed.activate() + await vi.advanceTimersByTimeAsync(0) + hostEmit()({ type: 'snapshot', sessions: [summary('session-1'), summary('session-2')] }) + expect([...feed.getSnapshot().keys()]).toEqual(['session-1', 'session-2']) + + // A restarted host restores its readable sessions after the stream reopens. + hostEmit()({ type: 'snapshot', sessions: [] }) + expect([...feed.getSnapshot().keys()]).toEqual(['session-1', 'session-2']) + + hostEmit()({ type: 'snapshot', sessions: [summary('session-1', 'working')] }) + expect(feed.getSnapshot().get('session-1')?.status).toBe('working') + expect(feed.getSnapshot().get('session-2')?.status).toBe('idle') + }) + + it('stops a pending reconnect when the feeds are reset between tests', async () => { + getStructuredAgentSessionStatusFeed(LOCAL).activate() + await vi.advanceTimersByTimeAsync(0) + hostEmit()({ type: 'end' }) + expect(vi.getTimerCount()).toBe(1) + + resetStructuredAgentSessionStatusFeedsForTests() + + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(10_000) + expect(mocks.subscribeStatus).toHaveBeenCalledOnce() + }) +}) diff --git a/src/renderer/src/runtime/structured-agent-session-status-feed.ts b/src/renderer/src/runtime/structured-agent-session-status-feed.ts new file mode 100644 index 00000000000..2b550679315 --- /dev/null +++ b/src/renderer/src/runtime/structured-agent-session-status-feed.ts @@ -0,0 +1,211 @@ +// One host status stream per runtime target, shared by every session-list projection. +// +// The feed is a read-only mirror: the host projects each session's status from its journal and +// this owner keeps the latest summary per session while anyone is looking. Losing the stream +// keeps the cached summaries and reconnects; a fresh snapshot merges over them. +// Which sessions are listed is the tab map's decision, so the feed never retracts a summary. + +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../shared/agent-session-wire' +import { AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' +import { + runtimeEnvironmentSupportsCapability, + type RuntimeClientTarget +} from './runtime-rpc-client' +import { subscribeStructuredAgentSessionStatus } from './structured-agent-session-client' + +export type StructuredAgentSessionStatusSnapshot = ReadonlyMap + +export type StructuredAgentSessionStatusFeedOwner = { + activate: () => () => void + getSnapshot: () => StructuredAgentSessionStatusSnapshot + subscribe: (listener: () => void) => () => void +} + +const RECONNECT_MAX_DELAY_MS = 5_000 + +/** `stop` is the map's own teardown, not part of the owner contract callers hold. */ +type OwnedStatusFeed = StructuredAgentSessionStatusFeedOwner & { stop: () => void } + +const owners = new Map() + +export function structuredAgentSessionStatusFeedKey(target: RuntimeClientTarget): string { + return target.kind === 'local' ? 'local' : `environment:${target.environmentId}` +} + +function createOwner(target: RuntimeClientTarget): OwnedStatusFeed { + let snapshot: StructuredAgentSessionStatusSnapshot = new Map() + const listeners = new Set<() => void>() + const activations = new Set() + let generation = 0 + let handle: { unsubscribe: () => void } | null = null + let reconnectTimer: ReturnType | null = null + let reconnectAttempt = 0 + + const emit = (): void => { + for (const listener of listeners) { + listener() + } + } + const setSnapshot = (next: StructuredAgentSessionStatusSnapshot): void => { + snapshot = next + emit() + } + const applyEvent = (event: AgentSessionStatusEvent): void => { + if (event.type === 'snapshot') { + reconnectAttempt = 0 + // Merged, not replaced: a restarted host restores its readable sessions asynchronously, so + // the first snapshot can be empty and dropping those rows flickers every one to no-status. + const next = new Map(snapshot) + for (const session of event.sessions) { + next.set(session.sessionId, session) + } + setSnapshot(next) + return + } + if (event.type === 'status') { + const next = new Map(snapshot) + next.set(event.session.sessionId, event.session) + setSnapshot(next) + } + } + const active = (candidate: number): boolean => activations.size > 0 && candidate === generation + const clearReconnect = (): void => { + if (reconnectTimer) { + clearTimeout(reconnectTimer) + reconnectTimer = null + } + } + const dropHandle = (): void => { + handle?.unsubscribe() + handle = null + } + let open = (): void => {} + const scheduleReconnect = (candidate: number): void => { + if (!active(candidate) || reconnectTimer) { + return + } + const delay = Math.min(250 * 2 ** reconnectAttempt, RECONNECT_MAX_DELAY_MS) + reconnectAttempt += 1 + reconnectTimer = setTimeout(() => { + reconnectTimer = null + if (active(candidate)) { + open() + } + }, delay) + } + const subscribeToHost = (candidate: number): void => { + void subscribeStructuredAgentSessionStatus( + target, + (event) => { + if (!active(candidate)) { + return + } + if (event.type === 'end') { + dropHandle() + scheduleReconnect(candidate) + return + } + applyEvent(event) + }, + () => { + if (active(candidate)) { + dropHandle() + scheduleReconnect(candidate) + } + }, + () => { + if (active(candidate)) { + dropHandle() + scheduleReconnect(candidate) + } + } + ) + .then((opened) => { + if (active(candidate)) { + handle = opened + } else { + opened.unsubscribe() + } + }) + .catch(() => scheduleReconnect(candidate)) + } + open = (): void => { + const candidate = ++generation + dropHandle() + if (target.kind !== 'environment') { + // A local host is this build; only a remote one can predate the method. + subscribeToHost(candidate) + return + } + const environmentId = target.environmentId + void runtimeEnvironmentSupportsCapability( + environmentId, + AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY + ) + .then((supported) => { + if (!active(candidate)) { + return + } + // A host without the method is terminal, not a fault: retrying would relay-probe + // forever. A failed probe is not an answer, so that path still reconnects. + if (supported) { + subscribeToHost(candidate) + return + } + console.warn('[structured-session-status] host too old for the status feed', environmentId) + }) + .catch(() => scheduleReconnect(candidate)) + } + const stop = (): void => { + generation += 1 + clearReconnect() + dropHandle() + reconnectAttempt = 0 + } + + return { + activate: () => { + const token = Symbol('status-feed') + activations.add(token) + if (activations.size === 1) { + open() + } + return () => { + activations.delete(token) + if (activations.size === 0) { + stop() + } + } + }, + getSnapshot: () => snapshot, + subscribe: (listener) => { + listeners.add(listener) + return () => listeners.delete(listener) + }, + stop + } +} + +export function getStructuredAgentSessionStatusFeed( + target: RuntimeClientTarget +): StructuredAgentSessionStatusFeedOwner { + const key = structuredAgentSessionStatusFeedKey(target) + let owner = owners.get(key) + if (!owner) { + owner = createOwner(target) + owners.set(key, owner) + } + return owner +} + +export function resetStructuredAgentSessionStatusFeedsForTests(): void { + // Dropping the map alone leaves a live subscription and its pending reconnect running + // into the next test, where they reopen a stream nothing is holding. + for (const owner of owners.values()) { + owner.stop() + } + owners.clear() +} diff --git a/src/shared/agent-session-wire.ts b/src/shared/agent-session-wire.ts index 40a408df899..86c0e795d84 100644 --- a/src/shared/agent-session-wire.ts +++ b/src/shared/agent-session-wire.ts @@ -12,8 +12,13 @@ import type { AgentJournalResolution, AgentJournalSubmission } from './agent-session-journal-types' -import type { AgentSessionHandoffStage, AgentSessionOwnerRuntimeKind } from './agent-session-record' +import type { + AgentSessionHandoffStage, + AgentSessionOwnerRuntimeKind, + AgentSessionRecord +} from './agent-session-record' import type { AgentProviderSessionMetadata } from './agent-session-resume' +import type { StructuredAgentSessionProjectedStatus } from './structured-agent-session-projection' export type AgentSessionHandoffDirection = 'to-tui' | 'to-native' export type AgentSessionHandoffMode = 'now' | 'after-turn' | 'stop-turn' @@ -159,6 +164,29 @@ export type AgentSessionSubscribeEvent = } | { type: 'end' } +// ─── Status feed ──────────────────────────────────────────────────────────── + +/** What a session list needs to know about one session. The host projects it + * from the journal so no client has to replay a transcript to learn whether a + * turn is running. Additive surface: an older host has no such method. */ +export type AgentSessionStatusSummary = { + sessionId: string + workspaceId: string + agent: AgentSessionRecord['provider'] + /** Null until the journal holds a persisted user or assistant message. */ + status: StructuredAgentSessionProjectedStatus | null + latestPrompt: string + providerSession?: AgentProviderSessionMetadata + updatedAt: number +} + +/** A summary outlives its provider child: an evicted idle session is still idle, so the host + * keeps the last projection and never retracts one. Tabs, not this feed, decide what is listed. */ +export type AgentSessionStatusEvent = + | { type: 'snapshot'; sessions: AgentSessionStatusSummary[] } + | { type: 'status'; session: AgentSessionStatusSummary } + | { type: 'end' } + // ─── Mutation envelope ────────────────────────────────────────────────────── /** diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index 76e1252640a..159bb84f7e3 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -133,6 +133,10 @@ export const CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY = // to stop provider children after the last surface closes without tying lifetime to a transport. export const STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY = 'agent-session.structured.hold.v1' as const +// Why: agentSession.subscribeStatus is additive to a surface that already shipped, so a host +// advertising agent-session.structured.v1 may still answer it with method_not_found. Clients must +// probe before subscribing or they reconnect forever and never show any status at all. +export const AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY = 'agent-session.status-feed.v1' as const // Why: adding kimi to RESUMABLE_TUI_AGENTS grows terminal.ensureAgentSession's enum, and an // older host answers the unknown member with invalid_argument — a code the launch fallback does // not retry on — so clients must probe before taking the host-authority path. @@ -229,6 +233,7 @@ export const RUNTIME_CAPABILITIES = [ AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY, + AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, AGENT_SESSION_KIMI_RESUME_RUNTIME_CAPABILITY, FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY, GITHUB_MARK_PR_READY_RUNTIME_CAPABILITY, diff --git a/src/shared/structured-agent-session-projection.test.ts b/src/shared/structured-agent-session-projection.test.ts index 048051e70a5..8bdce30577e 100644 --- a/src/shared/structured-agent-session-projection.test.ts +++ b/src/shared/structured-agent-session-projection.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import { AGENT_STATUS_MAX_FIELD_LENGTH } from './agent-status-field-normalization' import type { AgentJournalRenderItem } from './agent-session-journal-types' import { parsePaneKey } from './stable-pane-id' import { @@ -6,6 +7,7 @@ import { hasPersistedStructuredAgentSessionTurn, projectStructuredItemToNativeChat, projectStructuredAgentSessionStatus, + projectStructuredAgentSessionStatusSummary, structuredAgentSessionPaneKey } from './structured-agent-session-projection' @@ -44,6 +46,52 @@ describe('structured agent session status projection', () => { expect(projectStructuredAgentSessionStatus([running, completed])).toBe('idle') }) + it('summarizes status with the newest user prompt, and null before any persisted turn', () => { + const running = item('running', 3, { + kind: 'status', + text: 'Working', + turnLifecycle: { turnId: 'turn-1', state: 'running' } + }) + const first = item('first', 1, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'first' }] + }) + const second = item('second', 2, { + kind: 'message', + role: 'user', + blocks: [ + { type: 'text', text: 'second' }, + { type: 'text', text: 'line' } + ] + }) + + expect(projectStructuredAgentSessionStatusSummary([running])).toEqual({ + status: null, + latestPrompt: '' + }) + expect(projectStructuredAgentSessionStatusSummary([first, second, running])).toEqual({ + status: 'working', + latestPrompt: 'second line' + }) + expect(projectStructuredAgentSessionStatusSummary([first, second])).toEqual({ + status: 'idle', + latestPrompt: 'second line' + }) + }) + + it('bounds the wire prompt at the shared agent-status preview cap', () => { + const pasted = item('pasted', 1, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'x'.repeat(AGENT_STATUS_MAX_FIELD_LENGTH * 40) }] + }) + + expect(projectStructuredAgentSessionStatusSummary([pasted]).latestPrompt).toHaveLength( + AGENT_STATUS_MAX_FIELD_LENGTH + ) + }) + it('creates a deterministic pane identity for status stores', () => { const paneKey = structuredAgentSessionPaneKey('structured-agent-session-1', 'session-1') diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 94938dc955f..6a5f01ba9ea 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -1,3 +1,4 @@ +import { normalizePromptField } from './agent-status-field-normalization' import type { AgentJournalRenderItem } from './agent-session-journal-types' import type { NativeChatBlock, NativeChatMessage } from './native-chat-types' import { sha256 } from './sha256' @@ -162,6 +163,34 @@ export function projectStructuredAgentSessionStatus( return activeStructuredAgentSessionTurnId(items) ? 'working' : 'idle' } +/** The newest user prompt, as the sidebar quotes it. */ +export function latestStructuredAgentSessionPrompt( + items: readonly AgentJournalRenderItem[] +): string { + for (let index = items.length - 1; index >= 0; index -= 1) { + const body = items[index]?.body + if (body?.kind === 'message' && body.role === 'user') { + return body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n') + } + } + return '' +} + +/** One projection shared by host and client: null status means "no turn yet", not idle. + * The prompt is bounded to the same preview every other agent-status row carries — a send + * admits 256 KB, and one status frame carries every retained session at once. */ +export function projectStructuredAgentSessionStatusSummary( + items: readonly AgentJournalRenderItem[] +): { status: StructuredAgentSessionProjectedStatus | null; latestPrompt: string } { + if (!hasPersistedStructuredAgentSessionTurn(items)) { + return { status: null, latestPrompt: '' } + } + return { + status: projectStructuredAgentSessionStatus(items), + latestPrompt: normalizePromptField(latestStructuredAgentSessionPrompt(items)) + } +} + export function structuredAgentSessionPaneKey(tabId: string, sessionId: string): string { const bytes = sha256(new TextEncoder().encode(sessionId)) const hex = Array.from(bytes.slice(0, 16), (byte) => byte.toString(16).padStart(2, '0')).join('') diff --git a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts index a2a64897fcb..ecf3072b39a 100644 --- a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts @@ -23,7 +23,10 @@ import { setStructuredAgentSessionHost } from '../../../src/main/native-chat/age import { AgentSessionRecordStore } from '../../../src/main/runtime/agent-session-record-store' import { computeAgentSessionPayloadFingerprint } from '../../../src/shared/agent-session-mutation-envelope' import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' -import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version' +import { + AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, + STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY +} from '../../../src/shared/protocol-version' import { resolveBaselineReleaseRef } from './release-checkout' import { loadAgentSessionWireBuild, @@ -41,6 +44,7 @@ const WORKSPACE = 'workspace-1' const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60' const NOW = 1_800_000_000_000 const CLIENT_CAPABILITY_UPDATE_METHOD = 'runtime.clientCapabilities.update' +const STATUS_FEED_METHOD = 'agentSession.subscribeStatus' /** Every method the structured surface publishes: the host method it must reach, * and the result it must hand back. A gate that hides one method and leaks @@ -107,6 +111,12 @@ const STRUCTURED_CALLS: { // A subscription that opens with nothing to say answers with no reply at all, // so reaching the host is the only signal that the gate opened. { method: 'agentSession.subscribe', hostMethod: 'subscribe' }, + // The status feed opens with a snapshot of every session, so its first reply is the contract. + { + method: STATUS_FEED_METHOD, + hostMethod: 'subscribeStatus', + result: { type: 'snapshot', sessions: [] } + }, // Teardown runs through the runtime's subscription registry rather than the // host, so its reply is the only signal that the gate opened. { method: 'agentSession.unsubscribe', hostMethod: null, result: { unsubscribed: true } } @@ -320,6 +330,10 @@ function structuredHostStub(): Record> { readOptions: vi.fn(async () => ({ models: [], current: { model: 'gpt-live' } })), history: vi.fn(() => ({ ok: true, page: { items: [] } })), subscribe: vi.fn(() => () => undefined), + subscribeStatus: vi.fn((subscriber: { emit: (event: unknown) => void }) => { + subscriber.emit({ type: 'snapshot', sessions: [] }) + return () => undefined + }), unsubscribe: vi.fn() } } @@ -443,6 +457,14 @@ describe('cross-version structured agent sessions', () => { expect(baseline.capabilities.includes(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)).toBe( baselineStructuredMethods().length > 0 ) + // The status feed is additive to a surface that already shipped, so it carries its own + // capability or a client cannot tell "host too old" from "the call failed" — and it + // would relay-retry a method_not_found forever instead of degrading once. + for (const build of [current, baseline]) { + expect(build.capabilities.includes(AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY)).toBe( + build.methodNames.includes(STATUS_FEED_METHOD) + ) + } // Additive surface: bumping the protocol number would strand every paired // device on this release rather than degrade one feature. expect(current.protocolVersion).toBe(baseline.protocolVersion)