diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 92a4f2653e0..e77b8ef2200 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -10,12 +10,17 @@ import { createStructuredAgentSessionHostStatusFeed, type StructuredAgentSessionStatusSubscriber } from './structured-agent-session-status-feed' +import { + StructuredTurnCompletionFeed, + type StructuredTurnCompletionSubscriber +} from './structured-turn-completion-feed' /** Owns every host-to-client publication edge, including compatibility waits. */ export class StructuredAgentSessionClientDelivery { readonly subscribers: AgentSessionSubscribers readonly waitForSendSettlement: StructuredAgentSessionSendSettlement['wait'] private readonly statusFeed + private readonly turnCompletions private readonly sendSettlement constructor( @@ -24,6 +29,7 @@ export class StructuredAgentSessionClientDelivery { deps: () => StructuredAgentSessionHostDeps ) { this.statusFeed = createStructuredAgentSessionHostStatusFeed({ sessions, now, deps }) + this.turnCompletions = new StructuredTurnCompletionFeed({ sessions, now }) this.sendSettlement = new StructuredAgentSessionSendSettlement((sessionId) => this.requireJournal(sessionId) ) @@ -34,26 +40,38 @@ export class StructuredAgentSessionClientDelivery { }) } - publishStatus = (sessionId: string): void => this.statusFeed.publish(sessionId) + // Why every status edge only baselines: a status publication is not a journal commit, so a + // terminal turn first seen here is one this feed was not watching when it landed. Baselining is + // also how an attaching session gets its live-only start, since attach publishes status. + publishStatus = (sessionId: string): void => { + this.statusFeed.publish(sessionId) + this.turnCompletions.baseline(sessionId) + } publishStatusAndSettlement = (sessionId: string): void => { - this.statusFeed.publish(sessionId) + this.publishStatus(sessionId) const journal = this.sessions.get(sessionId)?.journal if (journal) { this.sendSettlement.publish(sessionId, journal) } } - publishRestored = (sessionId: string): void => + publishRestored = (sessionId: string): void => { this.statusFeed.publish(sessionId, undefined, { replay: true }) + this.turnCompletions.baseline(sessionId) + } subscribeStatus = (subscriber: StructuredAgentSessionStatusSubscriber): (() => void) => this.statusFeed.subscribe(subscriber) forgetStatus = (sessionId: string): void => this.statusFeed.forget(sessionId) + subscribeTurnCompletions = (subscriber: StructuredTurnCompletionSubscriber): (() => void) => + this.turnCompletions.subscribe(subscriber) + closeSession(sessionId: string): void { this.sendSettlement.closeSession(sessionId) this.statusFeed.close(sessionId) + this.turnCompletions.forget(sessionId) } closeAll(): void { @@ -62,6 +80,8 @@ export class StructuredAgentSessionClientDelivery { private publishJournal(sessionId: string, journal: AgentSessionJournal): void { this.statusFeed.publish(sessionId, journal) + // The one edge that may announce a completion: this is journal commit. + this.turnCompletions.observe(sessionId, journal) this.sendSettlement.publish(sessionId, journal) } 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 781745ec26d..1dadf5c48be 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 @@ -53,6 +53,7 @@ import { type StructuredAgentSessionRestartResume } from './structured-agent-session-restart-resume-host' import { structuredAgentSessionRestartResumeSurfaces } from './structured-agent-session-restart-resume-wiring' +import type { StructuredTurnCompletionSubscriber } from './structured-turn-completion-feed' export type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' export class StructuredAgentSessionHost { @@ -336,6 +337,11 @@ export class StructuredAgentSessionHost { subscribeStatus = (subscriber: StructuredAgentSessionStatusSubscriber): (() => void) => this.clientDelivery.subscribeStatus(subscriber) + /** Root turns settling from now on, so a client can raise attention for a chat it is not + * showing. Retains nothing and replays nothing — see the feed for why recovery is live-only. */ + subscribeTurnCompletions = (subscriber: StructuredTurnCompletionSubscriber): (() => void) => + this.clientDelivery.subscribeTurnCompletions(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-turn-completion-feed.test.ts b/src/main/native-chat/agent-session-wire/structured-turn-completion-feed.test.ts new file mode 100644 index 00000000000..db0cb8dd15d --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-turn-completion-feed.test.ts @@ -0,0 +1,324 @@ +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 { AgentJournalTurnLifecycle } from '../../../shared/agent-session-journal-types' +import { agentJournalTurnBody } from '../../../shared/agent-session-turn-record' +import type { StructuredTurnCompletionEvent } from '../../../shared/structured-turn-completion' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import { indexedStatusFeedSession as indexed } from './structured-agent-session-status-feed-test-session' +import { StructuredTurnCompletionFeed } from './structured-turn-completion-feed' + +const SESSION = 'completion-session' +const TURN_IDENTITY = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-1', + ordinal: 0 +} as const +const USER_IDENTITY = { ...TURN_IDENTITY, ordinal: 1 } as const + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-turn-completion-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) + }) +} + +type Journal = Awaited> + +/** The host's own session index, built for real rather than faked: `indexedStatusFeedSession` + * already produces the journal-plus-location shape the feed reads. */ +function sessionIndex(sessions: Map) { + return new Map([...sessions].map(([sessionId, session]) => [sessionId, indexed(session)])) +} + +function feedFor(sessions: Map) { + const feed = new StructuredTurnCompletionFeed({ + sessions: sessionIndex(sessions), + now: () => 9_000 + }) + const events: StructuredTurnCompletionEvent[] = [] + const dispose = feed.subscribe({ + id: 'window-1', + emit: (event) => events.push(event) + }) + return { feed, events, dispose } +} + +/** A root turn record, as the journal holds it. Child turns write no turn record at all. */ +async function writeTurn(journal: Journal, turn: AgentJournalTurnLifecycle): Promise { + await journal.appendItem(TURN_IDENTITY, agentJournalTurnBody(turn), { + fence: 1 + }) +} + +async function writeUserMessage(journal: Journal): Promise { + await journal.appendItem( + USER_IDENTITY, + { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'hello' }] + }, + { fence: 1 } + ) +} + +function completions(events: StructuredTurnCompletionEvent[]) { + return events.flatMap((event) => (event.type === 'completion' ? [event.completion] : [])) +} + +describe('StructuredTurnCompletionFeed', () => { + it('announces a root turn that settles with a recorded verdict, scoped for dedupe', async () => { + const journal = await openJournal() + const sessions = new Map([[SESSION, { journal }]]) + const { feed, events, dispose } = feedFor(sessions) + + feed.baseline(SESSION) + await writeUserMessage(journal) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + expect(completions(events)).toEqual([]) + + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 4_242 + }) + feed.observe(SESSION, journal) + + expect(completions(events)).toEqual([ + { + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' + }, + sessionId: SESSION, + turnId: 'turn-1', + outcome: 'success', + completedAt: 4_242 + } + ]) + dispose() + }) + + it.each([ + { outcome: 'failure' as const, state: 'completed' as const }, + { outcome: 'cancellation' as const, state: 'interrupted' as const } + ])('announces $outcome so a consumer can decline it', async ({ outcome, state }) => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + await writeTurn(journal, { + turnId: 'turn-1', + state, + outcome, + completedAt: 7 + }) + feed.observe(SESSION, journal) + + expect(completions(events).map((completion) => completion.outcome)).toEqual([outcome]) + dispose() + }) + + it('stays silent for a turn whose verdict was never recorded', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + // An older host, or an end this host inferred rather than heard. Absent is UNKNOWN, and + // `completed` on its own is also what a provider API error is recorded as. + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + completedAt: 7 + }) + feed.observe(SESSION, journal) + + expect(completions(events)).toEqual([]) + dispose() + }) + + describe('live-only recovery', () => { + it('baselines a session whose turn already finished before the feed saw it', async () => { + const journal = await openJournal() + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + // Restore, restart and rewind all arrive here. A restored session is not a completing one. + feed.baseline(SESSION) + feed.observe(SESSION, journal) + + expect(completions(events)).toEqual([]) + dispose() + }) + + it('baselines on the commit edge too, when that is the first sight of the session', async () => { + const journal = await openJournal() + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.observe(SESSION, journal) + + expect(completions(events)).toEqual([]) + dispose() + }) + + it('opens a subscriber on nothing, so a completion it missed is dropped not replayed', async () => { + const journal = await openJournal() + const sessions = new Map([[SESSION, { journal }]]) + const { feed, dispose } = feedFor(sessions) + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + feed.observe(SESSION, journal) + dispose() + + // A second window connects after the fact. Catch-up is deliberately absent: if this ever + // starts returning the earlier completion, recovery has silently become replay. + const late: StructuredTurnCompletionEvent[] = [] + const disposeLate = feed.subscribe({ + id: 'window-2', + emit: (event) => late.push(event) + }) + expect(late).toEqual([]) + + feed.observe(SESSION, journal) + expect(late).toEqual([]) + disposeLate() + }) + + it('never re-announces a turn it already accounted for', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + // An event-recovery snapshot and a rewind both re-publish a journal whose newest turn is + // one this feed has already reported. + feed.observe(SESSION, journal) + feed.observe(SESSION, journal) + feed.observe(SESSION, journal) + feed.baseline(SESSION) + + expect(completions(events)).toHaveLength(1) + dispose() + }) + + it('forgets a closed session rather than retaining its turn bookkeeping', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + feed.forget(SESSION) + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + // Re-attaching is a fresh baseline, so the turn that settled while closed stays quiet. + feed.observe(SESSION, journal) + + expect(completions(events)).toEqual([]) + dispose() + }) + }) + + it('emits nothing for a session this host does not hold', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + + feed.baseline('some-other-session') + feed.observe('some-other-session') + + expect(events).toEqual([]) + dispose() + }) + + it('drops a subscriber whose transport throws instead of poisoning the others', async () => { + const journal = await openJournal() + const { feed, events, dispose } = feedFor(new Map([[SESSION, { journal }]])) + feed.subscribe({ + id: 'dead-window', + emit: () => { + throw new Error('transport gone') + } + }) + + feed.baseline(SESSION) + await writeTurn(journal, { turnId: 'turn-1', state: 'running' }) + feed.observe(SESSION, journal) + await writeTurn(journal, { + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + completedAt: 1 + }) + feed.observe(SESSION, journal) + + expect(completions(events)).toHaveLength(1) + dispose() + }) + + it('ends the stream on unsubscribe so a client can tell teardown from silence', async () => { + const journal = await openJournal() + const { feed, events } = feedFor(new Map([[SESSION, { journal }]])) + + feed.unsubscribe('window-1') + + expect(events).toEqual([{ type: 'end' }]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-turn-completion-feed.ts b/src/main/native-chat/agent-session-wire/structured-turn-completion-feed.ts new file mode 100644 index 00000000000..3cf69b1d4ac --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-turn-completion-feed.ts @@ -0,0 +1,168 @@ +// The host's answer to "did a structured chat's root turn just finish, and did it go well". +// +// Derived here rather than in a renderer because the host owns journal commit and is the only +// party that can observe a session whose chat is not mounted — which is the whole case the +// indicators exist for. The verdict is A0's recorded `outcome`, read off the journal turn record; +// nothing here re-derives success from lifecycle state, because `completed` is also what the host +// writes for a turn the provider ended with an API error. +// +// RECOVERY IS LIVE-ONLY, BY DECISION. Nothing is persisted and nothing is queued, so nothing can +// strand — that is the reason for the choice. Two rules implement it, and both are pinned by +// tests because a later refactor turning either into catch-up would be silent: +// +// 1. A session BASELINES on the first edge that reaches this feed. Whatever terminal turn the +// journal already holds is recorded as accounted for and announced to nobody. A restored, +// rewound or restarted session is not a completing session. +// 2. A subscriber arrives to nothing. Unlike the status feed there is no opening snapshot, so a +// completion that happened while no client was connected is dropped rather than replayed. +// +// Only the journal-commit edge can announce anything. Every other publication edge — handoff, +// lease renewal, background-task ticks, restore — may only baseline a session it has never seen, +// never advance one it has. That asymmetry is what stops a status tick that happens to interleave +// with a commit from silently absorbing the completion it was not looking at. + +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentJournalTurnLifecycle } from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurnOutcome } from '../../../shared/agent-session-turn-record' +import { newestStructuredAgentSessionTurn } from '../../../shared/structured-agent-session-projection' +import type { StructuredTurnCompletionEvent } from '../../../shared/structured-turn-completion' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' + +export type StructuredTurnCompletionSubscriber = { + id: string + emit: (event: StructuredTurnCompletionEvent) => void +} + +type CompletionFeedSession = { + journal: AgentSessionJournal + params: { location: AgentSessionRecord['location'] } +} + +export type StructuredTurnCompletionFeedDeps = { + sessions: ReadonlyMap + now: () => number +} + +/** + * How many of a session's terminal turn ids stay accounted for. + * + * The set is what makes a re-publication of the same journal — an event-recovery snapshot, a + * rewind that leaves an older terminal turn newest — silent. Evicting the oldest entries only + * risks re-announcing a turn that was already announced this session and has since been rewound + * past this many turns, which no real session reaches. Bounded because the alternative is a map + * that grows for the life of a long chat. + */ +const MAX_ACCOUNTED_TURNS = 128 + +export class StructuredTurnCompletionFeed { + private readonly subscribers = new Map() + private readonly accountedBySession = new Map>() + + constructor(private readonly deps: StructuredTurnCompletionFeedDeps) {} + + /** Live only: a subscriber gets no snapshot, so it hears completions from now on and no earlier. */ + subscribe(subscriber: StructuredTurnCompletionSubscriber): () => void { + this.subscribers.set(subscriber.id, subscriber) + 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. + } + } + + /** + * Record where a session already is without announcing it. + * + * Every non-commit publication edge calls this, which is what gives an attaching or restoring + * session its baseline. It never advances a session this feed already knows, so it cannot + * absorb a completion that the commit edge has not reported yet. + */ + baseline(sessionId: string): void { + if (this.accountedBySession.has(sessionId)) { + return + } + this.observe(sessionId) + } + + /** The session is gone; its turn bookkeeping goes with it. */ + forget(sessionId: string): void { + this.accountedBySession.delete(sessionId) + } + + /** The journal-commit edge: the only one that can announce a completion. */ + observe(sessionId: string, journal?: AgentSessionJournal): void { + const session = this.deps.sessions.get(sessionId) + if (!session) { + return + } + const accounted = this.accountedBySession.get(sessionId) + const turn = this.newestTerminalTurn(journal ?? session.journal) + if (!accounted) { + // First sight of this session. Whatever it already finished happened before we were + // watching, so it is accounted for and announced to nobody. + this.accountedBySession.set(sessionId, new Set(turn ? [turn.turnId] : [])) + return + } + if (!turn || accounted.has(turn.turnId)) { + return + } + this.account(accounted, turn.turnId) + const outcome = readAgentJournalTurnOutcome(turn) + if (!outcome) { + // Unknown is not a completion: an older host, an end the host inferred rather than heard, + // or a verdict this build cannot place. Accounted for above so it is asked once. + return + } + this.broadcast({ + type: 'completion', + completion: { + scope: session.params.location, + sessionId, + turnId: turn.turnId, + outcome, + completedAt: turn.completedAt ?? this.deps.now() + } + }) + } + + /** The newest ROOT turn once it has stopped running. Child turns write no turn record at all, + * so reading the journal's own turn item is already root-only. */ + private newestTerminalTurn(journal: AgentSessionJournal): AgentJournalTurnLifecycle | null { + if (journal.isReadOnly) { + return null + } + const turn = newestStructuredAgentSessionTurn(journal.snapshot().items) + return turn && turn.state !== 'running' ? turn : null + } + + private account(accounted: Set, turnId: string): void { + accounted.add(turnId) + while (accounted.size > MAX_ACCOUNTED_TURNS) { + const oldest = accounted.values().next() + if (oldest.done) { + return + } + accounted.delete(oldest.value) + } + } + + private broadcast(event: StructuredTurnCompletionEvent): void { + // A Map skips entries deleted mid-iteration, so a failing subscriber can drop itself here. + for (const subscriber of this.subscribers.values()) { + try { + subscriber.emit(event) + } catch { + this.subscribers.delete(subscriber.id) + } + } + } +} diff --git a/src/main/notifications/notification-delivery-service.test.ts b/src/main/notifications/notification-delivery-service.test.ts index 1e1b09bfc42..d7579605d30 100644 --- a/src/main/notifications/notification-delivery-service.test.ts +++ b/src/main/notifications/notification-delivery-service.test.ts @@ -151,6 +151,47 @@ describe('createNotificationDeliveryService', () => { expect(harness.dispatchMobileNotification).not.toHaveBeenCalled() }) + describe('per-event mobile dedupe', () => { + it('fans out once per completion even when the coarse burst key differs', () => { + const harness = makeHarness(makeSettings()) + const service = createNotificationDeliveryService(harness.deps) + const key = 'local\u0000session-1\u0000turn-1' + + // Two windows raise the SAME completion, and the coarse per-workspace gate cannot hold the + // second: that key includes agentState, which the two windows read from their own row and + // can disagree about. Only the event identity makes this one push rather than two. + service.dispatch(makeRequest({ mobileDedupeKey: key, agentState: 'done' })) + service.dispatch(makeRequest({ mobileDedupeKey: key, agentState: 'working' })) + + expect(harness.dispatchMobileNotification).toHaveBeenCalledTimes(1) + }) + + it('holds a repeat of one completion open-endedly, not for a burst window', () => { + const harness = makeHarness(makeSettings()) + const service = createNotificationDeliveryService(harness.deps) + const key = 'local\u0000session-1\u0000turn-1' + + service.dispatch(makeRequest({ mobileDedupeKey: key, agentState: 'done' })) + // Long past the coarse burst window. The same completion is still the same completion. + now += 60_000 + service.dispatch(makeRequest({ mobileDedupeKey: key, agentState: 'working' })) + + expect(harness.dispatchMobileNotification).toHaveBeenCalledTimes(1) + }) + + it('still fans out for a genuinely different completion', () => { + const harness = makeHarness(makeSettings()) + const service = createNotificationDeliveryService(harness.deps) + + service.dispatch(makeRequest({ mobileDedupeKey: 'local\u0000session-1\u0000turn-1' })) + service.dispatch( + makeRequest({ mobileDedupeKey: 'local\u0000session-2\u0000turn-1', worktreeId: 'wt-2' }) + ) + + expect(harness.dispatchMobileNotification).toHaveBeenCalledTimes(2) + }) + }) + it('reports blocked-by-system on macOS when permission is undecided', async () => { const harness = makeHarness(makeSettings()) harness.deps.platform = 'darwin' diff --git a/src/main/notifications/notification-delivery-service.ts b/src/main/notifications/notification-delivery-service.ts index 7ca85c61f92..df3acc0dd3c 100644 --- a/src/main/notifications/notification-delivery-service.ts +++ b/src/main/notifications/notification-delivery-service.ts @@ -14,6 +14,7 @@ import type { import type { OrcaRuntimeService } from '../runtime/orca-runtime' import { buildNotificationOptions } from '../ipc/notification-options' import { reserveNotificationCooldown } from '../ipc/notification-burst-cooldown' +import { reserveNotificationEventOnce } from '../../shared/notification-event-dedupe' export type NotificationDeliveryDependencies = { readNotificationSettings: () => NotificationSettings @@ -48,6 +49,7 @@ export function createNotificationDeliveryService( ): NotificationDeliveryService { const recentDesktopNotifications = new Map() const recentMobileNotifications = new Map() + const mobileEventsSeen = new Set() const dedupeKeyFor = (request: NotificationDispatchRequest): string => request.worktreeId ?? request.worktreeLabel ?? 'global' @@ -71,7 +73,16 @@ export function createNotificationDeliveryService( const notificationOptions = buildNotificationOptions(request) // Why: desktop focus only means this computer sees the worktree; the paired phone may still need the alert. - if (deps.dispatchMobileNotification && request.source !== 'test') { + // + // Two gates, not one. A sender that names its event identity gets at-most-once per event, + // with no expiry: several windows can be connected to the same execution host and each + // dispatches the same completion, and only this process sees all of them. The coarse + // per-workspace burst gate below still applies on top, so naming an event never loosens + // the existing policy. + const mobileEventAdmitted = + request.mobileDedupeKey === undefined || + reserveNotificationEventOnce(mobileEventsSeen, request.mobileDedupeKey) + if (deps.dispatchMobileNotification && request.source !== 'test' && mobileEventAdmitted) { if ( reserveNotificationCooldown( recentMobileNotifications, diff --git a/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts index 9570fe0abb0..d5c0977674a 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts @@ -79,5 +79,6 @@ export const ADMISSION_METHODS = [ { method: 'agentSession.subscribe', params: { sessionId: SESSION } }, { method: 'agentSession.hold', params: { sessionId: SESSION, holderId: 'surface-1' } }, { method: 'agentSession.reveal', params: { sessionId: SESSION } }, - { method: 'agentSession.subscribeStatus', params: null } + { method: 'agentSession.subscribeStatus', params: null }, + { method: 'agentSession.subscribeTurnCompletion', params: null } ] as const diff --git a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts index aad992d4926..94fbe924fab 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts @@ -211,6 +211,7 @@ export function hostStub(): StructuredAgentSessionHost { subscribeStatus: vi.fn((subscriber: StructuredAgentSessionStatusSubscriber) => statusFeed().subscribe(subscriber) ), + subscribeTurnCompletions: vi.fn(() => () => undefined), unsubscribe: vi.fn(), release: vi.fn() }) 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 index f3d00232359..b7763a9a949 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-subscription-id.ts @@ -27,3 +27,9 @@ export function structuredAgentSessionSubscriptionId(ctx: RpcContext, sessionId: export function structuredAgentSessionStatusSubscriptionId(ctx: RpcContext): string { return withFrameId(ctx, `${SUBSCRIPTION_PREFIX}.status:${ctx.connectionId ?? 'local'}`) } + +/** The turn-completion feed, per connection like the status feed and namespaced apart from it so + * one client can hold both. */ +export function structuredAgentSessionTurnCompletionSubscriptionId(ctx: RpcContext): string { + return withFrameId(ctx, `${SUBSCRIPTION_PREFIX}.turnCompletion:${ctx.connectionId ?? 'local'}`) +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session-turn-completion-stream.ts b/src/main/runtime/rpc/methods/structured-agent-session-turn-completion-stream.ts new file mode 100644 index 00000000000..4f17d5147c8 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-turn-completion-stream.ts @@ -0,0 +1,33 @@ +// `agentSession.subscribeTurnCompletion` — root turns settling, so a client can raise attention +// for a chat it is not showing. +// +// Separate from the status stream because a status summary carries no turn identity and no +// provider verdict, and a completion must not be inferred from a status transition. Like the +// status stream it is per connection rather than per session, and unlike it the stream opens with +// nothing: recovery is live-only by decision, so a completion that happened while this client was +// away is dropped rather than replayed. + +import { defineStreamingMethod } from '../core' +import { requireStructuredHost as requireHost } from './structured-agent-session-gate' +import { bindStructuredAgentSessionStream } from './structured-agent-session-status-stream' +import { structuredAgentSessionTurnCompletionSubscriptionId } from './structured-agent-session-subscription-id' + +export const STRUCTURED_AGENT_SESSION_TURN_COMPLETION_METHODS = [ + defineStreamingMethod({ + name: 'agentSession.subscribeTurnCompletion', + params: null, + handler: async (_params, ctx, emit) => { + const host = requireHost(ctx) + const subscriptionId = structuredAgentSessionTurnCompletionSubscriptionId(ctx) + let dispose = (): void => {} + const stream = bindStructuredAgentSessionStream(ctx, subscriptionId, () => dispose()) + if (stream.isClosed()) { + return + } + dispose = host.subscribeTurnCompletions({ id: subscriptionId, emit }) + if (stream.isClosed()) { + dispose() + } + } + }) +] 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 08e947ae562..78fbc9ee156 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -162,7 +162,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(26) + expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(27) }) it('hides the surface from a declared client that did not advertise it', async () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index a6f09a18de4..33b16ed97b5 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -40,6 +40,7 @@ import { bindStructuredAgentSessionStream, STRUCTURED_AGENT_SESSION_STATUS_METHODS } from './structured-agent-session-status-stream' +import { STRUCTURED_AGENT_SESSION_TURN_COMPLETION_METHODS } from './structured-agent-session-turn-completion-stream' import { structuredAgentSessionSubscriptionBase as subscriptionBaseFor, structuredAgentSessionSubscriptionId as subscriptionIdFor @@ -328,5 +329,6 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ ...STRUCTURED_AGENT_SESSION_HOLD_METHODS, ...STRUCTURED_AGENT_SESSION_REVEAL_METHODS, ...STRUCTURED_AGENT_SESSION_RESTART_RESUME_METHODS, - ...STRUCTURED_AGENT_SESSION_STATUS_METHODS + ...STRUCTURED_AGENT_SESSION_STATUS_METHODS, + ...STRUCTURED_AGENT_SESSION_TURN_COMPLETION_METHODS ] diff --git a/src/renderer/src/attention/agent-attention-notification-delivery.ts b/src/renderer/src/attention/agent-attention-notification-delivery.ts new file mode 100644 index 00000000000..92ef6674b93 --- /dev/null +++ b/src/renderer/src/attention/agent-attention-notification-delivery.ts @@ -0,0 +1,52 @@ +/** + * Sending one attention delivery request to main, and the client-side follow-ups that go with it. + * + * Extracted so a non-terminal surface reuses this rather than restating it: the success sound and + * the blocked-permission fallback are policy, and a second copy of them would drift. What the + * request SAYS still belongs to each surface — a terminal arbitrates a title against a hook + * snapshot, a structured chat reads its projected row — so only the send lives here. + * + * Whether a banner actually appears is main's call, not this module's: the enabled/source + * preferences and the suppress-while-focused setting are applied there, after mobile fan-out. + * A caller must never promise the user a banner. + */ +import { playDesktopNotificationSound } from '@/lib/desktop-notification-sound' +import { showBlockedNotificationFallbackToast } from '@/lib/blocked-notification-fallback' +import type { NotificationDispatchRequest } from '../../../shared/notification-settings-types' + +export type AgentAttentionNotificationSound = { + customSoundId: string + customSoundVolume: number | null +} + +export function deliverAgentAttentionNotification( + request: NotificationDispatchRequest, + sound: AgentAttentionNotificationSound +): void { + void window.api.notifications + .dispatch(request) + .then((result) => { + if (result.delivered) { + void playDesktopNotificationSound(sound.customSoundId, sound.customSoundVolume) + return + } + // Why: macOS is silently swallowing notifications (permission off or prompt unanswered) — + // surface an in-app pointer at the fix instead of letting the alert vanish without a trace. + if (result.reason === 'blocked-by-system') { + showBlockedNotificationFallbackToast() + } + }) + .catch((err) => { + console.warn('Failed to dispatch notification:', err) + }) +} + +/** The sound preferences one delivery reads, defaulted the way the dispatch path always has. */ +export function readAgentAttentionNotificationSound(settings: { + notifications?: { customSoundId?: string; customSoundVolume?: number | null } | undefined +}): AgentAttentionNotificationSound { + return { + customSoundId: settings.notifications?.customSoundId ?? 'system', + customSoundVolume: settings.notifications?.customSoundVolume ?? null + } +} diff --git a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx index f2c77fb5c98..4d648aed5c8 100644 --- a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx +++ b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.test.tsx @@ -22,6 +22,8 @@ const mocks = vi.hoisted(() => ({ setState: (state: Partial & { testRuntimeOwner?: string | null }) => void }, subscribeStatus: vi.fn(), + subscribeTurnCompletion: vi.fn(), + unsubscribeTurnCompletion: vi.fn(), subscribeTranscript: vi.fn(), supportsCapability: vi.fn(), unsubscribe: vi.fn() @@ -58,7 +60,8 @@ vi.mock('@/runtime/runtime-rpc-client', async (importOriginal) => ({ vi.mock('@/runtime/structured-agent-session-client', () => ({ callStructuredAgentSession: vi.fn(), subscribeStructuredAgentSession: mocks.subscribeTranscript, - subscribeStructuredAgentSessionStatus: mocks.subscribeStatus + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus, + subscribeStructuredAgentSessionTurnCompletion: mocks.subscribeTurnCompletion })) import { @@ -66,6 +69,7 @@ import { StructuredAgentSessionStatusBridge } from './StructuredAgentSessionStatusBridge' import { resetStructuredAgentSessionStatusFeedsForTests } from '@/runtime/structured-agent-session-status-feed' +import { resetStructuredTurnCompletionFeedsForTests } from '@/runtime/structured-turn-completion-feed' const structuredTab = { id: 'structured-tab-1', @@ -115,7 +119,11 @@ describe('StructuredAgentSessionStatusBridge', () => { beforeEach(() => { vi.clearAllMocks() resetStructuredAgentSessionStatusFeedsForTests() + resetStructuredTurnCompletionFeedsForTests() mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + mocks.subscribeTurnCompletion.mockResolvedValue({ + unsubscribe: mocks.unsubscribeTurnCompletion + }) mocks.supportsCapability.mockResolvedValue(true) mocks.store?.setState({ agentStatusByPaneKey: {}, @@ -127,6 +135,7 @@ describe('StructuredAgentSessionStatusBridge', () => { afterEach(() => { cleanup() resetStructuredAgentSessionStatusFeedsForTests() + resetStructuredTurnCompletionFeedsForTests() }) it('reuses the structured-tab projection for an unchanged tab map', () => { @@ -531,6 +540,8 @@ describe('StructuredAgentSessionStatusBridge', () => { expect(statuses()).toEqual([]) await waitFor(() => expect(mocks.unsubscribe).toHaveBeenCalledOnce()) + // The completion stream is held by the same tab enumeration, so it goes with the tab. + await waitFor(() => expect(mocks.unsubscribeTurnCompletion).toHaveBeenCalledOnce()) }) it('reconnects after the host ends the stream', async () => { diff --git a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx index d7954f5d1f4..71b9256f992 100644 --- a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx +++ b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx @@ -20,6 +20,7 @@ import { useAppStore } from '@/store' import { getActiveRuntimeTarget, type RuntimeClientTarget } from '@/runtime/runtime-rpc-client' import { getStructuredAgentSessionStatusFeed } from '@/runtime/structured-agent-session-status-feed' import { getStructuredAgentSessionTabs, type StructuredTab } from './structured-agent-session-tabs' +import { StructuredTurnCompletionAttention } from './StructuredTurnCompletionAttention' // Re-exported so the bridge stays the one import site its consumers already know. export { getStructuredAgentSessionTabs } from './structured-agent-session-tabs' @@ -204,6 +205,11 @@ export function StructuredAgentSessionStatusBridge(): React.JSX.Element { {tabs.map((tab) => ( ))} + {/* Attention rides the same global tab enumeration: a chat that is not on screen is + precisely the one whose completion has to be able to light an indicator. */} + {tabs.map((tab) => ( + + ))} ) } diff --git a/src/renderer/src/components/native-chat/StructuredTurnCompletionAttention.tsx b/src/renderer/src/components/native-chat/StructuredTurnCompletionAttention.tsx new file mode 100644 index 00000000000..6ff84f901c2 --- /dev/null +++ b/src/renderer/src/components/native-chat/StructuredTurnCompletionAttention.tsx @@ -0,0 +1,43 @@ +/** + * One structured tab's subscription to its own session's turn completions. + * + * Mounted per structured tab by the status bridge, which renders for every structured tab in + * every workspace whether or not it is on screen — that is exactly why a backgrounded chat can + * light anything at all. Filtering by session here rather than looking a tab up from a session id + * keeps the addressing in one direction: the tab that owns the session is the tab that subscribes, + * so a completion can only ever be addressed to the surface key that tab publishes. + */ +import { useEffect, useMemo } from 'react' +import { useAppStore } from '@/store' +import { getRuntimeEnvironmentIdForWorktree } from '@/lib/worktree-runtime-owner' +import { getActiveRuntimeTarget } from '@/runtime/runtime-rpc-client' +import { getStructuredTurnCompletionFeed } from '@/runtime/structured-turn-completion-feed' +import { structuredAgentSessionPaneKey } from '../../../../shared/structured-agent-session-projection' +import { dispatchStructuredTurnCompletion } from './structured-turn-completion-attention' +import type { StructuredTab } from './structured-agent-session-tabs' + +export function StructuredTurnCompletionAttention({ tab }: { tab: StructuredTab }): null { + const environmentId = useAppStore((state) => + getRuntimeEnvironmentIdForWorktree(state, tab.worktreeId) + ) + const target = useMemo( + () => getActiveRuntimeTarget({ activeRuntimeEnvironmentId: environmentId }), + [environmentId] + ) + const feed = useMemo(() => getStructuredTurnCompletionFeed(target), [target]) + useEffect(() => feed.activate(), [feed]) + useEffect(() => { + const paneKey = structuredAgentSessionPaneKey(tab.id, tab.entityId) + return feed.subscribe((completion) => { + if (completion.sessionId !== tab.entityId) { + return + } + dispatchStructuredTurnCompletion(completion, { + workspaceId: tab.worktreeId, + paneKey, + label: tab.label + }) + }) + }, [feed, tab.entityId, tab.id, tab.label, tab.worktreeId]) + return null +} diff --git a/src/renderer/src/components/native-chat/structured-turn-completion-attention.test.ts b/src/renderer/src/components/native-chat/structured-turn-completion-attention.test.ts new file mode 100644 index 00000000000..a9666025d50 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-turn-completion-attention.test.ts @@ -0,0 +1,444 @@ +/** + * The acceptance matrix: a finished structured chat must light exactly what a finished PTY agent + * lights, on the same rules. The PTY control runs in the same mixed workspace and the same test + * run, because "the chat marks something" is only the claim if a terminal beside it marks the + * same things from the same store. + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentStatusEntry } from '../../../../shared/agent-status-types' +import type { NotificationDispatchRequest } from '../../../../shared/notification-settings-types' +import { makePaneKey } from '../../../../shared/stable-pane-id' +import { structuredAgentSessionPaneKey } from '../../../../shared/structured-agent-session-projection' +import type { StructuredTurnCompletion } from '../../../../shared/structured-turn-completion' +import { structuredTurnCompletionKey } from '../../../../shared/structured-turn-completion' + +const WORKSPACE = 'wt-primary' +const OTHER_WORKSPACE = 'wt-secondary' +const FOLDER_WORKSPACE = 'folder:folder-1' +const GROUP = 'group-1' +const CHAT_TAB = 'chat-tab' +const SESSION = 'session-1' +const CHAT_PANE_KEY = structuredAgentSessionPaneKey(CHAT_TAB, SESSION) +const TERMINAL_TAB = 'term-tab' +const LEAF = '11111111-1111-4111-8111-111111111111' +const TERMINAL_PANE_KEY = makePaneKey(TERMINAL_TAB, LEAF) + +const SCOPE = { + executionHostId: 'local', + wslDistro: null, + workspaceId: WORKSPACE, + workspaceKind: 'git-worktree' +} as const + +function completion(overrides: Partial = {}): StructuredTurnCompletion { + return { + scope: SCOPE, + sessionId: SESSION, + turnId: 'turn-1', + outcome: 'success', + completedAt: 1_700, + ...overrides + } +} + +function agentRow(paneKey: string, tabId: string, workspaceId: string): AgentStatusEntry { + const now = 1_000 + return { + state: 'done', + prompt: 'do the thing', + updatedAt: now, + stateStartedAt: now, + agentType: 'claude', + paneKey, + tabId, + worktreeId: workspaceId, + terminalTitle: 'claude', + stateHistory: [], + lastAssistantMessage: 'Done.' + } +} + +type Seed = { + /** Which tab the workspace's focused group is showing. */ + visible?: 'chat' | 'terminal' + /** In-app workspace selection; the default leaves the target workspace backgrounded. */ + activeWorktreeId?: string | null + /** The session the chat tab is currently bound to. */ + boundSession?: string + chatTab?: boolean + groupAttentionEnabled?: boolean + /** Which workspace holds the tabs. A `folder:` key exercises the folder-workspace paths. */ + workspaceId?: string +} + +type UnifiedTabRow = { + id: string + worktreeId: string + groupId: string + contentType: string + entityId?: string + agentSessionAgent?: string + label?: string +} + +type TabGroupRow = { + id: string + worktreeId: string + activeTabId: string + tabOrder: string[] +} + +function buildState(seed: Seed = {}) { + const workspace = seed.workspaceId ?? WORKSPACE + const unifiedTabs: UnifiedTabRow[] = [ + { + id: TERMINAL_TAB, + worktreeId: workspace, + groupId: GROUP, + contentType: 'terminal' + }, + ...(seed.chatTab === false + ? [] + : [ + { + id: CHAT_TAB, + worktreeId: workspace, + groupId: GROUP, + contentType: 'agent-session', + entityId: seed.boundSession ?? SESSION, + agentSessionAgent: 'claude', + label: 'Fix auth' + } + ]) + ] + const groups: TabGroupRow[] = [ + { + id: GROUP, + worktreeId: workspace, + activeTabId: seed.visible === 'chat' ? CHAT_TAB : TERMINAL_TAB, + tabOrder: [TERMINAL_TAB, CHAT_TAB] + } + ] + const activeGroupIdByWorktree: Record = { + [workspace]: GROUP + } + const groupsByWorktree: Record = { + [workspace]: groups + } + const unifiedTabsByWorktree: Record = { + [workspace]: unifiedTabs + } + const tabsByWorktree: Record = { + [workspace]: [{ id: TERMINAL_TAB, ptyId: 'pty-1' }] + } + return { + activeWorktreeId: seed.activeWorktreeId === undefined ? OTHER_WORKSPACE : seed.activeWorktreeId, + activeTabId: TERMINAL_TAB, + activeGroupIdByWorktree, + groupsByWorktree, + unifiedTabsByWorktree, + tabsByWorktree, + ptyIdsByTabId: { [TERMINAL_TAB]: ['pty-1'] }, + suppressedPtyExitIds: {}, + terminalLayoutsByTabId: { + [TERMINAL_TAB]: { + root: { type: 'leaf', leafId: LEAF }, + activeLeafId: LEAF, + expandedLeafId: null, + ptyIdsByLeafId: { [LEAF]: 'pty-1' } + } + }, + browserTabsByWorktree: {}, + unreadAgentCompletionPanes: {}, + unreadTerminalTabs: {}, + unreadTerminalPanes: {}, + retainedAgentsByPaneKey: {}, + agentStatusByPaneKey: { + [CHAT_PANE_KEY]: agentRow(CHAT_PANE_KEY, CHAT_TAB, workspace), + [TERMINAL_PANE_KEY]: agentRow(TERMINAL_PANE_KEY, TERMINAL_TAB, workspace) + }, + folderWorkspaces: [ + { + id: 'folder-1', + name: 'notes', + projectGroupId: 'pg-1', + executionHostId: 'local' + } + ], + projectGroups: [{ id: 'pg-1', name: 'Notebook', executionHostId: 'local' }], + worktreesByRepo: { + repo1: [ + { + id: WORKSPACE, + repoId: 'repo1', + displayName: 'master', + branch: 'master' + }, + { + id: OTHER_WORKSPACE, + repoId: 'repo1', + displayName: 'other', + branch: 'other' + } + ] + }, + repos: [{ id: 'repo1', displayName: 'orca', connectionId: null }], + settings: { + experimentalTerminalAttention: seed.groupAttentionEnabled !== false, + notifications: { customSoundId: 'system', customSoundVolume: null } + }, + markWorktreeUnread: vi.fn(), + markTerminalTabUnread: vi.fn(), + markTerminalPaneUnread: vi.fn(), + markAgentCompletionPaneUnread: vi.fn() + } +} + +let state: ReturnType + +vi.mock('@/store', () => ({ useAppStore: { getState: () => state } })) +vi.mock('@/lib/desktop-notification-sound', () => ({ + playDesktopNotificationSound: vi.fn() +})) +vi.mock('@/lib/blocked-notification-fallback', () => ({ + showBlockedNotificationFallbackToast: vi.fn() +})) + +const { dispatchStructuredTurnCompletion } = await import('./structured-turn-completion-attention') +const { dispatchTerminalNotification } = await import('../terminal-pane/use-notification-dispatch') + +/** Every marker write the attention sinks made, as one comparable record. */ +function marks(seeded: ReturnType): Record { + return { + workspace: seeded.markWorktreeUnread.mock.calls.map(([id]) => id), + subject: seeded.markAgentCompletionPaneUnread.mock.calls.map(([key]) => key), + group: seeded.markTerminalTabUnread.mock.calls.map(([id]) => id), + surface: seeded.markTerminalPaneUnread.mock.calls.map(([key]) => key) + } +} + +/** Which markers were written and how many times, with the addresses dropped. */ +function markerShape(record: Record): Record { + return Object.fromEntries(Object.entries(record).map(([key, value]) => [key, value.length])) +} + +const NOTHING_LIT = { workspace: [], subject: [], group: [], surface: [] } + +/** Held directly rather than read back off the window stub, so no cast is needed to see it. */ +const notificationDispatch = vi.fn<(request: NotificationDispatchRequest) => Promise>() + +function lastDispatchRequest(): NotificationDispatchRequest | undefined { + return notificationDispatch.mock.calls.at(-1)?.[0] +} + +function seed(options: Seed = {}): ReturnType { + state = buildState(options) + return state +} + +function completeChat( + workspaceId = WORKSPACE, + outcome: StructuredTurnCompletion['outcome'] = 'success' +): void { + dispatchStructuredTurnCompletion( + completion({ + outcome, + scope: { + ...SCOPE, + workspaceId, + workspaceKind: workspaceId.startsWith('folder:') ? 'folder' : 'git-worktree' + } + }), + { workspaceId, paneKey: CHAT_PANE_KEY, label: 'Fix auth' } + ) +} + +function completeTerminal(workspaceId = WORKSPACE): void { + dispatchTerminalNotification(workspaceId, { + source: 'agent-task-complete', + paneKey: TERMINAL_PANE_KEY, + agentStatusSnapshot: agentRow(TERMINAL_PANE_KEY, TERMINAL_TAB, workspaceId) + }) +} + +beforeEach(() => { + vi.clearAllMocks() + notificationDispatch.mockResolvedValue({ delivered: true }) + vi.stubGlobal('window', { + api: { notifications: { dispatch: notificationDispatch } } + }) + // Orca in the background: unread is exactly what a user who cannot see the app needs. + vi.stubGlobal('document', { + visibilityState: 'hidden', + hasFocus: vi.fn(() => false) + }) + seed() +}) + +describe('a successful structured turn on a backgrounded workspace', () => { + it('lights the same markers a PTY agent lights, in the same run and the same workspace', () => { + completeChat() + const chat = marks(state) + + const control = seed() + completeTerminal() + const terminal = marks(control) + + // Workspace bold, the amber pane dot, the tab dot and the surface marker — the chat writes + // every one the terminal does, each addressed to its own surface. + expect(chat).toEqual({ + workspace: [WORKSPACE], + subject: [CHAT_PANE_KEY], + group: [CHAT_TAB], + surface: [CHAT_PANE_KEY] + }) + expect(terminal).toEqual({ + workspace: [WORKSPACE], + subject: [TERMINAL_PANE_KEY], + group: [TERMINAL_TAB], + surface: [TERMINAL_PANE_KEY] + }) + // Stated as a shape comparison too, so a change that drops one of the four from the chat path + // fails here even if someone updates the literal above to match. + expect(markerShape(chat)).toEqual(markerShape(terminal)) + }) + + it('requests delivery carrying the completion identity for mobile dedupe', () => { + completeChat() + + expect(lastDispatchRequest()).toMatchObject({ + source: 'agent-task-complete', + worktreeId: WORKSPACE, + paneKey: CHAT_PANE_KEY, + worktreeLabel: 'master', + repoLabel: 'orca', + isActiveWorktree: false, + mobileDedupeKey: structuredTurnCompletionKey(completion()) + }) + }) + + it('keys mobile dedupe on scope AND session AND turn, never on the turn id alone', () => { + const base = completion() + expect(structuredTurnCompletionKey(base)).toBe(structuredTurnCompletionKey(completion())) + for (const other of [ + completion({ turnId: 'turn-2' }), + completion({ sessionId: 'session-2' }), + completion({ scope: { ...SCOPE, executionHostId: 'ssh:box' } }), + completion({ scope: { ...SCOPE, wslDistro: 'Ubuntu' } }) + ]) { + expect(structuredTurnCompletionKey(other)).not.toBe(structuredTurnCompletionKey(base)) + } + }) +}) + +describe('a turn that did not succeed', () => { + it.each(['failure', 'cancellation'] as const)('lights nothing for %s', (outcome) => { + completeChat(WORKSPACE, outcome) + + expect(marks(state)).toEqual(NOTHING_LIT) + expect(lastDispatchRequest()).toBeUndefined() + }) +}) + +describe('visibility', () => { + it('writes no unread when the chat itself is the focused surface on screen', () => { + seed({ visible: 'chat', activeWorktreeId: WORKSPACE }) + vi.stubGlobal('document', { + visibilityState: 'visible', + hasFocus: vi.fn(() => true) + }) + + completeChat() + + expect(marks(state)).toEqual(NOTHING_LIT) + // A viewed surface suppresses UNREAD, not the delivery request: main owns the + // suppress-while-focused preference, and a paired phone may still need the alert. + expect(lastDispatchRequest()).toMatchObject({ isActiveWorktree: true }) + }) + + it('still marks a hidden chat tab inside the focused workspace', () => { + // The distinction the acceptance criteria turn on: the workspace is focused and on screen, + // but its group is showing the terminal, so the chat is a hidden sibling. + seed({ visible: 'terminal', activeWorktreeId: WORKSPACE }) + vi.stubGlobal('document', { + visibilityState: 'visible', + hasFocus: vi.fn(() => true) + }) + + completeChat() + + expect(marks(state)).toEqual({ + workspace: [WORKSPACE], + subject: [CHAT_PANE_KEY], + group: [CHAT_TAB], + surface: [CHAT_PANE_KEY] + }) + }) + + it('marks a selected-but-backgrounded workspace, because selection is not visibility', () => { + seed({ visible: 'chat', activeWorktreeId: WORKSPACE }) + + completeChat() + + expect(marks(state).workspace).toEqual([WORKSPACE]) + }) +}) + +describe('workspace kinds', () => { + it('marks a folder workspace, which is no git worktree and has no branch to name', () => { + // Folder workspaces reach every one of these paths: the surface reads the same unified tab + // index, and the notification labels resolve through the folder catalog, not a repo. + seed({ workspaceId: FOLDER_WORKSPACE }) + + completeChat(FOLDER_WORKSPACE) + + expect(marks(state)).toEqual({ + workspace: [FOLDER_WORKSPACE], + subject: [CHAT_PANE_KEY], + group: [CHAT_TAB], + surface: [CHAT_PANE_KEY] + }) + expect(lastDispatchRequest()).toMatchObject({ + worktreeId: FOLDER_WORKSPACE, + worktreeLabel: 'notes' + }) + }) +}) + +describe('addressing', () => { + it('rejects a completion whose chat tab is gone', () => { + seed({ chatTab: false }) + + completeChat() + + expect(marks(state)).toEqual(NOTHING_LIT) + expect(lastDispatchRequest()).toBeUndefined() + }) + + it('rejects a completion for a session the tab has since been rebound away from', () => { + seed({ boundSession: 'session-other' }) + + completeChat() + + expect(marks(state)).toEqual(NOTHING_LIT) + }) + + it('holds the container marker behind the same setting the terminal path does', () => { + seed({ groupAttentionEnabled: false }) + completeChat() + const chat = marks(state) + + const control = seed({ groupAttentionEnabled: false }) + completeTerminal() + + // Workspace bold and the amber pane dot are unconditional; the tab dot is presentation + // policy, gated identically for both surface kinds. + expect(chat).toEqual({ + workspace: [WORKSPACE], + subject: [CHAT_PANE_KEY], + group: [], + surface: [] + }) + expect(marks(control).group).toEqual([]) + }) +}) diff --git a/src/renderer/src/components/native-chat/structured-turn-completion-attention.ts b/src/renderer/src/components/native-chat/structured-turn-completion-attention.ts new file mode 100644 index 00000000000..70af9da9119 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-turn-completion-attention.ts @@ -0,0 +1,121 @@ +/** + * What a settled structured turn earns: workspace unread, the chat's own attention markers, and a + * delivery request — decided by the same provider-neutral policy the terminal path uses, asked + * through the structured surface adapter. + * + * There is deliberately no structured-specific suppression, liveness or sibling rule here. Every + * such question is `AgentAttentionSurface`'s, and `createStructuredAttentionSurface` already + * answers all of them for a chat tab; this module only supplies the request and the sinks. The + * sinks are the same four the terminal path writes, so a chat and a terminal in one workspace hold + * one shared set of markers rather than two that can disagree. + * + * ONLY `success` LIGHTS ANYTHING. A failure or a cancellation is a real event the host reports, + * and it earns no indicator: a dot that says "done!" for an API error or for a stop the user asked + * for teaches the user to distrust every dot. The transport never emits an unknown verdict at all. + */ +import { useAppStore } from '@/store' +import { + applyAgentAttention, + resolveAgentAttention, + type AgentAttentionDeliveryRequest +} from '@/attention/agent-attention-policy' +import { + deliverAgentAttentionNotification, + readAgentAttentionNotificationSound +} from '@/attention/agent-attention-notification-delivery' +import { buildAgentNotificationId } from '../../../../shared/agent-notification-id' +import type { StructuredTurnCompletion } from '../../../../shared/structured-turn-completion' +import { structuredTurnCompletionKey } from '../../../../shared/structured-turn-completion' +import { getNotificationWorkspaceLabels } from '../terminal-pane/terminal-notification-state' +import { createStructuredAttentionSurface } from './structured-attention-surface' + +export type StructuredTurnCompletionTarget = { + workspaceId: string + /** The pane key the structured status producer publishes: `:`. */ + paneKey: string + /** Tab label, used only as the notification's title fallback. */ + label: string +} + +/** + * Raise attention for one settled root turn. + * + * Returns nothing: whether a banner appears is main's decision (enabled/source preferences and + * suppress-while-focused are applied there, after mobile fan-out), and whether unread was written + * is the policy's. A caller must not read a return value as "the user was told". + */ +export function dispatchStructuredTurnCompletion( + completion: StructuredTurnCompletion, + target: StructuredTurnCompletionTarget +): void { + if (completion.outcome !== 'success') { + return + } + const state = useAppStore.getState() + const decision = resolveAgentAttention( + { + subject: { workspaceId: target.workspaceId, surfaceKey: target.paneKey }, + reason: 'agent-completion', + settlesTurn: true, + // Why: the host derived this from its own journal commit for a session that runs with no + // renderer PTY, so the event itself is the out-of-band proof the subject just produced + // work. Admission still has to place the surface — a key naming a closed or rebound tab is + // rejected there, not here. + hasFreshActivityEvidence: true, + // Same gate as the terminal path: the container marker is presentation policy, so a chat + // tab lights its dot under exactly the setting a terminal tab does. + groupAttentionEnabled: state.settings?.experimentalTerminalAttention === true + }, + createStructuredAttentionSurface(state) + ) + if (!decision.admitted) { + return + } + + const sound = readAgentAttentionNotificationSound(state.settings ?? {}) + const row = state.agentStatusByPaneKey?.[target.paneKey] + // Shares the agent notification id shape with the terminal path so an unread agent event and + // its OS notification stay dismissible under one id; null when the row has no turn timing yet. + const notificationId = buildAgentNotificationId({ + worktreeId: target.workspaceId, + paneKey: target.paneKey, + stateStartedAt: row?.stateStartedAt + }) + const requestDelivery = (request: AgentAttentionDeliveryRequest): void => { + deliverAgentAttentionNotification( + { + source: 'agent-task-complete', + ...(notificationId ? { notificationId } : {}), + // Why: the host is the only party that can see every connected window, so it owns the + // one-mobile-push-per-completion decision. Each window still decides its own banner. + mobileDedupeKey: structuredTurnCompletionKey(completion), + worktreeId: request.workspaceId, + paneKey: request.subjectKey ?? undefined, + ...getNotificationWorkspaceLabels(state, request.workspaceId, target.label), + terminalTitle: target.label, + isActiveWorktree: request.workspaceIsActive, + ...(row + ? { + agentType: row.agentType, + agentState: row.state, + agentPrompt: row.prompt, + agentLastAssistantMessage: row.lastAssistantMessage + } + : {}) + }, + sound + ) + } + + applyAgentAttention(decision, { + unread: { + markWorkspaceUnread: state.markWorktreeUnread, + markSubjectUnread: state.markAgentCompletionPaneUnread, + // The container id is the chat's unified tab id, which is what SortableTab reads this + // marker back under; `markTerminalTabUnread` accepts both tab indexes for that reason. + markGroupUnread: state.markTerminalTabUnread, + markSurfaceUnread: state.markTerminalPaneUnread + }, + requestDelivery + }) +} diff --git a/src/renderer/src/components/terminal-pane/use-notification-dispatch.ts b/src/renderer/src/components/terminal-pane/use-notification-dispatch.ts index f15c5067925..304e844d99d 100644 --- a/src/renderer/src/components/terminal-pane/use-notification-dispatch.ts +++ b/src/renderer/src/components/terminal-pane/use-notification-dispatch.ts @@ -1,8 +1,6 @@ import { useCallback } from 'react' import { useAppStore } from '@/store' import { resolveCommittedTitleAgentType } from '@/lib/pane-agent-evidence' -import { playDesktopNotificationSound } from '@/lib/desktop-notification-sound' -import { showBlockedNotificationFallbackToast } from '@/lib/blocked-notification-fallback' import { buildAgentNotificationId } from '../../../../shared/agent-notification-id' import { shareCompatibleTitleIdentityGroup } from '../../../../shared/agent-title-owner' import { @@ -21,6 +19,10 @@ import { resolveAgentAttention, type AgentAttentionDeliveryRequest } from '@/attention/agent-attention-policy' +import { + deliverAgentAttentionNotification, + readAgentAttentionNotificationSound +} from '@/attention/agent-attention-notification-delivery' const AGENT_NOTIFICATION_SNAPSHOT_MAX_AGE_MS = 10_000 @@ -133,8 +135,7 @@ export function dispatchTerminalNotification( // Desktop settings are applied in main after independent mobile delivery. - const customSoundId = state.settings?.notifications?.customSoundId ?? 'system' - const customSoundVolume = state.settings?.notifications?.customSoundVolume ?? null + const sound = readAgentAttentionNotificationSound(state.settings ?? {}) // Why: pane keys are reused across turns. A rich OS notification must not // expose the previous turn's prompt if the current turn has no fresh hook snapshot yet. const agentSnapshot = agentStatus @@ -161,8 +162,8 @@ export function dispatchTerminalNotification( : null const requestDelivery = (request: AgentAttentionDeliveryRequest): void => { - void window.api.notifications - .dispatch({ + deliverAgentAttentionNotification( + { source: event.source, ...(notificationId ? { notificationId } : {}), worktreeId: request.workspaceId, @@ -171,22 +172,9 @@ export function dispatchTerminalNotification( terminalTitle: event.terminalTitle, isActiveWorktree: request.workspaceIsActive, ...agentSnapshot - }) - .then((result) => { - if (result.delivered) { - void playDesktopNotificationSound(customSoundId, customSoundVolume) - return - } - // Why: macOS is silently swallowing notifications (permission off or - // prompt unanswered) — surface an in-app pointer at the fix instead of - // letting the alert vanish without a trace. - if (result.reason === 'blocked-by-system') { - showBlockedNotificationFallbackToast() - } - }) - .catch((err) => { - console.warn('Failed to dispatch notification:', err) - }) + }, + sound + ) } applyAgentAttention(attentionDecision, { diff --git a/src/renderer/src/runtime/structured-agent-session-client.ts b/src/renderer/src/runtime/structured-agent-session-client.ts index 5ccbe6360f3..3d52a974a62 100644 --- a/src/renderer/src/runtime/structured-agent-session-client.ts +++ b/src/renderer/src/runtime/structured-agent-session-client.ts @@ -105,6 +105,24 @@ export function subscribeStructuredAgentSession( ) } +/** Root turns settling on one runtime from now on. Raw because the caller validates: an event + * this build cannot place must be dropped rather than handed on as a completion. */ +export function subscribeStructuredAgentSessionTurnCompletion( + target: RuntimeClientTarget, + onEvent: (event: unknown) => void, + onError: (error: unknown) => void, + onClose: () => void +): Promise<{ unsubscribe: () => void }> { + return subscribeStructuredAgentSessionMethod( + target, + 'agentSession.subscribeTurnCompletion', + {}, + onEvent, + onError, + onClose + ) +} + /** Every structured session's projected status on one runtime, as the host publishes it. */ export function subscribeStructuredAgentSessionStatus( target: RuntimeClientTarget, diff --git a/src/renderer/src/runtime/structured-turn-completion-feed.test.ts b/src/renderer/src/runtime/structured-turn-completion-feed.test.ts new file mode 100644 index 00000000000..d28f742c655 --- /dev/null +++ b/src/renderer/src/runtime/structured-turn-completion-feed.test.ts @@ -0,0 +1,180 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AGENT_SESSION_TURN_COMPLETION_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' +import type { StructuredTurnCompletion } from '../../../shared/structured-turn-completion' + +const mocks = vi.hoisted(() => ({ + subscribeCompletion: vi.fn(), + supportsCapability: vi.fn(), + unsubscribe: vi.fn() +})) + +vi.mock('./structured-agent-session-client', () => ({ + subscribeStructuredAgentSessionTurnCompletion: mocks.subscribeCompletion +})) + +vi.mock('./runtime-rpc-client', () => ({ + runtimeEnvironmentSupportsCapability: mocks.supportsCapability +})) + +import { + getStructuredTurnCompletionFeed, + resetStructuredTurnCompletionFeedsForTests +} from './structured-turn-completion-feed' + +const REMOTE = { kind: 'environment', environmentId: 'env-1' } as const +const LOCAL = { kind: 'local' } as const + +function completionFrame(turnId = 'turn-1'): unknown { + return { + type: 'completion', + completion: { + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'wt-1', + workspaceKind: 'git-worktree' + }, + sessionId: 'session-1', + turnId, + outcome: 'success', + completedAt: 1_700 + } + } +} + +/** The raw event callback the feed handed to the most recent subscription. */ +function hostEmit(index = 0): (event: unknown) => void { + const call = mocks.subscribeCompletion.mock.calls[index] + if (!call) { + throw new Error('completion feed not subscribed') + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: argument 1 of the mocked subscribe is always the event callback. + return call[1] as (event: unknown) => void +} + +describe('structured turn completion feed', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.clearAllMocks() + resetStructuredTurnCompletionFeedsForTests() + mocks.subscribeCompletion.mockResolvedValue({ + unsubscribe: mocks.unsubscribe + }) + mocks.supportsCapability.mockResolvedValue(true) + }) + + afterEach(() => { + resetStructuredTurnCompletionFeedsForTests() + vi.useRealTimers() + }) + + it('never subscribes, and never retries, against a host without the capability', async () => { + mocks.supportsCapability.mockResolvedValue(false) + getStructuredTurnCompletionFeed(REMOTE).activate() + await vi.advanceTimersByTimeAsync(30_000) + + expect(mocks.supportsCapability).toHaveBeenCalledWith( + 'env-1', + AGENT_SESSION_TURN_COMPLETION_RUNTIME_CAPABILITY + ) + expect(mocks.subscribeCompletion).not.toHaveBeenCalled() + }) + + it('does not probe a local host, which is this build', async () => { + getStructuredTurnCompletionFeed(LOCAL).activate() + await vi.advanceTimersByTimeAsync(0) + + expect(mocks.supportsCapability).not.toHaveBeenCalled() + expect(mocks.subscribeCompletion).toHaveBeenCalledTimes(1) + }) + + it('announces a completion to every listener', async () => { + const feed = getStructuredTurnCompletionFeed(LOCAL) + feed.activate() + const seen: StructuredTurnCompletion[] = [] + const other: StructuredTurnCompletion[] = [] + feed.subscribe((completion) => seen.push(completion)) + feed.subscribe((completion) => other.push(completion)) + await vi.advanceTimersByTimeAsync(0) + + hostEmit()(completionFrame()) + + expect(seen.map((completion) => completion.turnId)).toEqual(['turn-1']) + expect(other.map((completion) => completion.turnId)).toEqual(['turn-1']) + }) + + it('drops an event it cannot place rather than passing on a half-read completion', async () => { + const feed = getStructuredTurnCompletionFeed(LOCAL) + feed.activate() + const seen: StructuredTurnCompletion[] = [] + feed.subscribe((completion) => seen.push(completion)) + await vi.advanceTimersByTimeAsync(0) + + hostEmit()({ type: 'completion', completion: { sessionId: 'session-1' } }) + hostEmit()({ type: 'completion' }) + hostEmit()('nonsense') + + expect(seen).toEqual([]) + }) + + it('keeps every other listener when one throws', async () => { + const feed = getStructuredTurnCompletionFeed(LOCAL) + feed.activate() + const seen: StructuredTurnCompletion[] = [] + feed.subscribe(() => { + throw new Error('listener exploded') + }) + feed.subscribe((completion) => seen.push(completion)) + await vi.advanceTimersByTimeAsync(0) + + hostEmit()(completionFrame()) + + expect(seen).toHaveLength(1) + }) + + it('baselines on reconnect: nothing is queued or replayed across the gap', async () => { + const feed = getStructuredTurnCompletionFeed(LOCAL) + feed.activate() + const seen: StructuredTurnCompletion[] = [] + feed.subscribe((completion) => seen.push(completion)) + await vi.advanceTimersByTimeAsync(0) + + // The host ends the stream; the client reconnects. + hostEmit()({ type: 'end' }) + await vi.advanceTimersByTimeAsync(5_000) + expect(mocks.subscribeCompletion).toHaveBeenCalledTimes(2) + + // A reopened stream opens on nothing. If a completion that happened during the gap ever + // arrives here, live-only recovery has silently become catch-up. + expect(seen).toEqual([]) + hostEmit(1)(completionFrame('turn-after-reconnect')) + expect(seen.map((completion) => completion.turnId)).toEqual(['turn-after-reconnect']) + }) + + it('tears the stream down once nothing is activated, leaving no reconnect running', async () => { + const feed = getStructuredTurnCompletionFeed(LOCAL) + const release = feed.activate() + await vi.advanceTimersByTimeAsync(0) + release() + + expect(mocks.unsubscribe).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(30_000) + expect(mocks.subscribeCompletion).toHaveBeenCalledTimes(1) + }) + + it('shares one stream per runtime target across every activation', async () => { + const first = getStructuredTurnCompletionFeed(LOCAL) + const second = getStructuredTurnCompletionFeed(LOCAL) + expect(second).toBe(first) + + const releaseFirst = first.activate() + const releaseSecond = second.activate() + await vi.advanceTimersByTimeAsync(0) + + expect(mocks.subscribeCompletion).toHaveBeenCalledTimes(1) + releaseFirst() + expect(mocks.unsubscribe).not.toHaveBeenCalled() + releaseSecond() + expect(mocks.unsubscribe).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/renderer/src/runtime/structured-turn-completion-feed.ts b/src/renderer/src/runtime/structured-turn-completion-feed.ts new file mode 100644 index 00000000000..6a1c0de4a83 --- /dev/null +++ b/src/renderer/src/runtime/structured-turn-completion-feed.ts @@ -0,0 +1,212 @@ +// One host turn-completion stream per runtime target, shared by every surface that raises +// attention for a structured chat. +// +// Unlike the status feed this owner caches nothing. A completion is an edge, not a state: there is +// no snapshot to merge, nothing to re-read on reconnect, and nothing a late listener can be told +// about. Losing the stream reconnects and the client BASELINES — completions that landed while it +// was away are dropped, not queued. That is the decided recovery policy, and the host holds its +// own half of it; a test pins this side so a later refactor cannot quietly turn it into catch-up. + +import { AGENT_SESSION_TURN_COMPLETION_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' +import { + readStructuredTurnCompletionEvent, + type StructuredTurnCompletion +} from '../../../shared/structured-turn-completion' +import { + runtimeEnvironmentSupportsCapability, + type RuntimeClientTarget +} from './runtime-rpc-client' +import { subscribeStructuredAgentSessionTurnCompletion } from './structured-agent-session-client' + +export type StructuredTurnCompletionListener = (completion: StructuredTurnCompletion) => void + +export type StructuredTurnCompletionFeedOwner = { + /** Holds the stream open while any caller is activated; the last release tears it down. */ + activate: () => () => void + subscribe: (listener: StructuredTurnCompletionListener) => () => void +} + +const RECONNECT_MAX_DELAY_MS = 5_000 + +type OwnedCompletionFeed = StructuredTurnCompletionFeedOwner & { + stop: () => void +} + +const owners = new Map() + +export function structuredTurnCompletionFeedKey(target: RuntimeClientTarget): string { + return target.kind === 'local' ? 'local' : `environment:${target.environmentId}` +} + +function createOwner(target: RuntimeClientTarget): OwnedCompletionFeed { + const listeners = new Set() + const activations = new Set() + let generation = 0 + let handle: { unsubscribe: () => void } | null = null + let reconnectTimer: ReturnType | null = null + let reconnectAttempt = 0 + + 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 + } + const announce = (completion: StructuredTurnCompletion): void => { + // A Set skips entries deleted mid-iteration, so a listener that unsubscribes itself from + // inside its own callback is safe here. + for (const listener of listeners) { + try { + listener(completion) + } catch (error) { + // One surface throwing must not cost every other surface its completion. + console.warn('[structured-turn-completion] listener failed', error) + } + } + } + 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 loseConnection = (candidate: number): void => { + if (candidate !== generation) { + return + } + generation += 1 + dropHandle() + scheduleReconnect(generation) + } + const subscribeToHost = (candidate: number): void => { + void subscribeStructuredAgentSessionTurnCompletion( + target, + (raw) => { + if (!active(candidate)) { + return + } + // An event this build cannot place is dropped, never coerced: inventing a verdict is how + // a dot ends up claiming a failed turn finished well. + const event = readStructuredTurnCompletionEvent(raw) + if (!event) { + return + } + if (event.type === 'end') { + loseConnection(candidate) + return + } + reconnectAttempt = 0 + announce(event.completion) + }, + () => { + if (active(candidate)) { + loseConnection(candidate) + } + }, + () => { + if (active(candidate)) { + loseConnection(candidate) + } + } + ) + .then((opened) => { + if (active(candidate)) { + handle = opened + } else { + opened.unsubscribe() + } + }) + .catch(() => loseConnection(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_TURN_COMPLETION_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-turn-completion] host too old for the completion feed', + environmentId + ) + }) + .catch(() => loseConnection(candidate)) + } + const stop = (): void => { + generation += 1 + clearReconnect() + dropHandle() + reconnectAttempt = 0 + } + + return { + activate: () => { + const token = Symbol('turn-completion-feed') + activations.add(token) + if (activations.size === 1) { + open() + } + return () => { + activations.delete(token) + if (activations.size === 0) { + stop() + } + } + }, + subscribe: (listener) => { + listeners.add(listener) + return () => listeners.delete(listener) + }, + stop + } +} + +export function getStructuredTurnCompletionFeed( + target: RuntimeClientTarget +): StructuredTurnCompletionFeedOwner { + const key = structuredTurnCompletionFeedKey(target) + let owner = owners.get(key) + if (!owner) { + owner = createOwner(target) + owners.set(key, owner) + } + return owner +} + +export function resetStructuredTurnCompletionFeedsForTests(): 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/notification-event-dedupe.ts b/src/shared/notification-event-dedupe.ts new file mode 100644 index 00000000000..67fbfefd68a --- /dev/null +++ b/src/shared/notification-event-dedupe.ts @@ -0,0 +1,27 @@ +/** + * At-most-once reservation for a notification event that has its own identity. + * + * Distinct from the burst cooldown beside it, which answers "has this workspace been noisy + * lately" and expires on a timer. An event identity does not expire: the same completion is the + * same completion an hour later, so a time window would let a slow second window raise it twice. + * + * Bounded by count rather than by age, and insertion-ordered so eviction drops the oldest. The + * only thing eviction can cost is a duplicate for an event that was already announced and has + * since had this many newer events behind it, which no real sequence reaches. + */ +const MAX_REMEMBERED_EVENTS = 256 + +export function reserveNotificationEventOnce(seen: Set, key: string): boolean { + if (seen.has(key)) { + return false + } + seen.add(key) + while (seen.size > MAX_REMEMBERED_EVENTS) { + const oldest = seen.values().next() + if (oldest.done) { + break + } + seen.delete(oldest.value) + } + return true +} diff --git a/src/shared/notification-settings-types.ts b/src/shared/notification-settings-types.ts index 1c06b90d8db..b4f5f8f8e34 100644 --- a/src/shared/notification-settings-types.ts +++ b/src/shared/notification-settings-types.ts @@ -31,6 +31,12 @@ export type NotificationDispatchRequest = { worktreeId?: string /** Stable `${tabId}:${leafId}` terminal pane key for click-to-focus routing. */ paneKey?: string + /** + * Exact identity of the event behind this dispatch, when the sender has one, so mobile fan-out + * is at most once per event no matter how many windows dispatched it. Additive: without it the + * coarse per-workspace burst dedupe is the only mobile gate, exactly as before. + */ + mobileDedupeKey?: string repoLabel?: string worktreeLabel?: string /** Legacy senders may still provide this; project labels are now always shown. */ diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index a7c15bc1296..05be53d101e 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -186,6 +186,14 @@ export const STRUCTURED_AGENT_SESSION_RESUME_HISTORY_RUNTIME_CAPABILITY = // 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: agentSession.subscribeTurnCompletion is a whole new stream on a surface that already +// shipped, so a host advertising agent-session.status-feed.v1 may still answer it with +// method_not_found — and a client that cannot tell that apart from a transport fault reconnects +// forever. It is also the only way a client learns the host derives completions at all: the +// summary feed looks identical on a host that does not, so calling and reading the silence +// proves nothing. +export const AGENT_SESSION_TURN_COMPLETION_RUNTIME_CAPABILITY = + 'agent-session.turn-completion.v1' as const // The RPC is registered unconditionally; per-session rewind support is a separate check. export const AGENT_SESSION_REWIND_RUNTIME_CAPABILITY = 'agent-session.rewind.v1' as const // Readers must understand a monitoring roster with no available stop control. @@ -345,6 +353,7 @@ export const RUNTIME_CAPABILITIES = [ STRUCTURED_AGENT_SESSION_REVEAL_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_RESUME_HISTORY_RUNTIME_CAPABILITY, AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, + AGENT_SESSION_TURN_COMPLETION_RUNTIME_CAPABILITY, AGENT_SESSION_REWIND_RUNTIME_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY, AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index 7278e5eda8e..736f95536d4 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -587,6 +587,7 @@ export const RPC_PARAMS_BY_METHOD = { 'agentSession.setOption': SetOptionParams, 'agentSession.subscribe': SubscribeParams, 'agentSession.subscribeStatus': null, + 'agentSession.subscribeTurnCompletion': null, 'agentSession.unsubscribe': UnsubscribeParams, 'agentTeams.prepareLaunch': AgentTeamsPrepareLaunch, 'agentTeams.tmuxCompat': AgentTeamsTmuxCompat, diff --git a/src/shared/structured-turn-completion.test.ts b/src/shared/structured-turn-completion.test.ts new file mode 100644 index 00000000000..31c5ca11347 --- /dev/null +++ b/src/shared/structured-turn-completion.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it } from 'vitest' +import { + readStructuredTurnCompletionEvent, + structuredTurnCompletionKey +} from './structured-turn-completion' + +const SCOPE = { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' +} as const + +function frame(overrides: Record = {}): Record { + return { + type: 'completion', + completion: { + scope: SCOPE, + sessionId: 'session-1', + turnId: 'turn-1', + outcome: 'success', + completedAt: 1_700, + ...overrides + } + } +} + +describe('readStructuredTurnCompletionEvent', () => { + it('reads a completion', () => { + expect(readStructuredTurnCompletionEvent(frame())).toEqual({ + type: 'completion', + completion: { + scope: SCOPE, + sessionId: 'session-1', + turnId: 'turn-1', + outcome: 'success', + completedAt: 1_700 + } + }) + }) + + it('reads the stream end', () => { + expect(readStructuredTurnCompletionEvent({ type: 'end' })).toEqual({ + type: 'end' + }) + }) + + it('keeps reading a newer host that added fields it has never heard of', () => { + const event = readStructuredTurnCompletionEvent({ + ...frame({ tokensUsed: 4_096, scope: { ...SCOPE, datacentre: 'iad' } }), + priority: 'high' + }) + expect(event?.type).toBe('completion') + if (event?.type !== 'completion') { + throw new Error('a forward-compatible frame must still read') + } + // Unknown keys are ignored, and the scope is rebuilt from the members this build knows, so a + // newer host's extra scope field cannot silently change the dedupe identity either. + expect(event.completion.scope).toEqual(SCOPE) + }) + + it('refuses an outcome arm from a later vocabulary rather than inventing a verdict', () => { + // A degrade-to-`success` here would light a dot for something this build cannot place; + // refusing matches the answer it gives for a turn whose outcome was never recorded. + expect(readStructuredTurnCompletionEvent(frame({ outcome: 'partial-success' }))).toBeNull() + }) + + it.each([ + ['an absent outcome', frame({ outcome: undefined })], + ['an empty session id', frame({ sessionId: '' })], + ['an empty turn id', frame({ turnId: '' })], + [ + 'an unparseable execution host', + frame({ scope: { ...SCOPE, executionHostId: 'nonsense:x' } }) + ], + ['a non-numeric clock', frame({ completedAt: 'soon' })], + ['an infinite clock', frame({ completedAt: Number.POSITIVE_INFINITY })], + ['an unknown workspace kind', frame({ scope: { ...SCOPE, workspaceKind: 'shelf' } })], + ['a missing completion', { type: 'completion' }], + ['an unknown event type', { type: 'progress' }], + ['a non-object', 'completion'], + ['null', null] + ])('refuses %s', (_label, value) => { + expect(readStructuredTurnCompletionEvent(value)).toBeNull() + }) + + it('normalizes the execution host so the dedupe scope matches what this build writes', () => { + const event = readStructuredTurnCompletionEvent( + frame({ scope: { ...SCOPE, executionHostId: 'ssh:box-1' } }) + ) + if (event?.type !== 'completion') { + throw new Error('a valid ssh host must read') + } + expect(event.completion.scope.executionHostId).toBe('ssh:box-1') + }) +}) + +describe('structuredTurnCompletionKey', () => { + it('separates every part, so no two identities can collide by concatenation', () => { + const left = structuredTurnCompletionKey({ + scope: SCOPE, + sessionId: 'a', + turnId: 'b:c' + }) + const right = structuredTurnCompletionKey({ + scope: SCOPE, + sessionId: 'a:b', + turnId: 'c' + }) + expect(left).not.toBe(right) + }) +}) diff --git a/src/shared/structured-turn-completion.ts b/src/shared/structured-turn-completion.ts new file mode 100644 index 00000000000..aab7b373e54 Binary files /dev/null and b/src/shared/structured-turn-completion.ts differ