Show a card in the coordinator's chat when it starts a worker (#35263)

* Snapshot agent message relations and verify task display prefixes

* Show native chat agent messages in navigable cards

* Keep agent message cards derived from canonical input and sidebar state

* test: correct agent card contracts and cover raw queue editing

* test: repair agent card runtime and operation contracts

* fix: preserve queued image previews and typed test measurements

* fix: use checked worker fixture and owned sender button typography

* fix: complete agent card fixtures and native presentation

* test: preserve queued edit exports and current authorship

* test: use namespace type imports in queued edit fixture

* Show newly started workers in native chats

* Preserve sender progress and consolidate card contracts

* Simplify sender and notice contract instrumentation

* Remove duplicate sender and notice measurement mirrors

* Use a template literal for the dispatch boundary fixture

* Correct sender and notice verification phase accounting

* Consolidate agent message tests around user contracts

* Restore folded agent message preview on focus exit

* Consolidate agent message regression coverage

* Consolidate agent message fixture data

* Fix display text test lint diagnostics

* Use conversation view type for mail test records

* Reuse active copy control in card workflow test

* Use fixed timestamp in worker start context fixture

* Count fixture array reads without reflective access

* Use a typed store in runtime state fixture

* Fix worker card fixture types and CLI transport project includes
This commit is contained in:
Brennan Benson
2026-10-29 20:28:54 -07:00
committed by GitHub
parent 324f138de2
commit bf3e7ed405
41 changed files with 2591 additions and 572 deletions
+5
View File
@@ -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",
@@ -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 }))
@@ -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)
}
)
@@ -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)
@@ -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<StructuredAgentSessionHostDeps['probeOwner']>
probeOwner: NonNullable<StructuredAgentSessionHostDeps['probeOwner']>,
options: {
store?: Partial<AgentSessionRecordStore>
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<void>()
const getRecord = vi.fn((_sessionId: string) => record)
const reconcileOnRestart = vi.fn(
async (args: Parameters<AgentSessionRecordStore['reconcileOnRestart']>[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()
@@ -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<typeof createRestartReconciler>
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 {
@@ -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<typeof leases.createRestartReconciler>
private readonly startup: ReturnType<typeof leases.createStartupRecovery>
private readonly startup: ReturnType<typeof createStartupRecovery>
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)
@@ -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<StructuredAgentSessionHostSession>
) {
/** 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<void> => {
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 }
}
@@ -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.
}
}
@@ -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<unknown>(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<Record<string, unknown>>()),
RuntimeClient: function () {
return {
isRemote: false,
async call(method: string, params: Record<string, unknown> = {}) {
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<string, unknown>
_meta: { runtimeId: string }
}) => void
>()
vi.doMock('../../../../../cli/format', () => ({ printResult: output }))
const cliPath = '../../../../../cli/index'
const { main }: { main(argv: string[]): Promise<void> } = 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
)
@@ -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<unknown> {
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
}
@@ -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<typeof runtime.showManagedTerminalWorkspace>
) => Promise<placement.WorkerAgentPlacement['worktree']>
}
vi.spyOn<WorkspaceLookup, 'showManagedTerminalWorkspace'>(
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<void>()
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<typeof createMinimalPersistedTerminalTab>[] = []
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
})
}
)
@@ -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<typeof startLocalWorker>[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
})
}
@@ -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<unknown> {
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<ReturnType<typeof placeWorkerAgent>> | 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
}
@@ -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')])
@@ -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 (
<div className="space-y-4 p-4">
<MessageRow
message={started}
expandSignal={false}
onScrollMessageToTop={() => {}}
runtimeContext={{ worktreeId: workspace, worktreePath: null, target: { kind: 'local' } }}
/>
{(['task', 'coordinator', 'worker', 'generic'] as const).map((kind, i) => {
const from: AgentMessageSource = {
kind: 'agent',
@@ -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(
<MessageRow
message={message}
expandSignal={false}
onScrollMessageToTop={() => {}}
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(['', ' ', '<system-reminder>Internal noise</system-reminder>'])(
'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(<MessageRow message={message} expandSignal={false} onScrollMessageToTop={() => {}} />)
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(
<MessageRow
message={message}
expandSignal={false}
onScrollMessageToTop={() => {}}
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<number, number>()
const views = new Map<number, AgentMessageSenderView>()
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<AgentStatusEntry>) {
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) => (
<Probe key={id} id={id} />
{messages.map((message) => (
<MessageRow
key={message.id}
message={message}
runtimeContext={context}
expandSignal={false}
onScrollMessageToTop={scroll}
/>
))}
</>
)
// 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(
<>
<Probe id={0} />
<Probe id={9} workspace="remote" remote />
</>
)
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(<Probe id={0} />)
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 ? (
<AgentStateDot state={value.dot} progress={value.progress} size="md" title={null} />
) : null
}
render(<Child />)
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')
})
@@ -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({
/>
<div className="min-w-0 flex-1">
<div className="flex items-center justify-between gap-2 text-xs text-muted-foreground">
<span>{heading(from)}</span>
<span>{heading(from, message.agentStart === true)}</span>
<div className="flex items-center gap-1">
{agentMessageCanCopy(text) ? (
// Why: -my-1 keeps the 24px button inside the text-xs header line, so copy costs no row of its own.
@@ -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<HTMLDivElement | null>(null)
@@ -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 (
<div ref={rowRef} data-native-chat-message-id={message.id}>
<NativeChatAgentMessageCard
message={message}
from={message.from}
runtimeContext={runtimeContext}
onLinkClick={onLinkClick}
allowFileUriLinks={allowFileUriLinks}
onRefolded={returnToView}
/>
</div>
)
}
const notice = isSystem ? nativeChatNoticeBlock(blocks) : undefined
if (notice) {
return (
@@ -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<number, number>()
const views = new Map<number, AgentMessageSenderView>()
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<AgentStatusEntry>) {
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) => (
<Probe key={id} id={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(
<>
<Probe id={0} />
<Probe id={9} workspace="remote" remote />
</>
)
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(<Probe id={0} />)
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 ? (
<AgentStateDot state={value.dot} progress={value.progress} size="md" title={null} />
) : null
}
render(<Child />)
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')
})
@@ -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
)
}
@@ -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
@@ -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({
@@ -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)
}
@@ -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<string, string> | undefined {
let persisted: PersistedNativeChatSessionOptions | undefined
for (const [, mutation] of calls) {
persisted =
applyNativeChatSessionOptionSettingsMutation(
persisted,
mutation as Parameters<typeof applyNativeChatSessionOptionSettingsMutation>[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'] }
}
@@ -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}`] : []
})
@@ -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<string, string> | undefined {
let persisted: PersistedNativeChatSessionOptions | undefined
for (const [, mutation] of mocks.enqueueSettingsWrite.mock.calls) {
persisted =
applyNativeChatSessionOptionSettingsMutation(
persisted,
mutation as Parameters<typeof applyNativeChatSessionOptionSettingsMutation>[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'
})
@@ -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(<StructuredChatTab inWorkspace fallback />)
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()
+1
View File
@@ -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"
},
@@ -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 }
@@ -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',
@@ -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,
+20 -39
View File
@@ -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
@@ -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'
}
+3
View File
@@ -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
}
+2
View File
@@ -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. */
@@ -12,6 +12,7 @@ export function nativeChatRowSpeaksOrActs(
receipts: ReadonlyMap<string, NativeChatResolvedPrompt>
): boolean {
return (
!message.agentStart &&
message.role !== 'user' &&
message.role !== 'reasoning' &&
receipts.get(message.id)?.kind !== 'approval' &&
@@ -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) =>
@@ -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<AgentJournalRenderItem, Projection>()
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<AgentJournalRenderItem, Projection>,
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<string, NativeChatResolvedPrompt>
} {
@@ -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<typeof projectQuestions>
>()
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<AgentJournalRenderItem, Projection>()
const histories = new WeakMap<
readonly AgentJournalRenderItem[],
ReturnType<typeof projectQuestions>
>()
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()
@@ -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
})
}
}