diff --git a/config/tsconfig.node.json b/config/tsconfig.node.json index cd833e90c0a..b029a12c3d2 100644 --- a/config/tsconfig.node.json +++ b/config/tsconfig.node.json @@ -16,6 +16,11 @@ "./scripts/vitest-node-runtime-pool.ts", "./scripts/vitest-module-cache-resolution-key.ts", "../src/main/**/*", + "../src/cli/runtime/transport.ts", + "../src/cli/runtime/envelope-schema.ts", + "../src/cli/runtime/types.ts", + "../src/cli/runtime/runtime-access-denied.ts", + "../src/cli/runtime/runtime-pid-liveness.ts", "../src/renderer/src/store/slices/worktrees/create/worktree-create-payload.ts", "../src/renderer/src/components/native-chat/native-chat-resolution-receipt.ts", "../src/renderer/src/components/native-chat/structured-agent-question-projection.ts", diff --git a/mobile/src/session/use-mobile-structured-agent-session.test.tsx b/mobile/src/session/use-mobile-structured-agent-session.test.tsx index 2d5d7f2a47a..d69a78ed2df 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session.test.tsx @@ -285,6 +285,49 @@ describe('useMobileStructuredAgentSession', () => { ) }) + it('keeps creation facts out of the current phone conversation', async () => { + await act(async () => { + renderer = create(createElement(Harness)) + }) + await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) + const items: AgentJournalRenderItem[] = [ + { + itemId: 'before', + revision: 1, + sequence: 1, + observedAt: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'Coordinate' }] } + }, + { + itemId: 'start', + revision: 1, + sequence: 2, + observedAt: 2, + body: { + kind: 'agent-start', + dispatchId: 'dispatch', + task: 'Worker assignment', + worker: { + party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, + name: 'Worker' + } + } + }, + { + itemId: 'after', + revision: 1, + sequence: 3, + observedAt: 3, + body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'Answer' }] } + } + ] + await act(async () => listener?.(snapshotEvent(3, items))) + expect(hook?.session.messages.map(({ blocks }) => blocks)).toEqual([ + [{ type: 'text', text: 'Coordinate' }], + [{ type: 'text', text: 'Answer' }] + ]) + }) + it('re-holds before resubscribing after a reconnect', async () => { act(() => { renderer = create(createElement(Harness, { connected: true })) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts index ea66623138f..8016f61aa29 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import { projectStructuredAgentStartItem } from '../../../shared/structured-agent-start-projection' import { AGENT_JOURNAL_THREAD_SCOPE, AGENT_SESSION_JOURNAL_SCHEMA_VERSION @@ -505,3 +506,56 @@ it('preserves stored keys and reducer aliases across an item revision and a tomb expect(state.items.size).toBe(0) expect(state.tombstones.get('canonical')).toBe(9) }) + +const startWorker = { + party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, + name: 'Worker', + relation: 'worker' +} +it.each([ + { worker: startWorker, senders: [startWorker] }, + { + worker: { ...startWorker, relation: 'future-relation' }, + senders: [{ ...startWorker, relation: undefined }] + }, + { worker: { ...startWorker, name: 17 }, senders: [{ ...startWorker, name: null }] }, + { + worker: { ...startWorker, party: { ...startWorker.party, orcaSessionId: 'invalid' } }, + senders: [] + }, + { worker: null, senders: [] }, + { worker: { futureIdentity: 'new shape' }, senders: [] } +])('replays open worker metadata without damaging task/core: $worker', ({ worker, senders }) => { + const body = { kind: 'agent-start', dispatchId: 'dispatch', task: ' Full task\r\nEND ', worker } + const parsed = parseJournalRow( + JSON.stringify({ ...BASE, kind: 'item', itemId: 'start', revision: 1, body }) + ) + if (!parsed.ok) { + throw new Error('creation fact was refused') + } + expect(parsed.row).toMatchObject({ body }) + const state = createJournalReducerState('session-1', BASE.epoch) + applyJournalRow(state, parsed.row) + const item = state.items.get('start') + if (!item) { + throw new Error('creation fact was lost during replay') + } + const message = projectStructuredAgentStartItem(item) + expect(message?.blocks).toEqual([{ type: 'text', text: body.task }]) + expect(message?.from?.senders).toEqual(senders) +}) + +it.each([{ dispatchId: '' }, { dispatchId: null }, { task: null }, { task: 17 }])( + 'rejects damaged creation core while worker metadata stays open: %j', + (damage) => { + expect( + parse({ + ...BASE, + kind: 'item', + itemId: 'start', + revision: 1, + body: { kind: 'agent-start', dispatchId: 'dispatch', task: 'Task', worker: null, ...damage } + }) + ).toBe(false) + } +) diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index f3efaa62d34..23a2e6353d1 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -203,6 +203,10 @@ export class AgentSessionJournal { item = (itemId: string): AgentJournalRenderItem | null => this.state.items.get(itemId) ?? null + /** Remember publication even after a rewind removes the row. */ + hasRecordedItem = (itemId: string): boolean => + this.state.items.has(itemId) || this.state.tombstones.has(itemId) + visitItemsWithLinkage = (visit: JournalItemLinkageVisitor): void => { for (const item of this.state.items.values()) { visit(item.itemId, item.sequence, item.body, item) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.test.ts index 03ee850f74e..02dd3997146 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.test.ts @@ -6,7 +6,7 @@ import { AGENT_SESSION_RECORD_SCHEMA_VERSION, type AgentSessionRecord } from '../../../shared/agent-session-record' -import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import { closeTestJournalHostDatabases, openTestJournalHostDatabase @@ -60,15 +60,27 @@ function reservedRecord(): AgentSessionRecord { function runtimeState( record: AgentSessionRecord | null, - probeOwner: NonNullable + probeOwner: NonNullable, + options: { + store?: Partial + probeOwners?: StructuredAgentSessionHostDeps['probeOwners'] + now?: StructuredAgentSessionHostDeps['now'] + } = {} ) { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the runtime state reads only store, probeOwner and optional deps; the rest of the host's deps are unused here. + const journalDatabase = openTestJournalHostDatabase(stateDirectory) + const store = Object.assign( + AgentSessionRecordStore.open({ journalDatabase, hostId: 'local', legacySessionTabIds: [] }), + { getRecord: () => record, ...options.store } + ) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: these runtime-state cases supply every store/probe/clock dependency they invoke; other host dependencies are unused. const deps = { - store: { getRecord: () => record } as unknown as AgentSessionRecordStore, + store, adapter: {}, - journalDatabase: openTestJournalHostDatabase(stateDirectory), + journalDatabase, claimKeyId: 'key-1', probeOwner, + probeOwners: options.probeOwners, + now: options.now, logger: createStructuredAgentSessionLogger() } as StructuredAgentSessionHostDeps return new StructuredAgentSessionHostRuntimeState(deps, new Map()) @@ -127,6 +139,70 @@ describe('host runtime-state owner probe', () => { expect(probeOwner).not.toHaveBeenCalled() }) + it('constructs inert per-state reconcilers sharing only their own addressed in-flight work', async () => { + const record = liveRecord() + record.lease.unreconciled = true + const pending = Promise.withResolvers() + const getRecord = vi.fn((_sessionId: string) => record) + const reconcileOnRestart = vi.fn( + async (args: Parameters[0]) => { + await pending.promise + await args.probe(record) + await args.probeMany?.([record]) + record.lease.unreconciled = false + return new Map() + } + ) + const probeOwner = vi.fn(async () => ({ outcome: 'pid-absent' as const })) + const otherProbeOwner = vi.fn(async () => ({ outcome: 'pid-absent' as const })) + const probeOwners = vi.fn(async () => new Map()) + const otherProbeOwners = vi.fn(async () => new Map()) + const now = vi.fn(() => NOW) + const otherNow = vi.fn(() => NOW + 1) + const store = { getRecord, reconcileOnRestart } + const state = runtimeState(record, probeOwner, { store, probeOwners, now }) + const other = runtimeState(record, otherProbeOwner, { + store, + probeOwners: otherProbeOwners, + now: otherNow + }) + for (const dependency of [ + getRecord, + reconcileOnRestart, + probeOwner, + otherProbeOwner, + probeOwners, + otherProbeOwners, + now, + otherNow + ]) { + expect(dependency).not.toHaveBeenCalled() + } + + const first = state.reconcileLeases(record.sessionId) + const repeated = state.reconcileLeases(record.sessionId) + expect(reconcileOnRestart).toHaveBeenCalledTimes(1) + const separate = other.reconcileLeases(record.sessionId) + expect(reconcileOnRestart).toHaveBeenCalledTimes(2) + expect( + reconcileOnRestart.mock.calls.map(([args]) => ({ sessionId: args.sessionId, now: args.now })) + ).toEqual([ + { sessionId: record.sessionId, now: NOW }, + { sessionId: record.sessionId, now: NOW + 1 } + ]) + pending.resolve() + expect(await Promise.all([first, repeated, separate])).toEqual([null, null, null]) + for (const probe of [probeOwner, otherProbeOwner]) { + expect(probe).toHaveBeenCalledExactlyOnceWith(record) + } + for (const probe of [probeOwners, otherProbeOwners]) { + expect(probe).toHaveBeenCalledExactlyOnceWith([record]) + } + expect(now).toHaveBeenCalledOnce() + expect(otherNow).toHaveBeenCalledOnce() + expect(getRecord.mock.calls.every((args) => args[0] === record.sessionId)).toBe(true) + }) + it('does not force-close a provider for transient lease probe errors', async () => { const onEventSinkFailure = vi.fn() const log = recordingStructuredAgentSessionLogger() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts index e36a2b69d97..d4135b4dc01 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts @@ -1,3 +1,4 @@ +import { createRestartReconciler } from './structured-agent-session-restart-reconcile' import { StructuredAgentSessionDeliveryOrder } from './structured-agent-session-delivery-order' import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' import type { AgentSessionRecord } from '../../../shared/agent-session-record' @@ -27,6 +28,7 @@ export class StructuredAgentSessionHostRuntimeState { readonly deliveryOrder = new StructuredAgentSessionDeliveryOrder() readonly acquireAborts = new StructuredAgentSessionAcquireAborts() readonly startFlights = new StructuredAgentSessionStartFlights() + readonly reconcileLeases: ReturnType readonly startupAttempts: StructuredAgentSessionStartupAttempts private readonly leaseRenewer: StructuredAgentSessionLeaseRenewer private readonly onEventSinkFailure?: (sessionId: string, error: unknown) => void @@ -59,6 +61,12 @@ export class StructuredAgentSessionHostRuntimeState { // Ownership errors remain transient/visible; only sink I/O failure may force provider shutdown. logger: deps.logger }) + this.reconcileLeases = createRestartReconciler({ + store: deps.store, + probe: (record) => this.probeRecord(record), + ...(deps.probeOwners ? { probeMany: deps.probeOwners } : {}), + now: () => this.deps.now?.() ?? Date.now() + }) } startLeaseRenewal(): void { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index ce8916f826f..78dd68829eb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -1,4 +1,5 @@ import { continuationContextReader as contextReader } from './structured-agent-session-continuation-context' +import { structuredAgentStartJournal } from './structured-agent-start-journal' import type { AgentSessionRewindParams } from '../../../shared/agent-session-rewind' import { rewindStructuredAgentSession } from './structured-agent-session-rewind' import { StructuredConversationCommandController } from './structured-conversation-command-controller' @@ -6,7 +7,7 @@ import { StructuredConversationCommandController } from './structured-conversati import type { AgentSessionSubscribeInput } from './structured-agent-session-subscribers' import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record' -import * as leases from './structured-agent-session-restart-reconcile' +import { createStartupRecovery } from './structured-agent-session-restart-reconcile' import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue' import * as providerSupport from './structured-agent-session-provider-support' import * as reveal from './structured-agent-session-reveal' @@ -66,8 +67,7 @@ export class StructuredAgentSessionHost { private readonly subscribers = this.clientDelivery.subscribers private readonly tasks = new StructuredAgentSessionTaskQueue() private readonly runtimeState: StructuredAgentSessionHostRuntimeState - private readonly reconcileLeases: ReturnType - private readonly startup: ReturnType + private readonly startup: ReturnType readonly reconcileRestartLeases = () => this.startup.reconcileRestartLeases() private readonly lifetime: lifetime.StructuredAgentSessionConversationLifetime private readonly conversationDelivery: conversation.StructuredAgentSessionConversationDelivery @@ -93,12 +93,6 @@ export class StructuredAgentSessionHost { (id, e) => this.eventRecovery.recoverAfterSinkFailure(id, e), (expired) => this.lifetime.expireStartup(expired) ) - this.reconcileLeases = leases.createRestartReconciler({ - store: deps.store, - probe: (record) => this.runtimeState.probeRecord(record), - ...(deps.probeOwners ? { probeMany: deps.probeOwners } : {}), - now: () => this.now() - }) this.conversationDelivery = conversation.createStructuredAgentSessionConversationDelivery({ deps, acquireAborts: this.runtimeState.acquireAborts, @@ -115,7 +109,7 @@ export class StructuredAgentSessionHost { stopOutgrownAgent: (id) => this.lifetime.stopAgent(id, { cause: 'evict', resting: true }), clientDelivery: this.clientDelivery }) - this.startup = leases.createStartupRecovery(deps, this.reconcileLeases, this.runtimeState) + this.startup = createStartupRecovery(deps, this.runtimeState.reconcileLeases, this.runtimeState) this.eventRecovery = new StructuredAgentSessionEventRecovery({ deps, store: deps.store, @@ -183,7 +177,7 @@ export class StructuredAgentSessionHost { ...this.lifetimeContext(), subscribers: this.subscribers, tasks: this.tasks, - reconcileLeases: (sessionId) => this.reconcileLeases(sessionId), + reconcileLeases: (sessionId) => this.runtimeState.reconcileLeases(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), openProviderChild: this.eventRecovery.openProviderChild, openConversation: this.conversationDelivery.open @@ -320,6 +314,13 @@ export class StructuredAgentSessionHost { journalSnapshot = async (sessionId: string) => (await this.lifetime.conversation(sessionId)).journal.snapshot() + private readonly startJournal = structuredAgentStartJournal( + () => this.deps, + (sessionId) => this.lifetime.conversation(sessionId) + ) + captureAgentStartDestination = this.startJournal.captureAgentStartDestination + appendAgentStart = this.startJournal.appendAgentStart + currentWork = this.clientDelivery.readCurrentWork readCurrentRequests: BackgroundTasks['currentRequests'] = (...args) => this.backgroundTasks.currentRequests(...args) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-start-journal.ts b/src/main/native-chat/agent-session-wire/structured-agent-start-journal.ts new file mode 100644 index 00000000000..338f1e85890 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-start-journal.ts @@ -0,0 +1,59 @@ +import type { AgentJournalAgentStartItem } from '../../../shared/agent-session-journal-types' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { readJournalSessionEpoch } from '../agent-session-journal/journal-row-table' +import type { + StructuredAgentSessionHostDeps, + StructuredAgentSessionHostSession +} from './structured-agent-session-host-types' + +type AgentStartDestination = { sessionId: string; epoch: string; contextId: string | null } + +export function structuredAgentStartJournal( + deps: () => StructuredAgentSessionHostDeps, + conversation: (sessionId: string) => Promise +) { + /** Record only in this transcript; opening at rest neither follows clears nor starts its agent. */ + const captureAgentStartDestination = (sessionId: string): AgentStartDestination | null => { + const record = deps().store.getConversation(sessionId) + const epoch = readJournalSessionEpoch(deps().journalDatabase.db, sessionId) + return record && epoch + ? { sessionId, epoch, contextId: record.providerContextBoundary?.operationId ?? null } + : null + } + + const appendAgentStart = async ( + destination: AgentStartDestination, + body: AgentJournalAgentStartItem, + observedAt: number + ): Promise => { + const { sessionId } = destination + const stillSameContext = () => { + const current = captureAgentStartDestination(sessionId) + return current?.epoch === destination.epoch && current?.contextId === destination.contextId + } + if (!stillSameContext()) { + return + } + const { journal } = await conversation(sessionId) + const identity = { + provider: 'orca' as const, + clientMessageId: `agent-start:${body.dispatchId}` + } + await journal.appendResolvedItem( + () => { + if (!stillSameContext() || journal.hasRecordedItem(agentJournalItemKey(identity))) { + return null + } + return { identity, body } + }, + { + fence: deps().store.getConversation(sessionId)?.currentFence ?? 0, + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + observedAt + } + ) + } + + return { captureAgentStartDestination, appendAgentStart } +} diff --git a/src/main/runtime/orchestration/worker-start-card.ts b/src/main/runtime/orchestration/worker-start-card.ts new file mode 100644 index 00000000000..71be87fe604 --- /dev/null +++ b/src/main/runtime/orchestration/worker-start-card.ts @@ -0,0 +1,45 @@ +import { getStructuredAgentSessionHost } from '../../native-chat/agent-session-wire/structured-agent-session-registry' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { OrchestrationSessionCaller } from './orchestration-caller-identity' + +/** No publication work participates in launch settlement, including lookup and logging failures. */ +export function prepareWorkerStartCard( + runtime: OrcaRuntimeService, + callerSession: OrchestrationSessionCaller | undefined +): (args: { dispatchId: string; workerAddress: string; task: string }) => void { + try { + const host = getStructuredAgentSessionHost() + const destination = callerSession && host?.captureAgentStartDestination(callerSession.sessionId) + if (!host || !destination) { + return () => {} + } + return (args) => { + const observedAt = Date.now() + void Promise.resolve() + .then(async () => { + await host.appendAgentStart( + destination, + { + kind: 'agent-start', + dispatchId: args.dispatchId, + worker: runtime.orchestrationSenderNames.sender(args.workerAddress, args.dispatchId), + task: args.task + }, + observedAt + ) + }) + .catch(logStartCardFailure) + } + } catch (error) { + logStartCardFailure(error) + return () => {} + } +} + +function logStartCardFailure(error: unknown): void { + try { + console.warn('[orchestration] recording a started worker in its caller chat failed', error) + } catch { + // Display bookkeeping cannot change a successful launch. + } +} diff --git a/src/main/runtime/rpc/methods/orchestration/agent-message-command-contract.test.ts b/src/main/runtime/rpc/methods/orchestration/agent-message-command-contract.test.ts index d34bb6a65be..81d2920f7ba 100644 --- a/src/main/runtime/rpc/methods/orchestration/agent-message-command-contract.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/agent-message-command-contract.test.ts @@ -1,5 +1,16 @@ +import { randomUUID } from 'node:crypto' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { z } from 'zod' +import { UnixSocketTransport } from '../../unix-socket-transport' +import { sendRequest } from '../../../../../cli/runtime/transport' +import type { RuntimeMetadata } from '../../../../../shared/runtime-bootstrap' +import type { RuntimeOrchestrationEnvelope } from '../../../../../shared/runtime-rpc-envelope' +import { OrchestrationMutationExecutor } from '../../orchestration-mutation-executor' +import { CLIENT_HISTORY_READER } from '../../../../native-chat/agent-session-wire/agent-session-item-wire-form' import { afterEach, expect, it, vi } from 'vitest' import { createOrchestrationRpcHarness } from './rpc-test-harness' +import { partialRuntimeStore } from '../../../runtime-store-partial-fixture' import { resolveOrchestrationSessionCaller } from '../../orchestration-session-caller' import { setStructuredAgentSessionHost } from '../../../../native-chat/agent-session-wire/structured-agent-session-registry' import { @@ -28,10 +39,13 @@ import * as topology from './worker/worker-topology' const h = createOrchestrationRpcHarness() let rig: QueuedMessageTestRig | undefined +let transport: UnixSocketTransport | undefined const COORD = testOrcaSessionId('command-coordinator') const WORKER = testOrcaSessionId('command-worker') afterEach(async () => { setStructuredAgentSessionHost(null) + await transport?.stop() + transport = undefined await rig?.dispose() rig = undefined h.cleanup() @@ -43,7 +57,7 @@ afterEach(async () => { process.exitCode = 0 }) -it.each(['worker-start', 'dispatch'] as const)( +it.each(['worker-start', 'saved-task', 'dispatch'] as const)( 'real %s argv constructs, persists and delivers canonical input with a derived task display', async (command) => { rig = await createQueuedMessageTestRig({ sessionId: WORKER }) @@ -58,7 +72,10 @@ it.each(['worker-start', 'dispatch'] as const)( } }) expect(await current.host.attach(QUEUED_RIG_CALLER, attach)).toMatchObject({ ok: true }) - const { db, runtime } = h.setup(false) + const { db, runtime } = h.setup( + false, + partialRuntimeStore({ getSettings: () => ({ experimentalNativeChat: false }) }) + ) vi.spyOn(runtime, 'ensureStructuredAgentSessionHost').mockResolvedValue() vi.spyOn(runtime, 'validateOrchestrationAgentLauncher').mockImplementation(() => {}) type WorkspaceLookup = { @@ -118,43 +135,125 @@ it.each(['worker-start', 'dispatch'] as const)( vi.stubEnv('ORCA_PAIRING_CODE', '') vi.stubEnv('ORCA_ENVIRONMENT', '') const requests: string[] = [] + const requestBytes: number[] = [] + let rejectedFrames = 0 + const kind = process.platform === 'win32' ? 'named-pipe' : 'unix' + const endpoint = + kind === 'named-pipe' + ? `\\\\.\\pipe\\orca-start-${randomUUID()}` + : join(tmpdir(), `orca-start-${randomUUID()}.sock`) + transport = new UnixSocketTransport({ kind, endpoint }) + const executor = new OrchestrationMutationExecutor(runtime) + const requestSchema = z.object({ + id: z.string(), + authToken: z.string(), + method: z.string(), + params: z.unknown(), + orchestrationRequestId: z.string().optional() + }) + transport.onMessage((wire, reply) => { + if (!wire) { + rejectedFrames++ + reply( + JSON.stringify({ + id: 'unknown', + ok: false, + error: { code: 'bad_request', message: 'Oversized request' } + }) + ) + return + } + void (async () => { + const request = requestSchema.parse(JSON.parse(wire)) + requestBytes.push(Buffer.byteLength(wire)) + const admitted = await resolveOrchestrationSessionCaller( + runtime, + { + ...request, + orchestrationCompatibilityEvidence: { agentSessionId: COORD } + }, + undefined + ) + requests.push(request.method) + const rpc = h.findMethod(request.method) + const parsed = rpc.params?.parse(admitted.request.params) + const result = await executor.run( + admitted.request, + parsed, + (mutation) => + rpc.handler(parsed, { + runtime, + orchestrationMutation: mutation?.identity, + recordMutationReceipt: mutation?.recordReceipt, + replayedMutationReceipt: mutation?.replayedReceipt, + markMutationEffectPossible: mutation?.markEffectPossible, + orchestrationCaller: admitted.caller, + orchestrationCompatibilityEvidence: + admitted.request.orchestrationCompatibilityEvidence + }), + undefined, + admitted.caller?.sessionId + ) + reply( + JSON.stringify({ + id: request.id, + ok: true, + result, + _meta: { runtimeId: runtime.getRuntimeId() } + }) + ) + })().catch((error: unknown) => { + const request = requestSchema.parse(JSON.parse(wire)) + reply( + JSON.stringify({ + id: request.id, + ok: false, + error: { + code: 'fixture_failure', + message: String(error) + } + }) + ) + }) + }) + await transport.start() + const metadata: RuntimeMetadata = { + runtimeId: runtime.getRuntimeId(), + pid: process.pid, + startedAt: 1, + authToken: 'test-token', + transports: [{ kind, endpoint }] + } + const call = async ( + method: string, + params: unknown, + options?: RuntimeOrchestrationEnvelope + ) => { + const response = await sendRequest(metadata, method, params, 10000, options) + if (!response.ok) { + throw new Error(response.error.message) + } + return response + } vi.doMock('../../../../../cli/runtime-client', async (original) => ({ ...(await original>()), RuntimeClient: function () { - return { - isRemote: false, - async call(method: string, params: Record = {}) { - const admitted = await resolveOrchestrationSessionCaller( - runtime, - { - id: 'cli', - authToken: 'test-token', - method, - params, - orchestrationCompatibilityEvidence: { agentSessionId: COORD } - }, - undefined - ) - requests.push(method) - const rpc = h.findMethod(method) - const parsed = rpc.params?.parse(admitted.request.params) - return { - result: await rpc.handler(parsed, { - runtime, - orchestrationCaller: admitted.caller, - orchestrationCompatibilityEvidence: - admitted.request.orchestrationCompatibilityEvidence - }) - } - } - } + return { isRemote: false, call } } })) - const output = vi.fn() + const output = + vi.fn< + (response: { + id: string + ok: boolean + result: Record + _meta: { runtimeId: string } + }) => void + >() vi.doMock('../../../../../cli/format', () => ({ printResult: output })) const cliPath = '../../../../../cli/index' const { main }: { main(argv: string[]): Promise } = await import(cliPath) - const task = ' Current task 🐋\r\n\r\n=== TASK ===\nDo not clip the task. ' + const task = ` Current task 🐋\r\n"\t\u0001\n=== TASK ===\nDo not clip the task. ${command === 'dispatch' ? '' : '\n'.repeat(490000)}` let argv: string[] let openingSend: string | undefined if (command === 'dispatch') { @@ -181,11 +280,21 @@ it.each(['worker-start', 'dispatch'] as const)( 'codex', '--timeout-ms', '1000', + '--retry-request', + '11111111-1111-4111-8111-111111111111', '--json' ] } + if (command === 'saved-task') { + await main(['orchestration', 'task-create', '--spec', task, '--json']) + const created = db.listTasks({ runId: run.id }).find((entry) => entry.spec === task) + if (!created) { + throw new Error('serialized task-create did not store exact task') + } + argv.splice(argv.indexOf('--spec'), 2, '--task', created.id) + } const running = main(argv) - if (command === 'worker-start') { + if (command !== 'dispatch') { await eventually(() => expect(current.dispatch).toHaveBeenCalled()) const id = current.dispatch.mock.calls[0][0].clientMessageId await current.settleAccepted(id, 'task-turn') @@ -194,6 +303,75 @@ it.each(['worker-start', 'dispatch'] as const)( expect(requests).toContain( command === 'dispatch' ? 'orchestration.dispatch' : 'orchestration.workerStart' ) + if (command !== 'dispatch') { + await eventually(async () => + expect( + (await current.host.journalSnapshot(COORD)).items.filter( + (item) => item.body.kind === 'agent-start' + ) + ).toHaveLength(1) + ) + const fact = (await current.host.journalSnapshot(COORD)).items.find( + (item) => item.body.kind === 'agent-start' + ) + if (!fact) { + throw new Error('ready caller fact missing') + } + expect(fact.body).toMatchObject({ task }) + const callerHistory = await current.host.history({ sessionId: COORD, direction: 'tail' }) + expect(callerHistory).toMatchObject({ ok: true, page: { items: [fact] } }) + const formed = CLIENT_HISTORY_READER.form(fact) + expect(formed.item).toBe(fact) + expect(formed.item.shortened).toBeUndefined() + expect(formed.bytes).toBeGreaterThan(980000) + expect(formed.bytes).toBeLessThan(CLIENT_HISTORY_READER.budgetBytes) + expect(Math.max(...requestBytes)).toBeGreaterThan(980000) + expect(Math.max(...requestBytes)).toBeLessThan(1024 * 1024) + console.info('start input/reader bytes', { + command, + request: Math.max(...requestBytes), + storedTask: Buffer.byteLength(task), + row: formed.bytes, + readerBudget: CLIENT_HISTORY_READER.budgetBytes + }) + const placed = vi.mocked(placement.placeWorkerAgent).mock.calls.length + const printed = output.mock.calls.length + const resultBeforeReplay = output.mock.lastCall?.[0] + if (!resultBeforeReplay) { + throw new Error('successful worker-start response missing') + } + expect(resultBeforeReplay).toMatchObject({ + ok: true, + _meta: { runtimeId: runtime.getRuntimeId() }, + result: { mutation: { requestId: '11111111-1111-4111-8111-111111111111', replayed: false } } + }) + await main(argv) + expect(process.exitCode ?? 0).toBe(0) + expect(output).toHaveBeenCalledTimes(printed + 1) + expect(output.mock.lastCall?.[0]).toEqual({ + ...resultBeforeReplay, + id: expect.any(String), + result: { + ...resultBeforeReplay.result, + mutation: { requestId: '11111111-1111-4111-8111-111111111111', replayed: true } + } + }) + expect(output.mock.lastCall?.[0].id).not.toBe(resultBeforeReplay.id) + expect(placement.placeWorkerAgent).toHaveBeenCalledTimes(placed) + expect( + (await current.host.journalSnapshot(COORD)).items.filter( + (item) => item.body.kind === 'agent-start' + ) + ).toEqual([fact]) + const beforeWrites = db.listTasks({ runId: run.id }).length + const beforeRequests = requests.length + await expect( + call('orchestration.taskCreate', { spec: task + '\n'.repeat(100000), run: run.id }) + ).rejects.toThrow() + expect(rejectedFrames).toBe(1) + expect(requests).toHaveLength(beforeRequests) + expect(db.listTasks({ runId: run.id })).toHaveLength(beforeWrites) + } const snapshot = await current.host.journalSnapshot(WORKER) const history = await current.host.history({ sessionId: WORKER, direction: 'tail' }) if (!history.ok) { @@ -215,6 +393,9 @@ it.each(['worker-start', 'dispatch'] as const)( throw new Error('production host did not persist task') } const from = readAgentMessageSource(JSON.parse(JSON.stringify(body.from))) + if (command !== 'dispatch') { + expect(db.listTasks({ runId: run.id }).filter((entry) => entry.spec === task)).toHaveLength(1) + } expect(from?.senders[0].party.orcaSessionId).toBe(COORD) expect(from?.senders[0].relation).toBe('coordinator') expect(agentMessageDisplayText(body.blocks, from)).toBe(task) @@ -225,7 +406,7 @@ it.each(['worker-start', 'dispatch'] as const)( expect(latestStructuredAgentSessionPrompt(items)).toBe(raw) const prompt = projectStructuredAgentSessionStatusSummary(items).latestPrompt expect(isOrcaDispatchPrompt(prompt)).toBe(true) - expect(prompt).toContain('=== TASK === Current task') + expect(prompt).toContain('Current task') if (from?.orchestration?.message !== 'task') { throw new Error('task identity missing') } @@ -274,5 +455,6 @@ it.each(['worker-start', 'dispatch'] as const)( } else { expect(nativeChatMessageCopyText(current.dispatch.mock.calls[0][0].body)).toBe(raw) } - } + }, + 30000 ) diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts b/src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts index 4fb3ca60007..7231657c406 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts @@ -34,6 +34,7 @@ import { } from './federated-worker-start-receipts' import { parseTaskDeps } from '../worker/task-deps-argument' import { releaseRetriedWorkerTerminal } from '../worker/retried-worker-terminal-release' +import { prepareWorkerStartCard } from '../../../../orchestration/worker-start-card' export async function startFederatedWorker(args: { params: WorkerStartInput @@ -51,6 +52,7 @@ export async function startFederatedWorker(args: { callerSession?: OrchestrationSessionCaller }): Promise { const { params, runtime, db, task, runId, orchestrationMutation } = args + const recordStart = prepareWorkerStartCard(runtime, args.callerSession) if (!isWorkerStartTimeoutWithinTimerLimit(params.timeoutMs)) { throw new OrchestrationError( 'invalid_argument', @@ -162,6 +164,7 @@ export async function startFederatedWorker(args: { const createdTask = started.task const taskForRemote = task ?? createdTask db.recordWorkerStage({ dispatchId: started.dispatch.id, stage: 'remote_attach_requested' }) + let readyResult: unknown try { const remote = parseRemoteFederatedWorkerStartReceipt( await runtime.callOrchestrationWorkerServer( @@ -229,7 +232,7 @@ export async function startFederatedWorker(args: { }) const readyWorker = db.markWorkerDispatchReady(started.dispatch.id) runtime.ensureOrchestrationFederationRelay(runId) - return { + readyResult = { runId, taskId: taskForRemote.id, dispatchId: started.dispatch.id, @@ -242,33 +245,33 @@ export async function startFederatedWorker(args: { effects: remote.effects ?? [], residualResources: remote.residualResources ?? [] } - } - if (remote.state === 'outcome_unknown') { + } else if (remote.state === 'outcome_unknown') { const worker = db.markWorkerStartUnknown( started.dispatch.id, remote.failedStage ?? 'remote_attach', remote.lastError ?? 'The worker server reported an unknown start outcome.' ) return federatedUnknownReceipt(worker, taskForRemote.id, server.name, launch) - } - const worker = db.failWorkerStart( - started.dispatch.id, - remote.failedStage ?? 'remote_attach', - remote.lastError ?? `The worker server returned ${remote.state}.` - ) - return { - runId, - taskId: taskForRemote.id, - dispatchId: started.dispatch.id, - state: worker.state, - stage: worker.stage, - server: { environmentId: server.environmentId, name: server.name }, - failedStage: worker.stage, - lastError: worker.last_error, - setup: remote.setup, - launch, - effects: remote.effects ?? [], - residualResources: remote.residualResources ?? [] + } else { + const worker = db.failWorkerStart( + started.dispatch.id, + remote.failedStage ?? 'remote_attach', + remote.lastError ?? `The worker server returned ${remote.state}.` + ) + return { + runId, + taskId: taskForRemote.id, + dispatchId: started.dispatch.id, + state: worker.state, + stage: worker.stage, + server: { environmentId: server.environmentId, name: server.name }, + failedStage: worker.stage, + lastError: worker.last_error, + setup: remote.setup, + launch, + effects: remote.effects ?? [], + residualResources: remote.residualResources ?? [] + } } } catch (error) { const reason = error instanceof Error ? error.message : String(error) @@ -291,4 +294,12 @@ export async function startFederatedWorker(args: { const worker = db.markWorkerStartUnknown(started.dispatch.id, 'remote_attach', reason) return federatedUnknownReceipt(worker, taskForRemote.id, server.name, requestedLaunch) } + if (!params.terminal) { + recordStart({ + dispatchId: started.dispatch.id, + workerAddress: `dispatch:${started.dispatch.id}`, + task: taskForRemote.spec + }) + } + return readyResult } diff --git a/src/main/runtime/rpc/methods/orchestration/worker-start-card.test.ts b/src/main/runtime/rpc/methods/orchestration/worker-start-card.test.ts new file mode 100644 index 00000000000..93937515b97 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/worker-start-card.test.ts @@ -0,0 +1,764 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { createOrchestrationRpcHarness } from './rpc-test-harness' +import { partialRuntimeStore } from '../../../runtime-store-partial-fixture' +import { resolveOrchestrationSessionCaller } from '../../orchestration-session-caller' +import { setStructuredAgentSessionHost } from '../../../../native-chat/agent-session-wire/structured-agent-session-registry' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from '../../../../native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture' +import { + hostTestAttachParams, + hostTestOperationId +} from '../../../../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { testOrcaSessionId } from '../../../../../shared/orca-session-address-test-fixture' +import { projectStructuredAgentStartItem } from '../../../../../shared/structured-agent-start-projection' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemBody +} from '../../../../../shared/agent-session-journal-types' +import { + closeTestJournalHostDatabase, + openTestJournalHostDatabase +} from '../../../../native-chat/agent-session-journal/journal-host-database-test-support' +import * as journalSchema from '../../../../native-chat/agent-session-journal/journal-row-schema' +import { StructuredAgentSessionJournalProjections } from '../../../../native-chat/agent-session-wire/structured-agent-session-status-journal-projection' +import { installSqlVisitCounters } from '../../../dispatch-predicate-visits.test-fixture' +import { createMinimalPersistedTerminalTab } from '../../../../persistence/restoring-sessions/session-owner-fields' +import { getDefaultWorkspaceSession } from '../../../../../shared/constants' +import { readFile } from 'node:fs/promises' +import { projectStructuredAgentSessionStatusSummary } from '../../../../../shared/structured-agent-session-projection' +import { prepareWorkerStartCard } from '../../../orchestration/worker-start-card' +import * as placement from './worker/worker-start-agent-placement' +import * as topology from './worker/worker-topology' +import { + ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY, + ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY +} from '../../../../../shared/protocol-version' + +const h = createOrchestrationRpcHarness() +const COORD = testOrcaSessionId('start-card-caller') +const WORKER = testOrcaSessionId('start-card-worker') +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + setStructuredAgentSessionHost(null) + await rig?.dispose() + rig = undefined + h.cleanup() + vi.restoreAllMocks() +}) + +it.each([ + 'success', + 'failure', + 'readiness-failure', + 'append-throws', + 'append-pending', + 'logging-throws', + 'reuse', + 'dispatch', + 'context-only', + 'cleared-context', + 'federated', + 'federated-unknown' +] as const)( + 'real worker-start records creation independently of launch: %s', + async (scenario) => { + rig = await createQueuedMessageTestRig({ sessionId: WORKER }) + const current = rig + if (scenario === 'readiness-failure') { + current.dispatch.mockRejectedValueOnce(new Error('provider dispatch failed')) + } + const federated = scenario === 'federated' || scenario === 'federated-unknown' + setStructuredAgentSessionHost(current.host) + expect( + await current.host.attach( + QUEUED_RIG_CALLER, + hostTestAttachParams(null, { + envelope: { + sessionId: COORD, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: '' + } + }) + ) + ).toMatchObject({ ok: true }) + const { runtime, db } = h.setup( + false, + partialRuntimeStore({ getSettings: () => ({ experimentalNativeChat: false }) }) + ) + if (federated) { + vi.spyOn(runtime, 'resolveOrchestrationWorkerServer').mockReturnValue({ + environmentId: 'remote', + name: 'remote', + peerFingerprint: 'peer', + pairingRevision: 1 + }) + vi.spyOn(runtime, 'ensureOrchestrationFederationRelay').mockImplementation(() => {}) + vi.spyOn(runtime, 'callOrchestrationWorkerServer').mockImplementation( + async (_host, method, params) => { + if (method === 'status.get') { + return { + capabilities: [ + ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY, + ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY + ] + } + } + if (!params || typeof params !== 'object' || !('dispatchId' in params)) { + throw new Error('missing remote dispatch') + } + return { + dispatchId: params.dispatchId, + state: scenario === 'federated' ? 'ready' : 'outcome_unknown', + runtimeEpoch: 'remote-epoch', + worktreeId: 'remote-workspace', + terminalHandle: 'term_remote' + } + } + ) + } + vi.spyOn(runtime, 'ensureStructuredAgentSessionHost').mockResolvedValue() + vi.spyOn(runtime, 'validateOrchestrationAgentLauncher').mockImplementation(() => {}) + type WorkspaceLookup = { + showManagedTerminalWorkspace: ( + ...args: Parameters + ) => Promise + } + vi.spyOn( + runtime, + 'showManagedTerminalWorkspace' + ).mockResolvedValue({ id: 'workspace-1', repoId: 'repo' }) + vi.spyOn(runtime, 'getStructuredAgentSessionCreateSupport').mockResolvedValue({ + supported: true + }) + vi.spyOn(runtime, 'getOrchestrationDispatchAuthority').mockReturnValue({ + runtimeId: runtime.getRuntimeId(), + terminalHandle: 'term_worker', + ptyId: 'pty', + worktreeId: 'workspace-1', + paneKey: 'tab:11111111-1111-4111-8111-111111111111', + processIncarnation: 'fixture:1', + launchTokenHash: null, + hostScope: { kind: 'local', hostId: 'local' } + }) + vi.spyOn(topology, 'monitorWorkerSetup').mockImplementation(() => {}) + vi.spyOn(placement, 'placeWorkerAgent').mockImplementation(async (args) => { + if (scenario === 'failure') { + throw new Error('placement failed') + } + return { + mode: args.mode, + worktree: { id: 'workspace-1', repoId: 'repo' }, + terminalHandle: 'term_worker', + structuredSession: { + host: current.host, + identity: { + sessionId: WORKER, + handle: 'term_worker', + agent: args.agent ?? null, + paneKey: 'tab:11111111-1111-4111-8111-111111111111', + processIncarnation: 'fixture:1', + worktreeId: 'workspace-1', + hostScope: { kind: 'local', hostId: 'local' } + }, + fence: current.store.requireWritableRecord(WORKER).lease.runtimeFence + }, + setupReceipt: placement.EXISTING_WORKTREE_SETUP + } + }) + const run = db.createRun({ + objective: 'cards', + coordinatorHandle: null, + coordinatorPaneKey: null, + coordinatorOrcaSessionId: COORD + }) + const append = vi.spyOn(current.host, 'appendAgentStart') + const pending = Promise.withResolvers() + if (scenario === 'append-throws' || scenario === 'logging-throws') { + append.mockRejectedValueOnce(new Error('journal unavailable')) + } else if (scenario === 'append-pending') { + append.mockReturnValueOnce(pending.promise) + } + const warn = vi.spyOn(console, 'warn').mockImplementation((message) => { + if ( + scenario === 'logging-throws' && + message === '[orchestration] recording a started worker in its caller chat failed' + ) { + throw new Error('logger failed') + } + }) + const task = ` Full task 🐋\r\n${'task content\n'.repeat(9000)}END ` + const dispatch = scenario === 'dispatch' || scenario === 'context-only' + const method = dispatch ? 'orchestration.dispatch' : 'orchestration.workerStart' + const admitted = await resolveOrchestrationSessionCaller( + runtime, + { + id: 'start-card', + authToken: 'test-token', + method, + params: dispatch + ? { + task: db.createTask({ runId: run.id, spec: task }).id, + to: `orca_session_id:${WORKER}`, + inject: scenario === 'dispatch', + returnPreamble: true + } + : { + spec: task, + ...(scenario === 'reuse' ? {} : { agent: 'codex' }), + timeoutMs: 1000, + ...(federated ? { on: 'remote', worktree: 'id:remote-workspace' } : {}), + ...(scenario === 'reuse' ? { terminal: `orca_session_id:${WORKER}` } : {}) + }, + orchestrationCompatibilityEvidence: { agentSessionId: COORD } + }, + undefined + ) + const callerJournal = current.host.collaboratorsForTests().sessions.get(COORD)!.journal + const callerBodies: AgentJournalItemBody[] = [ + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'Coordinate this work' }] }, + { kind: 'tool-call', name: 'shell', input: { command: 'inspect' }, state: 'running' } + ] + for (const [index, body] of callerBodies.entries()) { + await callerJournal.appendItem( + { provider: 'orca', clientMessageId: `caller-${index}` }, + body, + { + fence: current.store.getConversation(COORD)!.currentFence, + turnScope: AGENT_JOURNAL_THREAD_SCOPE + } + ) + } + const before = await current.host.journalSnapshot(COORD) + const rpc = h.findMethod(method) + const running = rpc.handler(rpc.params?.parse(admitted.request.params), { + runtime, + orchestrationCaller: admitted.caller, + orchestrationCompatibilityEvidence: admitted.request.orchestrationCompatibilityEvidence, + ...(federated + ? { + orchestrationMutation: { + callerFingerprint: 'caller', + requestId: 'start-request', + method, + payloadHash: 'payload' + } + } + : {}) + }) + if (scenario !== 'failure' && scenario !== 'context-only' && !federated) { + await eventually(() => expect(current.dispatch).toHaveBeenCalled()) + expect(append).not.toHaveBeenCalled() + if (scenario === 'cleared-context') { + const getConversation = current.store.getConversation + vi.spyOn(current.store, 'getConversation').mockImplementation((id) => { + const record = getConversation(id) + return record && id === COORD + ? { + ...record, + providerContextBoundary: { + operationId: 'later-clear', + afterFence: record.currentFence, + clearedAt: 1700000000000 + } + } + : record + }) + } + if (scenario !== 'readiness-failure') { + await current.settleAccepted(current.dispatch.mock.calls[0][0].clientMessageId, 'task-turn') + } + } + const result = await running + if (scenario === 'readiness-failure') { + expect(result).not.toMatchObject({ state: 'ready' }) + } else if (!dispatch) { + expect(result).toMatchObject({ + state: + scenario === 'failure' + ? 'failed' + : scenario === 'federated-unknown' + ? 'outcome_unknown' + : 'ready' + }) + } + // A pending display append must not hold the successful RPC; release it only after the reply. + if (scenario === 'append-pending') { + expect(append).toHaveBeenCalledTimes(1) + pending.resolve() + } + if (scenario === 'append-throws' || scenario === 'logging-throws') { + await eventually(() => + expect(warn).toHaveBeenCalledWith( + '[orchestration] recording a started worker in its caller chat failed', + expect.any(Error) + ) + ) + } + if (scenario === 'success' || scenario === 'federated') { + await eventually(async () => + expect( + (await current.host.journalSnapshot(COORD)).items.some( + (item) => item.body.kind === 'agent-start' + ) + ).toBe(true) + ) + } + const snapshot = await current.host.journalSnapshot(COORD) + const cards = snapshot.items.filter((item) => item.body.kind === 'agent-start') + if (scenario !== 'success' && scenario !== 'federated') { + expect(cards).toHaveLength(0) + return + } + expect(cards).toHaveLength(1) + const [card] = cards + expect(snapshot.items.map(({ itemId }) => itemId)).toEqual([ + ...before.items.map(({ itemId }) => itemId), + card.itemId + ]) + expect(card.body).toMatchObject({ + kind: 'agent-start', + task, + ...(!federated + ? { worker: { party: { address: 'term_worker', terminalHandle: 'term_worker' } } } + : {}) + }) + if (card.body.kind !== 'agent-start') { + throw new Error('missing start body') + } + if (federated) { + expect(card.body.worker.party.address).toBe(`dispatch:${card.body.dispatchId}`) + } + const destination = append.mock.calls[0][0] + await Promise.all([ + current.host.appendAgentStart( + destination, + { ...card.body, task: 'replayed task' }, + card.observedAt + 10 + ), + current.host.appendAgentStart(destination, card.body, card.observedAt + 20) + ]) + expect( + (await current.host.journalSnapshot(COORD)).items.filter( + (item) => item.body.kind === 'agent-start' + ) + ).toEqual(cards) + expect(projectStructuredAgentStartItem(card)).toMatchObject({ + role: 'system', + agentStart: true, + timestamp: card.observedAt, + blocks: [{ type: 'text', text: task }] + }) + expect( + projectStructuredAgentSessionStatusSummary(snapshot.items, snapshot.submissions) + ).toEqual(projectStructuredAgentSessionStatusSummary(before.items, before.submissions)) + // Full task survives removal of the orchestration rows it came from. + db.resetAll() + expect( + (await current.host.journalSnapshot(COORD)).items.find((item) => item.itemId === card.itemId) + ).toEqual(card) + if (scenario === 'success') { + const context = await current.host.continuationContext({ + sessionId: COORD, + workspaceId: current.store.getConversation(COORD)!.location.workspaceId, + expectedCursor: snapshot.cursor + }) + expect(context).toMatchObject({ status: 'current' }) + if (context.status !== 'current' || !context.transcriptPath) { + throw new Error('saved context missing') + } + const saved = await readFile(context.transcriptPath, 'utf8') + expect(saved).toContain('Coordinate this work') + expect(saved).not.toContain(task) + expect(context.capturedText).not.toContain(task) + await current.quitRestartHostProcess() + closeTestJournalHostDatabase(current.root) + current.crashRestartHostProcess() + const reopened = await current.host.journalSnapshot(COORD) + expect(reopened.items.map(({ itemId }) => itemId)).toEqual( + snapshot.items.map(({ itemId }) => itemId) + ) + expect(reopened.items.filter(({ body }) => body.kind !== 'tool-call')).toEqual( + snapshot.items.filter(({ body }) => body.kind !== 'tool-call') + ) + const recovered = reopened.items.find(({ itemId }) => itemId === 'orca:caller-1') + expect(recovered).toEqual({ + ...snapshot.items[1], + revision: snapshot.items[1].revision + 1, + recovered: true, + recoveredAt: expect.any(Number), + body: { + kind: 'tool-call', + name: 'shell', + input: { command: 'inspect' }, + state: 'failed', + endedAs: 'interrupted' + } + }) + expect(reopened.cursor.epoch).toBe(snapshot.cursor.epoch) + expect(reopened.cursor.sequence).toBeGreaterThan(snapshot.cursor.sequence) + const history = await current.host.history({ sessionId: COORD, direction: 'tail' }) + expect(history).toMatchObject({ ok: true, page: { items: reopened.items } }) + } + const journal = current.host.collaboratorsForTests().sessions.get(COORD)!.journal + if (scenario === 'success') { + for (let index = 0; index < 3; index++) { + await journal.appendItem( + { provider: 'orca', clientMessageId: `pressure-${index}` }, + { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: 'x'.repeat(750000) }] + }, + { + fence: current.store.getConversation(COORD)!.currentFence, + turnScope: AGENT_JOURNAL_THREAD_SCOPE + } + ) + } + const tail = await current.host.history({ sessionId: COORD, direction: 'tail' }) + if (!tail.ok) { + throw new Error('pressure tail missing') + } + expect(tail.page.hasOlder).toBe(true) + expect(tail.page.items.some(({ itemId }) => itemId === card.itemId)).toBe(false) + const older = await current.host.history({ + sessionId: COORD, + direction: 'before', + cursor: tail.page.window.nextCursor + }) + if (!older.ok) { + throw new Error('pressure older page missing') + } + expect(older.page.items.find(({ itemId }) => itemId === card.itemId)).toEqual(card) + } + await journal.appendTombstone( + { provider: 'orca', clientMessageId: `agent-start:${card.body.dispatchId}` }, + { fence: current.store.getConversation(COORD)!.currentFence } + ) + if (scenario === 'success') { + await current.quitRestartHostProcess() + closeTestJournalHostDatabase(current.root) + current.crashRestartHostProcess() + } + await current.host.appendAgentStart(destination, card.body, card.observedAt + 30) + expect( + (await current.host.journalSnapshot(COORD)).items.some((item) => item.itemId === card.itemId) + ).toBe(false) + }, + 20_000 +) + +it.each([10, 100])( + 'counts populated producer source and journal operations for %i starts and duplicate publications', + async (count) => { + rig = await createQueuedMessageTestRig({ sessionId: COORD }) + const current = rig + setStructuredAgentSessionHost(current.host) + const session = getDefaultWorkspaceSession() + const { runtime, db } = h.setup( + false, + partialRuntimeStore({ + getSettings: () => ({ experimentalNativeChat: false }), + getWorkspaceSession: () => session + }) + ) + vi.spyOn(runtime, 'ensureStructuredAgentSessionHost').mockResolvedValue() + const run = db.createRun({ + objective: 'scale', + coordinatorHandle: null, + coordinatorPaneKey: null + }) + const tabs: ReturnType[] = [] + const facts: { dispatchId: string; workerAddress: string; task: string; name: string }[] = [] + for (let index = 0; index < 10 * count; index++) { + const handle = `term_scale_${index}` + const tabId = `scale-tab-${index}` + const leafId = `11111111-1111-4111-8111-${String(index).padStart(12, '0')}` + tabs.push( + createMinimalPersistedTerminalTab({ + worktreeId: 'workspace-1', + tabId, + ptyId: `pty-${index}`, + existingTabCount: index + }) + ) + runtime.handles.set(handle, { + handle, + runtimeId: runtime.getRuntimeId(), + rendererGraphEpoch: 1, + worktreeId: 'workspace-1', + tabId, + leafId, + ptyId: null, + ptyGeneration: 0, + issuedForPaneKey: `${tabId}:${leafId}` + }) + const task = db.createTask({ + runId: run.id, + spec: `Task ${index}`, + displayName: `Scale worker ${index}` + }) + const { dispatch } = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: {} + }) + db.prepareStartingWorkerAuthority({ + dispatchId: dispatch.id, + handle, + paneKey: `${tabId}:${leafId}`, + processIncarnation: `scale:${index}`, + worktreeId: 'workspace-1', + effects: [], + setupState: 'not_applicable' + }) + db.markWorkerDispatchReady(dispatch.id) + if (index >= 9 * count) { + facts.push({ + dispatchId: dispatch.id, + workerAddress: handle, + task: task.spec, + name: `Scale worker ${index}` + }) + } + } + let tabVisits = 0 + tabs.forEach((tab, index) => { + Object.defineProperty(tabs, index, { + get() { + tabVisits++ + return tab + } + }) + }) + session.tabsByWorktree['workspace-1'] = tabs + vi.mocked(runtime.getTerminalPaneKey).mockImplementation( + (handle) => runtime.handles.get(handle)?.issuedForPaneKey ?? null + ) + const journal = current.host.collaboratorsForTests().sessions.get(COORD)!.journal + for (let index = 0; index < 9 * count; index++) { + await journal.appendItem( + { provider: 'orca', clientMessageId: `history-${index}` }, + { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: `History ${index}` }] + }, + { + fence: current.store.getConversation(COORD)!.currentFence, + turnScope: AGENT_JOURNAL_THREAD_SCOPE + } + ) + } + const admitted = await resolveOrchestrationSessionCaller( + runtime, + { + id: 'scale', + authToken: 'test-token', + method: 'orchestration.workerStart', + params: { spec: 'Task' }, + orchestrationCompatibilityEvidence: { agentSessionId: COORD } + }, + undefined + ) + let dispatchVisits = 0 + let taskVisits = 0 + installSqlVisitCounters({ + start_dispatch_visit: () => { + dispatchVisits++ + return 1 + }, + start_task_visit: () => { + taskVisits++ + return 1 + } + }) + const prepareSource = db.db.prepare.bind(db.db) + const sourceQueries = vi + .spyOn(db.db, 'prepare') + .mockImplementation((query) => + prepareSource( + query + .replace( + 'WHERE assignee_handle = ?', + 'WHERE assignee_handle = ? AND start_dispatch_visit(rowid)' + ) + .replace( + 'FROM tasks WHERE id = ?', + 'FROM tasks WHERE id = ? AND start_task_visit(rowid)' + ) + ) + ) + const connection = openTestJournalHostDatabase(current.root).db + const epochReads = vi.spyOn( + connection.prepare('SELECT epoch FROM journal_sessions WHERE session_id = ?'), + 'get' + ) + const inserts = vi.spyOn( + connection.prepare( + 'INSERT INTO journal_rows (session_id, epoch, seq, ts, row_json) VALUES (?, ?, ?, ?, ?)' + ), + 'run' + ) + const journalQueries = vi.spyOn(connection, 'prepare') + const serialized = vi.spyOn(journalSchema, 'serializeJournalRow') + const activeSource = vi.spyOn(db, 'getActiveDispatchForTerminal') + const recordedSource = vi.spyOn(db, 'getDispatchContextById') + const sender = vi.spyOn(runtime.orchestrationSenderNames, 'sender') + const names = vi.spyOn(runtime.orchestrationSenderNames, 'nameOf') + const dedupe = vi.spyOn(journal, 'hasRecordedItem') + const scan = vi.spyOn(journal, 'visitItemsWithLinkage') + const projection = StructuredAgentSessionJournalProjections.prototype + const readProjection = projection.read + let projecting = false + vi.spyOn(projection, 'read').mockImplementation(function ( + this: StructuredAgentSessionJournalProjections, + ...args + ) { + const previous = projecting + projecting = true + try { + return readProjection.apply(this, args) + } finally { + projecting = previous + } + }) + const snapshotSources: string[] = [] + const readSnapshot = journal.snapshot.bind(journal) + const snapshot = vi.spyOn(journal, 'snapshot').mockImplementation(() => { + snapshotSources.push(projecting ? 'status-projection' : 'other') + return readSnapshot() + }) + const append = vi.spyOn(current.host, 'appendAgentStart') + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const before = journal.snapshot() + snapshot.mockClear() + const fail = prepareWorkerStartCard(runtime, admitted.caller) + expect(epochReads).toHaveBeenCalledTimes(1) + inserts.mockImplementationOnce(() => { + throw new Error('fixture insert failed') + }) + fail(facts[0]) + await eventually(() => expect(append).toHaveBeenCalledTimes(1)) + await Promise.allSettled(append.mock.results.map((result) => result.value)) + await eventually(() => + expect(warn).toHaveBeenCalledWith( + '[orchestration] recording a started worker in its caller chat failed', + expect.any(Error) + ) + ) + expect(epochReads).toHaveBeenCalledTimes(3) + expect(serialized).toHaveBeenCalledTimes(1) + expect(inserts).toHaveBeenCalledTimes(1) + expect(sourceQueries).toHaveBeenCalledTimes(2) + expect(dispatchVisits).toBe(1) + expect(taskVisits).toBe(1) + expect(tabVisits).toBe(9 * count + 1) + expect(journal.snapshot()).toEqual(before) + expect( + connection + .prepare('SELECT COUNT(*) AS count FROM journal_rows WHERE session_id = ?') + .get(COORD)?.count + ).toBe(before.cursor.sequence) + for (const counter of [ + epochReads, + serialized, + inserts, + sourceQueries, + journalQueries, + activeSource, + recordedSource, + sender, + names, + dedupe, + scan, + snapshot, + append, + warn + ]) { + counter.mockClear() + } + snapshotSources.length = 0 + tabVisits = dispatchVisits = taskVisits = 0 + const publishers = facts.map(() => prepareWorkerStartCard(runtime, admitted.caller)) + expect(epochReads).toHaveBeenCalledTimes(count) + facts.forEach((fact, index) => publishers[index](fact)) + await eventually(() => expect(append).toHaveBeenCalledTimes(count)) + await Promise.all(append.mock.results.map((result) => result.value)) + expect(epochReads).toHaveBeenCalledTimes(3 * count) + expect(serialized).toHaveBeenCalledTimes(count) + expect(inserts).toHaveBeenCalledTimes(count) + expect(snapshotSources).toEqual(['status-projection']) + // Duplicate callbacks still name the sender but must stop before row serialization/insertion. + facts.forEach((fact, index) => publishers[index](fact)) + await eventually(() => expect(append).toHaveBeenCalledTimes(2 * count)) + await Promise.all(append.mock.results.map((result) => result.value)) + expect(epochReads).toHaveBeenCalledTimes(5 * count) + expect(serialized).toHaveBeenCalledTimes(count) + expect(inserts).toHaveBeenCalledTimes(count) + expect(sender).toHaveBeenCalledTimes(2 * count) + expect(names).toHaveBeenCalledTimes(2 * count) + expect(activeSource).toHaveBeenCalledTimes(2 * count) + expect(recordedSource).not.toHaveBeenCalled() + expect(sourceQueries).toHaveBeenCalledTimes(4 * count) + expect(dispatchVisits).toBe(2 * count) + expect(taskVisits).toBe(2 * count) + const visitsPerPass = facts.reduce((sum, _fact, index) => sum + 9 * count + index + 1, 0) + expect(tabVisits).toBe(2 * visitsPerPass) + expect(tabVisits).toBeLessThanOrEqual(2 * count * tabs.length) + expect(dedupe).toHaveBeenCalledTimes(2 * count) + expect(scan).not.toHaveBeenCalled() + expect(snapshot).toHaveBeenCalledTimes(1) + expect(snapshotSources).toEqual(['status-projection']) + expect(journalQueries.mock.calls.filter(([query]) => /FROM journal_rows/.test(query))).toEqual( + [] + ) + expect(warn).not.toHaveBeenCalled() + for (const [query] of sourceQueries.mock.calls.slice(0, 2)) { + const plan = prepareSource(`EXPLAIN QUERY PLAN ${query}`) + .all('unused') + .map((row) => row.detail) + .join(' | ') + expect(plan).toMatch(/SEARCH (?:dispatch_contexts|tasks)/) + expect(plan).not.toMatch(/SCAN/) + } + const cards = journal.snapshot().items.filter(({ body }) => body.kind === 'agent-start') + expect(cards.map(({ body }) => body)).toEqual( + facts.map((fact) => ({ + kind: 'agent-start', + dispatchId: fact.dispatchId, + task: fact.task, + worker: { + party: { + address: fact.workerAddress, + terminalHandle: fact.workerAddress, + orcaSessionId: null + }, + name: fact.name + } + })) + ) + expect( + connection + .prepare('SELECT COUNT(*) AS count FROM journal_rows WHERE session_id = ?') + .get(COORD)?.count + ).toBe(before.cursor.sequence + count) + console.info('start population/operations', { + count, + sourceRecords: 10 * count, + history: 9 * count, + epochReads: epochReads.mock.calls.length, + serializedRows: serialized.mock.calls.length, + inserts: inserts.mock.calls.length, + dispatchVisits, + taskVisits, + tabVisits + }) + } +) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start-dispatch.ts b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start-dispatch.ts new file mode 100644 index 00000000000..cb6959ccd8b --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start-dispatch.ts @@ -0,0 +1,81 @@ +import type { TuiAgent } from '../../../../../../shared/tui-agent' +import type { WorkerStartModeReceipt } from '../../orchestration-worker-start-mode' +import type { OrchestrationWorkerLaunchReceipt } from './worker-launch-preferences' +import type { WorkerStartInput } from './worker-start-schema' +import type { startLocalWorker } from './local-worker-start' +import { resolveDispatchCreator } from '../runs/dispatch-creator' +import { parseTaskDeps } from './task-deps-argument' + +export function createLocalWorkerStartDispatch( + { + runtime, + db, + run, + coordinator, + callerSession, + existingTask, + orchestrationMutation, + helperCreator + }: Parameters[0], + { + params, + requestedWorktree, + createsWorktree, + mode, + resolvedWorktreeId, + repoSelector, + agent, + launchReceipt + }: { + params: WorkerStartInput + requestedWorktree: string + createsWorktree: boolean + mode: WorkerStartModeReceipt + resolvedWorktreeId: string | undefined + repoSelector: string | undefined + agent: TuiAgent | undefined + launchReceipt: OrchestrationWorkerLaunchReceipt + } +) { + const startOptions = { + worktree: requestedWorktree, + mode, + resolvedWorktreeId: resolvedWorktreeId ?? null, + name: params.name ?? null, + repo: repoSelector ?? params.repo ?? null, + baseBranch: params.baseBranch ?? null, + terminal: params.terminal ?? null, + agent: agent ?? null, + launch: launchReceipt, + timeoutMs: params.timeoutMs ?? 60_000, + setup: createsWorktree ? (params.setup ?? 'run') : 'not_applicable', + setupSource: createsWorktree + ? params.setup + ? 'explicit_request' + : 'orchestration_default' + : 'existing_worktree' + } + return db.createStartingWorkerDispatch({ + creator: resolveDispatchCreator(runtime, params.from, callerSession), + maxDepth: runtime.getNestedWorkerMaxDepth(), + requiredCreatorDispatchId: helperCreator?.id, + taskId: existingTask?.id, + taskSpec: params.spec, + taskTitle: params.taskTitle, + taskDeps: parseTaskDeps(params.deps), + // A helper's Task sits under its creator's Task, so the Run's Task tree shows who started it. + taskParentId: params.parent ?? helperCreator?.task_id, + taskRunId: run.id, + // A handle-less session creates root Tasks: Task lineage is recorded by terminal only. + taskCreatedByTerminalHandle: coordinator?.terminalHandle ?? undefined, + taskCreatedByPaneKey: coordinator?.paneKey ?? undefined, + taskCreatedByProcessIncarnation: coordinator?.terminalHandle + ? (runtime.getTerminalProcessIncarnation(coordinator.terminalHandle) ?? undefined) + : undefined, + taskCreatedByRunGeneration: run.consumer_generation, + retryOf: params.retryOf, + startOptions, + runtimeEpoch: runtime.getRuntimeId(), + mutationReceipt: orchestrationMutation + }) +} diff --git a/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts index f52fc2b0128..f77a33119ea 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts @@ -1,3 +1,4 @@ +import { createLocalWorkerStartDispatch } from './local-worker-start-dispatch' import { probeWorkerOpenCodeModelLaunchSupport } from './worker-opencode-model-preflight' import { resolveWorkerConfiguredAgentParams } from './worker-configured-agent-preflight' import { waitForWorkerAgentReady } from '../../../../launched-agent-composer-readiness' @@ -9,7 +10,6 @@ import type { OrchestrationCallerIdentity, OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity' -import { resolveDispatchCreator } from '../runs/dispatch-creator' import { resolveDispatchCallerWorktreeId } from '../../orchestration-caller-workspace' import { resolveWorkerStartModeOnHost, @@ -26,7 +26,6 @@ import { persistWorkerSetupWaitOutcome } from './worker-setup-gate' import { failWorkerStartWithReceipt } from './worker-start-receipt' -import { parseTaskDeps } from './task-deps-argument' import { assertExplicitWorkerUsable } from './explicit-worker-terminal-validation' import { resolveDispatchAssigneeParty } from '../../../../orchestration/orchestration-party' import { CHAT_WORKER_AUTHORITY, chatAssigneeOf, chatWorkerMode } from '../chat-assignee-admission' @@ -36,6 +35,7 @@ import { requireWorkerAuthority, type WorkerEffect } from './worker-topology' import { prepareLocalWorkerStart } from './worker-start-validation' import { deliverAndSettleWorkerStartReadiness } from './worker-start-readiness-settlement' import type { WorkerStartMutation } from './worker-start-mutation' +import { prepareWorkerStartCard } from '../../../../orchestration/worker-start-card' export async function startLocalWorker(args: { params: WorkerStartInput @@ -51,7 +51,8 @@ export async function startLocalWorker(args: { /** Set when a worker starts its own helper; the helper reports to this Dispatch. */ helperCreator?: DispatchContextRow }): Promise { - const { runtime, db, run, coordinator, callerSession, existingTask, orchestrationMutation } = args + const { runtime, db, run, coordinator, callerSession, orchestrationMutation } = args + const recordStart = prepareWorkerStartCard(runtime, callerSession) // `--terminal` names a party: a chat by its `/clear` root address, as its Dispatch does. const terminal = args.params.terminal && resolveDispatchAssigneeParty(args.params.terminal, db) const params = terminal ? { ...args.params, terminal: terminal.address } : args.params @@ -132,46 +133,15 @@ export async function startLocalWorker(args: { const hostMode = resolveWorkerStartModeOnHost(runtime, args.mode, resolvedWorktree?.id, agent) let mode = chatWorkerMode(await hostMode, chat) - const startOptions = { - worktree: requestedWorktree, + const started = createLocalWorkerStartDispatch(args, { + params, + requestedWorktree, + createsWorktree, mode, - resolvedWorktreeId: resolvedWorktree?.id ?? null, - name: params.name ?? null, - repo: creationTarget?.repoSelector ?? params.repo ?? null, - baseBranch: params.baseBranch ?? null, - terminal: params.terminal ?? null, - agent: agent ?? null, - launch: launch.receipt, - timeoutMs: params.timeoutMs ?? 60_000, - setup: createsWorktree ? (params.setup ?? 'run') : 'not_applicable', - setupSource: createsWorktree - ? params.setup - ? 'explicit_request' - : 'orchestration_default' - : 'existing_worktree' - } - const started = db.createStartingWorkerDispatch({ - creator: resolveDispatchCreator(runtime, params.from, callerSession), - maxDepth: runtime.getNestedWorkerMaxDepth(), - requiredCreatorDispatchId: args.helperCreator?.id, - taskId: existingTask?.id, - taskSpec: params.spec, - taskTitle: params.taskTitle, - taskDeps: parseTaskDeps(params.deps), - // A helper's Task sits under its creator's Task, so the Run's Task tree shows who started it. - taskParentId: params.parent ?? args.helperCreator?.task_id, - taskRunId: run.id, - // A handle-less session creates root Tasks: Task lineage is recorded by terminal only. - taskCreatedByTerminalHandle: coordinator?.terminalHandle ?? undefined, - taskCreatedByPaneKey: coordinator?.paneKey ?? undefined, - taskCreatedByProcessIncarnation: coordinator?.terminalHandle - ? (runtime.getTerminalProcessIncarnation(coordinator.terminalHandle) ?? undefined) - : undefined, - taskCreatedByRunGeneration: run.consumer_generation, - retryOf: params.retryOf, - startOptions, - runtimeEpoch: runtime.getRuntimeId(), - mutationReceipt: orchestrationMutation + resolvedWorktreeId: resolvedWorktree?.id, + repoSelector: creationTarget?.repoSelector, + agent, + launchReceipt: launch.receipt }) releaseRetriedWorkerTerminal(runtime, db, params) const effects: WorkerEffect[] = [] @@ -185,6 +155,7 @@ export async function startLocalWorker(args: { let terminalHandle = params.terminal let placed: Awaited> | undefined let failedStage = 'terminal_create' + let result: unknown try { placed = await placeWorkerAgent({ runtime, @@ -271,7 +242,7 @@ export async function startLocalWorker(args: { terminalOwnership: params.terminal ? 'external' : 'created' }) - return await deliverAndSettleWorkerStartReadiness({ + result = await deliverAndSettleWorkerStartReadiness({ runtime, db, run, @@ -297,7 +268,7 @@ export async function startLocalWorker(args: { } }) } catch (error) { - return await failWorkerStartWithReceipt({ + result = await failWorkerStartWithReceipt({ runtime, structuredSession: placed?.structuredSession ?? null, db, @@ -311,4 +282,19 @@ export async function startLocalWorker(args: { mode }) } + if ( + !params.terminal && + terminalHandle && + result && + typeof result === 'object' && + 'state' in result && + result.state === 'ready' + ) { + recordStart({ + dispatchId: started.dispatch.id, + workerAddress: terminalHandle, + task: task.spec + }) + } + return result } diff --git a/src/main/runtime/rpc/methods/structured-worker-read-cursor.test.ts b/src/main/runtime/rpc/methods/structured-worker-read-cursor.test.ts index a6c7f00cf59..cda7a72b117 100644 --- a/src/main/runtime/rpc/methods/structured-worker-read-cursor.test.ts +++ b/src/main/runtime/rpc/methods/structured-worker-read-cursor.test.ts @@ -6,6 +6,9 @@ import { agentSessionConversationView } from '../../../../shared/agent-session-c /** Mutate journals between reads to catch in-place revisions and mid-array insertions missed by oldest-item-only cursor identity; both can silently omit or duplicate output. */ import { beforeEach, describe, expect, it, vi } from 'vitest' +import { buildStructuredJournalArchive } from '../../orchestration/structured-worker-journal-archive' +import { projectSavedChatContextItems } from '../../../native-chat/agent-session-saved-prompt-text' +import { formatWorkerTranscriptMessage } from '../../../../shared/worker-transcript-text' import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' import type { StructuredWorkerIdentity } from '../../structured-worker-identity' @@ -91,6 +94,53 @@ describe('the structured worker-read cursor over a mutating journal', () => { hostRef.current = null }) + it('omits creation facts from worker output, consumed prefixes, saved context and archives', async () => { + const start: AgentJournalRenderItem = { + itemId: 'i2', + revision: 1, + sequence: 2, + observedAt: 1, + body: { + kind: 'agent-start', + dispatchId: 'dispatch', + task: 'Private worker assignment', + worker: { + party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, + name: 'Worker' + } + } + } + const items = [message('i1', 'first'), start, message('i3', 'answer')] + installJournal(items) + const first = await read(undefined, 1) + expect(textsOf(first)).toEqual(['first']) + installJournal([ + items[0], + { + ...start, + revision: 2, + body: + start.body.kind === 'agent-start' + ? { ...start.body, task: 'Revised display fact' } + : start.body + }, + items[2] + ]) + expect(textsOf(await read(first.cursor))).toEqual(['answer']) + const saved = projectSavedChatContextItems(items).map(formatWorkerTranscriptMessage).join('\n') + expect(saved).toBe('[assistant] first\n[assistant] answer') + const archive = buildStructuredJournalArchive({ + agent: 'claude', + processIncarnation: 'fixture', + items, + hasOlder: false + }) + expect(archive.messages.map(formatWorkerTranscriptMessage)).toEqual([ + '[assistant] first', + '[assistant] answer' + ]) + }) + it('refuses to resume when an already-delivered item was revised in place', async () => { // Revise an already-read item in place; old oldest-item identity silently skipped the revised text. installJournal([message('i1', 'hel'), message('i2', 'second')]) diff --git a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.fixture.tsx b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.fixture.tsx index 84338267128..0ce36d6f9ea 100644 --- a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.fixture.tsx +++ b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.fixture.tsx @@ -4,6 +4,8 @@ import { defineFixture } from '../../component-fixtures/component-fixture' import { makeTab } from '@/store/slices/store-test-helpers' import type { AgentMessageSource } from '../../../../shared/agent-session/agent-session-message-source' import { NativeChatAgentMessageCard } from './NativeChatAgentMessageCard' +import { MessageRow } from './NativeChatMessageRow' +import { projectStructuredAgentStartItem } from '../../../../shared/structured-agent-start-projection' import { NativeChatUserMessageRow } from './NativeChatUserMessageRow' const workspace = 'agent-message-fixture' @@ -11,6 +13,21 @@ const image = 'data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAPAAAAA8CAIAAADXHaAKAAAAvklEQVR4nO3SUQkAIBTAwBfUAAawP5YQhHFwAfaxWftAxnwvgIcMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSbm/mJH95bvMnAAAAABJRU5ErkJggg==' const prose = 'First paragraph with **real Markdown**, a [web link](https://example.com) and inline `code`.\n\nSecond paragraph keeps its own spacing. This longer line wraps differently in the narrow frame, so Show more is based on rendered lines.\n\n![Example image](data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAPAAAAA8CAIAAADXHaAKAAAAvklEQVR4nO3SUQkAIBTAwBfUAAawP5YQhHFwAfaxWftAxnwvgIcMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSTE0KYYmxdCkGJoUQ5NiaFIMTYqhSbm/mJH95bvMnAAAAABJRU5ErkJggg==)\n\nLast paragraph after the image.' +const started = projectStructuredAgentStartItem({ + itemId: 'started', + revision: 1, + sequence: 7, + observedAt: 1_800_000_000_000, + body: { + kind: 'agent-start', + dispatchId: 'fixture-dispatch', + task: prose, + worker: { + party: { address: 'term_sender', terminalHandle: 'term_sender', orcaSessionId: null }, + name: 'Parser worker' + } + } +})! function cards() { const human: NativeChatMessage = { id: 'human', @@ -21,6 +38,12 @@ function cards() { } return (
+ {}} + runtimeContext={{ worktreeId: workspace, worktreePath: null, target: { kind: 'local' } }} + /> {(['task', 'coordinator', 'worker', 'generic'] as const).map((kind, i) => { const from: AgentMessageSource = { kind: 'agent', diff --git a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.test.tsx b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.test.tsx index f4831cd700e..4f18542f634 100644 --- a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.test.tsx @@ -1,33 +1,199 @@ // @vitest-environment happy-dom import '@testing-library/jest-dom/vitest' -import { act, cleanup, fireEvent, renderHook, screen } from '@testing-library/react' +import { act, cleanup, fireEvent, screen } from '@testing-library/react' import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { render } from './native-chat-tooltip-test-harness' import { useAppStore } from '@/store' -import { makeTab, makeWorktree, TEST_REPO } from '@/store/slices/store-test-helpers' -import { readStoreListenerCount } from '@/store/store-listener-census' -import { getKnownExecutionHostIdForWorktree } from '@/lib/worktree-runtime-owner' -import { resolveStructuredAgentSessionOwner } from '@/runtime/structured-agent-session-owner' -import type { TerminalLayoutSnapshot } from '../../../../shared/terminal-tab-types' -import type { AgentStatusEntry } from '../../../../shared/agent-status-types' -import type { AgentMessageSenderView } from '../../../../shared/agent-session/agent-message-presentation' -import { runtimeEnvironmentStatusFromSnapshot } from '../../../../shared/runtime-host-status' -import * as presentation from '@/lib/program-status/agent-row-presentation' -import * as cache from '@/components/dashboard/worktree-agent-rows-cache' -import * as selectors from '@/components/sidebar/worktree-agent-row-selectors' -import { useAgentMessageSenderView } from './native-chat-agent-message-sender-view' -import { useAgentMessageSenderSource } from './native-chat-agent-message-sender-source' -import { AgentStateDot } from '@/components/AgentStateDot' +import { + initialStore, + from, + status, + resetSenderStore +} from './native-chat-agent-sender.test-fixture' import { AGENT_SESSION_WRITE_NOTICE_COPY } from '../../../../shared/agent-session-write-notice-copy' +import * as terminalLinks from '@/components/terminal-pane/terminal-handle-links' import * as markdown from './NativeChatMarkdown' import * as display from '../../../../shared/agent-session/agent-message-display-text' import type * as SonnerModule from 'sonner' import type * as RuntimeRpc from '@/runtime/runtime-rpc-client' -import type { AgentMessageSource } from '../../../../shared/agent-session/agent-session-message-source' import type { NativeChatMessage } from '../../../../shared/native-chat-types' +import { createNativeChatMessageListProjection } from './native-chat-message-list-projection' import { buildNativeChatTranscriptSlots } from './native-chat-transcript-slots' import { nativeChatSlotCopyText } from './native-chat-message-copy-text' import { NativeChatAgentMessageCard } from './NativeChatAgentMessageCard' +import { MessageRow } from './NativeChatMessageRow' +import { projectStructuredAgentStartItem } from '../../../../shared/structured-agent-start-projection' +import { + createStructuredQuestionProjection, + projectStructuredQuestionMessages +} from '../../../../shared/structured-agent-question-projection' +import { desktopStructuredTranscript } from './structured-agent-session-message-projection' +import { + EMPTY_STRUCTURED_AGENT_SESSION, + reduceStructuredAgentSession +} from '../../../../shared/structured-agent-session-reducer' +import type { AgentSessionHistoryPage } from '../../../../shared/agent-session-wire' +import { projectStructuredItemToNativeChat } from '../../../../shared/structured-agent-session-projection' +import { isAdmissibleAgentJournalRenderItem } from '../../../../shared/agent-session-journal-schemas' +import { agentMessageHeading } from '../../../../shared/agent-session/agent-message-presentation' +import { nativeChatTurnFoldRow } from '../../../../shared/native-chat/transcript/native-chat-turn-fold' +import { isNoiseMessage } from '../../../../shared/native-chat-noise' +import { nativeChatRowSpeaksOrActs } from '../../../../shared/native-chat/native-chat-trailing-run' +import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' + +it('reloads a positioned worker start into the shared card with full task, Copy and navigation', async () => { + const task = ` Task with [a link][target].\n\n${'More task text.\n'.repeat(20)}END \n\n[target]: https://example.com\n` + const item: AgentJournalRenderItem = { + itemId: 'start:dispatch', + revision: 1, + sequence: 7, + observedAt: 1, + body: { kind: 'agent-start', dispatchId: 'dispatch', worker: from.senders[0], task } + } + const reloaded: unknown = JSON.parse(JSON.stringify(item)) + if (!isAdmissibleAgentJournalRenderItem(reloaded)) { + throw new Error('start item did not survive admission') + } + const message = projectStructuredAgentStartItem(reloaded) + if (!message?.from) { + throw new Error('start item did not project') + } + expect(message).toMatchObject({ + role: 'system', + agentStart: true, + journalPosition: { sequence: 7, index: 0 }, + timestamp: 1 + }) + expect(agentMessageHeading(message.from, message.agentStart)).toBe('started') + expect(nativeChatTurnFoldRow(message, 'turn', false).outlivesTurn).toBe(true) + expect(isNoiseMessage(message)).toBe(false) + expect(nativeChatRowSpeaksOrActs(message, true, new Map())).toBe(false) + const future: unknown = { ...item, body: { kind: 'future-start' } } + if (!isAdmissibleAgentJournalRenderItem(future)) { + throw new Error('unknown journal kind was refused') + } + expect(projectStructuredItemToNativeChat(future)).toBeNull() + expect(projectStructuredAgentStartItem({ ...item, shortened: true })).toBeNull() + expect(projectStructuredItemToNativeChat(reloaded)).toBeNull() + height = 120 + const writeClipboardText = vi.fn(async () => {}) + vi.stubGlobal('api', { ui: { writeClipboardText } }) + render( + {}} + runtimeContext={{ worktreeId: 'wt-chat', worktreePath: null, target: { kind: 'local' } }} + /> + ) + expect(screen.getByText('Started worker')).toBeInTheDocument() + expect(screen.getByRole('link', { name: 'a link' })).toHaveAttribute( + 'href', + 'https://example.com' + ) + expect(screen.getByRole('button', { name: 'Show more' })).toHaveAttribute( + 'aria-expanded', + 'false' + ) + fireEvent.click(screen.getByRole('button', { name: 'Show more' })) + expect(screen.getByRole('button', { name: 'Show less' })).toHaveAttribute('aria-expanded', 'true') + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Copy message' }))) + expect(writeClipboardText).toHaveBeenCalledWith(task) + expect(nativeChatSlotCopyText(slots(message), message.id)).toBe(task) + fireEvent.click(screen.getByRole('button', { name: 'Show less' })) + expect(screen.getByRole('button', { name: 'Show more' })).toHaveAttribute( + 'aria-expanded', + 'false' + ) + navigation.rpc.mockResolvedValue({ location: { kind: 'terminal', handle: 'term_0' } }) + const focus = vi.spyOn(terminalLinks, 'focusRendererTerminalHandle').mockReturnValue(true) + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Build cards' }))) + expect(navigation.rpc).toHaveBeenCalledWith({ kind: 'local' }, 'orchestration.partyLocation', { + address: 'term_0' + }) + expect(focus).toHaveBeenCalledWith('term_0', null) +}) + +it.each(['', ' ', 'Internal noise'])( + 'retains a creation fact through real tool/start/answer slots for %j', + (task) => { + const message = projectStructuredAgentStartItem({ + itemId: 'fact', + revision: 1, + sequence: 2, + observedAt: 1, + body: { kind: 'agent-start', dispatchId: 'd', worker: from.senders[0], task } + })! + for (const working of [true, false]) { + const positioned = slots(message, true, working) + const fact = positioned.find( + (slot) => slot.kind === 'message' && slot.message.id === message.id + ) + expect(fact).toMatchObject({ kind: 'message', folded: false, drawsMessage: true }) + expect(nativeChatSlotCopyText(positioned, message.id)).toBe(task) + expect( + positioned.some((slot) => slot.kind === 'message' && slot.message.id === 'answer') + ).toBe(true) + } + render( {}} />) + expect(screen.getByText('Started worker')).toBeInTheDocument() + } +) + +it('renders unreadable worker metadata with the existing unnamed fallback and no navigation', () => { + const raw: unknown = { + itemId: 'future-worker', + revision: 1, + sequence: 1, + observedAt: 1, + body: { + kind: 'agent-start', + dispatchId: 'd', + task: 'Keep the complete task', + worker: { futureIdentity: 'new' } + } + } + if (!isAdmissibleAgentJournalRenderItem(raw)) { + throw new Error('open metadata was refused') + } + const message = projectStructuredAgentStartItem(raw)! + render( + {}} + runtimeContext={context} + /> + ) + expect(screen.getByText('Started worker')).toBeInTheDocument() + expect(screen.getByText('an agent')).toBeInTheDocument() + expect(screen.getByText('Keep the complete task')).toBeInTheDocument() + expect(screen.queryByRole('button', { name: 'an agent' })).toBeNull() +}) + +it.each([false, true])( + 'isolates canonical and desktop policy caches, desktop first=%s', + (desktopFirst) => { + const item: AgentJournalRenderItem = { + itemId: 'policy-start', + revision: 1, + sequence: 1, + observedAt: 1, + body: { kind: 'agent-start', dispatchId: 'd', worker: from.senders[0], task: 'Full task' } + } + const items = [item] + if (desktopFirst) { + desktopStructuredTranscript.messages(items) + } else { + projectStructuredQuestionMessages(items) + } + expect(projectStructuredQuestionMessages(items)).toEqual([]) + const messages = desktopStructuredTranscript.messages(items) + expect(messages).toHaveLength(1) + expect(desktopStructuredTranscript.messages(items)).toBe(messages) + expect(desktopStructuredTranscript.messages([...items])[0]).toBe(messages[0]) + } +) const navigation = vi.hoisted(() => ({ rpc: vi.fn(), toast: vi.fn() })) vi.mock('@/runtime/runtime-rpc-client', async (original) => ({ @@ -47,85 +213,13 @@ vi.mock('sonner', async (importOriginal) => { ) } }) -const initialStore = useAppStore.getState() -const LEAF = '11111111-1111-4111-8111-111111111111' -const renders = new Map() -const views = new Map() -function status(i: number, worktreeId = 'workspace'): AgentStatusEntry { - return { - paneKey: `tab-${i}:${LEAF}`, - worktreeId, - terminalHandle: `term_${i}`, - agentType: 'claude', - state: 'working', - prompt: 'work', - updatedAt: 2_000_000, - stateStartedAt: 2_000_000, - stateHistory: [] - } -} -function Probe({ - id, - workspace = 'workspace', - remote = false -}: { - id: number - workspace?: string - remote?: boolean -}) { - const value = useAgentMessageSenderView( - { - party: { address: `term_${id}`, terminalHandle: `term_${id}`, orcaSessionId: null }, - name: 'Sender' - }, - workspace, - remote ? { kind: 'environment', environmentId: 'remote' } : { kind: 'local' } - ) - views.set(id, value) - renders.set(id, (renders.get(id) ?? 0) + 1) - return null -} -function write(i: number, update: Partial) { - const state = useAppStore.getState() - const key = status(i).paneKey - act(() => - useAppStore.setState({ - agentStatusByPaneKey: { - ...state.agentStatusByPaneKey, - [key]: { ...state.agentStatusByPaneKey[key], ...update } - }, - agentStatusEpoch: state.agentStatusEpoch + 1 - }) - ) -} - let height = 48 let resize: () => void let disconnects = 0 beforeEach(() => { vi.useFakeTimers({ toFake: ['Date'] }) vi.setSystemTime(2_000_000) - renders.clear() - views.clear() - useAppStore.setState( - { - ...initialStore, - activeWorktreeId: 'workspace', - activeWorkspaceExecutionHostId: 'local', - tabsByWorktree: { workspace: [makeTab({ id: 'tab-0', worktreeId: 'workspace' })] }, - unifiedTabsByWorktree: {}, - agentStatusByPaneKey: { [status(0).paneKey]: status(0) }, - retainedAgentsByPaneKey: {}, - migrationUnsupportedByPtyId: {}, - runtimeAgentOrchestrationByPaneKey: {}, - runtimePaneTitlesByTabId: {}, - terminalLayoutsByTabId: {}, - ptyIdsByTabId: {}, - paneForegroundAgentByPaneKey: {}, - agentStatusEpoch: 1 - }, - true - ) + resetSenderStore() navigation.rpc.mockReset() navigation.toast.mockReset() height = 48 @@ -168,14 +262,6 @@ afterEach(() => { vi.useRealTimers() }) -const from: AgentMessageSource = { - kind: 'agent', - orchestration: null, - senders: ['Build cards', 'Other sender'].map((name, i) => ({ - name, - party: { address: `term_${i}`, terminalHandle: `term_${i}`, orcaSessionId: null } - })) -} const context = { worktreeId: 'workspace', worktreePath: null, target: { kind: 'local' as const } } function card(message: NativeChatMessage) { return ( @@ -188,17 +274,43 @@ function card(message: NativeChatMessage) { /> ) } -function slots(message: NativeChatMessage) { +function slots(message: NativeChatMessage, surrounding = false, working = false) { + const input: NativeChatMessage[] = surrounding + ? [ + { + id: 'tool', + role: 'assistant', + source: 'transcript', + timestamp: 0, + journalPosition: { sequence: 1, index: 0 }, + blocks: [{ type: 'tool-call', name: 'shell', input: { command: 'inspect' } }] + }, + message, + { + id: 'answer', + role: 'assistant', + source: 'transcript', + timestamp: 2, + journalPosition: { sequence: 3, index: 0 }, + blocks: [{ type: 'text', text: 'Final answer' }] + } + ] + : [message] + const messages = createNativeChatMessageListProjection()(input).conversation return buildNativeChatTranscriptSlots({ - messages: [message], - turnKeys: [undefined], - liveTurnKey: undefined, + messages, + turnKeys: messages.map(() => (surrounding ? 'turn' : undefined)), + liveTurnKey: working ? 'turn' : undefined, receipts: new Map(), - turnStatuses: { active: null, completedByTurn: {} }, + turnStatuses: { + active: working ? { startedAt: 1, thinking: false, workedSeconds: null } : null, + completedByTurn: + surrounding && !working ? { turn: { startedAt: 1, thinking: false, workedSeconds: 1 } } : {} + }, turnDiffs: new Map(), expandedTurnKeys: new Set(), - isWorking: false, - lifecycleWorking: false + isWorking: working, + lifecycleWorking: working }) } @@ -405,232 +517,183 @@ it.each(['task', 'mail'] as const)( } ) -it.each([100, 1000])( - 'shares actual row acquisition and isolates selected progress across %i cards', - (n) => { - let visits = 0 - const live = new Proxy( - Object.fromEntries( - Array.from({ length: n }, (_, i) => { - const entry = status(i) - return [entry.paneKey, entry] - }) - ), - { - ownKeys(target) { - const keys = Object.getOwnPropertyNames(target) - visits += keys.length - return keys +it.each([10, 100])( + 'bounds real publication work and settled body renders with %i starts', + async (count) => { + let itemVisits = 0 + let taskReads = 0 + let listVisits = 0 + let conversions = 0 + const projection = createStructuredQuestionProjection((item) => { + conversions++ + return projectStructuredAgentStartItem(item) + }) + const ordinary: AgentJournalRenderItem[] = Array.from({ length: 9 * count }, (_, index) => ({ + itemId: `ordinary-${index}`, + revision: 1, + sequence: index + 1, + observedAt: 1, + body: { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: `History ${index}` }] + } + })) + const starts: AgentJournalRenderItem[] = Array.from({ length: count }, (_, index) => { + const body = { + kind: 'agent-start' as const, + dispatchId: `d-${index}`, + worker: from.senders[0], + get task() { + taskReads++ + return `Task ${index}` + } + } + return { + itemId: `start-${index}`, + revision: 1, + sequence: ordinary.length + index + 1, + observedAt: 1, + get body() { + itemVisits++ + return body } } - ) - useAppStore.setState({ - agentStatusByPaneKey: live, - tabsByWorktree: { - workspace: Array.from({ length: n }, (_, i) => - makeTab({ id: `tab-${i}`, worktreeId: 'workspace' }) - ) - }, - terminalLayoutsByTabId: Object.fromEntries( - Array.from({ length: n }, (_, i): [string, TerminalLayoutSnapshot] => [ - `tab-${i}`, - { root: { type: 'leaf', leafId: LEAF }, activeLeafId: LEAF, expandedLeafId: null } - ]) - ) }) - const normalize = vi.spyOn(presentation, 'resolveRowProgramStatus') - const mounted = render( + const items = [...ordinary, ...starts] + const page: AgentSessionHistoryPage = { + sessionId: 'scale', + epoch: 'scale-epoch', + direction: 'tail', + items, + removedItemIds: [], + submissions: [], + window: { + oldest: { epoch: 'scale-epoch', sequence: 1 }, + newest: { epoch: 'scale-epoch', sequence: items.length }, + nextCursor: { epoch: 'scale-epoch', sequence: 1 } + }, + liveCursor: { epoch: 'scale-epoch', sequence: items.length }, + hasOlder: false, + hasNewer: false + } + let state = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, { + type: 'history-page', + page + }) + const readRows = () => { + const counted = state.items.slice() + counted.forEach((item, index) => { + Object.defineProperty(counted, index, { + get() { + listVisits++ + return item + } + }) + }) + return projection.messages(counted).filter(({ agentStart }) => agentStart) + } + let messages = readRows() + expect(conversions).toBe(items.length) + expect(taskReads).toBe(count) + // One existing question pass visits the source array once; the second pass visits its composed rows. + expect(listVisits).toBe(items.length) + expect(itemVisits).toBeLessThan(100 * items.length) + const markdownRender = vi.spyOn(markdown, 'NativeChatMarkdown') + const scroll = () => {} + const rows = () => ( <> - {Array.from({ length: n }, (_, id) => ( - + {messages.map((message) => ( + ))} ) - // Handle lookup, visible-authority acquisition and sidebar indexing each enumerate the N live entries. - expect(visits).toBe(n * 3) - expect(normalize).toHaveBeenCalledTimes(n) - for (let i = 0; i < n; i++) { - expect(views.get(i)?.dot).toBe('working') - } - const before = new Map(renders) - visits = 0 - normalize.mockClear() - act(() => useAppStore.setState({ sidebarBody: 'agents' })) - expect(visits).toBe(0) - expect(normalize).not.toHaveBeenCalled() - expect(renders).toEqual(before) - for (const progress of [40, 60]) { - normalize.mockClear() - const epoch = useAppStore.getState().agentStatusEpoch - act(() => - useAppStore.getState().setAgentProgramStatus(status(0).paneKey, { - state: 'working', - progress, - observedAt: 2_000_000 - }) - ) - expect(useAppStore.getState().agentStatusEpoch).toBe(epoch) - expect(normalize).toHaveBeenCalledTimes(n) - expect(views.get(0)?.progress).toBe(progress) - expect(renders.get(0)).toBe(before.get(0)! + (progress === 40 ? 1 : 2)) - for (let i = 1; i < n; i++) { - expect(renders.get(i)).toBe(before.get(i)) + const mounted = render(rows()) + await act(async () => {}) + markdownRender.mockClear() + navigation.rpc.mockClear() + itemVisits = taskReads = listVisits = conversions = 0 + // Real history publication with identical immutable rows must reuse their body construction. + state = reduceStructuredAgentSession(state, { + type: 'history-page', + page: { ...page, items: [...items] } + }) + messages = readRows() + await act(async () => mounted.rerender(rows())) + expect(conversions).toBe(0) + expect(taskReads).toBe(0) + expect(listVisits).toBe(items.length) + expect(itemVisits).toBeLessThan(100 * items.length) + expect(markdownRender).not.toHaveBeenCalled() + const unrelated: AgentJournalRenderItem = { + itemId: 'new-answer', + revision: 1, + sequence: items.length + 1, + observedAt: 2, + body: { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: 'Another row streams' }] } } - mounted.unmount() + state = reduceStructuredAgentSession(state, { + type: 'event', + event: { + type: 'batch', + sessionId: 'scale', + batch: { + cursor: { epoch: 'scale-epoch', sequence: items.length + 1 }, + items: [unrelated], + removedItemIds: [], + submissions: [] + } + } + }) + messages = readRows() + await act(async () => mounted.rerender(rows())) + expect(conversions).toBe(1) + expect(taskReads).toBe(0) + expect(markdownRender).not.toHaveBeenCalled() + conversions = 0 + // A relevant live revision updates its own body in a distinct act turn. + const revised: AgentJournalRenderItem = { + ...starts[0], + revision: 2, + body: { + kind: 'agent-start', + dispatchId: 'd-0', + worker: from.senders[0], + task: 'Changed task' + } + } + state = reduceStructuredAgentSession(state, { + type: 'event', + event: { + type: 'batch', + sessionId: 'scale', + batch: { + cursor: { epoch: 'scale-epoch', sequence: items.length + 2 }, + items: [revised], + removedItemIds: [], + submissions: [] + } + } + }) + messages = readRows() + await act(async () => mounted.rerender(rows())) + expect(conversions).toBe(1) + expect(markdownRender).toHaveBeenCalledTimes(1) + expect(screen.getByText('Changed task')).toBeInTheDocument() + expect( + navigation.rpc.mock.calls.filter(([, method]) => + /agentSession\.item|capabilit/i.test(String(method)) + ) + ).toEqual([]) } ) - -it('keeps other hosts isolated, retains completed senders, and releases disconnected subscriptions', () => { - const create = vi.spyOn(cache, 'createWorktreeAgentRowsCache') - const baseline = readStoreListenerCount() - expect(baseline).not.toBeNull() - const other = status(9, 'remote') - useAppStore.setState({ - repos: [TEST_REPO], - worktreesByRepo: { - [TEST_REPO.id]: [ - makeWorktree({ id: 'remote', repoId: TEST_REPO.id, hostId: 'runtime:remote' }) - ] - }, - tabsByWorktree: { - ...useAppStore.getState().tabsByWorktree, - remote: [makeTab({ id: 'tab-9', worktreeId: 'remote' })] - }, - agentStatusByPaneKey: { ...useAppStore.getState().agentStatusByPaneKey, [other.paneKey]: other } - }) - const mounted = render( - <> - - - - ) - expect(readStoreListenerCount()).toBeGreaterThan(baseline!) - const leased = create.mock.results.flatMap((result) => - result.type === 'return' && result.value.byWorktree.size ? [result.value] : [] - ) - expect(leased.flatMap((value) => [...value.byWorktree.keys()]).sort()).toEqual([ - 'remote', - 'workspace' - ]) - const before = new Map(renders) - write(9, { state: 'waiting' }) - expect(views.get(9)?.dot).toBe('waiting') - expect(renders.get(9)).toBe(before.get(9)! + 1) - expect(renders.get(0)).toBe(before.get(0)) - write(0, { - agentType: 'codex', - programStatus: { state: 'blocked', progress: 60, observedAt: 2_000_000 } - }) - expect(views.get(0)).toEqual({ agentType: 'codex', dot: 'permission', progress: 60 }) - expect(renders.get(9)).toBe(before.get(9)! + 1) - for (const value of leased) { - expect(value.lastComputedWorktreeIds.length).toBeLessThanOrEqual(1) - } - const entry = { ...status(0), state: 'done' as const } - act(() => - useAppStore.setState({ - agentStatusByPaneKey: { [other.paneKey]: other }, - retainedAgentsByPaneKey: { - [entry.paneKey]: { - entry, - tab: makeTab({ id: 'tab-0', worktreeId: 'workspace' }), - worktreeId: 'workspace', - agentType: 'claude', - startedAt: 1 - } - }, - runtimeStatusByEnvironmentId: new Map([ - [ - 'remote', - runtimeEnvironmentStatusFromSnapshot({ - environmentId: 'remote', - pairingRevision: 1, - sequence: 1, - checkedAt: 1, - verification: 'blocked', - transport: 'disconnected', - retired: true, - status: null - }) - ] - ]) - }) - ) - expect(views.get(0)?.dot).toBe('done') - act(() => useAppStore.setState({ activeWorkspaceExecutionHostId: 'runtime:remote' })) - expect(views.get(0)).toEqual({}) - act(() => useAppStore.setState({ activeWorkspaceExecutionHostId: 'local' })) - expect(views.get(0)?.dot).toBe('done') - const disconnected = new Map(renders) - const constructions = create.mock.calls.length - act(() => useAppStore.setState({ sidebarBody: 'agents' })) - expect(renders).toEqual(disconnected) - expect(create).toHaveBeenCalledTimes(constructions) - mounted.unmount() - for (const value of leased) { - expect(value.byWorktree.size).toBe(0) - } - expect(readStoreListenerCount()).toBe(baseline) - render() - expect(views.get(0)?.dot).toBe('done') -}) - -it('uses the child row own glyph even when its parent program needs permission', () => { - write(0, { - programStatus: { state: 'blocked', progress: 60, observedAt: 2_000_000 }, - subagents: [{ id: 'child', state: 'working', startedAt: 2_000_000, description: 'Child' }] - }) - const state = useAppStore.getState() - const projected = cache.selectWorktreeAgentRowsCached({ - state, - worktreeId: 'workspace', - now: 2_000_000, - generation: '', - orchestration: selectors.selectRuntimeAgentOrchestrationForWorktree(state, 'workspace') - }) - const child = projected.find((row) => row.childRow)! - expect(child).toBeDefined() - const locate = () => ({ host: 'local', worktreeId: 'workspace', paneKey: child.paneKey }) - let value: AgentMessageSenderView | undefined - function Child() { - value = useAgentMessageSenderSource(locate) - return value.dot ? ( - - ) : null - } - render() - expect(value?.dot).toBe('working') - expect(value?.progress).toBeUndefined() - expect(screen.queryByRole('progressbar')).toBeNull() - expect(screen.getByLabelText('Working')).toBeDefined() -}) - -it('refuses a target-less ambiguous workspace twin while an explicit local target resolves', () => { - const workspace = `${TEST_REPO.id}::/w` - const entry = status(0, workspace) - useAppStore.setState({ - repos: [TEST_REPO], - activeWorktreeId: 'different', - activeWorkspaceExecutionHostId: 'local', - detectedWorktreesByRepo: {}, - worktreesByRepo: { - [TEST_REPO.id]: [ - makeWorktree({ id: workspace, repoId: TEST_REPO.id, hostId: 'local' }), - makeWorktree({ id: workspace, repoId: TEST_REPO.id, hostId: 'runtime:e' }) - ] - }, - tabsByWorktree: { [workspace]: [makeTab({ id: 'tab-0', worktreeId: workspace })] }, - agentStatusByPaneKey: { [entry.paneKey]: entry } - }) - const state = useAppStore.getState() - expect(getKnownExecutionHostIdForWorktree(state, workspace)).toBe('local') - expect(resolveStructuredAgentSessionOwner(state, workspace)).toBeNull() - const view = renderHook(() => useAgentMessageSenderView(from.senders[0], workspace)) - expect(view.result.current).toEqual({}) - view.unmount() - const explicit = renderHook(() => - useAgentMessageSenderView(from.senders[0], workspace, { kind: 'local' }) - ) - expect(explicit.result.current.dot).toBe('working') -}) diff --git a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.tsx b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.tsx index 803decb0fed..35176a5142f 100644 --- a/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.tsx +++ b/src/renderer/src/components/native-chat/NativeChatAgentMessageCard.tsx @@ -28,8 +28,10 @@ import { NativeChatCopyButton } from './NativeChatCopyButton' import { NativeChatMessageTimestamp } from './NativeChatMessageTimestamp' import { NativeChatImageAttachments } from './NativeChatTranscriptChrome' -function heading(from: AgentMessageSource): string { - switch (agentMessageHeading(from)) { +function heading(from: AgentMessageSource, agentStart: boolean): string { + switch (agentMessageHeading(from, agentStart)) { + case 'started': + return translate('components.native-chat.agentMessage.startedWorker', 'Started worker') case 'task': return translate( 'components.native-chat.agentMessage.taskFromCoordinator', @@ -93,7 +95,7 @@ export function NativeChatAgentMessageCard({ />
- {heading(from)} + {heading(from, message.agentStart === true)}
{agentMessageCanCopy(text) ? ( // Why: -my-1 keeps the 24px button inside the text-xs header line, so copy costs no row of its own. diff --git a/src/renderer/src/components/native-chat/NativeChatMessageList.tsx b/src/renderer/src/components/native-chat/NativeChatMessageList.tsx index 0a31242b1d5..c84e62279db 100644 --- a/src/renderer/src/components/native-chat/NativeChatMessageList.tsx +++ b/src/renderer/src/components/native-chat/NativeChatMessageList.tsx @@ -11,7 +11,7 @@ import type { CommentMarkdownLinkClickHandler } from '@/components/sidebar/Comme import { translate } from '@/i18n/i18n' import type { NativeChatLiveSession } from './use-native-chat-live-session' import { useNativeChatTranscriptProjection } from './use-native-chat-transcript-projection' -import { structuredQuestionTranscript } from '../../../../shared/structured-agent-question-projection' +import { desktopStructuredTranscript } from './structured-agent-session-message-projection' import { nativeChatTaskListPredecessors } from '../../../../shared/native-chat/input/native-chat-task-list-history' import { NativeChatDockedTaskList } from './NativeChatDockedTaskList' import { useNativeChatTurnStatus } from './use-native-chat-turn-status' @@ -127,7 +127,8 @@ export function NativeChatMessageList({ const revealedDiff = navigationRequest?.kind === 'diff' ? navigationRequest.target : null const railJump = navigationRequest?.kind === 'rail' ? navigationRequest : null const receipts = useMemo( - () => (journalItems ? structuredQuestionTranscript(journalItems).receipts : new Map()), + () => + journalItems ? desktopStructuredTranscript.transcript(journalItems).receipts : new Map(), [journalItems] ) const scrollRef = useRef(null) diff --git a/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx b/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx index 553b735a051..67589047ffc 100644 --- a/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx +++ b/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx @@ -2,6 +2,7 @@ import { memo, useRef } from 'react' import type { NativeChatRewindSurface } from './use-native-chat-rewind' import type { CommentMarkdownLinkClickHandler } from '@/components/sidebar/CommentMarkdown' import { NativeChatUserMessageRow } from './NativeChatUserMessageRow' +import { NativeChatAgentMessageCard } from './NativeChatAgentMessageCard' import { NativeChatAssistantMessageRow } from './NativeChatAssistantMessageRow' import type { NativeChatMessage, @@ -102,10 +103,25 @@ export const MessageRow = memo(function MessageRow( const { scrollToTop, returnToView } = useNativeChatRowNavigation(rowRef, onScrollMessageToTop) // Return after all hooks to preserve hook order when content disappears. - if (run === undefined && !rowHasContent(content)) { + if (run === undefined && !rowHasContent(content) && !message.agentStart) { return null } + if (message.agentStart && message.from) { + return ( +
+ +
+ ) + } + const notice = isSystem ? nativeChatNoticeBlock(blocks) : undefined if (notice) { return ( diff --git a/src/renderer/src/components/native-chat/native-chat-agent-message-sender-view.test.tsx b/src/renderer/src/components/native-chat/native-chat-agent-message-sender-view.test.tsx new file mode 100644 index 00000000000..003c50637ce --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-agent-message-sender-view.test.tsx @@ -0,0 +1,308 @@ +// @vitest-environment happy-dom +import '@testing-library/jest-dom/vitest' +import { act, cleanup, renderHook, screen } from '@testing-library/react' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { render } from './native-chat-tooltip-test-harness' +import { useAppStore } from '@/store' +import { makeTab, makeWorktree, TEST_REPO } from '@/store/slices/store-test-helpers' +import { readStoreListenerCount } from '@/store/store-listener-census' +import { getKnownExecutionHostIdForWorktree } from '@/lib/worktree-runtime-owner' +import { resolveStructuredAgentSessionOwner } from '@/runtime/structured-agent-session-owner' +import type { TerminalLayoutSnapshot } from '../../../../shared/terminal-tab-types' +import type { AgentStatusEntry } from '../../../../shared/agent-status-types' +import type { AgentMessageSenderView } from '../../../../shared/agent-session/agent-message-presentation' +import { runtimeEnvironmentStatusFromSnapshot } from '../../../../shared/runtime-host-status' +import * as presentation from '@/lib/program-status/agent-row-presentation' +import * as cache from '@/components/dashboard/worktree-agent-rows-cache' +import * as selectors from '@/components/sidebar/worktree-agent-row-selectors' +import { useAgentMessageSenderView } from './native-chat-agent-message-sender-view' +import { useAgentMessageSenderSource } from './native-chat-agent-message-sender-source' +import { AgentStateDot } from '@/components/AgentStateDot' +import { + initialStore, + LEAF, + from, + status, + resetSenderStore +} from './native-chat-agent-sender.test-fixture' + +const renders = new Map() +const views = new Map() +function Probe({ + id, + workspace = 'workspace', + remote = false +}: { + id: number + workspace?: string + remote?: boolean +}) { + const value = useAgentMessageSenderView( + { + party: { address: `term_${id}`, terminalHandle: `term_${id}`, orcaSessionId: null }, + name: 'Sender' + }, + workspace, + remote ? { kind: 'environment', environmentId: 'remote' } : { kind: 'local' } + ) + views.set(id, value) + renders.set(id, (renders.get(id) ?? 0) + 1) + return null +} +function write(i: number, update: Partial) { + const state = useAppStore.getState() + const key = status(i).paneKey + act(() => + useAppStore.setState({ + agentStatusByPaneKey: { + ...state.agentStatusByPaneKey, + [key]: { ...state.agentStatusByPaneKey[key], ...update } + }, + agentStatusEpoch: state.agentStatusEpoch + 1 + }) + ) +} + +beforeEach(() => { + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(2_000_000) + renders.clear() + views.clear() + resetSenderStore() +}) +afterEach(() => { + cleanup() + useAppStore.setState(initialStore, true) + vi.restoreAllMocks() + vi.useRealTimers() +}) + +it.each([100, 1000])( + 'shares actual row acquisition and isolates selected progress across %i cards', + (n) => { + let visits = 0 + const live = new Proxy( + Object.fromEntries( + Array.from({ length: n }, (_, i) => { + const entry = status(i) + return [entry.paneKey, entry] + }) + ), + { + ownKeys(target) { + const keys = Object.getOwnPropertyNames(target) + visits += keys.length + return keys + } + } + ) + useAppStore.setState({ + agentStatusByPaneKey: live, + tabsByWorktree: { + workspace: Array.from({ length: n }, (_, i) => + makeTab({ id: `tab-${i}`, worktreeId: 'workspace' }) + ) + }, + terminalLayoutsByTabId: Object.fromEntries( + Array.from({ length: n }, (_, i): [string, TerminalLayoutSnapshot] => [ + `tab-${i}`, + { root: { type: 'leaf', leafId: LEAF }, activeLeafId: LEAF, expandedLeafId: null } + ]) + ) + }) + const normalize = vi.spyOn(presentation, 'resolveRowProgramStatus') + const mounted = render( + <> + {Array.from({ length: n }, (_, id) => ( + + ))} + + ) + // Handle lookup, visible-authority acquisition and sidebar indexing each enumerate the N live entries. + expect(visits).toBe(n * 3) + expect(normalize).toHaveBeenCalledTimes(n) + for (let i = 0; i < n; i++) { + expect(views.get(i)?.dot).toBe('working') + } + const before = new Map(renders) + visits = 0 + normalize.mockClear() + act(() => useAppStore.setState({ sidebarBody: 'agents' })) + expect(visits).toBe(0) + expect(normalize).not.toHaveBeenCalled() + expect(renders).toEqual(before) + for (const progress of [40, 60]) { + normalize.mockClear() + const epoch = useAppStore.getState().agentStatusEpoch + act(() => + useAppStore.getState().setAgentProgramStatus(status(0).paneKey, { + state: 'working', + progress, + observedAt: 2_000_000 + }) + ) + expect(useAppStore.getState().agentStatusEpoch).toBe(epoch) + expect(normalize).toHaveBeenCalledTimes(n) + expect(views.get(0)?.progress).toBe(progress) + expect(renders.get(0)).toBe(before.get(0)! + (progress === 40 ? 1 : 2)) + for (let i = 1; i < n; i++) { + expect(renders.get(i)).toBe(before.get(i)) + } + } + mounted.unmount() + } +) + +it('keeps other hosts isolated, retains completed senders, and releases disconnected subscriptions', () => { + const create = vi.spyOn(cache, 'createWorktreeAgentRowsCache') + const baseline = readStoreListenerCount() + expect(baseline).not.toBeNull() + const other = status(9, 'remote') + useAppStore.setState({ + repos: [TEST_REPO], + worktreesByRepo: { + [TEST_REPO.id]: [ + makeWorktree({ id: 'remote', repoId: TEST_REPO.id, hostId: 'runtime:remote' }) + ] + }, + tabsByWorktree: { + ...useAppStore.getState().tabsByWorktree, + remote: [makeTab({ id: 'tab-9', worktreeId: 'remote' })] + }, + agentStatusByPaneKey: { ...useAppStore.getState().agentStatusByPaneKey, [other.paneKey]: other } + }) + const mounted = render( + <> + + + + ) + expect(readStoreListenerCount()).toBeGreaterThan(baseline!) + const leased = create.mock.results.flatMap((result) => + result.type === 'return' && result.value.byWorktree.size ? [result.value] : [] + ) + expect(leased.flatMap((value) => [...value.byWorktree.keys()]).sort()).toEqual([ + 'remote', + 'workspace' + ]) + const before = new Map(renders) + write(9, { state: 'waiting' }) + expect(views.get(9)?.dot).toBe('waiting') + expect(renders.get(9)).toBe(before.get(9)! + 1) + expect(renders.get(0)).toBe(before.get(0)) + write(0, { + agentType: 'codex', + programStatus: { state: 'blocked', progress: 60, observedAt: 2_000_000 } + }) + expect(views.get(0)).toEqual({ agentType: 'codex', dot: 'permission', progress: 60 }) + expect(renders.get(9)).toBe(before.get(9)! + 1) + for (const value of leased) { + expect(value.lastComputedWorktreeIds.length).toBeLessThanOrEqual(1) + } + const entry = { ...status(0), state: 'done' as const } + act(() => + useAppStore.setState({ + agentStatusByPaneKey: { [other.paneKey]: other }, + retainedAgentsByPaneKey: { + [entry.paneKey]: { + entry, + tab: makeTab({ id: 'tab-0', worktreeId: 'workspace' }), + worktreeId: 'workspace', + agentType: 'claude', + startedAt: 1 + } + }, + runtimeStatusByEnvironmentId: new Map([ + [ + 'remote', + runtimeEnvironmentStatusFromSnapshot({ + environmentId: 'remote', + pairingRevision: 1, + sequence: 1, + checkedAt: 1, + verification: 'blocked', + transport: 'disconnected', + retired: true, + status: null + }) + ] + ]) + }) + ) + expect(views.get(0)?.dot).toBe('done') + act(() => useAppStore.setState({ activeWorkspaceExecutionHostId: 'runtime:remote' })) + expect(views.get(0)).toEqual({}) + act(() => useAppStore.setState({ activeWorkspaceExecutionHostId: 'local' })) + expect(views.get(0)?.dot).toBe('done') + const disconnected = new Map(renders) + const constructions = create.mock.calls.length + act(() => useAppStore.setState({ sidebarBody: 'agents' })) + expect(renders).toEqual(disconnected) + expect(create).toHaveBeenCalledTimes(constructions) + mounted.unmount() + for (const value of leased) { + expect(value.byWorktree.size).toBe(0) + } + expect(readStoreListenerCount()).toBe(baseline) + render() + expect(views.get(0)?.dot).toBe('done') +}) + +it('uses the child row own glyph even when its parent program needs permission', () => { + write(0, { + programStatus: { state: 'blocked', progress: 60, observedAt: 2_000_000 }, + subagents: [{ id: 'child', state: 'working', startedAt: 2_000_000, description: 'Child' }] + }) + const state = useAppStore.getState() + const projected = cache.selectWorktreeAgentRowsCached({ + state, + worktreeId: 'workspace', + now: 2_000_000, + generation: '', + orchestration: selectors.selectRuntimeAgentOrchestrationForWorktree(state, 'workspace') + }) + const child = projected.find((row) => row.childRow)! + expect(child).toBeDefined() + const locate = () => ({ host: 'local', worktreeId: 'workspace', paneKey: child.paneKey }) + let value: AgentMessageSenderView | undefined + function Child() { + value = useAgentMessageSenderSource(locate) + return value.dot ? ( + + ) : null + } + render() + expect(value?.dot).toBe('working') + expect(value?.progress).toBeUndefined() + expect(screen.queryByRole('progressbar')).toBeNull() + expect(screen.getByLabelText('Working')).toBeDefined() +}) + +it('refuses a target-less ambiguous workspace twin while an explicit local target resolves', () => { + const workspace = `${TEST_REPO.id}::/w` + const entry = status(0, workspace) + useAppStore.setState({ + repos: [TEST_REPO], + activeWorktreeId: 'different', + activeWorkspaceExecutionHostId: 'local', + detectedWorktreesByRepo: {}, + worktreesByRepo: { + [TEST_REPO.id]: [ + makeWorktree({ id: workspace, repoId: TEST_REPO.id, hostId: 'local' }), + makeWorktree({ id: workspace, repoId: TEST_REPO.id, hostId: 'runtime:e' }) + ] + }, + tabsByWorktree: { [workspace]: [makeTab({ id: 'tab-0', worktreeId: workspace })] }, + agentStatusByPaneKey: { [entry.paneKey]: entry } + }) + const state = useAppStore.getState() + expect(getKnownExecutionHostIdForWorktree(state, workspace)).toBe('local') + expect(resolveStructuredAgentSessionOwner(state, workspace)).toBeNull() + const view = renderHook(() => useAgentMessageSenderView(from.senders[0], workspace)) + expect(view.result.current).toEqual({}) + view.unmount() + const explicit = renderHook(() => + useAgentMessageSenderView(from.senders[0], workspace, { kind: 'local' }) + ) + expect(explicit.result.current.dot).toBe('working') +}) diff --git a/src/renderer/src/components/native-chat/native-chat-agent-sender.test-fixture.ts b/src/renderer/src/components/native-chat/native-chat-agent-sender.test-fixture.ts new file mode 100644 index 00000000000..7fb191d1fa6 --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-agent-sender.test-fixture.ts @@ -0,0 +1,50 @@ +import { useAppStore } from '@/store' +import { makeTab } from '@/store/slices/store-test-helpers' +import type { AgentStatusEntry } from '../../../../shared/agent-status-types' +import type { AgentMessageSource } from '../../../../shared/agent-session/agent-session-message-source' + +export const initialStore = useAppStore.getState() +export const LEAF = '11111111-1111-4111-8111-111111111111' +export function status(i: number, worktreeId = 'workspace'): AgentStatusEntry { + return { + paneKey: `tab-${i}:${LEAF}`, + worktreeId, + terminalHandle: `term_${i}`, + agentType: 'claude', + state: 'working', + prompt: 'work', + updatedAt: 2_000_000, + stateStartedAt: 2_000_000, + stateHistory: [] + } +} +export const from: AgentMessageSource = { + kind: 'agent', + orchestration: null, + senders: ['Build cards', 'Other sender'].map((name, i) => ({ + name, + party: { address: `term_${i}`, terminalHandle: `term_${i}`, orcaSessionId: null } + })) +} + +export function resetSenderStore(): void { + useAppStore.setState( + { + ...initialStore, + activeWorktreeId: 'workspace', + activeWorkspaceExecutionHostId: 'local', + tabsByWorktree: { workspace: [makeTab({ id: 'tab-0', worktreeId: 'workspace' })] }, + unifiedTabsByWorktree: {}, + agentStatusByPaneKey: { [status(0).paneKey]: status(0) }, + retainedAgentsByPaneKey: {}, + migrationUnsupportedByPtyId: {}, + runtimeAgentOrchestrationByPaneKey: {}, + runtimePaneTitlesByTabId: {}, + terminalLayoutsByTabId: {}, + ptyIdsByTabId: {}, + paneForegroundAgentByPaneKey: {}, + agentStatusEpoch: 1 + }, + true + ) +} diff --git a/src/renderer/src/components/native-chat/native-chat-message-copy-text.ts b/src/renderer/src/components/native-chat/native-chat-message-copy-text.ts index 358fd68e3ad..a62a46cdaae 100644 --- a/src/renderer/src/components/native-chat/native-chat-message-copy-text.ts +++ b/src/renderer/src/components/native-chat/native-chat-message-copy-text.ts @@ -10,7 +10,7 @@ export function nativeChatSlotCopyText( ): string | null { const slot = slots.find((entry) => entry.kind === 'message' && entry.message.id === messageId) return slot?.kind === 'message' - ? slot.message.from && slot.message.role === 'user' + ? slot.message.from && (slot.message.role === 'user' || slot.message.agentStart) ? agentMessageTranscriptText(slot.message.blocks, slot.message.from) : nativeChatMessageCopyText(slot.message, nativeChatStoppedBeforeStartLabel()) : null diff --git a/src/renderer/src/components/native-chat/native-chat-transcript-slots.ts b/src/renderer/src/components/native-chat/native-chat-transcript-slots.ts index 3155a4af1a9..5f26a704112 100644 --- a/src/renderer/src/components/native-chat/native-chat-transcript-slots.ts +++ b/src/renderer/src/components/native-chat/native-chat-transcript-slots.ts @@ -179,7 +179,9 @@ export function buildNativeChatTranscriptSlots( ) const drawsRow = receipt !== undefined || - (!folded && nativeChatRowRendersContent(message.blocks) && message.id !== liveReasoningId) + (!folded && + (message.agentStart || nativeChatRowRendersContent(message.blocks)) && + message.id !== liveReasoningId) const roster = sectionSlots.rosterAt(message.id) if (drawsRow || status !== undefined || turnDiff !== undefined) { slots.push({ diff --git a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts index 8cdaf67dd3e..6d9de3acc40 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts @@ -4,7 +4,12 @@ import type { } from '../../../../shared/agent-session-journal-types' import type { StructuredAgentSessionOptimisticMessage } from '../../../../shared/structured-agent-session-message-projection' import { projectStructuredAgentSessionMessages as projectMessages } from '../../../../shared/structured-agent-session-message-projection' -import { projectStructuredQuestionMessages } from '../../../../shared/structured-agent-question-projection' +import { createStructuredQuestionProjection } from '../../../../shared/structured-agent-question-projection' +import { projectStructuredAgentStartItem } from '../../../../shared/structured-agent-start-projection' + +export const desktopStructuredTranscript = createStructuredQuestionProjection( + projectStructuredAgentStartItem +) /** Keep host-recorded rejections in place as not sent unless a queue card owns them. */ export function projectStructuredAgentSessionMessages( @@ -12,5 +17,5 @@ export function projectStructuredAgentSessionMessages( optimistic: readonly StructuredAgentSessionOptimisticMessage[], submissions: readonly AgentJournalSubmission[] ) { - return projectMessages(items, optimistic, submissions, projectStructuredQuestionMessages) + return projectMessages(items, optimistic, submissions, desktopStructuredTranscript.messages) } diff --git a/src/renderer/src/components/native-chat/structured-session-option-defaults.test-fixture.ts b/src/renderer/src/components/native-chat/structured-session-option-defaults.test-fixture.ts new file mode 100644 index 00000000000..321e6af5216 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-session-option-defaults.test-fixture.ts @@ -0,0 +1,53 @@ +import { + applyNativeChatSessionOptionSettingsMutation, + resolveStructuredLaunchSeedOptions +} from '../../../../shared/native-chat/input/native-chat-session-option-defaults' +import type { PersistedNativeChatSessionOptions } from '../../../../shared/native-chat/input/native-chat-session-options' + +/** Replay every host mutation in order, exactly as the runtime does. */ +export function seededByNextLaunch( + calls: readonly unknown[][] +): Record | undefined { + let persisted: PersistedNativeChatSessionOptions | undefined + for (const [, mutation] of calls) { + persisted = + applyNativeChatSessionOptionSettingsMutation( + persisted, + mutation as Parameters[1] + ) ?? persisted + } + return resolveStructuredLaunchSeedOptions(persisted, 'codex') +} + +export const OPTIONS = { + models: [ + { + id: 'gpt-live', + label: 'GPT Live', + isDefault: true, + defaultEffort: 'medium', + efforts: [ + { value: 'medium', label: 'Medium' }, + { value: 'high', label: 'High' } + ] + }, + { + id: 'gpt-fast', + label: 'GPT Fast', + isDefault: false, + defaultEffort: 'low', + efforts: [ + { value: 'low', label: 'Low' }, + { value: 'medium', label: 'Medium' } + ] + } + ], + current: { model: 'gpt-live', effort: 'medium' } +} + +export const FAST_OPTIONS = { + ...OPTIONS, + models: OPTIONS.models.map((model) => ({ ...model, supportsFastMode: true })), + fastModeSupport: { supported: true }, + current: { ...OPTIONS.current, fastMode: false, confirmed: ['fastMode'] } +} diff --git a/src/renderer/src/components/native-chat/use-native-chat-hand-off.tsx b/src/renderer/src/components/native-chat/use-native-chat-hand-off.tsx index cf2c6cf057d..4551d2874b2 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-hand-off.tsx +++ b/src/renderer/src/components/native-chat/use-native-chat-hand-off.tsx @@ -145,6 +145,9 @@ export function useNativeChatHandOff({ ? (context.capturedText ?? '') : messages .flatMap((message) => { + if (message.agentStart) { + return [] + } const text = nativeChatMessageCopyText(message).trim() return text ? [`${message.role}: ${text}`] : [] }) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx index 74db048a063..98b753b3810 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.test.tsx @@ -5,6 +5,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const mocks = vi.hoisted(() => ({ call: vi.fn(), + startCapability: vi.fn(), + startDetail: vi.fn(), operationId: vi.fn(), enqueueSettingsWrite: vi.fn(), toastError: vi.fn() @@ -19,6 +21,8 @@ let submissions: AgentJournalSubmission[] = [] vi.mock('@/runtime/structured-agent-session-client', () => ({ subscribeStructuredAgentSessionStatus: vi.fn(async () => ({ unsubscribe: vi.fn() })), callStructuredAgentSession: mocks.call, + supportsStructuredAgentSessionItemDetail: mocks.startCapability, + readStructuredAgentSessionItem: mocks.startDetail, supportsStructuredAgentSessionQuietRepeatedStop: vi.fn(async () => false) })) @@ -60,25 +64,12 @@ import type { AgentJournalSubmission } from '../../../../shared/agent-session-journal-types' import { - applyNativeChatSessionOptionSettingsMutation, - resolveStructuredLaunchSeedOptions -} from '../../../../shared/native-chat/input/native-chat-session-option-defaults' -import type { PersistedNativeChatSessionOptions } from '../../../../shared/native-chat/input/native-chat-session-options' + OPTIONS, + FAST_OPTIONS, + seededByNextLaunch +} from './structured-session-option-defaults.test-fixture' import { useStructuredAgentSession } from './use-structured-agent-session' -/** Replay every host mutation in order, exactly as the runtime does. */ -function seededByNextLaunch(): Record | undefined { - let persisted: PersistedNativeChatSessionOptions | undefined - for (const [, mutation] of mocks.enqueueSettingsWrite.mock.calls) { - persisted = - applyNativeChatSessionOptionSettingsMutation( - persisted, - mutation as Parameters[1] - ) ?? persisted - } - return resolveStructuredLaunchSeedOptions(persisted, 'codex') -} - import { useAppStore } from '@/store' import { forgetStructuredSessionSelection } from '@/lib/structured-agent-session-selection-controller' @@ -106,39 +97,6 @@ afterEach(() => { const LOCAL_TARGET = { kind: 'local' } as const -const OPTIONS = { - models: [ - { - id: 'gpt-live', - label: 'GPT Live', - isDefault: true, - defaultEffort: 'medium', - efforts: [ - { value: 'medium', label: 'Medium' }, - { value: 'high', label: 'High' } - ] - }, - { - id: 'gpt-fast', - label: 'GPT Fast', - isDefault: false, - defaultEffort: 'low', - efforts: [ - { value: 'low', label: 'Low' }, - { value: 'medium', label: 'Medium' } - ] - } - ], - current: { model: 'gpt-live', effort: 'medium' } -} - -const FAST_OPTIONS = { - ...OPTIONS, - models: OPTIONS.models.map((model) => ({ ...model, supportsFastMode: true })), - fastModeSupport: { supported: true }, - current: { ...OPTIONS.current, fastMode: false, confirmed: ['fastMode'] } -} - function renderOptionsSession() { return renderHook(() => useStructuredAgentSession({ @@ -157,6 +115,53 @@ describe('useStructuredAgentSession working state', () => { mocks.call.mockResolvedValue(null) }) + it('renders full immutable creation facts without acquisition or pending work', async () => { + const previousItems = items + const previousSubmissions = submissions + items = [ + { + itemId: 'start', + revision: 1, + sequence: 1, + observedAt: 1, + body: { + kind: 'agent-start', + dispatchId: 'dispatch', + task: 'Full task', + worker: { + party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, + name: 'Worker' + } + } + } + ] + submissions = [] + let pending = 0 + mocks.startCapability.mockResolvedValue(true) + mocks.startDetail.mockImplementation(() => { + pending++ + return new Promise(() => {}) + }) + const mounted = renderOptionsSession() + try { + await act(async () => {}) + expect(mounted.result.current.messages).toEqual([ + expect.objectContaining({ agentStart: true, blocks: [{ type: 'text', text: 'Full task' }] }) + ]) + const first = mounted.result.current.messages[0] + items = [...items] + await act(async () => mounted.rerender()) + expect(mounted.result.current.messages[0]).toBe(first) + expect(mocks.startCapability).not.toHaveBeenCalled() + expect(mocks.startDetail).not.toHaveBeenCalled() + expect(pending).toBe(0) + } finally { + mounted.unmount() + items = previousItems + submissions = previousSubmissions + } + }) + it('reports work from an unanswered dispatch, and keeps the turn id provider-minted', () => { submissions = [ { @@ -502,7 +507,7 @@ describe('useStructuredAgentSession options', () => { expect(await result.current.setStructuredOption('model', 'gpt-fast')).toBe(true) }) - expect(seededByNextLaunch()).toEqual({ model: 'gpt-fast' }) + expect(seededByNextLaunch(mocks.enqueueSettingsWrite.mock.calls)).toEqual({ model: 'gpt-fast' }) expect(mocks.enqueueSettingsWrite).toHaveBeenCalledWith(LOCAL_TARGET, { type: 'apply-picks', agent: 'codex', @@ -558,7 +563,10 @@ describe('useStructuredAgentSession options', () => { }) // Without the model the launch resolves nothing, so the remembered effort would be dead. - expect(seededByNextLaunch()).toEqual({ model: 'gpt-live', effort: 'high' }) + expect(seededByNextLaunch(mocks.enqueueSettingsWrite.mock.calls)).toEqual({ + model: 'gpt-live', + effort: 'high' + }) }) it('remembers nothing when the transport leaves the pick unconfirmed', async () => { @@ -649,7 +657,7 @@ describe('useStructuredAgentSession options', () => { agent: 'codex', picks: [{ modelId: 'gpt-live', optionId: 'fastMode', value: true }] }) - expect(seededByNextLaunch()).toEqual({ + expect(seededByNextLaunch(mocks.enqueueSettingsWrite.mock.calls)).toEqual({ model: 'gpt-live', fastMode: 'true' }) diff --git a/src/renderer/src/components/native-chat/use-structured-native-chat-pane-commands.test.tsx b/src/renderer/src/components/native-chat/use-structured-native-chat-pane-commands.test.tsx index 51fda61124a..d24f9aaa868 100644 --- a/src/renderer/src/components/native-chat/use-structured-native-chat-pane-commands.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-native-chat-pane-commands.test.tsx @@ -161,6 +161,21 @@ function StructuredChatTab({ timestamp: 1, source: 'transcript', blocks: [{ type: 'text', text: 'Keep the task context' }] + }, + { + id: 'start', + role: 'system', + timestamp: 2, + source: 'transcript', + agentStart: true, + blocks: [{ type: 'text', text: 'Do not export this worker assignment' }] + }, + { + id: 'answer', + role: 'assistant', + timestamp: 3, + source: 'transcript', + blocks: [{ type: 'text', text: 'Keep the answer' }] } ] : [], @@ -437,7 +452,7 @@ describe('a structured chat tab', () => { view.rerender() await act(async () => openMenuHandOff()) expect(handOff.dialog.mock.lastCall?.[0].request.source).toMatchObject({ - capturedText: 'user: Keep the task context', + capturedText: 'user: Keep the task context\n\nassistant: Keep the answer', transcriptPath: undefined }) fireEvent.click(screen.getByText('Close hand-off')) @@ -445,7 +460,7 @@ describe('a structured chat tab', () => { handOff.options.mockResolvedValue({ status: 'current', transcriptPath: '/native.jsonl' }) await act(async () => openMenuHandOff()) expect(handOff.dialog.mock.lastCall?.[0].request.source).toMatchObject({ - capturedText: 'user: Keep the task context', + capturedText: 'user: Keep the task context\n\nassistant: Keep the answer', transcriptPath: undefined }) fireEvent.click(screen.getByText('Close hand-off')) @@ -476,6 +491,12 @@ describe('a structured chat tab', () => { expect(handOff.dialog.mock.lastCall?.[0].request.source.capturedText).toContain( 'Keep the task context' ) + expect(handOff.dialog.mock.lastCall?.[0].request.source.capturedText).toContain( + 'Keep the answer' + ) + expect(handOff.dialog.mock.lastCall?.[0].request.source.capturedText).not.toContain( + 'worker assignment' + ) fireEvent.click(screen.getByText('Close hand-off')) } else { expect(handOff.error).toHaveBeenCalledOnce() diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 782d9f94ed8..5fc2d3c990b 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -19768,6 +19768,7 @@ "goToAgent": "Go to agent", "messageFromCoordinator": "Message from coordinator", "showMore": "Show more", + "startedWorker": "Started worker", "taskFromCoordinator": "Task from coordinator", "updateFromWorker": "Update from worker" }, diff --git a/src/shared/agent-session-journal-identity.ts b/src/shared/agent-session-journal-identity.ts new file mode 100644 index 00000000000..35145ef5ecd --- /dev/null +++ b/src/shared/agent-session-journal-identity.ts @@ -0,0 +1,39 @@ +import type { AgentType } from './agent-status-types' +import type { AgentSessionProviderHandle } from './agent-session-provider-handle' + +/** Epoch-qualified position in one journal. `sequence` 0 means "before the first row". */ +export type AgentJournalCursor = { + epoch: string + sequence: number +} + +/** Persisted/wire handles must retain their shape; attach omits opaque. Derived via agentSessionJournalProviderHandle; opaque covers other transports or pending proof. */ +export type AgentSessionJournalProviderHandle = + | { kind: 'codex'; threadId: string } + | { kind: 'claude'; sessionId: string; leafUuid: string | null } + | { kind: 'opaque'; agent: AgentType; value: string } + +/** The narrow slice of the durable session record the journal needs. The full + * record (owner, lease, account home) belongs to the session store. */ +export type AgentSessionJournalIdentity = { + /** Orca agent-session id — the journal's primary key. */ + sessionId: string + /** Execution-host workspace key. Identical for a worktree, a folder + * workspace, a WSL distro, and an SSH host; never a path. */ + workspaceId: string + /** Execution host that owns the process, so a client restart adjudicates nothing. */ + hostId: string + agent: AgentType + /** The record's proved handle; null before the provider has proved one. */ + providerHandle: AgentSessionProviderHandle | null +} + +// Codex renumbers items on resume, so persisted item IDs are not identity; Claude preserves UUIDs on fork. + +export type AgentJournalItemIdentity = + | { provider: 'codex'; threadId: string; turnId: string; ordinal: number } + | { provider: 'claude'; sessionId: string; uuid: string } + /** A submission Orca minted before any provider echo existed. */ + | { provider: 'orca'; clientMessageId: string } + /** Bridge-era transcript record with no provider-stable identity. */ + | { provider: 'legacy'; agent: AgentType; sessionId: string; recordId: string } diff --git a/src/shared/agent-session-journal-schemas.test.ts b/src/shared/agent-session-journal-schemas.test.ts index 9a2b9194cc9..930a6a9c151 100644 --- a/src/shared/agent-session-journal-schemas.test.ts +++ b/src/shared/agent-session-journal-schemas.test.ts @@ -22,6 +22,15 @@ const RESOLUTION = { // Typed fixtures ensure the schema audit validates the canonical model. const CANONICAL_BODIES: AgentJournalItemBody[] = [ + { + kind: 'agent-start', + dispatchId: 'dispatch', + task: 'Complete task', + worker: { + party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, + name: 'Worker' + } + }, { kind: 'message', role: 'reasoning', diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index db7b615d479..5752081a18a 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -194,6 +194,11 @@ const TurnLifecycleFields = { const KnownItemBody = z.discriminatedUnion('kind', [ MessageBody, + z.object({ + kind: z.literal('agent-start'), + dispatchId: z.string().min(1), + task: z.string() + }), z.object({ kind: z.literal('tool-call'), ...ToolMetadata, diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 4358a77325c..7a8adc2fb6f 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -1,3 +1,10 @@ +import type { AgentJournalCursor, AgentJournalItemIdentity } from './agent-session-journal-identity' +export type { + AgentJournalCursor, + AgentJournalItemIdentity, + AgentSessionJournalIdentity, + AgentSessionJournalProviderHandle +} from './agent-session-journal-identity' import type { AgentJournalTurnScope } from './agent-session-journal-turn-scope' export type { AgentJournalTurnScope } from './agent-session-journal-turn-scope' import type { AgentSessionProviderContextBoundary } from './agent-session-provider-context' @@ -11,9 +18,11 @@ import type { AgentSessionQuestionAnswer } from './agent-session-question-answer import type { AgentJournalTurnOutcome } from './agent-turn-outcome' import type { NativeChatToolMetadata } from './native-chat/tools/native-chat-tool-identity' import type { AgentSessionContextUsage } from './agent-session-context-usage' -import type { AgentSessionProviderHandle } from './agent-session-provider-handle' import type { NativeChatBlock, NativeChatRole } from './native-chat-types' -import type { AgentMessageSource } from './agent-session/agent-session-message-source' +import type { + AgentMessageSender, + AgentMessageSource +} from './agent-session/agent-session-message-source' import type { AgentSessionExecutionOptions } from './agent-session-execution-options' export { type AgentType } @@ -32,43 +41,6 @@ export function journalRowSchemaVersion(bodies: readonly { kind: string }[]): nu : AGENT_SESSION_JOURNAL_PRE_TURN_SCHEMA_VERSION } -/** Epoch-qualified position in one journal. `sequence` 0 means "before the first row". */ -export type AgentJournalCursor = { - epoch: string - sequence: number -} - -/** Persisted/wire handles must retain their shape; attach omits opaque. Derived via agentSessionJournalProviderHandle; opaque covers other transports or pending proof. */ -export type AgentSessionJournalProviderHandle = - | { kind: 'codex'; threadId: string } - | { kind: 'claude'; sessionId: string; leafUuid: string | null } - | { kind: 'opaque'; agent: AgentType; value: string } - -/** The narrow slice of the durable session record the journal needs. The full - * record (owner, lease, account home) belongs to the session store. */ -export type AgentSessionJournalIdentity = { - /** Orca agent-session id — the journal's primary key. */ - sessionId: string - /** Execution-host workspace key. Identical for a worktree, a folder - * workspace, a WSL distro, and an SSH host; never a path. */ - workspaceId: string - /** Execution host that owns the process, so a client restart adjudicates nothing. */ - hostId: string - agent: AgentType - /** The record's proved handle; null before the provider has proved one. */ - providerHandle: AgentSessionProviderHandle | null -} - -// Codex renumbers items on resume, so persisted item IDs are not identity; Claude preserves UUIDs on fork. - -export type AgentJournalItemIdentity = - | { provider: 'codex'; threadId: string; turnId: string; ordinal: number } - | { provider: 'claude'; sessionId: string; uuid: string } - /** A submission Orca minted before any provider echo existed. */ - | { provider: 'orca'; clientMessageId: string } - /** Bridge-era transcript record with no provider-stable identity. */ - | { provider: 'legacy'; agent: AgentType; sessionId: string; recordId: string } - // ─── Bounded payloads ─────────────────────────────────────────────────────── /** A tool output or diff body. A row keeps the whole value (`truncated: false`). `truncated: true` @@ -348,7 +320,16 @@ export type AgentJournalStatusItem = * absent whenever nothing told the host one. */ export type AgentJournalTurnItem = { kind: 'turn' } & AgentJournalTurnLifecycle +/** Display-only creation fact; survives removal of the orchestration Task and Dispatch. */ +export type AgentJournalAgentStartItem = { + kind: 'agent-start' + dispatchId: string + worker: AgentMessageSender + task: string +} + export type AgentJournalItemBody = + | AgentJournalAgentStartItem | AgentJournalMessageItem | AgentJournalToolCallItem | AgentJournalDiffItem diff --git a/src/shared/agent-session/agent-message-presentation.ts b/src/shared/agent-session/agent-message-presentation.ts index 1aec05070c3..e79f6127252 100644 --- a/src/shared/agent-session/agent-message-presentation.ts +++ b/src/shared/agent-session/agent-message-presentation.ts @@ -2,7 +2,7 @@ import type { AgentMessageSender, AgentMessageSource } from './agent-session-mes import type { AgentStatusEntry } from '../agent-status-types' export const AGENT_MESSAGE_FOLDED_LINES = 3 -export type AgentMessageHeading = 'task' | 'coordinator' | 'worker' | 'generic' +export type AgentMessageHeading = 'started' | 'task' | 'coordinator' | 'worker' | 'generic' export type AgentMessageSenderView = Readonly<{ agentType?: AgentStatusEntry['agentType'] progress?: number @@ -20,7 +20,13 @@ export type AgentMessageSenderView = Readonly<{ | 'permission' }> -export function agentMessageHeading(from: AgentMessageSource): AgentMessageHeading { +export function agentMessageHeading( + from: AgentMessageSource, + agentStart = false +): AgentMessageHeading { + if (agentStart) { + return 'started' + } if (from.orchestration?.message === 'task') { return 'task' } diff --git a/src/shared/native-chat-noise.ts b/src/shared/native-chat-noise.ts index ef75c81f751..71ab0363327 100644 --- a/src/shared/native-chat-noise.ts +++ b/src/shared/native-chat-noise.ts @@ -11,6 +11,9 @@ function messageText(message: NativeChatMessage): string { /** Harness-injected user/system turns are transport machinery, not conversation. */ export function isNoiseMessage(message: NativeChatMessage): boolean { + if (message.agentStart) { + return false + } if (message.role !== 'user' && message.role !== 'system') { return false } diff --git a/src/shared/native-chat-types.ts b/src/shared/native-chat-types.ts index 2d3fc247bef..c1eed010930 100644 --- a/src/shared/native-chat-types.ts +++ b/src/shared/native-chat-types.ts @@ -233,6 +233,8 @@ export type NativeChatMessage = AgentJournalProducerLinkage & { command?: { name: string } /** On a user-role message another agent sent through Orca: who, as the journal recorded it. */ from?: AgentMessageSource + /** Projected creation fact, never a user turn or provider input. */ + agentStart?: true /** Accepted but not yet handed to the agent: drawn after everything the agent has done. */ queued?: true /** Unsent rows belong to no turn. Draw at journal position, or after the conversation when unplaced. */ diff --git a/src/shared/native-chat/native-chat-trailing-run.ts b/src/shared/native-chat/native-chat-trailing-run.ts index 3cccea1626f..ee8e6df41f2 100644 --- a/src/shared/native-chat/native-chat-trailing-run.ts +++ b/src/shared/native-chat/native-chat-trailing-run.ts @@ -12,6 +12,7 @@ export function nativeChatRowSpeaksOrActs( receipts: ReadonlyMap ): boolean { return ( + !message.agentStart && message.role !== 'user' && message.role !== 'reasoning' && receipts.get(message.id)?.kind !== 'approval' && diff --git a/src/shared/native-chat/transcript/native-chat-turn-fold.ts b/src/shared/native-chat/transcript/native-chat-turn-fold.ts index 4e2b4b48e14..7162ae62cd8 100644 --- a/src/shared/native-chat/transcript/native-chat-turn-fold.ts +++ b/src/shared/native-chat/transcript/native-chat-turn-fold.ts @@ -41,14 +41,16 @@ export function nativeChatTurnFoldRow( turnKey, role: message.role, rendersProse: nativeChatRowRendersProse(message), - draws: hasReceipt || nativeChatRowRendersContent(message.blocks), + draws: hasReceipt || message.agentStart === true || nativeChatRowRendersContent(message.blocks), // Use raw roster blocks: empty rosters render no block, making their plain-text twin the sole spawn record. - outlivesTurn: message.blocks.some( - (block) => - isSubagentGroupBlock(block) || - isBackgroundTaskBlock(block) || - isStoppedBeforeStartBlock(block) - ), + outlivesTurn: + message.agentStart === true || + message.blocks.some( + (block) => + isSubagentGroupBlock(block) || + isBackgroundTaskBlock(block) || + isStoppedBeforeStartBlock(block) + ), // A row about Orca's own stop is stored red for clients that predate it; it reports no failure. reportsFailure: message.blocks.some( (block) => diff --git a/src/shared/structured-agent-question-projection.ts b/src/shared/structured-agent-question-projection.ts index 7a2eac81770..0f536b5f91e 100644 --- a/src/shared/structured-agent-question-projection.ts +++ b/src/shared/structured-agent-question-projection.ts @@ -12,7 +12,7 @@ import { readAgentJournalTurn } from './agent-session-turn-record' import type { NativeChatResolvedPrompt } from './native-chat-resolution-receipt' type Projection = { message: NativeChatMessage | null; questionKey: string | null } -const projections = new WeakMap() +type ItemProjector = (item: AgentJournalRenderItem) => NativeChatMessage | null const pendingGroups = new WeakMap< AgentJournalQuestionItem, { @@ -57,13 +57,17 @@ function questionKey(questions: readonly { question: string }[]): string | null return texts.length > 0 && texts.every(Boolean) ? JSON.stringify(texts.sort()) : null } -function projectItem(item: AgentJournalRenderItem): Projection { +function projectItem( + item: AgentJournalRenderItem, + projections: WeakMap, + projectMessage: ItemProjector +): Projection { const cached = projections.get(item) if (cached) { return cached } const { body } = item - let message = projectStructuredItemToNativeChat(item) + let message = projectMessage(item) let key: string | null = null if (body.kind === 'question') { const questions = body.questions?.length ? body.questions : [{ question: body.question }] @@ -91,7 +95,10 @@ function projectItem(item: AgentJournalRenderItem): Projection { } /** Question presentation is client-local; archives and older RPC consumers keep their projection. */ -function projectQuestions(items: readonly AgentJournalRenderItem[]): { +function projectQuestions( + items: readonly AgentJournalRenderItem[], + project: (item: AgentJournalRenderItem) => Projection +): { messages: NativeChatMessage[] receipts: ReadonlyMap } { @@ -107,7 +114,7 @@ function projectQuestions(items: readonly AgentJournalRenderItem[]): { if (lifecycle) { turn = lifecycle.turnId } - const projection = projectItem(item) + const projection = project(item) rows.push({ item, projection, turn }) if (turn && item.body.kind === 'question' && projection.questionKey) { let questions = questionsByTurn.get(turn) @@ -165,22 +172,31 @@ function projectQuestions(items: readonly AgentJournalRenderItem[]): { return { messages, receipts } } -const histories = new WeakMap< - readonly AgentJournalRenderItem[], - ReturnType ->() - -export function structuredQuestionTranscript(items: readonly AgentJournalRenderItem[]) { - let projection = histories.get(items) - if (!projection) { - projection = projectQuestions(items) - histories.set(items, projection) +/** Each surface owns caches for the item interpretation it renders. */ +export function createStructuredQuestionProjection( + projectMessage: ItemProjector = projectStructuredItemToNativeChat +) { + const projections = new WeakMap() + const histories = new WeakMap< + readonly AgentJournalRenderItem[], + ReturnType + >() + const project = (item: AgentJournalRenderItem) => projectItem(item, projections, projectMessage) + const transcript = (items: readonly AgentJournalRenderItem[]) => { + let projection = histories.get(items) + if (!projection) { + projection = projectQuestions(items, project) + histories.set(items, projection) + } + return projection + } + return { + transcript, + messages: (items: readonly AgentJournalRenderItem[]) => transcript(items).messages } - return projection } -export function projectStructuredQuestionMessages( - items: readonly AgentJournalRenderItem[] -): NativeChatMessage[] { - return structuredQuestionTranscript(items).messages -} +export const { + transcript: structuredQuestionTranscript, + messages: projectStructuredQuestionMessages +} = createStructuredQuestionProjection() diff --git a/src/shared/structured-agent-start-projection.ts b/src/shared/structured-agent-start-projection.ts new file mode 100644 index 00000000000..32d3ae617a8 --- /dev/null +++ b/src/shared/structured-agent-start-projection.ts @@ -0,0 +1,30 @@ +import type { AgentJournalRenderItem } from './agent-session-journal-types' +import { agentJournalItemRowOrigin } from './agent-session-journal-position' +import { agentJournalLinkageFields } from './agent-session-journal-producer' +import { readAgentMessageSource } from './agent-session/agent-session-message-source' +import type { NativeChatMessage } from './native-chat-types' +import { projectStructuredItemToNativeChat } from './structured-agent-session-projection' + +/** Display-only creation facts are opt-in; canonical transcript readers keep their existing meaning. */ +export function projectStructuredAgentStartItem( + item: AgentJournalRenderItem +): NativeChatMessage | null { + if (item.body.kind !== 'agent-start') { + return projectStructuredItemToNativeChat(item) + } + if (item.shortened) { + return null + } + return { + ...agentJournalItemRowOrigin(item), + ...agentJournalLinkageFields(item), + role: 'system', + blocks: [{ type: 'text', text: item.body.task }], + agentStart: true, + from: readAgentMessageSource({ + kind: 'agent', + senders: [item.body.worker], + orchestration: null + }) + } +}