diff --git a/src/main/agent-hooks/server-ingest-structured-state-clock.test.ts b/src/main/agent-hooks/server-ingest-structured-state-clock.test.ts new file mode 100644 index 00000000000..eeb8ffab51e --- /dev/null +++ b/src/main/agent-hooks/server-ingest-structured-state-clock.test.ts @@ -0,0 +1,96 @@ +// The host's status row takes its state clock from the summary's `statusStartedAt`, the session's +// own lifecycle clock, so `worktree ps`, mobile and the dashboard date a parent the way the sidebar +// does. An older summary without the clock keeps the ingest's own continuity rule. + +import { beforeEach, describe, expect, it } from 'vitest' +import type { AgentSessionStatusSummary } from '../../shared/agent-session-wire' +import { makeStructuredAgentStatusSubject } from '../../shared/agent-status-subject' +import { AgentHookServer, _internals } from './server' + +const SESSION = 'b2c3d4e5-f6a7-4b8c-9d0e-1f2a3b4c5d6e' +const SUBJECT = makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'repo-1::/workspace/app', + workspaceKind: 'git-worktree' + }, + SESSION +) +const SETTLED = 22_000 + +function summary(over: Partial = {}): AgentSessionStatusSummary { + return { + sessionId: SESSION, + workspaceId: 'repo-1::/workspace/app', + agent: 'codex', + status: 'idle', + hostExecutionOwned: true, + latestPrompt: 'fan out', + updatedAt: SETTLED, + ...over + } +} + +function row(server: AgentHookServer) { + return server.getStatusSnapshot()[0] +} + +beforeEach(() => { + _internals.resetCachesForTests() +}) + +describe("the host row's state clock", () => { + it("dates a subagent's approval at the ask, and the parent's completion where it was", () => { + const server = new AgentHookServer() + server.ingestStructuredStatus(summary({ statusStartedAt: SETTLED }), SUBJECT) + server.ingestStructuredStatus( + summary({ status: 'attention', statusStartedAt: 27_000, updatedAt: 27_000 }), + SUBJECT + ) + expect(row(server)).toMatchObject({ + state: 'blocked', + stateStartedAt: 27_000, + mainAgent: { state: 'blocked', stateStartedAt: 27_000 } + }) + + server.ingestStructuredStatus(summary({ statusStartedAt: SETTLED, updatedAt: 28_500 }), SUBJECT) + expect(row(server)).toMatchObject({ + state: 'done', + stateStartedAt: SETTLED, + mainAgent: { state: 'done', stateStartedAt: SETTLED } + }) + }) + + it('settles a row child work held open on the parent clock, not the last child row', () => { + const server = new AgentHookServer() + server.ingestStructuredStatus( + summary({ status: 'working', statusStartedAt: 10_000, updatedAt: 10_000 }), + SUBJECT + ) + server.ingestStructuredStatus( + summary({ + statusStartedAt: SETTLED, + updatedAt: 24_000, + backgroundTasks: [{ id: 'child-1', kind: 'agent', state: 'working' }] + }), + SUBJECT + ) + expect(row(server)).toMatchObject({ state: 'working', stateStartedAt: 10_000 }) + + server.ingestStructuredStatus(summary({ statusStartedAt: SETTLED, updatedAt: 26_000 }), SUBJECT) + expect(row(server)).toMatchObject({ state: 'done', stateStartedAt: SETTLED }) + }) + + it("keeps the ingest's own continuity for an older host's summary, which carries no clock", () => { + const server = new AgentHookServer() + server.ingestStructuredStatus(summary(), SUBJECT) + server.ingestStructuredStatus(summary({ status: 'attention', updatedAt: 27_000 }), SUBJECT) + server.ingestStructuredStatus(summary({ updatedAt: 28_500 }), SUBJECT) + expect(row(server)).toMatchObject({ + state: 'done', + stateStartedAt: 28_500, + mainAgent: { state: 'done', stateStartedAt: 28_500 } + }) + }) +}) diff --git a/src/main/agent-hooks/server/server-ingest-structured.ts b/src/main/agent-hooks/server/server-ingest-structured.ts index b57b4d62040..b936d28b0e2 100644 --- a/src/main/agent-hooks/server/server-ingest-structured.ts +++ b/src/main/agent-hooks/server/server-ingest-structured.ts @@ -14,6 +14,10 @@ import { isAgentStatusHeldOpenByChildWork } from '../../../shared/agent-lead-status-fold' import { structuredAgentSessionAgentStatus } from '../../../shared/structured-agent-session-agent-status' +import { + structuredAgentSessionDatedMainAgent, + structuredAgentSessionRowStateStartedAt +} from '../../../shared/structured-agent-session-status-started-at' import { structuredStatusLegacyEvent } from './server-structured-status-row' import { AgentHookServerIngestTerminal } from './server-ingest-terminal' @@ -49,7 +53,7 @@ export abstract class AgentHookServerIngestStructured extends AgentHookServerIng // by the journal: a restart's republish is not a new main agent state either. const mainAgent = continueMainAgentStatus( priorStatus?.mainAgent, - agentStatus.mainAgent, + structuredAgentSessionDatedMainAgent(agentStatus.mainAgent, summary), summary.updatedAt ) const tabId = structuredAgentSessionTabId(parsed.sessionId) @@ -88,9 +92,10 @@ export abstract class AgentHookServerIngestStructured extends AgentHookServerIng // Continuity is the whole published work identity: `state` alone no longer means "a turn is // running", so monitoring that becomes a real turn must restart the clock, not inherit it. stateStartedAt: - priorStatus?.state === state && priorStatus.workingMode === workingMode + structuredAgentSessionRowStateStartedAt({ state, mainAgent }, summary) ?? + (priorStatus?.state === state && priorStatus.workingMode === workingMode ? priorStatus.stateStartedAt - : summary.updatedAt, + : summary.updatedAt), observation: { origin: 'structured', kind: 'transition', diff --git a/src/main/native-chat/agent-session-journal/journal-render-item.ts b/src/main/native-chat/agent-session-journal/journal-render-item.ts index 3324ffedbd1..8d821629c2b 100644 --- a/src/main/native-chat/agent-session-journal/journal-render-item.ts +++ b/src/main/native-chat/agent-session-journal/journal-render-item.ts @@ -21,6 +21,7 @@ export function journalRenderItem( body, sequence: row.seq, observedAt: row.ts, + ...(row.recovered ? { recoveredAt: row.ts } : {}), ...(row.recovered ? { recovered: row.recovered } : {}), ...agentJournalLinkageFields(row) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts new file mode 100644 index 00000000000..a239e4a5154 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovered-turn-clock.test.ts @@ -0,0 +1,210 @@ +// A turn that was running when its host went away ends when recovery settles it. That settlement is +// the edge the user needs to see — their work stopped — so the session reads as newly done then, +// and nothing along the way may call it a success. Every hop is the real one: durable journal, +// recovery settlement, status feed, the host's status row, and the turn-completion feed. + +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 { + AgentSessionStatusSummary, + AgentSessionTurnCompletionEvent +} from '../../../shared/agent-session-wire' +import { AgentHookServer, _internals } from '../../agent-hooks/server' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { settleStructuredAgentSessionDeadGeneration } from './structured-agent-session-dead-generation-settlement' +import { + settleStaleSessionStateOnAcquire, + type StructuredAgentSessionTurnVerdict +} from './structured-agent-session-stale-turn-verdict' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' +import { indexedStatusFeedSession } from './structured-agent-session-status-feed-test-session' +import { StructuredAgentSessionTurnCompletionFeed } from './structured-agent-session-turn-completion-feed' + +const SESSION = 'recovered-turn-session' +const THREAD = 'thread-1' +const TURN_STARTED = 1_000 +const EXIT_OBSERVED = 2_000 +const RECOVERED = 9_000 + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + _internals.resetCachesForTests() + root = await mkdtemp(join(tmpdir(), 'orca-recovered-turn-')) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +/** A session whose turn was running when its host went away, reopened by the next host. */ +async function sessionWithRunningTurn() { + let clock = TURN_STARTED + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + now: () => clock, + journalDir: join(root, SESSION) + }) + await journal.appendItem( + { provider: 'orca', clientMessageId: 'prompt-1' }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'long job' }] }, + { fence: 1 } + ) + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 9 }, + { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: TURN_STARTED }, + { fence: 1 } + ) + const server = new AgentHookServer() + const sessions = new Map([[SESSION, indexedStatusFeedSession({ journal })]]) + const feed = new StructuredAgentSessionStatusFeed({ + sessions, + getRecord: () => null, + now: () => clock, + statusSink: () => ({ + publish: (summary, subject) => server.ingestStructuredStatus(summary, subject), + forget: (subject) => server.dropStructuredStatus(subject) + }) + }) + const summaries: AgentSessionStatusSummary[] = [] + feed.subscribe({ + id: 'list-1', + emit: (event) => { + if (event.type === 'snapshot') { + summaries.push(...event.sessions) + } else if (event.type === 'status') { + summaries.push(event.session) + } + } + }) + const completions = new StructuredAgentSessionTurnCompletionFeed({ sessions, now: () => clock }) + const completionEvents: AgentSessionTurnCompletionEvent[] = [] + completions.subscribe({ id: 'dot-1', emit: (event) => completionEvents.push(event) }) + // Both feeds have seen the turn running, so its settlement is a transition they must judge. + completions.observe(SESSION, journal) + expect(summaries.at(-1)).toMatchObject({ status: 'working', statusStartedAt: TURN_STARTED }) + const publish = (): void => { + feed.publish(SESSION, journal) + completions.observe(SESSION, journal) + } + return { + journal, + server, + summaries, + completionEvents, + publish, + recoverAt: (at: number) => { + clock = at + } + } +} + +function settleDeadGeneration( + journal: AgentSessionJournal, + verdict: StructuredAgentSessionTurnVerdict +): Promise { + return settleStructuredAgentSessionDeadGeneration({ + journal, + sessionId: SESSION, + fence: 2, + settlementId: 'settle-1', + verdict, + pendingSubmissionReason: 'provider_exited_before_acknowledgement' + }) +} + +describe('a turn recovery settled after its host went away', () => { + it.each([ + ['an unverifiable end', { state: 'unverifiable' } as const], + ['an exit observed before the restart', { state: 'interrupted', completedAt: EXIT_OBSERVED }] + ] satisfies [string, StructuredAgentSessionTurnVerdict][])( + 'is done as of the recovery, never as a success: %s', + async (_label, verdict) => { + const session = await sessionWithRunningTurn() + session.recoverAt(RECOVERED) + expect(await settleDeadGeneration(session.journal, verdict)).toBe(true) + session.publish() + + expect(session.summaries.at(-1)).toMatchObject({ + status: 'idle', + statusStartedAt: RECOVERED + }) + expect(session.summaries.at(-1)).not.toHaveProperty('turnOutcome') + const [row] = session.server.getStatusSnapshot() + // A done row dated at the recovery is a completion the user has not read yet. + expect(row).toMatchObject({ + state: 'done', + stateStartedAt: RECOVERED, + mainAgent: { state: 'done', stateStartedAt: RECOVERED } + }) + expect(row?.mainAgent).not.toHaveProperty('outcome') + // The dot and the OS notification come only from a completion event, and none is sent. + expect(session.completionEvents).toEqual([]) + } + ) + + it('is dated the same way when a new provider child finds the turn still running', async () => { + const session = await sessionWithRunningTurn() + session.recoverAt(RECOVERED) + await settleStaleSessionStateOnAcquire({ + journal: session.journal, + sessionId: SESSION, + fence: 2, + acquisitionGeneration: 'generation-2' + }) + session.publish() + + expect(session.summaries.at(-1)).toMatchObject({ + status: 'idle', + statusStartedAt: RECOVERED + }) + expect(session.server.getStatusSnapshot()[0]).toMatchObject({ + state: 'done', + stateStartedAt: RECOVERED + }) + expect(session.completionEvents).toEqual([]) + }) + + // The control that keeps the silence above from being vacuous: a turn its provider finished does + // reach the completion feed through this same harness, dated by its own end. + it('leaves a turn its provider finished to the provider, dated by its own end', async () => { + const session = await sessionWithRunningTurn() + session.recoverAt(RECOVERED) + await session.journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 9 }, + { + kind: 'turn', + turnId: 'turn-1', + state: 'completed', + outcome: 'success', + startedAt: TURN_STARTED, + completedAt: EXIT_OBSERVED + }, + { fence: 1 } + ) + session.publish() + + expect(session.summaries.at(-1)).toMatchObject({ + status: 'idle', + statusStartedAt: EXIT_OBSERVED, + turnOutcome: 'success' + }) + expect(session.completionEvents).toEqual([ + expect.objectContaining({ + type: 'completion', + completion: expect.objectContaining({ outcome: 'success' }) + }) + ]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-clock.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-clock.test.ts new file mode 100644 index 00000000000..dcd06c0c7ef --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-clock.test.ts @@ -0,0 +1,130 @@ +// Which journal changes republish a session's status now that the summary carries its own state +// clock: a moved clock always does, and row activity alone does not once the clock dates the state. + +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 { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionStatusSummary } from '../../../shared/agent-session-wire' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' +import { indexedStatusFeedSession } from './structured-agent-session-status-feed-test-session' + +const SESSION = 'clock-session' +const THREAD = 'thread-1' + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-status-feed-clock-')) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +function codexItem(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal } +} + +function approval(title: string, answered = false): AgentJournalItemBody { + return { + kind: 'approval', + title, + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: answered + ? { state: 'resolved', selectedOptionId: 'yes', resolvedBy: 'user', resolvedAt: 2 } + : { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } +} + +async function openFeed() { + let clock = 1_000 + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + now: () => (clock += 100), + journalDir: join(root, SESSION) + }) + const feed = new StructuredAgentSessionStatusFeed({ + sessions: new Map([[SESSION, indexedStatusFeedSession({ journal })]]), + getRecord: () => null, + now: () => clock + }) + const published: AgentSessionStatusSummary[] = [] + feed.subscribe({ + id: 'list-1', + emit: (event) => { + if (event.type === 'status') { + published.push(event.session) + } + } + }) + const write = async (identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => { + await journal.appendItem(identity, body, { fence: 1 }) + feed.publish(SESSION, journal) + } + await write( + { provider: 'orca', clientMessageId: 'prompt-1' }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] } + ) + return { published, write } +} + +describe('status republication and the state clock', () => { + it('stays quiet for activity on a dated idle session, and speaks when the clock moves', async () => { + const { published, write } = await openFeed() + await write(codexItem(0), { + kind: 'turn', + turnId: 'turn-1', + state: 'completed', + completedAt: 1_150 + }) + const settled = published.length + expect(published.at(-1)).toMatchObject({ status: 'idle', statusStartedAt: 1_150 }) + + // A roster-style revision: new journal activity, same words, same clock. + await write(codexItem(5), { kind: 'status', text: 'Subagent running' }) + await write(codexItem(5), { kind: 'status', text: 'Subagent running' }) + expect(published).toHaveLength(settled) + + await write(codexItem(0), { + kind: 'turn', + turnId: 'turn-1', + state: 'completed', + completedAt: 1_900 + }) + expect(published.at(-1)).toMatchObject({ status: 'idle', statusStartedAt: 1_900 }) + }) + + it('republishes an attention clock that moves while the session stays in attention', async () => { + const { published, write } = await openFeed() + await write(codexItem(0), { + kind: 'turn', + turnId: 'turn-1', + state: 'running', + startedAt: 1_100 + }) + await write(codexItem(1), approval('first')) + const first = published.at(-1)?.statusStartedAt + await write(codexItem(2), approval('second')) + const second = published.at(-1) + expect(second).toMatchObject({ status: 'attention', statusStartedAt: first }) + + await write(codexItem(1), approval('first', true)) + expect(published.at(-1)).toMatchObject({ status: 'attention' }) + expect(published.at(-1)?.statusStartedAt).toBeGreaterThan(first ?? Infinity) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index 6ec254f1032..29b0f4cd40d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -13,6 +13,7 @@ import { agentProviderSessionsEqual } from '../../../shared/agent-session-resume' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { normalizeOptionalField } from '../../../shared/agent-status-field-normalization' +import { isAgentStatusHeldOpenByChildWork } from '../../../shared/agent-lead-status-fold' import { AGENT_MODEL_MAX_LENGTH } from '../../../shared/agent-status-types' import { agentSessionBackgroundTasksEqual, @@ -21,6 +22,7 @@ import { type AgentSessionStatusSummary } from '../../../shared/agent-session-wire' import { projectStructuredAgentSessionStatusSummary } from '../../../shared/structured-agent-session-projection' +import { structuredAgentSessionAgentStatus } from '../../../shared/structured-agent-session-agent-status' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { structuredAgentSessionProviderSessionMetadata } from './structured-agent-session-history-result' import { @@ -64,8 +66,13 @@ function summariesEqual(a: AgentSessionStatusSummary, b: AgentSessionStatusSumma a.status === b.status && a.hostExecutionOwned === b.hostExecutionOwned && a.rewindBlockedReason === b.rewindBlockedReason && - // Settled activity changes ranking; streaming active turns must stay quiet. - (a.status !== 'idle' || a.updatedAt === b.updatedAt) && + // A moved state clock changes ranking; row activity alone, including a subagent's, does not. + // An idle state the journal cannot date still republishes, since readers date it by `updatedAt`, + // and so does one live child work holds open: readers take each publish as its evidence. + a.statusStartedAt === b.statusStartedAt && + (a.status !== 'idle' || + a.updatedAt === b.updatedAt || + (a.statusStartedAt !== undefined && !isIdleHeldOpenByChildWork(b))) && a.latestPrompt === b.latestPrompt && a.model === b.model && a.toolName === b.toolName && @@ -77,6 +84,19 @@ function summariesEqual(a: AgentSessionStatusSummary, b: AgentSessionStatusSumma ) } +function isIdleHeldOpenByChildWork(summary: AgentSessionStatusSummary): boolean { + return ( + summary.status === 'idle' && + isAgentStatusHeldOpenByChildWork( + structuredAgentSessionAgentStatus({ + status: summary.status, + backgroundTasks: summary.backgroundTasks, + turnOutcome: summary.turnOutcome + }) + ) + ) +} + /** Wire the host's own deps into a feed; keeps the host at one call site. * `deps` is a thunk because the host builds the feed in a field initializer, * before its constructor parameters are assigned. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts new file mode 100644 index 00000000000..cf908932667 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts @@ -0,0 +1,348 @@ +// A subagent's work must not re-date the session that spawned it. +// +// The status row takes its completion stamp and acknowledgement clock from the summary's +// `statusStartedAt`. Subagents write into the same journal and keep going after the session's +// own agent has settled, so every hop below is the real one — provider translator, deferred +// sink, durable journal, status feed. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + AgentSessionBackgroundTask, + AgentSessionStatusEvent +} from '../../../shared/agent-session-wire' +import { AGENT_STATUS_STALE_AFTER_MS } from '../../../shared/agent-status-types' +import { projectStructuredAgentSessionStatusSummary } from '../../../shared/structured-agent-session-projection' +import { AgentHookServer, _internals } from '../../agent-hooks/server' +import { createClaudeJournalTranslator } from '../../claude/claude-structured-journal-translation' +import { createCodexJournalTranslator } from '../../codex/codex-structured-journal-translation' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import { createDeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' +import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' +import { indexedStatusFeedSession } from './structured-agent-session-status-feed-test-session' + +const SESSION = 'recency-session' +const CODEX_THREAD = 'thread-parent' +const CODEX_CHILD = 'thread-child' + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + _internals.resetCachesForTests() + root = await mkdtemp(join(tmpdir(), 'orca-subagent-recency-')) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +/** A journal, a feed over it, and a sink that publishes into the feed. One clock serves the + * journal and every provider event, and it advances on every read: on the wall clock a burst + * of appends can share a millisecond, and a `Math.max` over one number moves nothing, so a + * contaminated clock would pass. */ +async function openSession() { + let clock = 10_000 + const tick = (): number => (clock += 1_000) + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'claude', + providerHandle: { kind: 'codex', threadId: CODEX_THREAD } + }, + now: tick, + journalDir: join(root, SESSION) + }) + // The roster the provider adapter reports, and the host's status row the feed writes into. + const roster: { tasks: AgentSessionBackgroundTask[] } = { tasks: [] } + const server = new AgentHookServer() + const feed = new StructuredAgentSessionStatusFeed({ + sessions: new Map([[SESSION, indexedStatusFeedSession({ journal, hasProviderChild: true })]]), + getRecord: () => null, + now: () => 1, + readBackgroundTasks: () => ({ state: 'monitoring', tasks: roster.tasks }), + statusSink: () => ({ + publish: (summary, subject) => server.ingestStructuredStatus(summary, subject), + forget: (subject) => server.dropStructuredStatus(subject) + }) + }) + const events: AgentSessionStatusEvent[] = [] + feed.subscribe({ id: 'list-1', emit: (event) => events.push(event) }) + const deferred = createDeferredStructuredAgentSessionEventSink() + deferred.bind({ journal, fence: 1, publish: () => feed.publish(SESSION, journal) }) + const drain = async (): Promise => { + expect(await deferred.drained()).toEqual({ ok: true }) + } + /** The prompt as the host journals it; provider user frames never become user rows. */ + const prompt = (clientMessageId: string, text: string) => + journal.appendItem( + { provider: 'orca', clientMessageId }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + { fence: 1 } + ) + const latestStatus = () => { + const event = events.findLast((candidate) => candidate.type === 'status') + if (event?.type !== 'status') { + throw new Error('status publication missing') + } + return event.session + } + /** The journal's projection now, whether or not the feed republished it. */ + const projected = () => { + const snapshot = journal.snapshot() + return projectStructuredAgentSessionStatusSummary(snapshot.items, snapshot.submissions, 1) + } + return { + journal, + roster, + server, + projected, + tick, + sink: deferred.sink, + events, + prompt, + drain, + latestStatus, + close: deferred.close + } +} + +function claudeFrame(message: Record, startsTurn = false) { + return { + type: 'message' as const, + sessionId: SESSION, + ...(startsTurn ? { startsTurn: true as const } : {}), + message: { session_id: 'claude-session', ...message } + } +} + +function claudeUserTurn(uuid: string, text: string) { + return claudeFrame( + { + type: 'user', + uuid, + parent_tool_use_id: null, + message: { role: 'user', content: [{ type: 'text', text }] } + }, + true + ) +} + +function claudeResult(uuid: string) { + return claudeFrame({ type: 'result', subtype: 'success', uuid, result: 'ok' }) +} + +function claudeTask(subtype: string, fields: Record) { + return claudeFrame({ type: 'system', subtype, ...fields }) +} + +/** The real Codex translator over the session's sink, its clock shared with the journal's. */ +function codexTranslator(session: Awaited>) { + const translator = createCodexJournalTranslator({ + sink: session.sink, + sessionId: SESSION, + primaryThreadId: () => CODEX_THREAD, + now: session.tick, + schedule: (run) => { + run() + return () => {} + } + }) + const on = (threadId: string, method: string, params: Record = {}) => + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId, + method, + params: { threadId, ...params }, + observedAt: session.tick() + }) + const item = (threadId: string, method: string, turnId: string, body: Record) => + on(threadId, method, { turnId, item: body }) + return { translator, on, item } +} + +describe("a subagent's work and the recency of the session that spawned it", () => { + it("holds an idle Claude session's clock while its backgrounded subagent reports and finishes", async () => { + const session = await openSession() + const translator = createClaudeJournalTranslator({ sink: session.sink }) + const handle = (event: ReturnType): void => + translator.handle({ ...event, observedAt: session.tick() }) + await session.prompt('prompt-1', 'fan out') + handle(claudeUserTurn('user-1', 'fan out')) + handle( + claudeFrame({ + type: 'assistant', + uuid: 'assistant-1', + parent_tool_use_id: null, + message: { + role: 'assistant', + content: [ + { type: 'text', text: 'delegating' }, + { type: 'tool_use', id: 'toolu_1', name: 'Task', input: { description: 'x' } } + ] + } + }) + ) + handle( + claudeTask('task_started', { + task_id: 'task-1', + tool_use_id: 'toolu_1', + task_type: 'local_agent', + description: 'Watch the build', + is_backgrounded: true + }) + ) + handle( + claudeFrame({ + type: 'user', + uuid: 'user-2', + parent_tool_use_id: null, + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: 'toolu_1', content: 'launched' }] + } + }) + ) + handle(claudeResult('result-1')) + await session.drain() + const settled = session.latestStatus() + expect(settled).toMatchObject({ status: 'idle', statusStartedAt: expect.any(Number) }) + const evidenceClock = session.journal.lastActivityAt() + const published = session.events.length + const sequence = session.journal.cursor().sequence + + // The session's own agent has settled. Its backgrounded child renames itself and then + // finishes, and each edge revises the session's roster row. + handle( + claudeTask('task_updated', { task_id: 'task-1', patch: { description: 'Build watched' } }) + ) + handle( + claudeTask('task_notification', { + task_id: 'task-1', + tool_use_id: 'toolu_1', + status: 'completed', + summary: 'green' + }) + ) + await session.drain() + + // Controls, so the holds below are not vacuous: the child's edges DID reach the journal, + // and they moved its evidence clock. + expect(session.journal.cursor().sequence).toBeGreaterThan(sequence) + expect(session.journal.lastActivityAt()).toBeGreaterThan(evidenceClock) + expect(session.events).toHaveLength(published) + expect(session.projected().statusStartedAt).toBe(settled.statusStartedAt) + + // The session's own next turn still moves it. + await session.prompt('prompt-2', 'thanks') + handle(claudeUserTurn('user-3', 'thanks')) + handle(claudeResult('result-2')) + await session.drain() + expect(session.latestStatus().statusStartedAt).toBeGreaterThan(settled.statusStartedAt ?? 0) + translator.dispose() + session.close() + }) + + // A row live child work holds open is dated by when the host saw it, and mobile decays a working + // row whose evidence is older than the staleness window. The child's own rows are what keep it. + it("keeps the host's evidence fresh while a live subagent holds an idle Claude session open", async () => { + const session = await openSession() + const translator = createClaudeJournalTranslator({ sink: session.sink }) + const handle = (event: ReturnType): void => + translator.handle({ ...event, observedAt: session.tick() }) + await session.prompt('prompt-1', 'fan out') + handle(claudeUserTurn('user-1', 'fan out')) + handle( + claudeFrame({ + type: 'assistant', + uuid: 'assistant-1', + parent_tool_use_id: null, + message: { + role: 'assistant', + content: [{ type: 'tool_use', id: 'toolu_1', name: 'Task', input: { description: 'x' } }] + } + }) + ) + session.roster.tasks = [{ id: 'task-1', kind: 'agent', state: 'working' }] + handle(claudeResult('result-1')) + await session.drain() + const settled = session.latestStatus() + expect(settled).toMatchObject({ status: 'idle', statusStartedAt: expect.any(Number) }) + const [heldOpen] = session.server.getStatusSnapshot() + expect(heldOpen).toMatchObject({ state: 'working', mainAgent: { state: 'done' } }) + + const later = (heldOpen?.evidenceObservedAt ?? 0) + AGENT_STATUS_STALE_AFTER_MS + 1 + vi.spyOn(Date, 'now').mockReturnValue(later) + handle( + claudeFrame({ + type: 'assistant', + uuid: 'child-assistant-1', + parent_tool_use_id: 'toolu_1', + message: { role: 'assistant', content: [{ type: 'text', text: 'still reviewing' }] } + }) + ) + await session.drain() + + expect(session.server.getStatusSnapshot()[0]).toMatchObject({ + state: 'working', + evidenceObservedAt: later, + stateStartedAt: heldOpen?.stateStartedAt, + mainAgent: { state: 'done', stateStartedAt: settled.statusStartedAt } + }) + translator.dispose() + session.close() + }) + + it("holds an idle Codex session's clock while its subagent streams rows and spends tokens", async () => { + const session = await openSession() + const { on, item } = codexTranslator(session) + await session.prompt('prompt-1', 'fan out') + on(CODEX_THREAD, 'turn/started', { turn: { id: 'parent-turn' } }) + const spawn = { + type: 'subAgentActivity', + id: 'spawn-child', + kind: 'started', + agentThreadId: CODEX_CHILD, + agentPath: '/root/review' + } + item(CODEX_THREAD, 'item/started', 'parent-turn', spawn) + item(CODEX_THREAD, 'item/completed', 'parent-turn', spawn) + on(CODEX_CHILD, 'turn/started', { turn: { id: 'child-turn' } }) + on(CODEX_THREAD, 'turn/completed', { turn: { id: 'parent-turn', status: 'completed' } }) + await session.drain() + const settled = session.latestStatus() + expect(settled).toMatchObject({ status: 'idle', statusStartedAt: expect.any(Number) }) + const evidenceClock = session.journal.lastActivityAt() + const published = session.events.length + const sequence = session.journal.cursor().sequence + + // The parent's turn is over; its child runs on, writing prose and spending tokens. + item(CODEX_CHILD, 'item/completed', 'child-turn', { + type: 'agentMessage', + id: 'child-message', + text: 'reviewing' + }) + on(CODEX_CHILD, 'thread/tokenUsage/updated', { + turnId: 'child-turn', + tokenUsage: { total: { totalTokens: 900 } } + }) + on(CODEX_CHILD, 'turn/completed', { turn: { id: 'child-turn', status: 'completed' } }) + await session.drain() + + expect(session.journal.cursor().sequence).toBeGreaterThan(sequence) + expect(session.journal.lastActivityAt()).toBeGreaterThan(evidenceClock) + // Whatever else a child row republishes, none of it re-dates the session. + for (const event of session.events.slice(published)) { + expect(event).toMatchObject({ session: { statusStartedAt: settled.statusStartedAt } }) + } + expect(session.projected().statusStartedAt).toBe(settled.statusStartedAt) + session.close() + }) +}) diff --git a/src/renderer/src/components/activity/ActivityPrototypePage-test-fixtures.ts b/src/renderer/src/components/activity/ActivityPrototypePage-test-fixtures.ts index d484807b3ae..181f96869a2 100644 --- a/src/renderer/src/components/activity/ActivityPrototypePage-test-fixtures.ts +++ b/src/renderer/src/components/activity/ActivityPrototypePage-test-fixtures.ts @@ -164,6 +164,7 @@ export function makeActivityResult(args: { export function makeThreads(result: ReturnType) { return buildAgentPaneThreads({ events: result.events, - liveAgentByPaneKey: result.liveAgentByPaneKey + liveAgentByPaneKey: result.liveAgentByPaneKey, + paneEntryByPaneKey: result.paneEntryByPaneKey }) } diff --git a/src/renderer/src/components/activity/ActivityPrototypePage.test.ts b/src/renderer/src/components/activity/ActivityPrototypePage.test.ts index be5710c6ee6..2a960cb464e 100644 --- a/src/renderer/src/components/activity/ActivityPrototypePage.test.ts +++ b/src/renderer/src/components/activity/ActivityPrototypePage.test.ts @@ -29,6 +29,7 @@ import { PANE_KEY_2, PANE_KEY_3 } from './ActivityPrototypePage-test-fixtures' +import { activityThreadStatusId } from './activity-thread-presentation' describe('buildActivityEvents', () => { it('keeps every pane visible before applying the global activity cap', () => { @@ -134,6 +135,8 @@ describe('buildActivityEvents', () => { expect(result.events).toHaveLength(1) expect(result.liveAgentByPaneKey[PANE_KEY]).toBeUndefined() + // The pane's own row is still `working`, but only a fresh turn may say so. + expect(activityThreadStatusId(makeThreads(result)[0])).toBe('done') }) it('creates a thread for a fresh running agent with no historical events', () => { diff --git a/src/renderer/src/components/activity/activity-answered-ask-timeline.test.tsx b/src/renderer/src/components/activity/activity-answered-ask-timeline.test.tsx new file mode 100644 index 00000000000..e7ac1d3b7d3 --- /dev/null +++ b/src/renderer/src/components/activity/activity-answered-ask-timeline.test.tsx @@ -0,0 +1,264 @@ +// @vitest-environment happy-dom + +// A settled parent whose subagent asks and is answered. Its done keeps the turn's own end time, +// which the blocked ask postdates, so the Activity row must read the pane's row for its state and +// order the timeline by when each switch was seen, not by the states' own times. + +import { act, cleanup, render, renderHook, waitFor } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../../shared/agent-session-wire' +import type { Tab } from '../../../../shared/tab-types' +import type { AppState } from '@/store/types' +import { makeRepo, makeWorktree } from './ActivityPrototypePage-test-fixtures' +import { activityThreadRowCopy, activityThreadStatusId } from './activity-thread-presentation' +import { clearActivityThread } from './activity-clear-completed' +import { countActivityUnread } from './useActivityUnreadCount' +import { useAgentPaneThreads } from './use-agent-pane-threads' + +type TestStore = { + getState: () => AppState + setState: (state: Partial & { testRuntimeOwner?: string | null }) => void +} + +const mocks = vi.hoisted(() => { + const hoisted: { store: TestStore | null; subscribeStatus: Mock; unsubscribe: Mock } = { + store: null, + subscribeStatus: vi.fn(), + unsubscribe: vi.fn() + } + return hoisted +}) + +vi.mock('@/store', async () => { + const { createTestStore } = await import('@/store/slices/store-test-helpers') + const useAppStore = createTestStore() + mocks.store = useAppStore + return { useAppStore } +}) + +vi.mock('@/lib/worktree-runtime-owner', () => ({ + getRuntimeEnvironmentIdForWorktree: (state: { testRuntimeOwner?: string | null }) => + state.testRuntimeOwner ?? null +})) + +vi.mock('@/runtime/structured-agent-session-client', () => ({ + callStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus +})) + +import { StructuredAgentSessionStatusBridge } from '../native-chat/StructuredAgentSessionStatusBridge' +import { resetStructuredAgentSessionStatusFeedsForTests } from '@/runtime/structured-agent-session-status-feed' + +const structuredTab = { + id: 'structured-tab-1', + worktreeId: 'wt-1', + groupId: 'group-1', + contentType: 'agent-session', + entityId: 'session-1', + label: 'Claude Chat', + customLabel: null, + color: null, + sortOrder: 0, + createdAt: 0, + isPinned: false, + agentSessionAgent: 'claude' +} satisfies Tab + +const SETTLED = 22_000 +const ASKED = 27_000 +const ANSWERED = 28_500 + +function summary(overrides: Partial = {}): AgentSessionStatusSummary { + return { + sessionId: 'session-1', + workspaceId: 'wt-1', + agent: 'claude', + status: 'idle', + hostExecutionOwned: true, + latestPrompt: 'fan out', + statusStartedAt: SETTLED, + updatedAt: SETTLED, + ...overrides + } +} + +function store(): TestStore { + if (!mocks.store) { + throw new Error('store missing') + } + return mocks.store +} + +function paneKey(): string { + const [key] = Object.keys(store().getState().agentStatusByPaneKey) + if (!key) { + throw new Error('status row missing') + } + return key +} + +function acknowledge(at: number): void { + store().setState({ acknowledgedAgentsByPaneKey: { [paneKey()]: at } }) +} + +/** The Activity page's own pipeline over the store the bridge wrote. */ +function renderActivity() { + return renderHook(() => + useAgentPaneThreads({ + query: '', + readFilter: 'all', + groupBy: 'none', + selectedPaneKey: null, + showChildAgents: true + }) + ) +} + +function thread(activity: ReturnType) { + const [only, ...rest] = activity.result.current.allThreads + expect(rest).toHaveLength(0) + if (!only) { + throw new Error('activity thread missing') + } + return only +} + +async function connect(): Promise<(event: AgentSessionStatusEvent) => void> { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + const emit: (event: AgentSessionStatusEvent) => void = mocks.subscribeStatus.mock.calls[0]?.[1] + return (event) => act(() => emit(event)) +} + +describe("an answered subagent ask on a settled parent's Activity row", () => { + beforeEach(() => { + vi.clearAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + vi.spyOn(Date, 'now').mockReturnValue(ANSWERED + 1_000) + const worktree = makeWorktree() + store().setState({ + agentStatusByPaneKey: {}, + acknowledgedAgentsByPaneKey: {}, + activityClearedAtByPaneKey: {}, + retainedAgentsByPaneKey: {}, + testRuntimeOwner: null, + repos: [makeRepo()], + worktreesByRepo: { [worktree.repoId]: [worktree] }, + unifiedTabsByWorktree: { 'wt-1': [structuredTab] } + }) + }) + + afterEach(() => { + cleanup() + vi.restoreAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + }) + + it('reads done, lists Blocked then done, and leaves the answer read', async () => { + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary()] }) + acknowledge(SETTLED + 1_000) + emit({ + type: 'status', + session: summary({ status: 'attention', statusStartedAt: ASKED, updatedAt: ASKED }) + }) + // The user reads the ask and answers it; the parent returns to its own turn's end. + acknowledge(ASKED + 500) + emit({ type: 'status', session: summary({ updatedAt: ANSWERED }) }) + + const row = thread(renderActivity()) + expect(activityThreadStatusId(row)).toBe('done') + expect(activityThreadRowCopy(row).needsAttention).toBe(false) + // Newest first: the answer's done, the ask, then the turn's original done. + expect(row.events.map((event) => event.state)).toEqual(['done', 'blocked', 'done']) + expect(row.events.map((event) => [event.state, event.timestamp, event.observedAt])).toEqual([ + ['done', SETTLED, ANSWERED], + ['blocked', ASKED, ASKED], + ['done', SETTLED, SETTLED] + ]) + expect(row.latestEvent?.observedAt).toBe(ANSWERED) + // Unread keys on the turn's own end, which the user had already read. + expect(row.events.map((event) => event.unread)).toEqual([false, false, false]) + expect(row.unread).toBe(false) + expect(countActivityUnread(store().getState())).toBe(0) + + // Clearing the answered row must also pass the ask, which is dated after the done. + act(() => { + expect(clearActivityThread(row)).toBe(true) + }) + expect(renderActivity().result.current.allThreads).toHaveLength(0) + }) + + it('keeps each answered done between the asks around it', async () => { + const asks = [ASKED, ASKED + 400, ASKED + 800] + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary()] }) + for (const askedAt of asks) { + emit({ + type: 'status', + session: summary({ status: 'attention', statusStartedAt: askedAt, updatedAt: askedAt }) + }) + emit({ type: 'status', session: summary({ updatedAt: askedAt + 200 }) }) + } + + // Every done repeats the turn's end, so only when each was seen keeps them apart and in order; + // the oldest of the six falls to the per-pane cap. + expect(thread(renderActivity()).events.map((event) => [event.state, event.observedAt])).toEqual( + [ + ['done', ASKED + 1_000], + ['blocked', ASKED + 800], + ['done', ASKED + 600], + ['blocked', ASKED + 400], + ['done', ASKED + 200] + ] + ) + }) + + it('reads done after a clear hid that done, while the later ask stays listed', async () => { + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary()] }) + acknowledge(SETTLED + 1_000) + const beforeAsk = renderActivity() + let cleared = false + act(() => { + cleared = clearActivityThread(thread(beforeAsk)) + }) + expect(cleared).toBe(true) + beforeAsk.unmount() + expect(store().getState().activityClearedAtByPaneKey).toEqual({ [paneKey()]: SETTLED }) + + emit({ + type: 'status', + session: summary({ status: 'attention', statusStartedAt: ASKED, updatedAt: ASKED }) + }) + acknowledge(ASKED + 500) + emit({ type: 'status', session: summary({ updatedAt: ANSWERED }) }) + + const row = thread(renderActivity()) + // The cleared done stays cleared; only the ask, seen after the clear, is listed. + expect(row.events.map((event) => event.state)).toEqual(['blocked']) + expect(activityThreadStatusId(row)).toBe('done') + expect(activityThreadRowCopy(row).needsAttention).toBe(false) + expect(row.unread).toBe(false) + expect(countActivityUnread(store().getState())).toBe(0) + + // Once a second ask moves the answered done into history, it stays cleared there too. + emit({ + type: 'status', + session: summary({ + status: 'attention', + statusStartedAt: ANSWERED + 200, + updatedAt: ANSWERED + 200 + }) + }) + expect(thread(renderActivity()).events.map((event) => event.state)).toEqual([ + 'blocked', + 'blocked' + ]) + }) +}) diff --git a/src/renderer/src/components/activity/activity-clear-completed-hmr.test.ts b/src/renderer/src/components/activity/activity-clear-completed-hmr.test.ts index b1698551c5d..1ea5d3b2c94 100644 --- a/src/renderer/src/components/activity/activity-clear-completed-hmr.test.ts +++ b/src/renderer/src/components/activity/activity-clear-completed-hmr.test.ts @@ -89,6 +89,7 @@ function doneEvent(interrupted: boolean): ActivityEvent { id: 'evt', state: 'done', timestamp: 5_000, + observedAt: 5_000, worktree: makeWorktree(), repo: null, entry: { ...makeRetained('t-done:1').entry, interrupted }, diff --git a/src/renderer/src/components/activity/activity-clear-completed.test.ts b/src/renderer/src/components/activity/activity-clear-completed.test.ts index 65152720567..e0cee5aa9d8 100644 --- a/src/renderer/src/components/activity/activity-clear-completed.test.ts +++ b/src/renderer/src/components/activity/activity-clear-completed.test.ts @@ -86,6 +86,7 @@ function doneEvent(interrupted: boolean): ActivityEvent { id: 'evt', state: 'done', timestamp: 5_000, + observedAt: 5_000, worktree: makeWorktree(), repo: null, entry: { interrupted } as ActivityEvent['entry'], diff --git a/src/renderer/src/components/activity/activity-event-build-cache.ts b/src/renderer/src/components/activity/activity-event-build-cache.ts index 346390d5619..c5ff8e2fb5d 100644 --- a/src/renderer/src/components/activity/activity-event-build-cache.ts +++ b/src/renderer/src/components/activity/activity-event-build-cache.ts @@ -55,7 +55,7 @@ export function resolvePaneBuild( request: PaneBuildRequest, cache: ActivityEventBuildCache | undefined, seenCacheKeys: Set | null -): { events: ActivityEvent[]; live: ActivityLiveAgentSnapshot | null } { +): { events: ActivityEvent[]; live: ActivityLiveAgentSnapshot | null; rowEntry: AgentStatusEntry } { seenCacheKeys?.add(request.cacheKey) const cached = cache?.panes.get(request.cacheKey) const inputsUnchanged = @@ -84,7 +84,7 @@ export function resolvePaneBuild( cached.live.timestamp === liveTimestamp) if (inputsUnchanged && liveMatchesCache) { - return { events: cached.events, live: cached.live } + return { events: cached.events, live: cached.live, rowEntry } } // The live turn is itself an event, so a live change always rebuilds the pane's events. @@ -125,5 +125,5 @@ export function resolvePaneBuild( live, rowEntry }) - return { events, live } + return { events, live, rowEntry } } diff --git a/src/renderer/src/components/activity/activity-event-builder-sources.ts b/src/renderer/src/components/activity/activity-event-builder-sources.ts index 1049ed95acf..16010a2e081 100644 --- a/src/renderer/src/components/activity/activity-event-builder-sources.ts +++ b/src/renderer/src/components/activity/activity-event-builder-sources.ts @@ -20,7 +20,7 @@ export function appendUnsupportedAndRetainedEvents(context: { entry: AgentStatusEntry, terminalPtyId?: string | null ) => { worktree: Worktree; repo: Repo | null; knownWorktree: boolean } - pushPaneEvents: (paneEvents: ActivityEvent[]) => void + pushPaneEvents: (paneEvents: ActivityEvent[], rowEntry: AgentStatusEntry) => void }): void { const { args, @@ -45,7 +45,11 @@ export function appendUnsupportedAndRetainedEvents(context: { continue } const owner = resolveOwner(tabEntry, entry, unsupported.ptyId) - const { events: paneEvents, live } = resolvePaneBuild( + const { + events: paneEvents, + live, + rowEntry + } = resolvePaneBuild( { cacheKey, source: unsupported, @@ -67,7 +71,7 @@ export function appendUnsupportedAndRetainedEvents(context: { if (live) { liveAgentByPaneKey[entry.paneKey] = live } - pushPaneEvents(paneEvents) + pushPaneEvents(paneEvents, rowEntry) } for (const [paneKey, retained] of Object.entries(args.retainedAgentsByPaneKey)) { @@ -82,7 +86,7 @@ export function appendUnsupportedAndRetainedEvents(context: { if (!owner.knownWorktree) { continue } - const { events: paneEvents } = resolvePaneBuild( + const { events: paneEvents, rowEntry } = resolvePaneBuild( { cacheKey: `retained:${paneKey}`, source: retained, @@ -100,6 +104,6 @@ export function appendUnsupportedAndRetainedEvents(context: { cache, seenCacheKeys ) - pushPaneEvents(paneEvents) + pushPaneEvents(paneEvents, rowEntry) } } diff --git a/src/renderer/src/components/activity/activity-event-builder.ts b/src/renderer/src/components/activity/activity-event-builder.ts index 879b4c56c64..2f45a7e34bc 100644 --- a/src/renderer/src/components/activity/activity-event-builder.ts +++ b/src/renderer/src/components/activity/activity-event-builder.ts @@ -53,8 +53,10 @@ export function buildActivityEvents( ): { events: ActivityEvent[] liveAgentByPaneKey: Record + paneEntryByPaneKey: Record } { const events: ActivityEvent[] = [] + const paneEntryByPaneKey: Record = {} const seenEventIds = new Set() const tabContext = buildActivityTabContext(args.tabsByWorktree, args.unifiedTabsByWorktree) const tabHostIndex = buildActivityTabHostIndex(args.unifiedTabsByWorktree) @@ -62,9 +64,10 @@ export function buildActivityEvents( const liveAgentByPaneKey: Record = {} const seenCacheKeys = cache ? new Set() : null - const pushPaneEvents = (paneEvents: ActivityEvent[]): void => { + const pushPaneEvents = (paneEvents: ActivityEvent[], rowEntry: AgentStatusEntry): void => { // Why: a paneKey can appear in more than one source (live + retained overlap); // event ids stay globally unique so the first source wins, as before. + paneEntryByPaneKey[rowEntry.paneKey] ??= rowEntry for (const event of paneEvents) { if (seenEventIds.has(event.id)) { continue @@ -95,7 +98,11 @@ export function buildActivityEvents( // Only fresh live turns contribute working activity; history cannot establish liveness. // The freshness check runs on the raw entry (orchestration merges never change state/timing fields). const liveState = freshActivityLiveAgentState(entry, args.now) - const { events: paneEvents, live } = resolvePaneBuild( + const { + events: paneEvents, + live, + rowEntry + } = resolvePaneBuild( { cacheKey: `live:${paneKey}`, source: entry, @@ -116,7 +123,7 @@ export function buildActivityEvents( if (live) { liveAgentByPaneKey[paneKey] = live } - pushPaneEvents(paneEvents) + pushPaneEvents(paneEvents, rowEntry) } appendUnsupportedAndRetainedEvents({ @@ -138,5 +145,5 @@ export function buildActivityEvents( } } } - return { events: capActivityEvents(events), liveAgentByPaneKey } + return { events: capActivityEvents(events), liveAgentByPaneKey, paneEntryByPaneKey } } diff --git a/src/renderer/src/components/activity/activity-event-cap.ts b/src/renderer/src/components/activity/activity-event-cap.ts index d5799c22f92..2aca1800ee0 100644 --- a/src/renderer/src/components/activity/activity-event-cap.ts +++ b/src/renderer/src/components/activity/activity-event-cap.ts @@ -9,7 +9,7 @@ export function capActivityEvents(events: ActivityEvent[]): ActivityEvent[] { const working = events.filter((event) => event.state === 'working') const sorted = events .filter((event) => event.state !== 'working') - .sort((a, b) => b.timestamp - a.timestamp) + .sort((a, b) => b.observedAt - a.observedAt) const perPaneCount = new Map() const includedEventIds = new Set() const capped: ActivityEvent[] = [] @@ -42,5 +42,5 @@ export function capActivityEvents(events: ActivityEvent[]): ActivityEvent[] { includedEventIds.add(event.id) capped.push(event) } - return [...capped, ...working].sort((a, b) => b.timestamp - a.timestamp) + return [...capped, ...working].sort((a, b) => b.observedAt - a.observedAt) } diff --git a/src/renderer/src/components/activity/activity-pane-events.ts b/src/renderer/src/components/activity/activity-pane-events.ts index 9170bcf8898..6a147beac81 100644 --- a/src/renderer/src/components/activity/activity-pane-events.ts +++ b/src/renderer/src/components/activity/activity-pane-events.ts @@ -62,8 +62,14 @@ type PaneEventInputs = { export function buildPaneActivityEvents(args: PaneEventInputs): ActivityEvent[] { const events: ActivityEvent[] = [] const seenIds = new Set() - const append = (state: ActivityEventState, timestamp: number, entry: AgentStatusEntry): void => { - const id = `agent:${entry.paneKey}:${state}:${timestamp}` + const append = ( + state: ActivityEventState, + timestamp: number, + observedAt: number, + entry: AgentStatusEntry + ): void => { + // Why observedAt: an answered ask returns done to its turn's end, repeating that done's time. + const id = `agent:${entry.paneKey}:${state}:${observedAt}` if (seenIds.has(id)) { return } @@ -72,6 +78,7 @@ export function buildPaneActivityEvents(args: PaneEventInputs): ActivityEvent[] id, state, timestamp, + observedAt, worktree: args.worktree, repo: args.repo, entry, @@ -93,6 +100,7 @@ export function buildPaneActivityEvents(args: PaneEventInputs): ActivityEvent[] append( history.state as ActivityEventState, history.startedAt, + history.observedAt ?? history.startedAt, historyEntrySnapshot(args.entry, history) ) } @@ -108,6 +116,11 @@ export function buildPaneActivityEvents(args: PaneEventInputs): ActivityEvent[] if (args.entry.stateStartedAt <= args.clearedAt) { return events } - append(currentState, args.entry.stateStartedAt, args.entry) + append( + currentState, + args.entry.stateStartedAt, + args.entry.stateObservedAt ?? args.entry.stateStartedAt, + args.entry + ) return events } diff --git a/src/renderer/src/components/activity/activity-thread-builder.ts b/src/renderer/src/components/activity/activity-thread-builder.ts index a1076ea0022..d0f198dbeba 100644 --- a/src/renderer/src/components/activity/activity-thread-builder.ts +++ b/src/renderer/src/components/activity/activity-thread-builder.ts @@ -3,6 +3,7 @@ import { paneTitleForEvent, statusPreviewForEntry } from './activity-thread-presentation' +import type { AgentStatusEntry } from '../../../../shared/agent-status-types' import type { ActivityEvent, ActivityLiveAgentSnapshot, @@ -51,6 +52,7 @@ function reuseThreadIfEqual( previous.agentType === next.agentType && previous.currentAgentState === next.currentAgentState && previous.currentAgentEntry === next.currentAgentEntry && + previous.paneEntry === next.paneEntry && previous.responsePreview === next.responsePreview && previous.latestTimestamp === next.latestTimestamp && previous.latestEvent === next.latestEvent && @@ -67,6 +69,7 @@ export function buildAgentPaneThreads( args: { events: ActivityEvent[] liveAgentByPaneKey: Record + paneEntryByPaneKey?: Record generatedTitlesEnabled?: boolean }, reuseCache?: AgentPaneThreadReuseCache @@ -86,6 +89,7 @@ export function buildAgentPaneThreads( agentType: event.agentType, currentAgentState: null, currentAgentEntry: null, + paneEntry: args.paneEntryByPaneKey?.[paneKey], responsePreview: statusPreviewForEntry(event.entry, event.state), latestTimestamp: event.timestamp, latestEvent: event, @@ -99,7 +103,9 @@ export function buildAgentPaneThreads( existing.unread = existing.unread || event.unread existing.migrationUnsupportedPtyId = existing.migrationUnsupportedPtyId ?? event.migrationUnsupportedPtyId - if (!existing.latestEvent || event.timestamp > existing.latestEvent.timestamp) { + // Why max, not the latest event's: "Clear completed" cuts off at this, and must pass every event. + existing.latestTimestamp = Math.max(existing.latestTimestamp, event.timestamp) + if (!existing.latestEvent || event.observedAt > existing.latestEvent.observedAt) { existing.latestEvent = event existing.paneTitle = paneTitleForEvent(event, generatedTitlesEnabled) existing.agentType = event.agentType @@ -109,7 +115,6 @@ export function buildAgentPaneThreads( event.state, existing.responsePreview ) - existing.latestTimestamp = event.timestamp } } @@ -125,6 +130,7 @@ export function buildAgentPaneThreads( agentType: liveAgent.agentType, currentAgentState: liveAgent.state, currentAgentEntry: liveAgent.entry, + paneEntry: args.paneEntryByPaneKey?.[paneKey], responsePreview: statusPreviewForEntry(liveAgent.entry, liveAgent.entry.state), latestTimestamp: liveAgent.timestamp, latestEvent: null, @@ -153,7 +159,7 @@ export function buildAgentPaneThreads( .map((thread) => { const next: AgentPaneThread = { ...thread, - events: [...thread.events].sort((a, b) => b.timestamp - a.timestamp) + events: [...thread.events].sort((a, b) => b.observedAt - a.observedAt) } return reuseThreadIfEqual(reuseCache?.previousByPaneKey.get(thread.paneKey), next) }) diff --git a/src/renderer/src/components/activity/activity-thread-child-agent.test.ts b/src/renderer/src/components/activity/activity-thread-child-agent.test.ts index 7c687c72037..214e0e559cd 100644 --- a/src/renderer/src/components/activity/activity-thread-child-agent.test.ts +++ b/src/renderer/src/components/activity/activity-thread-child-agent.test.ts @@ -55,6 +55,7 @@ function makeEventFor(entry: AgentStatusEntry): ActivityEvent { id: `event-${entry.paneKey}`, state: 'done', timestamp: 1000, + observedAt: 1000, unread: false, worktree, repo: null, diff --git a/src/renderer/src/components/activity/activity-thread-presentation.ts b/src/renderer/src/components/activity/activity-thread-presentation.ts index 1aa27d321dc..5541b3da493 100644 --- a/src/renderer/src/components/activity/activity-thread-presentation.ts +++ b/src/renderer/src/components/activity/activity-thread-presentation.ts @@ -11,7 +11,12 @@ import { formatUiRelativeTime } from '@/i18n/relative-time-format' import { translate } from '@/i18n/i18n' import type { AgentStatusEntry, AgentStatusState } from '../../../../shared/agent-status-types' import type { TerminalTab } from '../../../../shared/terminal-tab-types' -import type { ActivityEvent, AgentPaneThread } from './activity-thread-types' +import { isHistoricalActivityState } from './activity-event-state' +import type { + ActivityEvent, + ActivityLiveAgentState, + AgentPaneThread +} from './activity-thread-types' const ACTIVITY_THREAD_RESPONSE_RENDER_PREVIEW_MAX_LENGTH = 320 @@ -117,13 +122,33 @@ export type ActivityThreadStatusId = AgentDotState /** Single classifier behind grouping, labels, and clear-completed; the only place the * interrupted predicate is spelled. */ export function activityThreadStatusId(thread: AgentPaneThread): ActivityThreadStatusId { - const state = thread.currentAgentState ?? thread.latestEvent?.state ?? 'done' - if (!thread.currentAgentState && state === 'done' && thread.latestEvent?.entry.interrupted) { + const paneEntry = paneActivityEntry(thread) + const state = threadCurrentState(thread) ?? 'done' + const interrupted = paneEntry ? paneEntry.interrupted : thread.latestEvent?.entry.interrupted + if (!thread.currentAgentState && state === 'done' && interrupted) { return 'interrupted' } return state } +// Why the pane's row: an answered ask's done predates the blocked event, and a clear can hide it. +function paneActivityEntry(thread: AgentPaneThread): AgentStatusEntry | null { + return thread.paneEntry && isHistoricalActivityState(thread.paneEntry.state) + ? thread.paneEntry + : null +} + +function threadCurrentState( + thread: AgentPaneThread +): ActivityLiveAgentState | AgentStatusState | null { + return ( + thread.currentAgentState ?? + paneActivityEntry(thread)?.state ?? + thread.latestEvent?.state ?? + null + ) +} + // Interrupted rows deliberately keep the done glyph (#2569). export function threadAgentState(thread: AgentPaneThread): AgentDotState { const id = activityThreadStatusId(thread) @@ -197,7 +222,7 @@ export function activityThreadRowCopy(thread: AgentPaneThread): ActivityThreadRo const renderedPreview = activityThreadResponseRenderPreview({ responsePreview: thread.responsePreview }) - const liveState = thread.currentAgentState ?? thread.latestEvent?.state ?? null + const liveState = threadCurrentState(thread) const toolPreviewState = liveState === 'monitoring' ? null : liveState const state = threadAgentState(thread) const needsAttention = state === 'waiting' || state === 'blocked' || state === 'permission' diff --git a/src/renderer/src/components/activity/activity-thread-types.ts b/src/renderer/src/components/activity/activity-thread-types.ts index 7d32f1a2b39..5d0b9e11acb 100644 --- a/src/renderer/src/components/activity/activity-thread-types.ts +++ b/src/renderer/src/components/activity/activity-thread-types.ts @@ -20,7 +20,10 @@ export type ActivityLiveAgentState = ActivityHookLiveAgentState | 'monitoring' export type ActivityEvent = { id: string state: ActivityEventState + /** The state's own start time; unread and "Clear completed" compare against it. */ timestamp: number + /** When Orca saw the switch into this state; orders the timeline and keys the event. */ + observedAt: number worktree: Worktree repo: Repo | null entry: AgentStatusEntry @@ -51,6 +54,8 @@ export type AgentPaneThread = { agentType: AgentType currentAgentState: ActivityLiveAgentState | null currentAgentEntry: AgentStatusEntry | null + /** The pane's own status row, live or not; its state outranks the newest event's. */ + paneEntry?: AgentStatusEntry responsePreview: string latestTimestamp: number latestEvent: ActivityEvent | null diff --git a/src/renderer/src/components/activity/use-agent-pane-threads.ts b/src/renderer/src/components/activity/use-agent-pane-threads.ts index 1934488439e..f9575f02a76 100644 --- a/src/renderer/src/components/activity/use-agent-pane-threads.ts +++ b/src/renderer/src/components/activity/use-agent-pane-threads.ts @@ -126,7 +126,11 @@ export function useAgentPaneThreads(args: { const threadReuseCacheRef = useRef>(undefined!) threadReuseCacheRef.current ??= createAgentPaneThreadReuseCache() - const { events: allEvents, liveAgentByPaneKey } = useMemo( + const { + events: allEvents, + liveAgentByPaneKey, + paneEntryByPaneKey + } = useMemo( () => buildActivityEvents( { @@ -157,11 +161,12 @@ export function useAgentPaneThreads(args: { { events: allEvents, liveAgentByPaneKey, + paneEntryByPaneKey, generatedTitlesEnabled: storeData.generatedTitlesEnabled }, threadReuseCacheRef.current ), - [allEvents, liveAgentByPaneKey, storeData.generatedTitlesEnabled] + [allEvents, liveAgentByPaneKey, paneEntryByPaneKey, storeData.generatedTitlesEnabled] ) const selectedPaneKeyIsLive = diff --git a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx index 8cd966ef9af..a90a4bfc146 100644 --- a/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx +++ b/src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge.tsx @@ -13,6 +13,10 @@ import { import { mainAgentStatusEqual, agentSubagentsEqual } from '../../../../shared/agent-status-types' import { structuredAgentSessionPaneKey } from '../../../../shared/structured-agent-session-projection' import { structuredAgentSessionAgentStatus } from '../../../../shared/structured-agent-session-agent-status' +import { + structuredAgentSessionDatedMainAgent, + structuredAgentSessionRowStateStartedAt +} from '../../../../shared/structured-agent-session-status-started-at' import { getRuntimeEnvironmentIdForWorktree } from '@/lib/worktree-runtime-owner' import { useAppStore } from '@/store' import { getActiveRuntimeTarget, type RuntimeClientTarget } from '@/runtime/runtime-rpc-client' @@ -73,7 +77,7 @@ function projectStatus( // Same continuity rule as the host ingest, on the main agent's own clock. const mainAgent = continueMainAgentStatus( current?.mainAgent, - agentStatus.mainAgent, + structuredAgentSessionDatedMainAgent(agentStatus.mainAgent, summary), summary.updatedAt ) const desired = { @@ -130,11 +134,12 @@ function projectStatus( // Same continuity key as the host ingest: monitoring and working are distinct published // states, so the timer beside the label must restart when the label changes. stateStartedAt: - desired.state !== 'done' && + structuredAgentSessionRowStateStartedAt(desired, summary) ?? + (desired.state !== 'done' && current?.state === desired.state && current.workingMode === desired.workingMode ? current.stateStartedAt - : summary.updatedAt, + : summary.updatedAt), // Same rule as the host ingest: the journal clock stopped when the lead's turn did, so a // row held open by child work alone is dated by when this client saw it instead. evidenceObservedAt: isAgentStatusHeldOpenByChildWork(desired) ? Date.now() : summary.updatedAt diff --git a/src/renderer/src/components/native-chat/structured-agent-session-status-bridge-clock.test.tsx b/src/renderer/src/components/native-chat/structured-agent-session-status-bridge-clock.test.tsx new file mode 100644 index 00000000000..c829c9ad350 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-agent-session-status-bridge-clock.test.tsx @@ -0,0 +1,215 @@ +// @vitest-environment happy-dom + +// The row's state clock, as the host dates it. A subagent's rows move the summary's `updatedAt` +// but never its `statusStartedAt`, so a settled parent must keep its completion stamp and stay +// read while its child works on; an older host that publishes no clock keeps the old dating. + +import { act, cleanup, render, waitFor } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../../../shared/agent-session-wire' +import { agentEntryCompletionAt } from '../../../../shared/agent-completion-time' +import type { AgentStatusEntry } from '../../../../shared/agent-status-types' +import type { Tab } from '../../../../shared/tab-types' +import type { AppState } from '@/store/types' +import { countActivityUnread } from '../activity/useActivityUnreadCount' + +type TestStore = { + getState: () => AppState + setState: (state: Partial & { testRuntimeOwner?: string | null }) => void +} + +const mocks = vi.hoisted(() => { + const hoisted: { store: TestStore | null; subscribeStatus: Mock; unsubscribe: Mock } = { + store: null, + subscribeStatus: vi.fn(), + unsubscribe: vi.fn() + } + return hoisted +}) + +vi.mock('@/store', async () => { + const { createTestStore } = await import('@/store/slices/store-test-helpers') + const useAppStore = createTestStore() + mocks.store = useAppStore + return { useAppStore } +}) + +vi.mock('@/lib/worktree-runtime-owner', () => ({ + getRuntimeEnvironmentIdForWorktree: (state: { testRuntimeOwner?: string | null }) => + state.testRuntimeOwner ?? null +})) + +vi.mock('@/runtime/structured-agent-session-client', () => ({ + callStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus +})) + +import { StructuredAgentSessionStatusBridge } from './StructuredAgentSessionStatusBridge' +import { resetStructuredAgentSessionStatusFeedsForTests } from '@/runtime/structured-agent-session-status-feed' + +const structuredTab = { + id: 'structured-tab-1', + worktreeId: 'wt-1', + groupId: 'group-1', + contentType: 'agent-session', + entityId: 'session-1', + label: 'Claude Chat', + customLabel: null, + color: null, + sortOrder: 0, + createdAt: 0, + isPinned: false, + agentSessionAgent: 'claude' +} satisfies Tab + +const SETTLED = 22_000 +const ACKNOWLEDGED = 23_000 + +function summary(overrides: Partial = {}): AgentSessionStatusSummary { + return { + sessionId: 'session-1', + workspaceId: 'wt-1', + agent: 'claude', + status: 'idle', + hostExecutionOwned: true, + latestPrompt: 'fan out', + updatedAt: SETTLED, + ...overrides + } +} + +function row(): AgentStatusEntry { + const [entry] = Object.values(mocks.store?.getState().agentStatusByPaneKey ?? {}) + if (!entry) { + throw new Error('status row missing') + } + return entry +} + +function acknowledge(at: number): void { + mocks.store?.setState({ acknowledgedAgentsByPaneKey: { [row().paneKey]: at } }) +} + +function unread(): number { + const state = mocks.store?.getState() + if (!state) { + throw new Error('store missing') + } + return countActivityUnread(state, ACKNOWLEDGED + 60_000) +} + +async function connect(): Promise<(event: AgentSessionStatusEvent) => void> { + render() + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + const emit: (event: AgentSessionStatusEvent) => void = mocks.subscribeStatus.mock.calls[0]?.[1] + return (event) => act(() => emit(event)) +} + +describe("the structured row's state clock", () => { + beforeEach(() => { + vi.clearAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + mocks.store?.setState({ + agentStatusByPaneKey: {}, + acknowledgedAgentsByPaneKey: {}, + testRuntimeOwner: null, + unifiedTabsByWorktree: { 'wt-1': [structuredTab] } + }) + }) + + afterEach(() => { + cleanup() + resetStructuredAgentSessionStatusFeedsForTests() + }) + + it('keeps a settled parent read while its subagent writes, until its own next turn ends', async () => { + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary({ statusStartedAt: SETTLED })] }) + expect(row()).toMatchObject({ state: 'done', stateStartedAt: SETTLED }) + acknowledge(ACKNOWLEDGED) + expect(unread()).toBe(0) + + // The child's rows move the evidence clock, and a host may still republish for them. + for (const updatedAt of [24_000, 25_000, 26_000]) { + emit({ type: 'status', session: summary({ statusStartedAt: SETTLED, updatedAt }) }) + } + expect(row()).toMatchObject({ state: 'done', updatedAt: 26_000, stateStartedAt: SETTLED }) + expect(unread()).toBe(0) + + emit({ + type: 'status', + session: summary({ status: 'working', statusStartedAt: 30_000, updatedAt: 30_000 }) + }) + emit({ type: 'status', session: summary({ statusStartedAt: 31_000, updatedAt: 31_000 }) }) + expect(row()).toMatchObject({ state: 'done', stateStartedAt: 31_000 }) + expect(unread()).toBe(1) + }) + + it("keeps the old dating for an older host's summary, which carries no clock", async () => { + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary()] }) + acknowledge(ACKNOWLEDGED) + expect(unread()).toBe(0) + emit({ type: 'status', session: summary({ updatedAt: 26_000 }) }) + expect(row()).toMatchObject({ state: 'done', stateStartedAt: 26_000 }) + expect(unread()).toBe(1) + }) + + it("dates a subagent's approval at the ask and leaves the parent's completion where it was", async () => { + const emit = await connect() + emit({ type: 'snapshot', sessions: [summary({ statusStartedAt: SETTLED })] }) + acknowledge(ACKNOWLEDGED) + + emit({ + type: 'status', + session: summary({ status: 'attention', statusStartedAt: 27_000, updatedAt: 27_000 }) + }) + expect(row()).toMatchObject({ + state: 'blocked', + stateStartedAt: 27_000, + mainAgent: { state: 'blocked', stateStartedAt: 27_000 } + }) + expect(unread()).toBe(1) + + // Answered: the parent is idle again, still dated by its own last turn. + acknowledge(28_000) + emit({ type: 'status', session: summary({ statusStartedAt: SETTLED, updatedAt: 28_500 }) }) + expect(row()).toMatchObject({ state: 'done', stateStartedAt: SETTLED }) + expect(agentEntryCompletionAt(row())).toBe(SETTLED) + expect(unread()).toBe(0) + }) + + it('leaves a row child work holds open on its own continuity, and settles it on the parent clock', async () => { + const emit = await connect() + const child = { id: 'child-1', kind: 'agent', state: 'working' } as const + emit({ + type: 'snapshot', + sessions: [summary({ status: 'working', statusStartedAt: 10_000, updatedAt: 10_000 })] + }) + emit({ + type: 'status', + session: summary({ statusStartedAt: SETTLED, updatedAt: 24_000, backgroundTasks: [child] }) + }) + // Still working as far as the row shows, so that state keeps the clock it started with. + expect(row()).toMatchObject({ + state: 'working', + stateStartedAt: 10_000, + mainAgent: { state: 'done', stateStartedAt: SETTLED } + }) + + emit({ + type: 'status', + session: summary({ + statusStartedAt: SETTLED, + updatedAt: 26_000, + backgroundTasks: [{ ...child, state: 'done' }] + }) + }) + expect(row()).toMatchObject({ state: 'done', stateStartedAt: SETTLED }) + }) +}) diff --git a/src/renderer/src/store/slices/agent-status-live-entry-builder.ts b/src/renderer/src/store/slices/agent-status-live-entry-builder.ts index 64bd5cc1a77..28bf952460c 100644 --- a/src/renderer/src/store/slices/agent-status-live-entry-builder.ts +++ b/src/renderer/src/store/slices/agent-status-live-entry-builder.ts @@ -1,10 +1,9 @@ import type { AppState } from '../types' import { resolveAgentStatusLiveEntryMainAgent } from './agent-status-live-entry-main-agent' +import { resolveAgentStatusLiveEntryStateHistory } from './agent-status-live-entry-state-history' import { - AGENT_STATE_HISTORY_MAX, agentSubagentsEqual, type MigrationUnsupportedPtyEntry, - type AgentStateHistoryEntry, type AgentStatusEntry } from '../../../../shared/agent-status-types' import { @@ -79,34 +78,8 @@ export function buildAgentStatusLiveEntry( return { entry: null, reason: 'stale' } } const effectiveTitle = terminalTitle ?? existing?.terminalTitle - let history: AgentStateHistoryEntry[] = existing?.stateHistory ?? [] - let lastCompletedAssistantMessage = existing?.lastCompletedAssistantMessage - const boundaryLandsOnRealDone = - existing?.state === 'done' && - existing.sessionBoundary !== true && - payload.state === 'done' && - payload.sessionBoundary === true - if ( - existing && - (existing.state !== payload.state || boundaryLandsOnRealDone) && - !(existing.state === 'done' && existing.sessionBoundary === true) - ) { - history = [ - ...history, - { - state: existing.state, - prompt: existing.prompt, - startedAt: existing.stateStartedAt, - interrupted: existing.interrupted - } - ] - if (history.length > AGENT_STATE_HISTORY_MAX) { - history = history.slice(history.length - AGENT_STATE_HISTORY_MAX) - } - if (existing.state === 'done') { - lastCompletedAssistantMessage = existing.lastAssistantMessage - } - } + const { history, lastCompletedAssistantMessage, stateObservedAt } = + resolveAgentStatusLiveEntryStateHistory(existing, payload, updatedAt) const identity = resolveAgentStatusIdentity({ existing: existing ? { @@ -231,6 +204,7 @@ export function buildAgentStatusLiveEntry( : {}), ...(metadata?.structuredHostOwned === true ? { structuredHostOwned: true as const } : {}), stateStartedAt, + stateObservedAt, agentType: identity.agentType, model: payload.model ?? (existing?.agentType === identity.agentType ? existing.model : undefined), diff --git a/src/renderer/src/store/slices/agent-status-live-entry-state-history.ts b/src/renderer/src/store/slices/agent-status-live-entry-state-history.ts new file mode 100644 index 00000000000..6976b0b990b --- /dev/null +++ b/src/renderer/src/store/slices/agent-status-live-entry-state-history.ts @@ -0,0 +1,54 @@ +import { + AGENT_STATE_HISTORY_MAX, + type AgentStateHistoryEntry, + type AgentStatusEntry +} from '../../../../shared/agent-status-types' +import type { AgentStatusPayload } from './agent-status-contract' + +/** The history a live entry carries after this write, and when its state was first observed. A + * state switch moves the previous state into history; a session boundary is never recorded. */ +export function resolveAgentStatusLiveEntryStateHistory( + existing: AgentStatusEntry | undefined, + payload: Pick, + updatedAt: number +): { + history: AgentStateHistoryEntry[] + lastCompletedAssistantMessage: string | undefined + stateObservedAt: number | undefined +} { + let history: AgentStateHistoryEntry[] = existing?.stateHistory ?? [] + let lastCompletedAssistantMessage = existing?.lastCompletedAssistantMessage + const boundaryLandsOnRealDone = + existing?.state === 'done' && + existing.sessionBoundary !== true && + payload.state === 'done' && + payload.sessionBoundary === true + const switchesState = existing?.state !== payload.state || boundaryLandsOnRealDone + if ( + existing && + switchesState && + !(existing.state === 'done' && existing.sessionBoundary === true) + ) { + history = [ + ...history, + { + state: existing.state, + prompt: existing.prompt, + startedAt: existing.stateStartedAt, + observedAt: existing.stateObservedAt, + interrupted: existing.interrupted + } + ] + if (history.length > AGENT_STATE_HISTORY_MAX) { + history = history.slice(history.length - AGENT_STATE_HISTORY_MAX) + } + if (existing.state === 'done') { + lastCompletedAssistantMessage = existing.lastAssistantMessage + } + } + return { + history, + lastCompletedAssistantMessage, + stateObservedAt: switchesState ? updatedAt : existing?.stateObservedAt + } +} diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index d58ad231fc9..29271faa2b1 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -282,6 +282,7 @@ export const AgentJournalRenderItemSchema = z.object({ sequence: z.number().int(), observedAt: z.number(), recovered: z.literal(true).optional(), + recoveredAt: z.number().optional(), ...AgentJournalProducerLinkageFields }) diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 9d7f6002539..a13f3b1f676 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -334,6 +334,8 @@ export type AgentJournalRenderItem = AgentJournalProducerLinkage & { observedAt: number /** Set when the row was appended by crash reconciliation rather than live. */ recovered?: true + /** When crash reconciliation wrote this revision; present exactly when `recovered` is. */ + recoveredAt?: number } // ─── Submissions ──────────────────────────────────────────────────────────── diff --git a/src/shared/agent-session-wire.ts b/src/shared/agent-session-wire.ts index a79e708d8fd..0a764b6fc23 100644 --- a/src/shared/agent-session-wire.ts +++ b/src/shared/agent-session-wire.ts @@ -231,6 +231,10 @@ export type AgentSessionStatusSummary = { backgroundTasks?: AgentSessionBackgroundTask[] providerSession?: AgentProviderSessionMetadata updatedAt: number + /** When the session's own agent entered `status`, dated by its own lifecycle edges and never by + * row activity: `updatedAt` also moves for a subagent's rows. Absent from older hosts, and when + * the journal records no such edge; readers then keep dating the state themselves. */ + statusStartedAt?: number } /** A summary outlives its provider child: an evicted idle session is still idle, so the host diff --git a/src/shared/agent-state-history.ts b/src/shared/agent-state-history.ts new file mode 100644 index 00000000000..4833d351c66 --- /dev/null +++ b/src/shared/agent-state-history.ts @@ -0,0 +1,22 @@ +import type { AgentStatusState } from './agent-status-types' + +/** A snapshot of a previous agent state, used to render activity blocks. + * Why: intentionally narrower than AgentStatusEntry — tool/assistant context is + * per-turn, not meaningful on a historical snapshot, and would bloat memory. + * Coalesced-turn output lives in AgentStatusEntry.lastCompletedAssistantMessage, + * one copy per pane, so it can't multiply by AGENT_STATE_HISTORY_MAX. */ +export type AgentStateHistoryEntry = { + state: AgentStatusState + prompt: string + /** When this state was first reported. */ + startedAt: number + /** `updatedAt` of the write that switched into this state. `startedAt` can predate an earlier + * entry (an answered ask returns to its turn's end), so this orders and identifies the switch. */ + observedAt?: number + /** True when this `done` was a cancellation (agent hook like Claude `is_interrupt`, + * or Orca's guarded fallback). Always falsy for non-`done` states so retention logic can preserve it. */ + interrupted?: boolean +} + +/** Maximum number of history entries kept per agent to bound memory. */ +export const AGENT_STATE_HISTORY_MAX = 20 diff --git a/src/shared/agent-status-types.ts b/src/shared/agent-status-types.ts index e198f366b2b..dcd447e6bef 100644 --- a/src/shared/agent-status-types.ts +++ b/src/shared/agent-status-types.ts @@ -4,6 +4,7 @@ import type { AgentProviderSessionMetadata } from './agent-session-resume' import type { AgentMainAgentStatus } from './main-agent-status' +import type { AgentStateHistoryEntry } from './agent-state-history' import { isAgentJournalTurnOutcome } from './agent-turn-outcome' import type { OrchestrationFleetAttention } from './orchestration-fleet-attention' import type { AgentStatusRowFacets } from './agent-status-observation' @@ -25,6 +26,7 @@ export type { MigrationUnsupportedPtyEntry } from './agent-status-ipc-payload' export { mainAgentStatusEqual, type AgentMainAgentStatus } from './main-agent-status' +export { AGENT_STATE_HISTORY_MAX, type AgentStateHistoryEntry } from './agent-state-history' export const AGENT_STATUS_STATES = ['working', 'blocked', 'waiting', 'done'] as const export type AgentStatusState = (typeof AGENT_STATUS_STATES)[number] @@ -36,24 +38,6 @@ export type AgentWorkingMode = 'monitoring' export type WellKnownAgentType = TuiAgent | 'unknown' export type AgentType = WellKnownAgentType | (string & {}) -/** A snapshot of a previous agent state, used to render activity blocks. - * Why: intentionally narrower than AgentStatusEntry — tool/assistant context is - * per-turn, not meaningful on a historical snapshot, and would bloat memory. - * Coalesced-turn output lives in AgentStatusEntry.lastCompletedAssistantMessage, - * one copy per pane, so it can't multiply by AGENT_STATE_HISTORY_MAX. */ -export type AgentStateHistoryEntry = { - state: AgentStatusState - prompt: string - /** When this state was first reported. */ - startedAt: number - /** True when this `done` was a cancellation (agent hook like Claude `is_interrupt`, - * or Orca's guarded fallback). Always falsy for non-`done` states so retention logic can preserve it. */ - interrupted?: boolean -} - -/** Maximum number of history entries kept per agent to bound memory. */ -export const AGENT_STATE_HISTORY_MAX = 20 - export type AgentStatusOrchestrationContext = { taskId: string dispatchId: string @@ -105,6 +89,8 @@ export type AgentStatusEntry = { /** Timestamp (ms) when the current `state` was first reported. * Why: separate from updatedAt so tool/prompt pings (which reset updatedAt) don't move it. */ stateStartedAt: number + /** `updatedAt` of the write that switched into `state`; see AgentStateHistoryEntry.observedAt. */ + stateObservedAt?: number agentType?: AgentType /** Provider model currently used by this session. */ model?: string diff --git a/src/shared/structured-agent-session-projection.test.ts b/src/shared/structured-agent-session-projection.test.ts index 9c3a64a9010..3c16614cf45 100644 --- a/src/shared/structured-agent-session-projection.test.ts +++ b/src/shared/structured-agent-session-projection.test.ts @@ -155,7 +155,8 @@ describe('structured agent session status projection', () => { }) expect(projectStructuredAgentSessionStatusSummary([first, second, running])).toEqual({ status: 'working', - latestPrompt: 'second line' + latestPrompt: 'second line', + statusStartedAt: 3 }) expect(projectStructuredAgentSessionStatusSummary([first, second])).toEqual({ status: 'idle', @@ -176,11 +177,13 @@ describe('structured agent session status projection', () => { // The first send has no journalled message until the provider replays it. expect(projectStructuredAgentSessionStatusSummary([], pending)).toEqual({ status: 'working', - latestPrompt: '' + latestPrompt: '', + statusStartedAt: 1 }) expect(projectStructuredAgentSessionStatusSummary([asked], pending)).toEqual({ status: 'working', - latestPrompt: 'go' + latestPrompt: 'go', + statusStartedAt: 1 }) }) @@ -260,7 +263,8 @@ describe('structured agent session status projection', () => { latestPrompt: 'look at the sidebar', toolName: 'Read', toolInput: '/repo/src/WorktreeCard.tsx', - lastAssistantMessage: 'Reading the card first.' + lastAssistantMessage: 'Reading the card first.', + statusStartedAt: 2 }) }) @@ -327,7 +331,8 @@ describe('structured agent session status projection', () => { expect(projectStructuredAgentSessionStatusSummary([ask, abandoned, running])).toEqual({ status: 'working', - latestPrompt: 'go' + latestPrompt: 'go', + statusStartedAt: 3 }) }) diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 4b5bfb5d098..6e61603c758 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -29,6 +29,8 @@ import { import type { NativeChatBlock, NativeChatMessage } from './native-chat-types' import { sha256 } from './sha256' +import { structuredAgentSessionStatusStartedAt } from './structured-agent-session-status-started-at' +import { isUnansweredStructuredAgentSessionDispatch } from './structured-agent-session-unanswered-dispatch' // Re-exported so the live-turn readers' existing consumers keep one import site. export { @@ -207,14 +209,8 @@ export function hasUnansweredStructuredAgentSessionDispatch( submissions: readonly AgentJournalSubmission[], currentFence?: number | null ): boolean { - return submissions.some( - (submission) => - (currentFence == null || submission.fence >= currentFence) && - (submission.dispatchState === 'pending' || - (submission.dispatchState === 'unknown' && - submission.recovered !== true && - // Older hosts publish the recovery reason but omit the optional marker. - submission.reason !== 'host_restarted_before_acknowledgement')) + return submissions.some((submission) => + isUnansweredStructuredAgentSessionDispatch(submission, currentFence) ) } @@ -313,6 +309,7 @@ export type StructuredAgentSessionStatusProjection = { lastAssistantMessage?: string /** The newest settled turn's provider verdict; present only while `status` is idle. */ turnOutcome?: AgentJournalTurnOutcome + statusStartedAt?: number } /** One projection shared by host and client: null status means "no turn yet", not idle. @@ -353,13 +350,20 @@ export function projectStructuredAgentSessionStatusSummary( // `readAgentJournalTurnOutcome` already answers null for anything it cannot place. const turnOutcome = status === 'idle' ? readAgentJournalTurnOutcome(newestStructuredAgentSessionTurn(items)) : null + const statusStartedAt = structuredAgentSessionStatusStartedAt( + status, + items, + submissions, + currentFence + ) return { status, latestPrompt: normalizePromptField(latestStructuredAgentSessionPrompt(items)), ...(toolName ? { toolName } : {}), ...(toolInput ? { toolInput } : {}), ...(lastAssistantMessage ? { lastAssistantMessage } : {}), - ...(turnOutcome ? { turnOutcome } : {}) + ...(turnOutcome ? { turnOutcome } : {}), + ...(statusStartedAt !== undefined ? { statusStartedAt } : {}) } } diff --git a/src/shared/structured-agent-session-status-started-at.test.ts b/src/shared/structured-agent-session-status-started-at.test.ts new file mode 100644 index 00000000000..85a9d475ea8 --- /dev/null +++ b/src/shared/structured-agent-session-status-started-at.test.ts @@ -0,0 +1,190 @@ +import { describe, expect, it } from 'vitest' +import type { + AgentJournalItemBody, + AgentJournalRenderItem, + AgentJournalSubmission +} from './agent-session-journal-types' +import { projectStructuredAgentSessionStatusSummary } from './structured-agent-session-projection' +import { structuredAgentSessionRowStateStartedAt } from './structured-agent-session-status-started-at' + +function item( + itemId: string, + sequence: number, + body: AgentJournalItemBody, + extra: Partial = {} +): AgentJournalRenderItem { + return { itemId, sequence, revision: 1, observedAt: sequence * 100, body, ...extra } +} + +function submission( + clientMessageId: string, + submittedAt: number, + dispatchState: AgentJournalSubmission['dispatchState'] = 'pending' +): AgentJournalSubmission { + return { + clientMessageId, + fence: 1, + payloadFingerprint: clientMessageId, + dispatchState, + providerItemId: null, + reason: null, + submittedAt, + resolvedAt: dispatchState === 'pending' ? null : submittedAt + 1 + } +} + +const ask = item('ask', 1, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'go' }] +}) +const childLinkage = { agentId: 'child-1', parentAgentId: 'root', producerKind: 'agent' } as const + +function pendingApproval(itemId: string, sequence: number, extra: Partial) { + return item( + itemId, + sequence, + { + kind: 'approval', + title: 'Run command?', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + extra + ) +} + +function summaryOf(items: AgentJournalRenderItem[], submissions: AgentJournalSubmission[] = []) { + return projectStructuredAgentSessionStatusSummary(items, submissions, 1) +} + +describe('when the session entered its status', () => { + it('dates idle by when its own newest turn ended, whatever a subagent wrote after', () => { + const turn = item('turn', 2, { + kind: 'turn', + turnId: 't1', + state: 'completed', + outcome: 'success', + startedAt: 150, + completedAt: 900 + }) + const childProse = item( + 'child', + 3, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'still going' }] }, + { ...childLinkage, observedAt: 5_000 } + ) + expect(summaryOf([ask, turn])).toMatchObject({ status: 'idle', statusStartedAt: 900 }) + expect(summaryOf([ask, turn, childProse])).toMatchObject({ + status: 'idle', + statusStartedAt: 900 + }) + }) + + it('dates a turn recovery settled by when recovery wrote it, not the end it restates', () => { + const settled = item( + 'turn', + 2, + { kind: 'turn', turnId: 't1', state: 'interrupted', startedAt: 150, completedAt: 400 }, + { recovered: true, recoveredAt: 7_000 } + ) + const unverifiable = item( + 'turn', + 2, + { kind: 'turn', turnId: 't1', state: 'unverifiable', startedAt: 150 }, + { recovered: true, recoveredAt: 7_000 } + ) + expect(summaryOf([ask, settled]).statusStartedAt).toBe(7_000) + expect(summaryOf([ask, unverifiable]).statusStartedAt).toBe(7_000) + }) + + it('leaves an idle state undated when the journal records no end for it', () => { + const undated = item('turn', 2, { kind: 'turn', turnId: 't1', state: 'unverifiable' }) + expect(summaryOf([ask, undated])).not.toHaveProperty('statusStartedAt') + expect(summaryOf([ask])).not.toHaveProperty('statusStartedAt') + }) + + it('dates working by the running turn, which a mid-turn send joins rather than restarts', () => { + const running = item('turn', 2, { + kind: 'turn', + turnId: 't1', + state: 'running', + requestedAt: 120, + startedAt: 150 + }) + const steer = submission('steer', 800) + expect(summaryOf([ask, running])).toMatchObject({ status: 'working', statusStartedAt: 120 }) + expect(summaryOf([ask, running], [steer])).toMatchObject({ + status: 'working', + statusStartedAt: 120 + }) + const unrequested = item('turn', 2, { + kind: 'turn', + turnId: 't1', + state: 'running', + startedAt: 150 + }) + expect(summaryOf([ask, unrequested]).statusStartedAt).toBe(150) + }) + + it('dates working before any turn opens by the earliest send still unanswered', () => { + const settled = item('turn', 2, { + kind: 'turn', + turnId: 't1', + state: 'completed', + completedAt: 300 + }) + const sends = [ + submission('answered', 100, 'accepted'), + submission('second', 700), + submission('first', 500) + ] + expect(summaryOf([ask, settled], sends)).toMatchObject({ + status: 'working', + statusStartedAt: 500 + }) + }) + + it("dates attention by the session's own oldest ask, and by a subagent's only when that alone holds it", () => { + const running = item('turn', 2, { + kind: 'turn', + turnId: 't1', + state: 'running', + startedAt: 150 + }) + const childAsk = pendingApproval('child-ask', 3, { ...childLinkage, observedAt: 400 }) + const ownAsk = pendingApproval('own-ask', 4, { observedAt: 600 }) + const laterOwnAsk = pendingApproval('later-own-ask', 5, { observedAt: 800 }) + expect(summaryOf([ask, running, childAsk])).toMatchObject({ + status: 'attention', + statusStartedAt: 400 + }) + expect(summaryOf([ask, running, childAsk, ownAsk, laterOwnAsk])).toMatchObject({ + status: 'attention', + statusStartedAt: 600 + }) + }) +}) + +describe('the row clock a status writer takes from the host', () => { + it("dates the row only while it shows the main agent's own state", () => { + const dated = { statusStartedAt: 900 } + expect( + structuredAgentSessionRowStateStartedAt( + { state: 'done', mainAgent: { state: 'done' } }, + dated + ) + ).toBe(900) + // Child work holds the row open; the main agent's clock does not date that state. + expect( + structuredAgentSessionRowStateStartedAt( + { state: 'working', mainAgent: { state: 'done' } }, + dated + ) + ).toBeUndefined() + expect( + structuredAgentSessionRowStateStartedAt({ state: 'done', mainAgent: { state: 'done' } }, {}) + ).toBeUndefined() + }) +}) diff --git a/src/shared/structured-agent-session-status-started-at.ts b/src/shared/structured-agent-session-status-started-at.ts new file mode 100644 index 00000000000..54f5b44764f --- /dev/null +++ b/src/shared/structured-agent-session-status-started-at.ts @@ -0,0 +1,100 @@ +// When the session's own agent entered the status it is in, read off its own lifecycle edges: its +// turn records, its sends, and the prompts holding it. Row timestamps never date it. Subagents +// write into the same journal and keep going after the session settles, and a clock over rows +// re-dates an idle session with every one of theirs. +// +// Turn records need no producer filter: a turn is the session's own unit of work and never carries +// subagent linkage (see the header of `structured-agent-session-live-turn.ts`). + +import type { AgentJournalRenderItem, AgentJournalSubmission } from './agent-session-journal-types' +import { isRootAgentJournalItem } from './agent-session-journal-producer' +import { readAgentJournalTurn } from './agent-session-turn-record' +import type { StructuredAgentSessionProjectedStatus } from './structured-agent-session-projection' +import type { AgentSessionStatusSummary } from './agent-session-wire' +import type { AgentMainAgentStatus } from './main-agent-status' +import type { AgentStatusState } from './agent-status-types' +import { isUnansweredStructuredAgentSessionDispatch } from './structured-agent-session-unanswered-dispatch' + +/** Undefined when the journal records no edge that dates `status`; readers keep their own rule. */ +export function structuredAgentSessionStatusStartedAt( + status: StructuredAgentSessionProjectedStatus, + items: readonly AgentJournalRenderItem[], + submissions: readonly AgentJournalSubmission[], + currentFence?: number | null +): number | undefined { + if (status === 'attention') { + return oldestPendingPromptAt(items) + } + const turnItem = newestTurnItem(items) + const turn = readAgentJournalTurn(turnItem?.body) + if (status === 'idle') { + if (!turnItem || !turn || turn.state === 'running') { + return undefined + } + // A turn recovery settled ended when that settle was written: when the user learns it stopped. + return turnItem.recoveredAt ?? turn.completedAt + } + if (turnItem && turn?.state === 'running') { + // A mid-turn send joins the running turn; it does not restart the stretch. + return turn.requestedAt ?? turn.startedAt ?? turnItem.observedAt + } + let earliest: number | undefined + for (const submission of submissions) { + if ( + isUnansweredStructuredAgentSessionDispatch(submission, currentFence) && + (earliest === undefined || submission.submittedAt < earliest) + ) { + earliest = submission.submittedAt + } + } + return earliest +} + +function newestTurnItem(items: readonly AgentJournalRenderItem[]): AgentJournalRenderItem | null { + for (let index = items.length - 1; index >= 0; index -= 1) { + const item = items[index] + if (item && readAgentJournalTurn(item.body)) { + return item + } + } + return null +} + +/** The session's own ask when it has one; a subagent's only when that alone holds the session. */ +function oldestPendingPromptAt(items: readonly AgentJournalRenderItem[]): number | undefined { + let own: number | undefined + let subagent: number | undefined + for (const item of items) { + if ( + (item.body.kind !== 'approval' && item.body.kind !== 'question') || + item.body.resolution.state !== 'pending' + ) { + continue + } + if (isRootAgentJournalItem(item)) { + own = Math.min(own ?? item.observedAt, item.observedAt) + } else { + subagent = Math.min(subagent ?? item.observedAt, item.observedAt) + } + } + return own ?? subagent +} + +/** The main agent's own status, dated by the host when it published a clock. */ +export function structuredAgentSessionDatedMainAgent( + mainAgent: T, + summary: Pick +): T & { stateStartedAt?: number } { + return summary.statusStartedAt === undefined + ? mainAgent + : { ...mainAgent, stateStartedAt: summary.statusStartedAt } +} + +/** The row's own start, when the host dated it: the row is showing the main agent's state rather + * than one child work holds open. Undefined leaves the writer's own continuity rule in charge. */ +export function structuredAgentSessionRowStateStartedAt( + row: { state: AgentStatusState; mainAgent: Pick }, + summary: Pick +): number | undefined { + return row.state === row.mainAgent.state ? summary.statusStartedAt : undefined +} diff --git a/src/shared/structured-agent-session-unanswered-dispatch.ts b/src/shared/structured-agent-session-unanswered-dispatch.ts new file mode 100644 index 00000000000..36c70a357cb --- /dev/null +++ b/src/shared/structured-agent-session-unanswered-dispatch.ts @@ -0,0 +1,17 @@ +import type { AgentJournalSubmission } from './agent-session-journal-types' + +/** One send the provider has neither opened a turn for nor refused; the rule is explained on + * `hasUnansweredStructuredAgentSessionDispatch`, which asks it of every send. */ +export function isUnansweredStructuredAgentSessionDispatch( + submission: AgentJournalSubmission, + currentFence?: number | null +): boolean { + return ( + (currentFence == null || submission.fence >= currentFence) && + (submission.dispatchState === 'pending' || + (submission.dispatchState === 'unknown' && + submission.recovered !== true && + // Older hosts publish the recovery reason but omit the optional marker. + submission.reason !== 'host_restarted_before_acknowledgement')) + ) +} diff --git a/tests/e2e/codex-child-approval-activity-row.unit.test.ts b/tests/e2e/codex-child-approval-activity-row.unit.test.ts new file mode 100644 index 00000000000..b9210bf79e8 --- /dev/null +++ b/tests/e2e/codex-child-approval-activity-row.unit.test.ts @@ -0,0 +1,294 @@ +// @vitest-environment happy-dom + +// A Codex parent settles, its subagent asks for approval, and the user answers. Every hop is the +// real one: provider translator, deferred sink, durable journal and host status feed, then the +// renderer's status bridge, agent-status store and Activity pipeline. The answer returns the +// session to its own turn's end, so the row must read done, list the ask before the done, and +// leave nothing unread that the user had already read. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { createElement } from 'react' +import { act, cleanup, render, renderHook, waitFor } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import type { + AgentSessionStatusEvent, + AgentSessionStatusSummary +} from '../../src/shared/agent-session-wire' +import type { AgentJournalItemBody } from '../../src/shared/agent-session-journal-types' +import { parseAgentJournalItemKey } from '../../src/shared/agent-session-journal-item-key' +import type { Tab } from '../../src/shared/tab-types' +import type { AppState } from '../../src/renderer/src/store/types' +import { createCodexJournalTranslator } from '../../src/main/codex/codex-structured-journal-translation' +import { CODEX_COMMAND_APPROVAL_METHOD } from '../../src/main/codex/codex-structured-prompt-replies' +import { createTrackedJournalOpener } from '../../src/main/native-chat/agent-session-journal/journal-store-test-open' +import { createDeferredStructuredAgentSessionEventSink } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-event-sink' +import { StructuredAgentSessionStatusFeed } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-status-feed' +import { indexedStatusFeedSession } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-test-session' +import { + makeRepo, + makeWorktree +} from '../../src/renderer/src/components/activity/ActivityPrototypePage-test-fixtures' +import { + activityThreadRowCopy, + activityThreadStatusId +} from '../../src/renderer/src/components/activity/activity-thread-presentation' +import { countActivityUnread } from '../../src/renderer/src/components/activity/useActivityUnreadCount' +import { useAgentPaneThreads } from '../../src/renderer/src/components/activity/use-agent-pane-threads' + +type TestStore = { + getState: () => AppState + setState: (state: Partial & { testRuntimeOwner?: string | null }) => void +} + +const mocks = vi.hoisted(() => { + const hoisted: { store: TestStore | null; subscribeStatus: Mock; unsubscribe: Mock } = { + store: null, + subscribeStatus: vi.fn(), + unsubscribe: vi.fn() + } + return hoisted +}) + +vi.mock('@/store', async () => { + const { createTestStore } = await import('@/store/slices/store-test-helpers') + const useAppStore = createTestStore() + mocks.store = useAppStore + return { useAppStore } +}) + +vi.mock('@/lib/worktree-runtime-owner', () => ({ + getRuntimeEnvironmentIdForWorktree: (state: { testRuntimeOwner?: string | null }) => + state.testRuntimeOwner ?? null +})) + +vi.mock('@/runtime/structured-agent-session-client', () => ({ + callStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSession: vi.fn(), + subscribeStructuredAgentSessionStatus: mocks.subscribeStatus +})) + +import { StructuredAgentSessionStatusBridge } from '../../src/renderer/src/components/native-chat/StructuredAgentSessionStatusBridge' +import { resetStructuredAgentSessionStatusFeedsForTests } from '../../src/renderer/src/runtime/structured-agent-session-status-feed' + +const SESSION = 'codex-child-approval' +const CODEX_THREAD = 'thread-parent' +const CODEX_CHILD = 'thread-child' + +const structuredTab = { + id: 'structured-tab-1', + worktreeId: 'wt-1', + groupId: 'group-1', + contentType: 'agent-session', + entityId: SESSION, + label: 'Codex Chat', + customLabel: null, + color: null, + sortOrder: 0, + createdAt: 0, + isPinned: false, + agentSessionAgent: 'codex' +} satisfies Tab + +let root: string +const journals = createTrackedJournalOpener() + +function store(): TestStore { + if (!mocks.store) { + throw new Error('store missing') + } + return mocks.store +} + +beforeEach(async () => { + vi.clearAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + mocks.subscribeStatus.mockResolvedValue({ unsubscribe: mocks.unsubscribe }) + root = await mkdtemp(join(tmpdir(), 'orca-codex-child-approval-')) + const worktree = makeWorktree() + store().setState({ + agentStatusByPaneKey: {}, + acknowledgedAgentsByPaneKey: {}, + activityClearedAtByPaneKey: {}, + retainedAgentsByPaneKey: {}, + testRuntimeOwner: null, + repos: [makeRepo()], + worktreesByRepo: { [worktree.repoId]: [worktree] }, + unifiedTabsByWorktree: { 'wt-1': [structuredTab] } + }) +}) + +afterEach(async () => { + cleanup() + vi.restoreAllMocks() + resetStructuredAgentSessionStatusFeedsForTests() + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +/** The host half: journal, feed and the real Codex translator over one advancing clock. */ +async function openHost() { + let clock = 10_000 + const tick = (): number => (clock += 1_000) + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'wt-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: CODEX_THREAD } + }, + now: tick, + journalDir: join(root, SESSION) + }) + const feed = new StructuredAgentSessionStatusFeed({ + sessions: new Map([[SESSION, indexedStatusFeedSession({ journal, hasProviderChild: true })]]), + getRecord: () => null, + now: () => 1, + readBackgroundTasks: () => ({ state: 'monitoring', tasks: [] }) + }) + const events: AgentSessionStatusEvent[] = [] + feed.subscribe({ id: 'renderer', emit: (event) => events.push(event) }) + const deferred = createDeferredStructuredAgentSessionEventSink() + const publish = (): void => feed.publish(SESSION, journal) + deferred.bind({ journal, fence: 1, publish }) + const prompts: string[] = [] + const translator = createCodexJournalTranslator({ + sink: deferred.sink, + sessionId: SESSION, + primaryThreadId: () => CODEX_THREAD, + now: tick, + bindPromptItemId: (journalItemId) => prompts.push(journalItemId), + schedule: (run) => { + run() + return () => {} + } + }) + const on = (threadId: string, method: string, params: Record = {}) => + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId, + method, + params: { threadId, ...params }, + observedAt: tick() + }) + const drain = async (): Promise => { + expect(await deferred.drained()).toEqual({ ok: true }) + } + /** What the host's answer path commits before it tells Codex. */ + const answer = async (itemId: string): Promise => { + const identity = parseAgentJournalItemKey(itemId) + const asked = journal.snapshot().items.find((item) => item.itemId === itemId)?.body + if (!identity || asked?.kind !== 'approval') { + throw new Error(`approval ${itemId} missing`) + } + const resolved: AgentJournalItemBody = { + ...asked, + resolution: { + state: 'resolved', + selectedOptionId: 'accept', + resolvedBy: 'user', + resolvedAt: tick() + } + } + await journal.appendItem(identity, resolved, { fence: 1 }) + publish() + translator.resolvePrompt(itemId) + } + return { journal, translator, on, drain, answer, prompts, events, tick, close: deferred.close } +} + +describe("a Codex subagent's answered approval on the settled parent's Activity row", () => { + it('reads done, lists the ask before the done, and leaves the answer read', async () => { + const host = await openHost() + render(createElement(StructuredAgentSessionStatusBridge)) + await waitFor(() => expect(mocks.subscribeStatus).toHaveBeenCalledOnce()) + const toRenderer: (event: AgentSessionStatusEvent) => void = + mocks.subscribeStatus.mock.calls[0]?.[1] + let forwarded = 0 + const deliver = (): AgentSessionStatusSummary => { + act(() => { + for (const event of host.events.slice(forwarded)) { + toRenderer(event) + } + }) + forwarded = host.events.length + const latest = host.events.findLast((event) => event.type === 'status') + if (latest?.type !== 'status') { + throw new Error('status publication missing') + } + return latest.session + } + const paneKey = (): string => Object.keys(store().getState().agentStatusByPaneKey)[0] ?? '' + + await host.journal.appendItem( + { provider: 'orca', clientMessageId: 'prompt-1' }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'fan out' }] }, + { fence: 1 } + ) + host.on(CODEX_THREAD, 'turn/started', { turn: { id: 'parent-turn' } }) + const spawn = { + type: 'subAgentActivity', + id: 'spawn-child', + kind: 'started', + agentThreadId: CODEX_CHILD, + agentPath: '/root/review' + } + host.on(CODEX_THREAD, 'item/started', { turnId: 'parent-turn', item: spawn }) + host.on(CODEX_THREAD, 'item/completed', { turnId: 'parent-turn', item: spawn }) + host.on(CODEX_CHILD, 'turn/started', { turn: { id: 'child-turn' } }) + host.on(CODEX_THREAD, 'turn/completed', { turn: { id: 'parent-turn', status: 'completed' } }) + await host.drain() + const settled = deliver() + expect(settled).toMatchObject({ status: 'idle', statusStartedAt: expect.any(Number) }) + store().setState({ acknowledgedAgentsByPaneKey: { [paneKey()]: host.tick() } }) + + host.translator.handle({ + type: 'prompt', + sessionId: SESSION, + threadId: CODEX_CHILD, + method: CODEX_COMMAND_APPROVAL_METHOD, + params: { command: 'pnpm test', availableDecisions: ['accept', 'decline'] }, + codexItemId: 'child-exec', + promptKey: 'child-approval' + }) + await host.drain() + const asked = deliver() + expect(asked).toMatchObject({ status: 'attention' }) + expect(asked.statusStartedAt).toBeGreaterThan(settled.statusStartedAt ?? Infinity) + // The user reads the ask, then answers it. + store().setState({ acknowledgedAgentsByPaneKey: { [paneKey()]: host.tick() } }) + const [approval] = host.prompts + await host.answer(approval ?? '') + host.on(CODEX_CHILD, 'turn/completed', { turn: { id: 'child-turn', status: 'completed' } }) + await host.drain() + const answered = deliver() + // The host rule under test elsewhere: the answer never re-dates the session's done. + expect(answered).toMatchObject({ status: 'idle', statusStartedAt: settled.statusStartedAt }) + expect(answered.updatedAt).toBeGreaterThan(asked.updatedAt) + + const activity = renderHook(() => + useAgentPaneThreads({ + query: '', + readFilter: 'all', + groupBy: 'none', + selectedPaneKey: null, + showChildAgents: true + }) + ) + const [row, ...others] = activity.result.current.allThreads + expect(others).toHaveLength(0) + if (!row) { + throw new Error('activity thread missing') + } + expect(activityThreadStatusId(row)).toBe('done') + expect(activityThreadRowCopy(row).needsAttention).toBe(false) + expect(row.events.map((event) => event.state)).toEqual(['done', 'blocked', 'done']) + expect(row.events.map((event) => event.unread)).toEqual([false, false, false]) + expect(countActivityUnread(store().getState())).toBe(0) + host.translator.dispose() + host.close() + }) +})