From bd1b8075da45fe159982ee5b92b6cc2f5f40df3c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 18 Sep 2026 03:27:22 -0400 Subject: [PATCH] feat(attention): light the unread indicators when a structured chat finishes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A finished structured native chat showed nothing outside its own Agent Activity badge. The workspace never went bold, the tab never took a dot, and the amber pane dot never appeared — three of the four indicators a finished PTY agent lights. The row already knew the chat had settled; nothing converted "this agent finished" into "you should look at this". Add that missing step. The execution host derives a completion at journal commit and publishes it on a new capability-negotiated stream; each connected window hands it to the attention policy already shared with the terminal path, asked through the structured surface adapter. Workspace bold, amber pane dot and tab dot now light on exactly the rules a PTY agent's completion lights them on. Only `success` lights anything. A failure or a cancellation is announced and earns no indicator, and a turn whose provider verdict was never recorded is not announced at all — a dot that says "done!" for an API error teaches the user to distrust every dot. Recovery is live-only by decision: nothing is persisted and nothing is queued, so nothing can strand. A session baselines on the first edge the feed sees, and a subscriber opens on nothing, so restore, rewind, restart and a reconnect after a gap all announce nothing. Both halves are pinned by tests, because a later refactor turning either into catch-up would otherwise be silent. --- ...tructured-agent-session-client-delivery.ts | 26 +- .../structured-agent-session-host.ts | 6 + .../structured-turn-completion-feed.test.ts | 324 +++++++++++++ .../structured-turn-completion-feed.ts | 168 +++++++ .../notification-delivery-service.test.ts | 41 ++ .../notification-delivery-service.ts | 13 +- ...ession-gate-classification.test-fixture.ts | 3 +- ...ructured-agent-session-rpc.test-fixture.ts | 1 + ...tructured-agent-session-subscription-id.ts | 6 + ...ed-agent-session-turn-completion-stream.ts | 33 ++ .../methods/structured-agent-session.test.ts | 2 +- .../rpc/methods/structured-agent-session.ts | 4 +- .../agent-attention-notification-delivery.ts | 52 ++ ...tructuredAgentSessionStatusBridge.test.tsx | 13 +- .../StructuredAgentSessionStatusBridge.tsx | 6 + .../StructuredTurnCompletionAttention.tsx | 43 ++ ...ructured-turn-completion-attention.test.ts | 444 ++++++++++++++++++ .../structured-turn-completion-attention.ts | 121 +++++ .../use-notification-dispatch.ts | 32 +- .../structured-agent-session-client.ts | 18 + .../structured-turn-completion-feed.test.ts | 180 +++++++ .../structured-turn-completion-feed.ts | 212 +++++++++ src/shared/notification-event-dedupe.ts | 27 ++ src/shared/notification-settings-types.ts | 6 + src/shared/protocol-version.ts | 9 + .../rpc-params-catalog.generated.ts | 1 + src/shared/structured-turn-completion.test.ts | 112 +++++ src/shared/structured-turn-completion.ts | Bin 0 -> 4397 bytes 28 files changed, 1873 insertions(+), 30 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-turn-completion-feed.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-turn-completion-feed.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-turn-completion-stream.ts create mode 100644 src/renderer/src/attention/agent-attention-notification-delivery.ts create mode 100644 src/renderer/src/components/native-chat/StructuredTurnCompletionAttention.tsx create mode 100644 src/renderer/src/components/native-chat/structured-turn-completion-attention.test.ts create mode 100644 src/renderer/src/components/native-chat/structured-turn-completion-attention.ts create mode 100644 src/renderer/src/runtime/structured-turn-completion-feed.test.ts create mode 100644 src/renderer/src/runtime/structured-turn-completion-feed.ts create mode 100644 src/shared/notification-event-dedupe.ts create mode 100644 src/shared/structured-turn-completion.test.ts create mode 100644 src/shared/structured-turn-completion.ts 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 0000000000000000000000000000000000000000..aab7b373e549dae369de0cbeaeca25a08d9c0105 GIT binary patch literal 4397 zcmbtXU2h{b6y>?U;_{TFN~Qvdmq3sTRBYMZ7Hzu%X@y#6#%a8rv4ibt+Afv&5&S}a z3FlsWCex)`A<>7n@p$g%Ip^LS?CMtzC$tK+0Oz+EO>#Mv7|{$yrSy`nHU% z*AMH}yKaO4SQDpSTO1&8C(HJe&TxQ7)~AhL(0Ca+6}n@C&oFmW!IEwX@( zt!wLw*6K!6bY$!bXIw40tGWoIS2h^7J(VyO+uj{=hH!|)IMfoUq`97?y* zC{gOnaW$3Dj;qOKL5F=Ufd-C8z+fuQr?uwyR-+d69m*1r7)LE#OQv!qI?I$^llzTU z7GbX1n%m9DVPqyKBL0Y?nAEne&v2*+#cDU2U0mxsTEcG(x?XW!2YaU3I3MW)-P1$r z&gkA%quse00+5q_sqc{N-A2<8n@6|0Y`K%KT&V;tow?#bOw4krW(*uZ^Z@uZsy6q! zxA6jL9anuo#ewdu&xDJ8bbCaHKOUW&U;K3X_RY!RtBZ5|xj21${{89eqq8X0)4ZfuEoh&{@|2M#23d5~R(aFE(H6kT?gE!LJti0{007~+wr59C(8R0Gs3V8(*23pB}B#4!@ zFUceFw$iIy7}g%eE6E{}zVfQlv*>1OCW1HP5(!Y!W@N1NTXo5#VVnT_DG34SunLxI z-6+8;Rtaav@1QI9g<$Vg@Hp;73a17oR`qzqYBQmS2~Wh2F5@Ik$vHpiUf3zYNM51= z1?3`Rz4Ip~`a$)SefDv1KquX#2ci`!&b3C{qPo_WE3pI^8_-4it$fhZZ(&VZgVM~% z2MHX~Vz!STxJXKwxRVe9FmZDO(Dfl*f1+|=&p12xzKAgh>B#}@Zcw?6))KSl;m_d@P2f82)p2XNoJVTSC0gBTAl zI)5ED;3uyIeKB@nY`f<#K}bf6#>5@J8(o<=<7W{QYBZ(M%GLP3G5T$i6G-^vPo-`Rcoz4noi{1g z)G2E?7Yz~vEDf|IYn$j8F!tndX1#uv{tO!Fe!oiXVX{8^3f2~hhqu!tFjx*)wHMF7 zmz1UmYNdbX7cxd+#jI%jc`{Ojwk(0TiHvoEz=}+=L#|oxGAZza_Y$}+Diu{Wl!lC~ z)(ZeSOBDy(Tk8Z(o{7<&;Sb%I>><;A6GJD`D%mdz++}>JSfpWF>@*I}a^T7Dzn8oO z?%f+{KN6_&ccR@#+R?3(3z_3Z&%$&(V9Snn>1NNv!JUp#vWT