diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts index 763df3080c38..2fc2bc28d735 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts @@ -19,6 +19,7 @@ export function setStructuredAgentSessionTabVisibility( tabId?: string ) => Promise } + onSessionTabHidden?: (sessionId: string) => void } restartResume: { dismiss: (sessionIds: readonly string[]) => Promise } }, @@ -97,14 +98,20 @@ export function createStructuredAgentSessionTabSurface( getSessionTabId: (sessionId: string): string | null => host.deps.store.getSessionTabId(sessionId), showSessionTabs: (sessionIds: readonly string[]) => host.deps.store.showSessionTabs(sessionIds), + notifySessionTabHidden: (sessionId: string): void => notifyTabHidden(host, sessionId), setSessionTabVisibility: async ( sessionId: string, visible: boolean, - tabId?: string + tabId?: string, + /** A close that may still put the tab back sends the hidden notice once it settles. */ + options?: { deferHiddenNotice?: boolean } ): Promise => { await setStructuredAgentSessionTabVisibility(host, sessionId, visible, tabId) if (!visible) { unopened.delete(sessionId) + if (!options?.deferHiddenNotice) { + notifyTabHidden(host, sessionId) + } } // The tab edge of the row's lifetime; the handle close is the other. if (!visible && !sessions.get(sessionId)?.child) { @@ -113,3 +120,19 @@ export function createStructuredAgentSessionTabSurface( } } } + +function notifyTabHidden( + host: Parameters[0], + sessionId: string +): void { + try { + host.deps.onSessionTabHidden?.(sessionId) + } catch (error) { + // Bookkeeping never gates closing a chat. + host.deps.logger.warn('a chat tab close listener failed', { + scope: 'tab-hidden-listener', + sessionId, + error + }) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts index bbc80c17fdf7..c1eee4eabc9d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts @@ -141,6 +141,8 @@ export type StructuredAgentSessionHostDeps = { idleSweep?: { intervalMs?: number; idleMs?: number } /** Whether an orchestration dispatch still owns this session's worker; absent answers no. */ hasOpenDispatch?: (record: AgentSessionRecord) => boolean + /** A chat tab left the screen: closed, or its workspace removed. Advisory; a throw is logged. */ + onSessionTabHidden?: (sessionId: string) => void /** Where every failure the host carries on past is reported. Required: a host without one would * drop exactly the failures nobody sees in the UI. */ logger: StructuredAgentSessionLogger diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 786577a37998..333836379e45 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -232,6 +232,7 @@ export class StructuredAgentSessionHost { getSessionTabId = this.tabs.getSessionTabId showSessionTabs = this.tabs.showSessionTabs setSessionTabVisibility = this.tabs.setSessionTabVisibility + notifySessionTabHidden = this.tabs.notifySessionTabHidden /** The records file could not be read this launch, so chats it holds are not listed yet. */ legacyRecordImportOwed = (): boolean => this.deps.journalDatabase.legacyRecordImportOwed === true /** This runtime holds a chat: a record, or the records file's chats still owed their copy. */ diff --git a/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts b/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts index 2e52e6f07483..74301c16b013 100644 --- a/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts +++ b/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts @@ -15,6 +15,7 @@ import type { ProjectExecutionRuntimeResolution } from '../../shared/project-exe import { resolveLocalProjectRuntimeForWorktreeId } from '../local-project-runtime-resolution' import type { RuntimePtyWorktreeRecord } from './runtime-terminal-state-records' import { + localOrchestrationCliCommand, resolveTerminalOrchestrationCliCommand, runtimeOrchestrationCliCommand, type OrchestrationCliCommand @@ -22,6 +23,8 @@ import { import type { FleetAgentStatusEvidence } from '../../shared/orchestration-fleet-agent-status-evidence' import { readOrchestrationFleetAgentStatusSnapshot } from './orchestration-fleet-agent-status-snapshot' import { resolveStructuredWorkerAuthority } from './structured-worker-authority' +import { isStructuredWorkerHandle } from './structured-worker-identity' +import { parseOrcaSessionAddress } from '../../shared/orca-session-address' import { matchesProcessIncarnation } from './orchestration/worker-terminal-process-liveness' export class OrcaRuntimeWithGetOrchestrationDispatchAuthority extends OrcaRuntimeWithVerifyOrchestrationCompatibilityCaller { @@ -250,6 +253,10 @@ export class OrcaRuntimeWithGetOrchestrationDispatchAuthority extends OrcaRuntim } getTerminalOrchestrationCliCommand(handle: string): OrchestrationCliCommand { + // A structured session runs in this process: it is told what the structured mail lane types. + if (isStructuredWorkerHandle(handle) || parseOrcaSessionAddress(handle)) { + return localOrchestrationCliCommand() + } let pty: RuntimePtyWorktreeRecord | null = null try { const ptyId = this.resolveLeafForHandle(handle)?.ptyId diff --git a/src/main/runtime/orca-runtime-get-pty-record-for-pane-key.ts b/src/main/runtime/orca-runtime-get-pty-record-for-pane-key.ts index 730b46346e26..f5ad8fa45b82 100644 --- a/src/main/runtime/orca-runtime-get-pty-record-for-pane-key.ts +++ b/src/main/runtime/orca-runtime-get-pty-record-for-pane-key.ts @@ -10,11 +10,14 @@ import { } from './structured-worker-authority' import type { StructuredPointerTarget } from './orchestration/structured-mailbox-pointer-delivery' import { + chatDispatchMailTarget, handleLessCoordinatorSessionId, structuredSessionAddressTarget, structuredSessionMailTarget, structuredSessionIdleEdgeMailboxes } from './orchestration/structured-session-mail-target' +import { exitedChatDispatchesForSession } from './orchestration/chat-assignee' +import { OPERATOR_CLOSE_EXIT_CAUSE } from '../../shared/terminal-exit-cause' import { resolveTerminalIdentityFromProbes, type RuntimeTerminalIdentity @@ -221,6 +224,24 @@ export class OrcaRuntimeWithGetPtyRecordForPaneKey extends OrcaRuntimeWithPruneM } } + /** + * A chat tab left the screen. If that chat worked a Dispatch and has now ended, the Dispatch + * settles as a terminal the operator closed does; a /clear moves the tab, so its successor keeps + * the work. + */ + onStructuredSessionTabHidden(sessionId: string): void { + try { + const db = this.getExistingOrchestrationDb() + for (const dispatch of db ? exitedChatDispatchesForSession(sessionId, db) : []) { + this.failActiveDispatchOnExit(dispatch.assignee_handle, null, 0, OPERATOR_CLOSE_EXIT_CAUSE) + } + } catch (error) { + console.warn('[orchestration] settling a closed chat worker failed', { + error: error instanceof Error ? error.message : String(error) + }) + } + } + /** Settlement drops anything parked for the session; nothing will ever redrive it again. */ forgetStructuredSessionMail(sessionId: string): void { this.orchestrationStructuredMailboxPointerDelivery.forgetSession(sessionId) @@ -247,10 +268,15 @@ export class OrcaRuntimeWithGetPtyRecordForPaneKey extends OrcaRuntimeWithPruneM return this.resolveStructuredWorkerDirectMailboxTarget(mailboxHandle) } const dispatchId = mailboxHandle.slice('dispatch:'.length) - const assignee = this._orchestrationDb?.getDispatchContextById?.(dispatchId)?.assignee_handle + const dispatch = this._orchestrationDb?.getDispatchContextById?.(dispatchId) + const assignee = dispatch?.assignee_handle if (!assignee) { return null } + const chat = chatDispatchMailTarget(dispatch, this._orchestrationDb) + if (chat !== undefined) { + return chat + } const sessionId = this.liveStructuredWorkerSessionId(assignee) return sessionId ? { sessionId, dispatchId } : null } diff --git a/src/main/runtime/orca-runtime-get-worktree-ps.ts b/src/main/runtime/orca-runtime-get-worktree-ps.ts index 705ce13a1fc6..fc9888ae14aa 100644 --- a/src/main/runtime/orca-runtime-get-worktree-ps.ts +++ b/src/main/runtime/orca-runtime-get-worktree-ps.ts @@ -231,6 +231,8 @@ export class OrcaRuntimeWithGetWorktreePs extends OrcaRuntimeWithStartTuiIdleVis ) }, ...(this.structuredAgentStatusSinkFn ? { statusSink: this.structuredAgentStatusSinkFn } : {}), + // A closed chat settles the Dispatch it was working, as a closed terminal does. + onSessionTabHidden: (sessionId) => this.onStructuredSessionTabHidden(sessionId), attentionDelivery: createStructuredAttentionMobileDelivery({ readNotificationSettings: () => this.requireStore().getSettings().notifications, readWorkspaceLabels: (scope) => diff --git a/src/main/runtime/orca-runtime-stop-requested-pty-ids.ts b/src/main/runtime/orca-runtime-stop-requested-pty-ids.ts index cfb9f26e4ad5..a8231eda03ee 100644 --- a/src/main/runtime/orca-runtime-stop-requested-pty-ids.ts +++ b/src/main/runtime/orca-runtime-stop-requested-pty-ids.ts @@ -4,6 +4,7 @@ import { createStructuredMailboxPointerHost } from './orchestration/structured-m import { localOrchestrationCliCommand } from './orchestration/cli-command' import { RuntimeOrchestrationSenderNames } from './runtime-orchestration-sender-names' import { isStructuredWorkerHandle } from './structured-worker-identity' +import { ORCA_SESSION_ADDRESS_PREFIX } from '../../shared/orca-session-address' import { resolveStructuredWorkerAuthority } from './structured-worker-authority' import { OrcaRuntimeWithRuntimeId } from './orca-runtime-runtime-id' import { RuntimeTerminalAgentPresence } from './runtime-terminal-agent-presence' @@ -224,7 +225,8 @@ export class OrcaRuntimeWithStopRequestedPtyIds extends OrcaRuntimeWithRuntimeId getDb: () => this._orchestrationDb, getTerminalHandleForPaneKey: (paneKey) => this.getTerminalHandleForPaneKey(paneKey), hasTerminalHandle: (handle) => this.handles.has(handle), - isStructuredWorkerHandle: (handle) => isStructuredWorkerHandle(handle), + isStructuredSessionOwner: (handle) => + isStructuredWorkerHandle(handle) || handle.startsWith(ORCA_SESSION_ADDRESS_PREFIX), canProbePtyLiveness: () => Boolean(this.ptyController?.probePtyLiveness), controllerKnowsPtyIsLive: (ptyId) => this.controllerKnowsPtyIsLive(ptyId), isLeafPtyProvenAbsent: (ptyId) => this.isLeafPtyProvenAbsent(ptyId) diff --git a/src/main/runtime/orchestration/agent-facing-parity.test.ts b/src/main/runtime/orchestration/agent-facing-parity.test.ts index 6aee5f6b955d..a5b382c2f46b 100644 --- a/src/main/runtime/orchestration/agent-facing-parity.test.ts +++ b/src/main/runtime/orchestration/agent-facing-parity.test.ts @@ -72,14 +72,16 @@ function runtime(prompts: string[]): OrcaRuntimeService { const fake: Pick< OrcaRuntimeService, 'getNestedWorkerMaxDepth' | 'getTerminalOrchestrationCliCommand' | 'sendTerminalAgentPrompt' - > = { - getNestedWorkerMaxDepth: () => 2, - getTerminalOrchestrationCliCommand: () => 'orca', - sendTerminalAgentPrompt: async (handle, text) => { - prompts.push(text) - return { handle, accepted: true, bytesWritten: text.length } + > & { orchestrationSenderNames: Pick } = + { + orchestrationSenderNames: { nameOf: () => null }, + getNestedWorkerMaxDepth: () => 2, + getTerminalOrchestrationCliCommand: () => 'orca', + sendTerminalAgentPrompt: async (handle, text) => { + prompts.push(text) + return { handle, accepted: true, bytesWritten: text.length } + } } - } // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: preamble delivery reads only the members the fake implements. return fake as OrcaRuntimeService } @@ -101,6 +103,7 @@ async function renderPreamble(worker: 'chat' | 'terminal'): Promise { terminalHandle: worker === 'chat' ? CHAT_WORKER_HANDLE : TERMINAL_HANDLE, dispatchId: 'ctx_1', dispatchDepth: 1, + runId: 'run_1', taskId: 'task_1', taskSpec: 'do it', coordinatorHandle: 'term_coord', diff --git a/src/main/runtime/orchestration/chat-assignee.test.ts b/src/main/runtime/orchestration/chat-assignee.test.ts new file mode 100644 index 000000000000..5b1ac6252a54 --- /dev/null +++ b/src/main/runtime/orchestration/chat-assignee.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { testOrcaSessionId } from '../../../shared/orca-session-address-test-fixture' +import { sessionRecord } from '../rpc/orchestration-session-caller-test-fixture' +import { chatAssigneeSessionId, observeChatAssignee } from './chat-assignee' +import type { AgentSessionRecordReader } from './structured-session-lineage' + +const CHAT = testOrcaSessionId('7e3b9d15-2c4a-4f86-a0b1-5c9e2d7f3b64') + +function store(record: AgentSessionRecord, visible: boolean): AgentSessionRecordReader { + return { + getRecord: (sessionId) => (sessionId === record.sessionId ? record : null), + listRecords: () => [record], + getVisibleSessionTabIndex: () => ({ present: true, sessionIds: visible ? [CHAT] : [] }) + } +} + +describe('a chat assignee', () => { + it('is named only by an Orca session ID handle', () => { + expect(chatAssigneeSessionId(`orca_session_id:${CHAT}`)).toBe(CHAT) + expect(chatAssigneeSessionId('term_worker')).toBeNull() + expect(chatAssigneeSessionId(null)).toBeNull() + }) + + it('is live at rest: a released lease is not an exit', () => { + const resting = sessionRecord(CHAT, { lease: { claimStatus: 'released' } }) + expect(observeChatAssignee(CHAT, null, store(resting, true))).toMatchObject({ + status: 'live', + session: { sessionId: CHAT } + }) + }) + + it('has exited once its chat is closed', () => { + expect(observeChatAssignee(CHAT, null, store(sessionRecord(CHAT), false))).toEqual({ + status: 'exited', + reason: 'The chat was closed.' + }) + }) + + it('is unverifiable on another host, with nothing to read, or with its successor unknown, never exited', () => { + const remote = sessionRecord(CHAT, { location: { executionHostId: 'ssh:box' } }) + expect(observeChatAssignee(CHAT, null, store(remote, true))).toMatchObject({ + status: 'unverifiable' + }) + expect(observeChatAssignee(CHAT, null, null)).toMatchObject({ status: 'unverifiable' }) + const cleared: AgentSessionRecord = { + ...sessionRecord(CHAT), + conversationCommand: { + command: 'clear', + state: 'completed', + replacementSessionId: 'clear-missing', + operationId: 'op', + callerKey: 'caller', + phase: 'committed' + } + } + expect(observeChatAssignee(CHAT, null, store(cleared, true))).toMatchObject({ + status: 'unverifiable', + reason: expect.stringContaining('after a /clear cannot be verified') + }) + const unknown = testOrcaSessionId('3f9a1c7e-6b2d-4e85-a0c4-9d1e7b3f5a26') + expect(observeChatAssignee(unknown, null, store(sessionRecord(CHAT), true))).toMatchObject({ + status: 'unverifiable' + }) + }) +}) diff --git a/src/main/runtime/orchestration/chat-assignee.ts b/src/main/runtime/orchestration/chat-assignee.ts new file mode 100644 index 000000000000..da1f7be01178 --- /dev/null +++ b/src/main/runtime/orchestration/chat-assignee.ts @@ -0,0 +1,109 @@ +/** + * A chat as a Dispatch assignee. Its Dispatch row names it by its `/clear` root address, + * `orca_session_id:`, and holds no pane or process: whether it can still work is read off the + * session records the same way mail to it is, so the two can never disagree. + */ + +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + ORCA_SESSION_ADDRESS_PREFIX, + parseOrcaSessionAddress, + type OrcaSessionId +} from '../../../shared/orca-session-address' +import type { OrchestrationDb } from './db' +import { DISPATCH_CONTEXT_COLUMN_LIST } from './db/row-column-lists' +import type { DispatchContextRow } from './types' +import { structuredSessionMailReach } from './structured-session-mail-address' +import { + lineageLiveSession, + readAgentSessionRecordStore, + type AgentSessionRecordReader +} from './structured-session-lineage' + +/** The chat a Dispatch assignee handle names; null for a terminal or a structured worker. */ +export function chatAssigneeSessionId( + assigneeHandle: string | null | undefined +): OrcaSessionId | null { + return assigneeHandle ? parseOrcaSessionAddress(assigneeHandle) : null +} + +export type ChatAssigneeObservation = + /** `session` is the conversation's live session: its `/clear` successor, if it has one. */ + | { status: 'live'; session: AgentSessionRecord } + | { status: 'exited'; reason: string } + | { status: 'unverifiable'; reason: string } + +/** + * A chat at rest is live: the send that reaches it starts its agent. Only a closed chat has + * exited; a record this host lacks, or another host, is unverifiable: missing evidence is no exit. + */ +export function observeChatAssignee( + sessionId: OrcaSessionId, + db: OrchestrationDb | null | undefined, + store: AgentSessionRecordReader | null = readAgentSessionRecordStore() +): ChatAssigneeObservation { + if (!store) { + return { + status: 'unverifiable', + reason: 'The agent-session host is not installed in this runtime generation.' + } + } + const record = store.getRecord(sessionId) + if (!record) { + return { status: 'unverifiable', reason: 'No durable record backs this chat.' } + } + const reach = structuredSessionMailReach(store, record, db) + switch (reach.kind) { + case 'reachable': + return { status: 'live', session: reach.session } + case 'other-host': + return { status: 'unverifiable', reason: 'The chat runs on another host.' } + case 'unverifiable': + return { + status: 'unverifiable', + reason: `The session continuing this chat after a /clear cannot be verified: ${reach.reason}` + } + case 'ended': + switch (reach.reason) { + case 'closed': + return { status: 'exited', reason: 'The chat was closed.' } + case 'worker-identity-lost': + // Not reachable for a chat today (its Dispatch records no worker incarnation); still no exit. + return { + status: 'unverifiable', + reason: + 'The session that continues this chat is a worker whose identity this host lost.' + } + } + } +} + +/** + * The unsettled Dispatches of the chat whose current session's tab was hidden, once that chat has + * exited, re-derived from the records at the notice. Only that chat: another chat may be mid-close + * with its tab hidden, and a close can still put its tab back. + */ +export function exitedChatDispatchesForSession( + hiddenSessionId: string, + db: OrchestrationDb, + store: AgentSessionRecordReader | null = readAgentSessionRecordStore() +): DispatchContextRow[] { + if (!store) { + return [] + } + const rows = db.db + .prepare( + `SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts + WHERE status IN ('pending', 'dispatched') AND substr(assignee_handle, 1, ?) = ?` + ) + .all(ORCA_SESSION_ADDRESS_PREFIX.length, ORCA_SESSION_ADDRESS_PREFIX) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns Dispatch rows; the adapter exposes unknown. + return (rows as DispatchContextRow[]).filter((dispatch) => { + const chat = chatAssigneeSessionId(dispatch.assignee_handle) + return ( + chat !== null && + lineageLiveSession(store, chat)?.sessionId === hiddenSessionId && + observeChatAssignee(chat, db, store).status === 'exited' + ) + }) +} diff --git a/src/main/runtime/orchestration/db/dispatch-context/dispatch-context-store.ts b/src/main/runtime/orchestration/db/dispatch-context/dispatch-context-store.ts index b51e5b4a8aa2..c2a63204a8c3 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/dispatch-context-store.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/dispatch-context-store.ts @@ -66,7 +66,10 @@ export function createDispatchContext( launchTokenHash: launchTokenHash ?? null, assigneeHandle, assigneePaneKey: assigneePaneKey ?? null, - assigneeOrcaSessionId: dispatchAssigneeOrcaSessionId(processIncarnation), + assigneeOrcaSessionId: dispatchAssigneeOrcaSessionId({ + handle: assigneeHandle, + processIncarnation + }), processIncarnation: processIncarnation ?? null, creatorDispatchId, ...recordedCreatorIdentity(params.creator), diff --git a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-authority.ts b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-authority.ts index fa826c7e1f69..e82ae3fad03e 100644 --- a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-authority.ts +++ b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-authority.ts @@ -7,8 +7,9 @@ export function prepareStartingWorkerAuthority( params: { dispatchId: string handle: string - paneKey: string - processIncarnation: string + /** Both null for a chat, which its Orca session ID identifies instead. */ + paneKey: string | null + processIncarnation: string | null launchTokenHash?: string worktreeId: string effects: unknown[] @@ -41,7 +42,7 @@ export function prepareStartingWorkerAuthority( `Dispatch ${params.dispatchId} already has a different launch-token commitment.` ) } - const existing = this.findActiveDispatchForAssignee(params.handle, params.paneKey) + const existing = this.findActiveDispatchForAssignee(params.handle, params.paneKey ?? undefined) if (existing && existing.id !== params.dispatchId) { throw new Error( `Terminal ${params.handle} already has an active dispatch (${existing.id} for task ${existing.task_id})` @@ -61,7 +62,10 @@ export function prepareStartingWorkerAuthority( .run( params.handle, params.paneKey, - dispatchAssigneeOrcaSessionId(params.processIncarnation), + dispatchAssigneeOrcaSessionId({ + handle: params.handle, + processIncarnation: params.processIncarnation + }), params.processIncarnation, params.hostScope ?? null, params.launchTokenHash ?? null, @@ -118,19 +122,24 @@ export function prepareStartingWorkerAuthority( ownership: 'owned' }) } else { - const transferable = this.findTransferableWorkerTerminalResource({ - terminalHandle: params.handle, - paneKey: params.paneKey, - processIncarnation: params.processIncarnation, - hostScope: params.hostScope ?? null - }) - if (transferable) { + // A chat has no process to take over: it is always the user's own, external resource. + const exact = + params.paneKey && params.processIncarnation + ? { paneKey: params.paneKey, processIncarnation: params.processIncarnation } + : null + const transferable = exact + ? this.findTransferableWorkerTerminalResource({ + terminalHandle: params.handle, + ...exact, + hostScope: params.hostScope ?? null + }) + : undefined + if (exact && transferable) { this.transferWorkerTerminalResourceStatement({ resourceId: transferable.id, toDispatchId: params.dispatchId, terminalHandle: params.handle, - paneKey: params.paneKey, - processIncarnation: params.processIncarnation, + ...exact, endpointId, endpointIncarnation: params.processIncarnation, hostScope: params.hostScope ?? null diff --git a/src/main/runtime/orchestration/db/worker-terminal/failed-start-dispatch-identity.ts b/src/main/runtime/orchestration/db/worker-terminal/failed-start-dispatch-identity.ts index 86657dfb0be4..36a45bf91b16 100644 --- a/src/main/runtime/orchestration/db/worker-terminal/failed-start-dispatch-identity.ts +++ b/src/main/runtime/orchestration/db/worker-terminal/failed-start-dispatch-identity.ts @@ -27,7 +27,10 @@ export function recordFailedStartDispatchIdentity( .run( resource.terminal_handle, resource.pane_key, - dispatchAssigneeOrcaSessionId(resource.process_incarnation), + dispatchAssigneeOrcaSessionId({ + handle: resource.terminal_handle, + processIncarnation: resource.process_incarnation + }), resource.process_incarnation, resource.host_scope, worker.dispatch_id diff --git a/src/main/runtime/orchestration/dispatch-assignee-orca-session-id.ts b/src/main/runtime/orchestration/dispatch-assignee-orca-session-id.ts index 52c75ece7329..5b70d2b2380e 100644 --- a/src/main/runtime/orchestration/dispatch-assignee-orca-session-id.ts +++ b/src/main/runtime/orchestration/dispatch-assignee-orca-session-id.ts @@ -1,11 +1,17 @@ -import type { OrcaSessionId } from '../../../shared/orca-session-address' +import { parseOrcaSessionAddress, type OrcaSessionId } from '../../../shared/orca-session-address' import { structuredWorkerOrcaSessionIdForIncarnation } from '../structured-worker-identity' import { canonicalOrcaSessionId } from './canonical-orca-session-id' -/** The Orca session id a Dispatch row stores for the structured worker a process incarnation names. */ -export function dispatchAssigneeOrcaSessionId( +/** + * The Orca session id a Dispatch row stores for its assignee: the structured worker its process + * incarnation names, or the chat its `orca_session_id:` handle names. A PTY has none. + */ +export function dispatchAssigneeOrcaSessionId(assignee: { + handle: string processIncarnation: string | null | undefined -): OrcaSessionId | null { - const orcaSessionId = structuredWorkerOrcaSessionIdForIncarnation(processIncarnation) +}): OrcaSessionId | null { + const orcaSessionId = + structuredWorkerOrcaSessionIdForIncarnation(assignee.processIncarnation) ?? + parseOrcaSessionAddress(assignee.handle) return orcaSessionId === null ? null : canonicalOrcaSessionId(orcaSessionId) } diff --git a/src/main/runtime/orchestration/dispatch-task-source.ts b/src/main/runtime/orchestration/dispatch-task-source.ts new file mode 100644 index 000000000000..865d2d931638 --- /dev/null +++ b/src/main/runtime/orchestration/dispatch-task-source.ts @@ -0,0 +1,26 @@ +import type { AgentMessageSource } from '../../../shared/agent-session-message-source' +import { agentMessageSender, type SenderNameResolver } from './agent-message-sender' +import type { OrchestrationDb } from './db' +import type { DispatchContextRow } from './types' + +/** Who a Dispatch's task is from, as the worker's chat shows it: the coordinator, and the + * records the task joins back to. */ +export function dispatchTaskSource(args: { + db: OrchestrationDb + dispatch: Pick + /** The coordinator's address; a Dispatch with no declared caller names no sender. */ + from: string | undefined + senderName: SenderNameResolver +}): AgentMessageSource { + const { dispatch } = args + return { + kind: 'agent', + senders: args.from ? [agentMessageSender(args.from, args.db, args.senderName)] : [], + orchestration: { + message: 'task', + runId: dispatch.run_id, + taskId: dispatch.task_id, + dispatchId: dispatch.id + } + } +} diff --git a/src/main/runtime/orchestration/mailbox-delivery-target.ts b/src/main/runtime/orchestration/mailbox-delivery-target.ts index 5d412bb12e84..4faba0773a25 100644 --- a/src/main/runtime/orchestration/mailbox-delivery-target.ts +++ b/src/main/runtime/orchestration/mailbox-delivery-target.ts @@ -5,8 +5,8 @@ type OrchestrationMailboxDeliveryTargetDependencies = { getDb: () => OrchestrationDb | null getTerminalHandleForPaneKey: (paneKey: string) => string | null hasTerminalHandle: (handle: string) => boolean - /** A structured worker has no PTY handle; its own lane delivers, so this must not claim it. */ - isStructuredWorkerHandle: (handle: string) => boolean + /** A structured worker or a chat has no PTY handle; its own lane delivers, so this must not claim it. */ + isStructuredSessionOwner: (handle: string) => boolean canProbePtyLiveness: () => boolean controllerKnowsPtyIsLive: (ptyId: string) => boolean isLeafPtyProvenAbsent: (ptyId: string) => Promise @@ -21,7 +21,7 @@ export class OrchestrationMailboxDeliveryTarget { if (this.deps.hasTerminalHandle(handle)) { return handle } - if (this.deps.isStructuredWorkerHandle(handle)) { + if (this.deps.isStructuredSessionOwner(handle)) { return null } const db = this.deps.getDb() @@ -39,7 +39,7 @@ export class OrchestrationMailboxDeliveryTarget { if (!ownerHandle) { return null } - if (this.deps.isStructuredWorkerHandle(ownerHandle)) { + if (this.deps.isStructuredSessionOwner(ownerHandle)) { // The structured lane owns this mailbox; nothing here can type into it. return null } diff --git a/src/main/runtime/orchestration/orchestration-party.ts b/src/main/runtime/orchestration/orchestration-party.ts index 26107a1474bc..962e94ff5f73 100644 --- a/src/main/runtime/orchestration/orchestration-party.ts +++ b/src/main/runtime/orchestration/orchestration-party.ts @@ -34,7 +34,8 @@ export const ORCHESTRATION_TARGET_PARAM: Readonly< 'orchestration.dispatch': 'to', 'orchestration.inbox': 'terminal', 'orchestration.sessionAddress': 'sessionId', - 'orchestration.partyLocation': 'address' + 'orchestration.partyLocation': 'address', + 'orchestration.workerStart': 'terminal' } /** The party an Orca session id names. Throws when it is a worker this host lost the identity of. */ @@ -104,18 +105,13 @@ export function resolveDeclaredCallerParty( return party } -/** A Dispatch assignee: a terminal or a structured worker, never a chat yet. */ +/** + * A Dispatch assignee: a terminal, a structured worker, or a chat named by its Orca session ID. A + * chat's party is its `/clear` root address, which its own session id resolves to when it reports. + */ export function resolveDispatchAssigneeParty( address: string, db: OrchestrationDb | null | undefined ): OrchestrationParty { - const party = resolveOrchestrationParty(address, db) - if (party.terminalHandle === null) { - throw new OrchestrationError( - CODES.chatNotDispatchable, - `Agent session ${party.orcaSessionId} is a chat, and a chat can't receive a dispatch yet. Start a worker with worker-start instead. No effects were applied.`, - NO_EFFECTS - ) - } - return party + return resolveOrchestrationParty(address, db) } diff --git a/src/main/runtime/orchestration/send-agent-turn-boundary.test.ts b/src/main/runtime/orchestration/send-agent-turn-boundary.test.ts index 334195a5c88e..f509f859635a 100644 --- a/src/main/runtime/orchestration/send-agent-turn-boundary.test.ts +++ b/src/main/runtime/orchestration/send-agent-turn-boundary.test.ts @@ -112,7 +112,9 @@ describe('agent turn send boundary', () => { 'runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts', 'runtime/rpc/methods/orchestration/runs/dispatch-methods.ts', 'runtime/orchestration/coordinator-task-dispatch.ts', - 'runtime/rpc/methods/orchestration/federation/federation.ts' + 'runtime/rpc/methods/orchestration/federation/federation.ts', + // A chat assignee's task, from `dispatch --inject` and `worker-start --terminal`. + 'runtime/rpc/methods/orchestration/chat-task-delivery.ts' ].sort() ) }) diff --git a/src/main/runtime/orchestration/structured-session-mail-target.ts b/src/main/runtime/orchestration/structured-session-mail-target.ts index 2d512d58e023..c40de4899b0d 100644 --- a/src/main/runtime/orchestration/structured-session-mail-target.ts +++ b/src/main/runtime/orchestration/structured-session-mail-target.ts @@ -22,7 +22,8 @@ import { readAgentSessionRecordStore, type AgentSessionRecordReader } from './structured-session-lineage' -import type { RunRow } from './types' +import type { DispatchContextRow, RunRow } from './types' +import { chatAssigneeSessionId } from './chat-assignee' /** * The session a Run's coordinator binding names when that binding has no handle. A structured @@ -61,6 +62,22 @@ export function structuredSessionMailTarget( : null } +/** + * The target of a `dispatch:` mailbox whose assignee is a chat: its live session, under the + * Dispatch's budget. `undefined` when the assignee is not a chat, so a worker keeps its own lookup. + */ +export function chatDispatchMailTarget( + dispatch: Pick, + db: OrchestrationDb | null | undefined +): StructuredPointerTarget | null | undefined { + const sessionId = chatAssigneeSessionId(dispatch.assignee_handle) + if (!sessionId) { + return undefined + } + const target = structuredSessionMailTarget(sessionId, db) + return target ? { ...target, dispatchId: dispatch.id } : null +} + /** * The target of an `orca_session_id:` mailbox; `undefined` when the handle is not a session address at * all, so other address forms keep their own resolution. @@ -77,8 +94,8 @@ export function structuredSessionAddressTarget( } /** - * Every mailbox a session reads for itself: the Runs it coordinates, its own direct mail and, for a - * worker, its active Dispatch; a `/clear` successor reads its worker's. Re-derived from the database + * Every mailbox a session reads for itself: the Runs it coordinates, its own direct mail and its + * active Dispatch, a worker's or a chat's; a `/clear` successor reads its root's. Re-derived from the database * on each idle edge rather than remembered, so mail that arrived while the session could not take * it (mid-turn, closed) is found again. */ @@ -91,10 +108,9 @@ export function structuredSessionOwnedMailboxes(sessionId: string, db: Orchestra if (db.getUnreadDirectMessageTypes(party.address).length > 0) { mailboxes.push(party.address) } - // Why: Dispatch mail parked on a session a /clear ended waits for an edge that never comes. - const dispatch = party.terminalHandle - ? db.findActiveDispatchForAssignee(party.terminalHandle, party.paneKey ?? undefined) - : undefined + // Why: Dispatch mail parked on a session a /clear or restart ended waits for an edge that never + // comes. A party's address is what its Dispatch names it by: a worker's handle, a chat's root. + const dispatch = db.findActiveDispatchForAssignee(party.address, party.paneKey ?? undefined) if (dispatch) { mailboxes.push(`dispatch:${dispatch.id}`) } diff --git a/src/main/runtime/rpc/methods/orchestration-chat-assignee-journal.ts b/src/main/runtime/rpc/methods/orchestration-chat-assignee-journal.ts new file mode 100644 index 000000000000..3dc39e31e152 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-chat-assignee-journal.ts @@ -0,0 +1,41 @@ +import type { AgentType } from '../../../../shared/native-chat-types' +import { chatAssigneeSessionId } from '../../orchestration/chat-assignee' +import { OrchestrationError } from '../../orchestration/orchestration-error' +import type { OrchestrationDb } from '../../orchestration/db' +import { readAgentSessionRecordStore } from '../../orchestration/structured-session-lineage' +import type { StructuredWorkerIdentity } from '../../structured-worker-identity' + +/** What a worker transcript read needs of the session it reads, and what its cursor is bound to. */ +export type StructuredJournalSource = Pick< + StructuredWorkerIdentity, + 'sessionId' | 'processIncarnation' | 'paneKey' +> + +/** + * A chat assignee's journal, read as a worker's is: its whole `/clear` lineage from its root, which + * is also what the cursor is bound to. Null when the assignee is not a chat. + */ +export function chatAssigneeJournal( + db: OrchestrationDb, + dispatchId: string +): { source: StructuredJournalSource; agent: AgentType } | null { + const handle = + db.getWorkerDispatch(dispatchId)?.agent_terminal_handle ?? + db.getDispatchContextById(dispatchId)?.assignee_handle + const chat = chatAssigneeSessionId(handle) + if (!handle || !chat) { + return null + } + const root = readAgentSessionRecordStore()?.getRecord(chat) + if (!root) { + throw new OrchestrationError( + 'transcript_required', + `The transcript for Dispatch ${dispatchId} could not be read; this host has no record of its chat.` + ) + } + // The chat's address stands in for a worker's pane and incarnation: it is its whole identity. + return { + source: { sessionId: chat, processIncarnation: handle, paneKey: handle }, + agent: root.provider + } +} diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-lifecycle.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-lifecycle.ts index 216763b73d1b..73cf06277692 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-lifecycle.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-lifecycle.ts @@ -47,6 +47,10 @@ import type { StructuredWorkerIdentity } from '../../structured-worker-identity' import type { WorkerTerminalReleaseState } from '../../orchestration/worker-terminal-ownership' import { releaseStructuredWorkerSession } from './orchestration-structured-worker-session' import { closeStructuredAgentSessionChild } from '../../structured-agent-session-close' +import { + chatAssigneeJournal, + type StructuredJournalSource +} from './orchestration-chat-assignee-journal' export { observeStructuredWorker, structuredWorkerOwned, type StructuredWorkerObservation } @@ -129,7 +133,10 @@ export async function readStructuredWorkerOutput(args: { limit?: number }): Promise { const identity = resolveStructuredWorkerForDispatch(args.db, args.dispatchId) - if (!identity) { + const journal = identity + ? { source: identity, agent: structuredWorkerAgent(identity) } + : chatAssigneeJournal(args.db, args.dispatchId) + if (!journal) { return null } if (args.source === 'terminal') { @@ -141,11 +148,11 @@ export async function readStructuredWorkerOutput(args: { ) } return readStructuredWorkerJournal({ - identity, + identity: journal.source, dispatchId: args.dispatchId, workerState: args.workerState, liveness: args.liveness, - agent: structuredWorkerAgent(identity), + agent: journal.agent, ...(args.cursor === undefined ? {} : { cursor: args.cursor }), ...(args.limit === undefined ? {} : { limit: args.limit }) }) @@ -153,7 +160,7 @@ export async function readStructuredWorkerOutput(args: { /** Journal page in the shape `worker-read --source transcript` already serves. */ export async function readStructuredWorkerJournal(args: { - identity: StructuredWorkerIdentity + identity: StructuredJournalSource dispatchId: string workerState: string liveness: StructuredWorkerObservation['status'] @@ -223,7 +230,7 @@ export async function readStructuredWorkerJournal(args: { * The oldest item stays in the anchor as the window-slide detector: a slide shifts every index. */ function structuredJournalPrefixIdentity(args: { - identity: StructuredWorkerIdentity + identity: StructuredJournalSource page: StructuredLineageJournalPage position: number }): string { diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts index 02797d54d6c8..390148eb6c4b 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts @@ -1,6 +1,7 @@ import { join } from 'node:path' import { beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentJournalSubmission } from '../../../../shared/agent-session-journal-types' +import type { AgentMessageSource } from '../../../../shared/agent-session-message-source' import { getAppEnvironment } from '../../../../shared/app-environment' import { DISPATCH_REJECTED_WRITE_FAILED } from '../../../../shared/structured-agent-session-dispatch-rejection' @@ -270,8 +271,25 @@ describe('structured worker dispatch preamble', () => { } } + const TASK_FROM: AgentMessageSource = { + kind: 'agent', + senders: [ + { + party: { address: 'term_coord', terminalHandle: 'term_coord', orcaSessionId: null }, + name: 'Coordinator' + } + ], + orchestration: { message: 'task', runId: 'r1', taskId: 't1', dispatchId: 'd1' } + } + const send = (host: PreambleHost) => - sendStructuredWorkerPreamble({ host, sessionId: 's1', dispatchId: 'd1', preamble: 'spec' }) + sendStructuredWorkerPreamble({ + host, + sessionId: 's1', + dispatchId: 'd1', + preamble: 'spec', + from: TASK_FROM + }) // No source and no person: a restart or a close rejects it, and orchestration re-derives it. it('sends the preamble as no person’s', async () => { @@ -282,6 +300,13 @@ describe('structured worker dispatch preamble', () => { expect(sent.mock.calls[0]?.[1]).not.toHaveProperty('userSend') }) + it('names who the task is from on the turn it sends', async () => { + const host = hostWithSubmission({ dispatchState: 'accepted', reason: null }) + const sent = vi.spyOn(host, 'send') + await send(host) + expect(sent.mock.calls[0]?.[1]).toMatchObject({ body: { from: TASK_FROM } }) + }) + it('reports the preamble delivered only on an accepted submission', async () => { await expect( send(hostWithSubmission({ dispatchState: 'accepted', reason: null })) diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts index 8bec6cfee309..3b1bc09f51ec 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts @@ -13,6 +13,7 @@ import { randomUUID } from 'node:crypto' import { isDefinitiveAgentSessionCreateRefusal } from '../../../../shared/agent-session-definitive-refusal' +import type { AgentMessageSource } from '../../../../shared/agent-session-message-source' import type { AgentJournalMessageItem, AgentJournalSubmission @@ -227,11 +228,14 @@ export async function sendStructuredWorkerPreamble(args: { sessionId: string dispatchId: string preamble: string + /** Who the task is from (`dispatchTaskSource`), shown on the worker's turn. */ + from: AgentMessageSource }): Promise<'accepted' | 'pending'> { const body: AgentJournalMessageItem = { kind: 'message', role: 'user', - blocks: [{ type: 'text', text: args.preamble }] + blocks: [{ type: 'text', text: args.preamble }], + from: args.from } const fence = args.host.deps.store.getRecord(args.sessionId)?.lease.runtimeFence if (fence === undefined) { @@ -260,7 +264,8 @@ export async function sendStructuredWorkerPreamble(args: { } } -function preambleDispatchState( +/** A sent preamble's verdict: acknowledged, or held for its agent; anything else throws. */ +export function preambleDispatchState( submission: AgentJournalSubmission | undefined ): 'accepted' | 'pending' { if (submission?.dispatchState === 'accepted' || submission?.dispatchState === 'pending') { diff --git a/src/main/runtime/rpc/methods/orchestration/chat-assignee-admission.ts b/src/main/runtime/rpc/methods/orchestration/chat-assignee-admission.ts new file mode 100644 index 000000000000..7717413a9af6 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/chat-assignee-admission.ts @@ -0,0 +1,83 @@ +import type { AgentSessionRecord } from '../../../../../shared/agent-session-record' +import { + formatOrcaSessionAddress, + type OrcaSessionId +} from '../../../../../shared/orca-session-address' +import { ORCHESTRATION_SESSION_CALLER_ERROR_CODES as CODES } from '../../../../../shared/orchestration-session-caller-codes' +import type { OrcaRuntimeService } from '../../../orca-runtime' +import type { WorkerStartModeReceipt } from '../orchestration-worker-start-mode' +import { observeChatAssignee } from '../../../orchestration/chat-assignee' +import type { OrchestrationDb } from '../../../orchestration/db' +import { OrchestrationError } from '../../../orchestration/orchestration-error' +import type { OrchestrationParty } from '../../../orchestration/orchestration-party' +import { readAgentSessionRecordStore } from '../../../orchestration/structured-session-lineage' +import { refuseUndeliverableSessionRecipient } from './messaging/session-recipient' + +const NO_EFFECTS = 'No effects were applied.' + +/** A chat has no pane or process; its Dispatch names it by its Orca session ID alone. */ +export const CHAT_WORKER_AUTHORITY = { paneKey: null, processIncarnation: null } as const + +/** The chat a Dispatch assignee party names; null for a terminal or a structured worker. */ +export function chatAssigneeOf(party: OrchestrationParty): OrcaSessionId | null { + return party.terminalHandle === null ? party.orcaSessionId : null +} + +/** + * Refuses a chat assignee before anything is created, by the same reach rules as mail to it: its + * task and the coordinator's mail reach it the same way. Returns the session running it now. + */ +export async function admitChatAssignee( + runtime: Pick, + sessionId: OrcaSessionId, + db: OrchestrationDb +): Promise { + // A failed install leaves no store, which the refusal below names. + await runtime.ensureStructuredAgentSessionHost().catch(() => undefined) + const store = readAgentSessionRecordStore() + const refusal = refuseUndeliverableSessionRecipient( + { sessionId, address: formatOrcaSessionAddress(sessionId) }, + store, + db, + NO_EFFECTS + ) + if (refusal) { + throw new OrchestrationError(refusal.code, refusal.message, { effectsApplied: false }) + } + const observed = observeChatAssignee(sessionId, db, store) + if (observed.status !== 'live') { + throw new OrchestrationError( + CODES.notLive, + `Agent session ${sessionId} cannot take a task here: ${observed.reason} ${NO_EFFECTS}`, + { effectsApplied: false } + ) + } + return observed.session +} + +/** A coordinator can't be its own worker: it would answer its own task forever. */ +export function refuseChatSelfAssignment(args: { + sessionId: OrcaSessionId + coordinatorSessionId: OrcaSessionId | null | undefined + remedy: string +}): void { + if (args.sessionId === args.coordinatorSessionId) { + throw new OrchestrationError( + 'terminal_is_coordinator', + `${formatOrcaSessionAddress(args.sessionId)} is this coordinator's own Orca session ID. ${args.remedy}` + ) + } +} + +/** A chat adopted as a worker is given the task, not launched; the decided mode stands. */ +export function chatWorkerMode( + mode: WorkerStartModeReceipt, + chat: OrcaSessionId | null +): WorkerStartModeReceipt { + return chat + ? { + ...mode, + detail: `--terminal names the chat ${formatOrcaSessionAddress(chat)}; its task goes to that chat and no agent is launched.` + } + : mode +} diff --git a/src/main/runtime/rpc/methods/orchestration/chat-task-delivery.ts b/src/main/runtime/rpc/methods/orchestration/chat-task-delivery.ts new file mode 100644 index 000000000000..8070bd2508f7 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/chat-task-delivery.ts @@ -0,0 +1,85 @@ +/** + * A chat assignee's task, sent as any agent's message into a chat is: the composer's queued send, + * held as a card while the chat is busy, recording the Dispatch it is from. The operation id is + * derived from the Dispatch, so a second send for one Dispatch replays instead of queueing twice. + */ + +import { createHash } from 'node:crypto' +import type { AgentJournalMessageItem } from '../../../../../shared/agent-session-journal-types' +import type { AgentMessageSource } from '../../../../../shared/agent-session-message-source' +import { getStructuredAgentSessionHost } from '../../../../native-chat/agent-session-wire/structured-agent-session-registry' +import { chatAssigneeSessionId, observeChatAssignee } from '../../../orchestration/chat-assignee' +import type { OrchestrationDb } from '../../../orchestration/db' +import { exposeUtcTimestamp } from '../../../orchestration/db/utc-timestamp' +import { OrchestrationError } from '../../../orchestration/orchestration-error' +import { sendAgentTurn } from '../../../orchestration/send-agent-turn' +import { structuredPointerCallerKey } from '../../../orchestration/structured-mailbox-pointer-host' +import type { DispatchContextRow } from '../../../orchestration/types' +import { preambleDispatchState } from '../orchestration-structured-worker-session' + +/** Handed over in every case: started, still starting its agent, or held as a card. */ +export type ChatTaskDelivery = 'accepted' | 'pending' | 'queued' + +type ChatTaskDispatch = Pick + +/** The host's `<13-digit ms>-<32 hex>` shape, stamped at the Dispatch's creation. */ +export function chatTaskOperationId( + dispatch: Pick +): string { + const createdAt = Date.parse(exposeUtcTimestamp(dispatch.created_at) ?? dispatch.created_at) + if (!Number.isSafeInteger(createdAt) || createdAt < 0) { + throw new Error(`Dispatch ${dispatch.id} has an unreadable creation time.`) + } + const digest = createHash('sha256').update(dispatch.id).digest('hex').slice(0, 32) + return `${String(createdAt).padStart(13, '0')}-${digest}` +} + +/** Sends the task into the chat's live session. Throws when nothing was handed over. */ +export async function sendChatTask(args: { + db: OrchestrationDb + dispatch: ChatTaskDispatch + /** Who the task is from (`dispatchTaskSource`), shown on the chat's card or turn. */ + from: AgentMessageSource + preamble: string +}): Promise { + const sessionId = chatAssigneeSessionId(args.dispatch.assignee_handle) + const host = getStructuredAgentSessionHost() + const observed = + sessionId && host ? observeChatAssignee(sessionId, args.db, host.deps.store) : null + if (!host || observed?.status !== 'live') { + throw new OrchestrationError( + 'dispatch_preamble_undelivered', + `The task was not delivered: ${observed && observed.status !== 'live' ? observed.reason : 'the chat cannot be reached here.'}` + ) + } + const body: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: args.preamble }], + from: args.from + } + const outcome = await sendAgentTurn({ + kind: 'structured-session', + host, + sessionId: observed.session.sessionId, + callerKey: structuredPointerCallerKey(args.dispatch.id), + turn: { + body, + // As a person's message is: a busy chat queues it as a card, sent when the queue reaches it. + delivery: 'queue', + operationId: chatTaskOperationId(args.dispatch), + expectedRuntimeFence: observed.session.lease.runtimeFence + } + }) + switch (outcome.kind) { + case 'refused': + throw new OrchestrationError( + 'dispatch_preamble_undelivered', + `The task was refused: ${outcome.refusal.message}` + ) + case 'queued': + return 'queued' + case 'sent': + return preambleDispatchState(outcome.submission) + } +} diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts b/src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts index df6886448503..ab95c74f889f 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts @@ -227,7 +227,7 @@ export function applyFederatedFleetObservations( } // Recompute every projection derived from the host's verdict. -function refreshFleetWorkerVerdict( +export function refreshFleetWorkerVerdict( worker: OrchestrationFleetPage['workers'][number], durable: ReadonlyMap ): void { diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/session-recipient.ts b/src/main/runtime/rpc/methods/orchestration/messaging/session-recipient.ts index 58c4a0e4ba73..b5325d70b25b 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/session-recipient.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/session-recipient.ts @@ -67,40 +67,44 @@ export function readSessionRecipient( : null } -/** Null when mail to this session can be stored and delivered here; otherwise why not. */ +/** + * Null when mail to this session can be stored and delivered here; otherwise why not. A Dispatch + * to a chat asks the same question, since its task and its mail reach the chat the same way. + */ export function refuseUndeliverableSessionRecipient( recipient: SessionRecipient, store: AgentSessionRecordReader | null, - db: OrchestrationDb + db: OrchestrationDb, + noEffect = 'No message was sent.' ): SessionRecipientRefusal | null { const { sessionId } = recipient if (!store) { return { code: CODES.unknown, - message: `Agent session ${sessionId} cannot be verified: this Orca is not running its agent-session host. No message was sent.` + message: `Agent session ${sessionId} cannot be verified: this Orca is not running its agent-session host. ${noEffect}` } } const found = lookupOrcaAgentSession(store, sessionId) if (found.kind === 'provider-id') { - return providerIdRefusal(sessionId, found.orcaSessionId, store) + return providerIdRefusal(sessionId, found.orcaSessionId, store, noEffect) } if (found.kind === 'unknown') { return { code: CODES.unknown, - message: `No Orca agent session ${sessionId} exists on this host. No message was sent.` + message: `No Orca agent session ${sessionId} exists on this host. ${noEffect}` } } const reach = structuredSessionMailReach(store, found.record, db) if (reach.kind === 'other-host') { return { code: CODES.hostBoundary, - message: `Agent session ${sessionId} runs on another host; mail reaches a session only on the host that runs it. Send from that host. No message was sent.` + message: `Agent session ${sessionId} runs on another host; mail reaches a session only on the host that runs it. Send from that host. ${noEffect}` } } if (reach.kind === 'unverifiable') { return { code: CODES.notLive, - message: `The session continuing agent session ${sessionId} after a /clear cannot be verified: ${reach.reason} No message was sent.` + message: `The session continuing agent session ${sessionId} after a /clear cannot be verified: ${reach.reason} ${noEffect}` } } if (reach.kind === 'ended') { @@ -108,8 +112,8 @@ export function refuseUndeliverableSessionRecipient( code: CODES.notLive, message: reach.reason === 'worker-identity-lost' - ? `Agent session ${sessionId} is a structured worker whose worker identity this host no longer has, so it can never read that mail. No message was sent.` - : `Agent session ${sessionId} has ended: its chat was closed. No message was sent.` + ? `Agent session ${sessionId} is a structured worker whose worker identity this host no longer has, so it can never read that mail. ${noEffect}` + : `Agent session ${sessionId} has ended: its chat was closed. ${noEffect}` } } return null @@ -118,7 +122,8 @@ export function refuseUndeliverableSessionRecipient( function providerIdRefusal( id: string, orcaSessionId: string, - store: AgentSessionRecordReader | null + store: AgentSessionRecordReader | null, + noEffect = 'No message was sent.' ): SessionRecipientRefusal { // The conversation's Orca session ID, which a `/clear`ed session keeps; not the live session's. const root = isOrcaSessionId(orcaSessionId) @@ -127,6 +132,6 @@ function providerIdRefusal( const address = `${ORCA_SESSION_ADDRESS_PREFIX}${root}` return { code: CODES.providerId, - message: `${id} is the provider's own session id, which changes on /clear. This session's Orca session ID is ${address}; address it by that instead. No message was sent.` + message: `${id} is the provider's own session id, which changes on /clear. This session's Orca session ID is ${address}; address it by that instead. ${noEffect}` } } diff --git a/src/main/runtime/rpc/methods/orchestration/runs/dispatch-methods.ts b/src/main/runtime/rpc/methods/orchestration/runs/dispatch-methods.ts index 43686705b013..6834aeb22b9f 100644 --- a/src/main/runtime/rpc/methods/orchestration/runs/dispatch-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/runs/dispatch-methods.ts @@ -14,6 +14,13 @@ import { orcaSessionIdOrHandle, resolveDispatchAssigneeParty } from '../../../../orchestration/orchestration-party' +import { + admitChatAssignee, + chatAssigneeOf, + refuseChatSelfAssignment +} from '../chat-assignee-admission' +import { sendChatTask } from '../chat-task-delivery' +import { dispatchTaskSource } from '../../../../orchestration/dispatch-task-source' export const ORCHESTRATION_DISPATCH_METHODS = [ defineMethod({ @@ -49,7 +56,13 @@ export const ORCHESTRATION_DISPATCH_METHODS = [ runId: run.id }) } - const assignee = params.to ? resolveDispatchAssigneeParty(params.to, db).address : undefined + const assigneeParty = params.to ? resolveDispatchAssigneeParty(params.to, db) : undefined + const assignee = assigneeParty?.address + // A chat has no pane, process or agent to probe: it is reached as mail to it is. + const chatAssignee = assigneeParty ? chatAssigneeOf(assigneeParty) : null + if (chatAssignee) { + await admitChatAssignee(runtime, chatAssignee, db) + } // Why: dry-run previews the preamble without mutating state, so it skips the ready-status check and uses a placeholder dispatchId. if (params.dryRun) { @@ -84,9 +97,10 @@ export const ORCHESTRATION_DISPATCH_METHODS = [ ) } - const dispatchAuthority = runtime.getOrchestrationDispatchAuthority(to) - const assigneePaneKey = - dispatchAuthority?.paneKey ?? runtime.getTerminalPaneKey(to) ?? undefined + const dispatchAuthority = chatAssignee ? null : runtime.getOrchestrationDispatchAuthority(to) + const assigneePaneKey = chatAssignee + ? undefined + : (dispatchAuthority?.paneKey ?? runtime.getTerminalPaneKey(to) ?? undefined) const processIncarnation = dispatchAuthority?.paneKey && dispatchAuthority.processIncarnation ? dispatchAuthority.processIncarnation @@ -98,6 +112,14 @@ export const ORCHESTRATION_DISPATCH_METHODS = [ runtime.getTerminalPaneKey(params.from) ?? null) : null + if (params.inject && chatAssignee) { + // The session wins over `--from`, so a chat calling with none is still refused here. + refuseChatSelfAssignment({ + sessionId: chatAssignee, + coordinatorSessionId: orchestrationCaller?.orcaSessionId, + remedy: 'Dispatch to a different agent, or use worker-start to create one.' + }) + } if ( params.inject && params.from && @@ -113,14 +135,14 @@ export const ORCHESTRATION_DISPATCH_METHODS = [ } // Why: injecting the preamble into a bare shell dumps it as shell commands (gibberish), so require a detected agent first. - if (params.inject) { + if (params.inject && !chatAssignee) { const hasAgent = await runtime.isTerminalRunningAgent(to) if (!hasAgent) { throw injectRejectedError(to, 'no_agent_detected') } } - if (params.inject && (!assigneePaneKey || !processIncarnation)) { + if (params.inject && !chatAssignee && (!assigneePaneKey || !processIncarnation)) { throw new OrchestrationError( 'stable_pane_required', `Terminal ${to} has no stable pane/process incarnation for lifecycle authority.` @@ -154,19 +176,34 @@ export const ORCHESTRATION_DISPATCH_METHODS = [ let prompt if (params.inject) { try { - prompt = await sendAgentTurn({ - kind: 'terminal', - runtime, - handle: to, - turn: { - purpose: 'dispatch-preamble', - body: preamble, - operationId: orchestrationMutation?.requestId ?? ctx.id - } - }) + if (chatAssignee) { + const from = dispatchTaskSource({ + db, + dispatch: ctx, + from: params.from, + senderName: (party, reported) => + runtime.orchestrationSenderNames.nameOf(party, reported) + }) + await sendChatTask({ db, dispatch: ctx, from, preamble }) + } else { + prompt = await sendAgentTurn({ + kind: 'terminal', + runtime, + handle: to, + turn: { + purpose: 'dispatch-preamble', + body: preamble, + operationId: orchestrationMutation?.requestId ?? ctx.id + } + }) + } injected = true } catch (err) { - db.failDispatch(ctx.id, err instanceof Error ? err.message : String(err)) + // Unknown is not undelivered: the chat may be running it, so the Dispatch stays open, as + // an unknown worker-start does, for the worker's report or the coordinator's abandon. + if (!(err instanceof OrchestrationError && err.code === 'operation_unknown')) { + db.failDispatch(ctx.id, err instanceof Error ? err.message : String(err)) + } throw err } } diff --git a/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts b/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts index 7234da7be4bb..64b7b255c48e 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts @@ -11,6 +11,11 @@ import { buildDispatchPreamble } from '../../../../orchestration/preamble' import { sendAgentTurn } from '../../../../orchestration/send-agent-turn' import { createWorkerBriefWriteGuard } from '../../../../launched-agent-write-guard' import { sendStructuredWorkerPreamble } from '../../orchestration-structured-worker-session' +import { chatAssigneeSessionId } from '../../../../orchestration/chat-assignee' +import { OrchestrationError } from '../../../../orchestration/orchestration-error' +import { sendChatTask, type ChatTaskDelivery } from '../chat-task-delivery' +import { dispatchTaskSource } from '../../../../orchestration/dispatch-task-source' +import type { AgentMessageSource } from '../../../../../../shared/agent-session-message-source' import type { WorkerTurnStartObservation } from './worker-start-turn-observation' import type { createStructuredWorkerSessionForWorktree } from './worker-topology' @@ -23,7 +28,8 @@ type StructuredSession = Awaited args.runtime.orchestrationSenderNames.nameOf(party, reported) + }) +} diff --git a/src/main/runtime/rpc/methods/orchestration/worker/explicit-worker-terminal-validation.ts b/src/main/runtime/rpc/methods/orchestration/worker/explicit-worker-terminal-validation.ts index f9b9e195e21a..a3ee27827f5b 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/explicit-worker-terminal-validation.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/explicit-worker-terminal-validation.ts @@ -2,6 +2,15 @@ import type { OrcaRuntimeService } from '../../../../orca-runtime' import { OrchestrationError } from '../../../../orchestration/orchestration-error' import { isStructuredWorkerHandle } from '../../../../structured-worker-identity' import type { OrchestrationCallerIdentity } from '../../../../orchestration/orchestration-caller-identity' +import type { OrchestrationDb } from '../../../../orchestration/db' +import type { OrcaSessionId } from '../../../../../../shared/orca-session-address' +import { formatOrcaSessionAddress } from '../../../../../../shared/orca-session-address' +import { + admitChatAssignee, + chatAssigneeOf, + refuseChatSelfAssignment +} from '../chat-assignee-admission' +import type { OrchestrationParty } from '../../../../orchestration/orchestration-party' /** * Admits a caller-supplied `--terminal` as this dispatch's worker pane. @@ -52,3 +61,40 @@ export async function assertExplicitWorkerTerminalUsable(args: { ) } } + +/** A `--terminal` party: a terminal pane, or a chat named by its Orca session ID. */ +export async function assertExplicitWorkerUsable(args: { + runtime: OrcaRuntimeService + db: OrchestrationDb + terminal: OrchestrationParty + from: string + coordinator: OrchestrationCallerIdentity | null + resolvedWorktreeId: string | undefined +}): Promise { + const chat = chatAssigneeOf(args.terminal) + await (chat + ? assertExplicitWorkerChatUsable({ ...args, chat }) + : assertExplicitWorkerTerminalUsable({ ...args, terminal: args.terminal.address })) +} + +/** The same refusals for a chat named by its Orca session ID: itself, unreachable, another worktree. */ +async function assertExplicitWorkerChatUsable(args: { + runtime: Pick + db: OrchestrationDb + chat: OrcaSessionId + coordinator: OrchestrationCallerIdentity | null + resolvedWorktreeId: string | undefined +}): Promise { + refuseChatSelfAssignment({ + sessionId: args.chat, + coordinatorSessionId: args.coordinator?.orcaSessionId, + remedy: 'Pass --terminal for a different agent, or omit it so worker-start creates one.' + }) + const session = await admitChatAssignee(args.runtime, args.chat, args.db) + if (session.location.workspaceId !== args.resolvedWorktreeId) { + throw new OrchestrationError( + 'terminal_worktree_mismatch', + `${formatOrcaSessionAddress(args.chat)} does not belong to worktree ${args.resolvedWorktreeId}.` + ) + } +} diff --git a/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts index 7a6168dbcd20..b46e8d744265 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts @@ -26,7 +26,9 @@ import { } from './worker-setup-gate' import { failWorkerStartWithReceipt } from './worker-start-receipt' import { parseTaskDeps } from './task-deps-argument' -import { assertExplicitWorkerTerminalUsable } from './explicit-worker-terminal-validation' +import { assertExplicitWorkerUsable } from './explicit-worker-terminal-validation' +import { resolveDispatchAssigneeParty } from '../../../../orchestration/orchestration-party' +import { CHAT_WORKER_AUTHORITY, chatAssigneeOf, chatWorkerMode } from '../chat-assignee-admission' import { recordCreatedWorkerTerminalCustody } from './created-worker-terminal-custody' import { tearDownFailedWorkerStart } from './failed-worker-start-teardown' import { requireWorkerAuthority, type WorkerEffect } from './worker-topology' @@ -52,8 +54,12 @@ export async function startLocalWorker(args: { /** Settings-driven; the executing host still gets to refuse below. */ mode: WorkerStartModeReceipt }): Promise { - const { params, runtime, db, run, coordinator, callerSession, existingTask } = args + const { runtime, db, run, coordinator, callerSession, existingTask } = args const { orchestrationMutation } = args + // `--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 + const chat = terminal ? chatAssigneeOf(terminal) : null const coordinatorPane = coordinator?.paneKey ?? null const requestedWorktree = params.worktree ?? 'current' const createsWorktree = requestedWorktree === 'new-child' || requestedWorktree === 'new-top-level' @@ -113,16 +119,18 @@ export async function startLocalWorker(args: { : requestedWorktree === 'current' ? await runtime.showManagedTerminalWorkspace(`id:${coordinatorWorktreeId}`) : await runtime.showManagedTerminalWorkspace(requestedWorktree) - if (params.terminal) { - await assertExplicitWorkerTerminalUsable({ + if (terminal) { + await assertExplicitWorkerUsable({ runtime, - terminal: params.terminal, + db, + terminal, from: params.from, coordinator, resolvedWorktreeId: resolvedWorktree?.id }) } - let mode = await resolveWorkerStartModeOnHost(runtime, args.mode, resolvedWorktree?.id, agent) + const hostMode = resolveWorkerStartModeOnHost(runtime, args.mode, resolvedWorktree?.id, agent) + let mode = chatWorkerMode(await hostMode, chat) const startOptions = { worktree: requestedWorktree, @@ -225,11 +233,14 @@ export async function startLocalWorker(args: { effects, timeoutMs: params.timeoutMs ?? 60_000 }) - : await waitForWorkerAgentReady(runtime, terminalHandle, { - agent, - reusesTerminal: Boolean(params.terminal), - timeoutMs: params.timeoutMs ?? 60_000 - }) + : chat + ? // A chat takes its task as a queued send whatever it is doing, so nothing is waited on. + null + : await waitForWorkerAgentReady(runtime, terminalHandle, { + agent, + reusesTerminal: Boolean(params.terminal), + timeoutMs: params.timeoutMs ?? 60_000 + }) if (wait) { persistWorkerSetupWaitOutcome({ ...setupStage, wait }) if (!wait.satisfied) { @@ -245,7 +256,10 @@ export async function startLocalWorker(args: { ) } } - const terminalAuthority = requireWorkerAuthority(runtime, terminalHandle) + // A chat has no pane or process; its Dispatch names it by its Orca session ID alone. + const terminalAuthority = chat + ? CHAT_WORKER_AUTHORITY + : requireWorkerAuthority(runtime, terminalHandle) db.prepareStartingWorkerAuthority({ dispatchId: started.dispatch.id, handle: terminalHandle, diff --git a/src/main/runtime/rpc/methods/orchestration/worker/session-worker-observation.ts b/src/main/runtime/rpc/methods/orchestration/worker/session-worker-observation.ts new file mode 100644 index 000000000000..62f5635847bf --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/worker/session-worker-observation.ts @@ -0,0 +1,91 @@ +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import type { OrcaSessionId } from '../../../../../../shared/orca-session-address' +import { + chatAssigneeSessionId, + observeChatAssignee, + type ChatAssigneeObservation +} from '../../../../orchestration/chat-assignee' +import type { OrchestrationDb } from '../../../../orchestration/db' +import { + holdStructuredWorker, + observeStructuredSession +} from '../../../../structured-worker-authority' +import { resolveStructuredWorkerForDispatch } from '../../orchestration-structured-worker-lifecycle' + +export type SessionWorkerObservation = { + terminal: null + exact: boolean + status: 'identity_changed' | 'live' | 'exited' | 'unverifiable' + reason?: string + addressable?: boolean + terminalHandle: null +} + +/** + * A worker that is an agent session, a structured worker or a chat, observed off the session + * records instead of a terminal; null for a PTY worker. + */ +export async function inspectSessionWorker( + runtime: Pick, + db: OrchestrationDb, + dispatchId: string, + terminalHandle: string +): Promise { + const chat = chatAssigneeSessionId(terminalHandle) + if (chat) { + // The address is the chat's whole identity, so it is always exact. After a restart nothing + // has installed the host yet, and its records are what this reads. + await runtime.ensureStructuredAgentSessionHost().catch(() => undefined) + const observed = observeChatAssignee(chat, db) + return { + terminal: null, + exact: true, + status: observed.status, + ...(observed.status === 'unverifiable' ? { reason: observed.reason } : {}), + terminalHandle: null + } + } + const structured = resolveStructuredWorkerForDispatch(db, dispatchId) + if (!structured) { + return null + } + // Exactness is the recorded pane and lineage, which the runtime getters answer from the + // structured registry; there is no terminal to show. + // + // `agentWait` is deliberately ABSENT rather than null. Null is the contract's "Orca looked and + // found no wait", and nothing here looks: a structured worker parks on a journal question item, + // which no terminal prompt scan can see. Reporting null would tell a coordinator the worker is + // not waiting, which is the one thing the field's own documentation forbids inferring. + const exact = db.isDispatchProcessCurrent({ + dispatchId, + paneKey: structured.paneKey, + processIncarnation: structured.processIncarnation + }) + // One lineage walk, so status and `addressable` judge the same session even mid-`/clear`. + const hold = holdStructuredWorker(structured, db) + const observation = + hold.kind === 'unverifiable' + ? { status: 'unverifiable' as const, reason: hold.reason } + : observeStructuredSession(hold.running.sessionId) + const addressable = hold.kind === 'unverifiable' ? null : hold.kind === 'held' + return { + terminal: null, + exact, + status: exact ? observation.status : 'identity_changed', + ...(exact && observation.reason ? { reason: observation.reason } : {}), + ...(exact && addressable !== null ? { addressable } : {}), + terminalHandle: null + } +} + +/** A fleet page's chat verdicts, read off the session records a restart leaves uninstalled. */ +export async function chatAssigneeObserver( + runtime: Pick, + db: OrchestrationDb, + rows: readonly { agentTerminalHandle: string | null }[] +): Promise<(sessionId: OrcaSessionId) => ChatAssigneeObservation> { + if (rows.some((row) => chatAssigneeSessionId(row.agentTerminalHandle))) { + await runtime.ensureStructuredAgentSessionHost().catch(() => undefined) + } + return (sessionId) => observeChatAssignee(sessionId, db) +} diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts index bf2fef1bd702..9c77fd8895fe 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-method.ts @@ -21,6 +21,7 @@ import { readWorkerListSnapshot } from './worker-list-snapshot-store' import { projectWorkerFleet, type WorkerListPageParams } from './worker-list-projection' +import { chatAssigneeObserver } from './session-worker-observation' import { exposeWorkerTerminalResource } from './worker-release-completion' import { WORKER_TERMINAL_LIST_STATES, WorkerListParams } from './worker-release-schemas' @@ -203,7 +204,8 @@ async function projectWorkerListPageWithFilteredSnapshot( statuses, limit, now: authorityNow, - completeProjection: args.completeProjection + completeProjection: args.completeProjection, + observeChat: await chatAssigneeObserver(runtime, db, pageRows) }) const federated = params.includeRemote ? await readFederatedFleetSnapshots({ diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-projection.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-projection.ts index fc3ce32567b3..63006a391245 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-projection.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-projection.ts @@ -1,11 +1,18 @@ import { ORCHESTRATION_FLEET_PAGE_MAX, projectOrchestrationFleet, - type FleetDurableWorker + type FleetDurableWorker, + type OrchestrationFleetWorker } from '../../../../../../shared/orchestration-fleet-projection' import { resolveFleetWorkerOutcome } from '../../../../../../shared/orchestration-fleet-outcome-resolution' import type { WorkerTerminalListState } from '../../../../orchestration/worker-terminal-ownership' import type { OrchestrationDb } from '../../../../orchestration/db' +import type { OrcaSessionId } from '../../../../../../shared/orca-session-address' +import { + chatAssigneeSessionId, + type ChatAssigneeObservation +} from '../../../../orchestration/chat-assignee' +import { refreshFleetWorkerVerdict } from '../federation/federated-fleet-snapshot' export type WorkerListPageParams = { run?: string @@ -21,6 +28,8 @@ export function projectWorkerFleet(args: { limit: number now: number completeProjection?: boolean + /** A chat assignee's verdict; no agent-status row names a chat, so its reach stands in. */ + observeChat?: (sessionId: OrcaSessionId) => ChatAssigneeObservation }) { const workers: FleetDurableWorker[] = args.rows.map((row) => { return { @@ -49,7 +58,7 @@ export function projectWorkerFleet(args: { }) const durable = new Map(workers.map((worker) => [worker.dispatchId, worker])) if (!args.completeProjection) { - return { + const fleet = { ...projectOrchestrationFleet({ workers, statuses: args.statuses, @@ -58,6 +67,8 @@ export function projectWorkerFleet(args: { }), durable } + applyChatAssigneeLiveness(fleet, args.observeChat, args.now) + return fleet } const projections: ReturnType['workers'] = [] @@ -71,9 +82,36 @@ export function projectWorkerFleet(args: { }).workers ) } - return { + const fleet = { workers: projections, page: { limit: workers.length, total: workers.length, hasMore: false, nextCursor: null }, durable } + applyChatAssigneeLiveness(fleet, args.observeChat, args.now) + return fleet +} + +/** Read off the session records, as worker-show and mail read it: at rest is live, closed exited. */ +function applyChatAssigneeLiveness( + fleet: { workers: OrchestrationFleetWorker[]; durable: ReadonlyMap }, + observeChat: ((sessionId: OrcaSessionId) => ChatAssigneeObservation) | undefined, + now: number +): void { + for (const worker of observeChat ? fleet.workers : []) { + const chat = chatAssigneeSessionId(fleet.durable.get(worker.dispatchId)?.agentTerminalHandle) + // A durable death certificate (a stop, a release) outranks a read of the records. + if (!chat || !observeChat || worker.liveness.verdict === 'exited') { + continue + } + const observed = observeChat(chat) + worker.liveness = + observed.status === 'live' + ? { verdict: 'live', observedAt: now, source: 'execution_host' } + : observed.status === 'exited' + ? { verdict: 'exited', source: 'execution_host' } + : { verdict: 'unverifiable', reason: 'host_unavailable' } + worker.evidence.liveStatus = observed.status === 'live' ? 'fresh' : 'unavailable' + worker.evidence.lastObservedAt = observed.status === 'live' ? now : null + refreshFleetWorkerVerdict(worker, fleet.durable) + } } diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts index 40b856ef1657..a47dca5ec937 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts @@ -5,11 +5,8 @@ import { OrchestrationError } from '../../../../orchestration/orchestration-erro import { parseWorkerTerminalHostScope } from '../../../../orchestration/worker-terminal-process-liveness' import type { OrchestrationFleetWorker } from '../../../../../../shared/orchestration-fleet-projection' import { projectWorkerFleet } from './worker-list-projection' -import { resolveStructuredWorkerForDispatch } from '../../orchestration-structured-worker-lifecycle' -import { - holdStructuredWorker, - observeStructuredSession -} from '../../../../structured-worker-authority' +import { observeChatAssignee } from '../../../../orchestration/chat-assignee' +import { inspectSessionWorker } from './session-worker-observation' import type { DispatchContextRow, FederatedDispatchRow, @@ -42,35 +39,9 @@ export async function inspectWorkerTerminal( if (!terminalHandle) { return { terminal: null, exact: false, status: 'unattached', terminalHandle: null } } - const structured = resolveStructuredWorkerForDispatch(db, dispatchId) - if (structured) { - // Exactness is the recorded pane and lineage, which the runtime getters answer from the - // structured registry; there is no terminal to show. - // - // `agentWait` is deliberately ABSENT rather than null. Null is the contract's "Orca looked and - // found no wait", and nothing here looks: a structured worker parks on a journal question item, - // which no terminal prompt scan can see. Reporting null would tell a coordinator the worker is - // not waiting, which is the one thing the field's own documentation forbids inferring. - const exact = db.isDispatchProcessCurrent({ - dispatchId, - paneKey: structured.paneKey, - processIncarnation: structured.processIncarnation - }) - // One lineage walk, so status and `addressable` judge the same session even mid-`/clear`. - const hold = holdStructuredWorker(structured, db) - const observation = - hold.kind === 'unverifiable' - ? { status: 'unverifiable' as const, reason: hold.reason } - : observeStructuredSession(hold.running.sessionId) - const addressable = hold.kind === 'unverifiable' ? null : hold.kind === 'held' - return { - terminal: null, - exact, - status: exact ? observation.status : 'identity_changed', - ...(exact && observation.reason ? { reason: observation.reason } : {}), - ...(exact && addressable !== null ? { addressable } : {}), - terminalHandle: null - } + const session = await inspectSessionWorker(runtime, db, dispatchId, terminalHandle) + if (session) { + return session } let effectiveHandle = terminalHandle let terminal = await runtime.showTerminal(effectiveHandle).catch(() => null) @@ -282,7 +253,8 @@ export function projectFleetWorkerPage( attentionFacts: db.getWorkerAttentionFactsForDispatches([dispatchId], now), statuses: runtime.getOrchestrationFleetAgentStatusSnapshot(), limit: 1, - now + now, + observeChat: (sessionId) => observeChatAssignee(sessionId, db) }) } diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-start-readiness-settlement.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-start-readiness-settlement.ts index 3075483f0b4f..87466499afe3 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-start-readiness-settlement.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-start-readiness-settlement.ts @@ -3,6 +3,7 @@ import type { OrchestrationDb } from '../../../../orchestration/db' import type { RunRow, TaskRow } from '../../../../orchestration/types' import type { WorkerStartModeReceipt } from '../../orchestration-worker-start-mode' import { deliverWorkerDispatchPreamble } from './deliver-worker-dispatch-preamble' +import { chatAssigneeSessionId } from '../../../../orchestration/chat-assignee' import type { OrchestrationWorkerLaunchReceipt } from './worker-launch-preferences' import { describeUnobservedWorkerTurnStart, @@ -55,6 +56,7 @@ export async function deliverAndSettleWorkerStartReadiness(args: { terminalHandle, dispatchId: args.dispatchId, dispatchDepth: args.dispatchDepth, + runId: run.id, taskId: task.id, taskSpec: task.spec, coordinatorHandle: args.coordinatorHandle, @@ -124,8 +126,10 @@ export async function deliverAndSettleWorkerStartReadiness(args: { residualResources: JSON.parse(worker.residual_resources) as unknown[], nextCommands: [ `orca orchestration worker-show --dispatch ${args.dispatchId} --json`, - // A structured worker has no screen to read. - ...(structuredSession ? [] : [`orca terminal read --terminal ${terminalHandle} --screen`]), + // A structured worker or a chat has no screen to read. + ...(structuredSession || chatAssigneeSessionId(terminalHandle) + ? [] + : [`orca terminal read --terminal ${terminalHandle} --screen`]), `orca orchestration worker-abandon --dispatch ${args.dispatchId} --json` ], ...(args.terminalRevealWarning ? { warning: args.terminalRevealWarning } : {}) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/workers.ts b/src/main/runtime/rpc/methods/orchestration/worker/workers.ts index be4e40ab2b64..5dd3f0c598b0 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/workers.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/workers.ts @@ -13,6 +13,8 @@ import { resolveWorkerStartReadinessTimeoutMs } from '../../../../../../shared/orchestration-timing-budgets' import { assertWorkerStartTaskSpecWithinPromptBudget } from './worker-start-prompt-budget' +import { ORCA_SESSION_ADDRESS_PREFIX } from '../../../../../../shared/orca-session-address' +import { ORCHESTRATION_SESSION_CALLER_ERROR_CODES as SESSION_CODES } from '../../../../../../shared/orchestration-session-caller-codes' export const ORCHESTRATION_WORKER_START_METHODS = [ defineMethod({ @@ -54,6 +56,14 @@ export const ORCHESTRATION_WORKER_START_METHODS = [ params, settings: readWorkerStartModeSettings(runtime) }) + if (params.on && params.terminal?.startsWith(ORCA_SESSION_ADDRESS_PREFIX)) { + // Refused here, before any remote call: an Orca session ID names a chat only on its own host. + throw new OrchestrationError( + SESSION_CODES.hostBoundary, + `${params.terminal} names a chat by its Orca session ID, which identifies it only on the host that runs it. Run worker-start on that host. No effects were applied.`, + { effectsApplied: false } + ) + } if (params.on) { // A remote worker is always a terminal agent; the mode receipt rides along so the // coordinator still learns why its structured default did not apply. diff --git a/src/main/runtime/rpc/orchestration-party-addressing.test.ts b/src/main/runtime/rpc/orchestration-party-addressing.test.ts index 0bdcbb0cbcc7..95c85720f78b 100644 --- a/src/main/runtime/rpc/orchestration-party-addressing.test.ts +++ b/src/main/runtime/rpc/orchestration-party-addressing.test.ts @@ -108,7 +108,8 @@ describe('every target param resolves both spellings of a party to one canonical 'orchestration.dispatch', 'orchestration.inbox', 'orchestration.sessionAddress', - 'orchestration.partyLocation' + 'orchestration.partyLocation', + 'orchestration.workerStart' ].sort() ) }) @@ -176,10 +177,31 @@ describe('every target param resolves both spellings of a party to one canonical } ) - it('dispatch: refuses a chat assignee with no row written', async () => { + it('dispatch: a chat named by its session address is the assignee, by its Orca session ID', async () => { const runId = await chatRun() const task = h.db.createTask({ runId, spec: 'work' }) + const { dispatch } = await as(SESSION_X, 'orchestration.dispatch', { + task: task.id, + to: ADDRESS_Z + }) + + expect(dispatch).toMatchObject({ + assignee_handle: ADDRESS_Z, + assignee_orca_session_id: SESSION_Z, + assignee_pane_key: null, + process_incarnation: null + }) + expect(await as(SESSION_Z, 'orchestration.check', { peek: true })).toMatchObject({ + dispatchId: idOf(dispatch) + }) + }) + + it('dispatch: refuses a chat on another host with no row written', async () => { + const runId = await chatRun() + const task = h.db.createTask({ runId, spec: 'work' }) + h.records.set(SESSION_Z, sessionRecord(SESSION_Z, { location: { executionHostId: 'ssh:box' } })) + const response = await call(SESSION_X, 'orchestration.dispatch', { task: task.id, to: ADDRESS_Z @@ -187,11 +209,7 @@ describe('every target param resolves both spellings of a party to one canonical expect(response).toMatchObject({ ok: false, - error: { - code: CODES.chatNotDispatchable, - message: `Agent session ${SESSION_Z} is a chat, and a chat can't receive a dispatch yet. Start a worker with worker-start instead. No effects were applied.`, - data: { effectsApplied: false } - } + error: { code: CODES.hostBoundary, data: { effectsApplied: false } } }) expect(h.db.db.prepare('SELECT COUNT(*) AS n FROM dispatch_contexts').get()).toEqual({ n: 0 }) expect(h.db.getTask(task.id)?.status).toBe('ready') diff --git a/src/main/runtime/rpc/orchestration-session-caller.test.ts b/src/main/runtime/rpc/orchestration-session-caller.test.ts index 38e26730a1e4..ba923c96329a 100644 --- a/src/main/runtime/rpc/orchestration-session-caller.test.ts +++ b/src/main/runtime/rpc/orchestration-session-caller.test.ts @@ -51,7 +51,6 @@ const NAMES_NO_RESOLVED_PARTY: Readonly> = { 'orchestration.run from': 'retired; refused before any handler', 'orchestration.runShow from': 'reads a Run by id; `from` is unused', 'orchestration.dispatchShow from': '`from` only fills the preview preamble text', - 'orchestration.workerStart terminal': 'adopts an existing PTY pane, which a session never has', 'orchestration.federationAttachStart terminal': 'names the remote worker terminal', 'orchestration.workerTerminalUserInput terminal': 'names the worker terminal', 'orchestration.workerTerminalUserInput sessionId': diff --git a/src/main/runtime/runtime-rpc-dispatch-inject-long-poll.test.ts b/src/main/runtime/runtime-rpc-dispatch-inject-long-poll.test.ts new file mode 100644 index 000000000000..e609ef178a69 --- /dev/null +++ b/src/main/runtime/runtime-rpc-dispatch-inject-long-poll.test.ts @@ -0,0 +1,62 @@ +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it, vi } from 'vitest' +import { OrcaRuntimeService } from './orca-runtime' +import { readRuntimeMetadata } from './runtime-metadata' +import { classifyRuntimeLongPoll, OrcaRuntimeRpcServer } from './runtime-rpc' +import { openFramedSession, sleep } from './runtime-rpc-test-harness' + +// `dispatch --inject` into a chat waits for its agent to accept the turn, which can outlast the +// socket's idle timer; keepalives hold the connection open as they do for worker-start. A terminal +// inject doesn't wait, so it stays a short RPC. +describe('orchestration.dispatch --inject as a long poll', () => { + it('is a keepalive-backed wait only when it injects into a chat', () => { + const dispatch = (params: Record) => + classifyRuntimeLongPoll({ + id: 'req_dispatch', + authToken: 'token', + method: 'orchestration.dispatch', + params + }) + expect(dispatch({ task: 'task_1', to: 'orca_session_id:x', inject: true })).toBe('wait') + expect(dispatch({ task: 'task_1', to: 'orca_session_id:x' })).toBeNull() + // A terminal inject writes and returns; it keeps a short-RPC slot. + expect(dispatch({ task: 'task_1', to: 'term_worker', inject: true })).toBeNull() + }) + + it('emits keepalives while an injected dispatch blocks', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-')) + const runtime = new OrcaRuntimeService() + const server = new OrcaRuntimeRpcServer({ runtime, userDataPath, keepaliveIntervalMs: 30 }) + vi.spyOn(server['dispatcher'], 'dispatch').mockImplementation(async (request) => { + await sleep(120) + return { + id: request.id, + ok: true, + result: { dispatch: { id: 'ctx_1' }, injected: true }, + _meta: { runtimeId: runtime.getRuntimeId() } + } + }) + await server.start() + try { + const metadata = readRuntimeMetadata(userDataPath) + const session = openFramedSession(metadata!.transports[0]!.endpoint, { + id: 'req_dispatch', + authToken: metadata!.authToken, + method: 'orchestration.dispatch', + params: { task: 'task_1', to: 'orca_session_id:x', inject: true } + }) + await session.done + + expect( + session.frames.filter((frame) => frame._keepalive === true).length + ).toBeGreaterThanOrEqual(2) + expect(session.frames.filter((frame) => frame.ok !== undefined)).toEqual([ + expect.objectContaining({ id: 'req_dispatch', ok: true }) + ]) + } finally { + await server.stop() + } + }) +}) diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-long-poll.ts b/src/main/runtime/runtime-rpc/runtime-rpc-long-poll.ts index c1ddeb78a6ed..23befc86b8de 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-long-poll.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-long-poll.ts @@ -1,4 +1,5 @@ import type { RpcRequest } from '../rpc/core' +import { ORCA_SESSION_ADDRESS_PREFIX } from '../../../shared/orca-session-address' export const KEEPALIVE_INTERVAL_MS = 10_000 @@ -37,6 +38,21 @@ export function classifyRuntimeLongPoll(request: RpcRequest): RuntimeLongPollCla ) { return 'wait' } + // An injected task into a chat waits for its agent to accept the turn, up to 60 s; the reply + // must outlive the 30 s idle timer or a delivered task reads as a dead runtime. A terminal + // inject doesn't wait, so it keeps its short-RPC slot. + if ( + request.method === 'orchestration.dispatch' && + typeof request.params === 'object' && + request.params !== null && + 'inject' in request.params && + request.params.inject === true && + 'to' in request.params && + typeof request.params.to === 'string' && + request.params.to.startsWith(ORCA_SESSION_ADDRESS_PREFIX) + ) { + return 'wait' + } if (request.method === 'browser.clientHost.attach') { return 'browser-host' } diff --git a/src/main/runtime/structured-agent-session-close.test.ts b/src/main/runtime/structured-agent-session-close.test.ts index 500a6cd1bebd..067e3b3b663e 100644 --- a/src/main/runtime/structured-agent-session-close.test.ts +++ b/src/main/runtime/structured-agent-session-close.test.ts @@ -58,6 +58,9 @@ type HostOptions = { indexThrows?: boolean } +/** The hide a close sends; its hidden notice waits for the close's outcome. */ +const HIDE_PENDING_CLOSE = [SESSION, false, undefined, { deferHiddenNotice: true }] + function installHost(options: HostOptions = {}) { const entry = record(SESSION) const held = new Set([SESSION]) @@ -72,6 +75,7 @@ function installHost(options: HostOptions = {}) { visible.delete(sessionId) } }) + const notifySessionTabHidden = vi.fn() const close = vi.fn(async (sessionId: string) => { if (options.closeThrows) { throw options.closeThrows @@ -102,9 +106,10 @@ function installHost(options: HostOptions = {}) { }, hasSession: (sessionId: string) => held.has(sessionId), setSessionTabVisibility, + notifySessionTabHidden, close } - return { close, setSessionTabVisibility, visible, log } + return { close, setSessionTabVisibility, notifySessionTabHidden, visible, log } } describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { @@ -127,7 +132,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { expect(host.visible.has(SESSION)).toBe(false) expect(retire).toHaveBeenCalledWith(SESSION) // The hide is the only visibility write a settled close performs. - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('restores the tab when the close throws and the child is still there', async () => { @@ -140,9 +145,14 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { expect(outcome.reason).toBe('provider round trip failed') expect(host.visible.has(SESSION)).toBe(true) expect(host.setSessionTabVisibility.mock.calls).toEqual([ - [SESSION, false], + HIDE_PENDING_CLOSE, [SESSION, true, TAB_ID] ]) + // Sent once, after the tab is back: a chat worker's Dispatch reads the chat as open. + expect(host.notifySessionTabHidden.mock.calls).toEqual([[SESSION]]) + expect(host.notifySessionTabHidden.mock.invocationCallOrder[0]).toBeGreaterThan( + host.setSessionTabVisibility.mock.invocationCallOrder[1]! + ) }) it('restores the tab when the post-close observation is not `exited`', async () => { @@ -154,7 +164,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { expect(outcome.closeAttempted).toBe(true) expect(host.visible.has(SESSION)).toBe(true) expect(host.setSessionTabVisibility.mock.calls).toEqual([ - [SESSION, false], + HIDE_PENDING_CLOSE, [SESSION, true, TAB_ID] ]) }) @@ -168,7 +178,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { expect(outcome.stopped).toBe(false) expect(host.visible.has(SESSION)).toBe(false) - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('does not put the tab back when the caller is discarding the workspace anyway', async () => { @@ -183,7 +193,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { expect(outcome.stopped).toBe(false) expect(host.visible.has(SESSION)).toBe(false) - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('does not publish a tab for a session that was already hidden', async () => { @@ -192,7 +202,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { await closeStructuredAgentSessionChild(SESSION) expect(host.visible.has(SESSION)).toBe(false) - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('does not roll back a visibility write that never landed', async () => { @@ -206,7 +216,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { reason: 'visibility write failed' }) expect(host.close).not.toHaveBeenCalled() - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('keeps the original failure when the restore itself throws', async () => { @@ -233,7 +243,7 @@ describe('closeStructuredAgentSessionChild tab-visibility rollback', () => { await closeStructuredAgentSessionChild(SESSION) - expect(host.setSessionTabVisibility.mock.calls).toEqual([[SESSION, false]]) + expect(host.setSessionTabVisibility.mock.calls).toEqual([HIDE_PENDING_CLOSE]) }) it('reports no close attempt when no host is installed', async () => { diff --git a/src/main/runtime/structured-agent-session-close.ts b/src/main/runtime/structured-agent-session-close.ts index 8a21f2a229b9..fad9b6b19bff 100644 --- a/src/main/runtime/structured-agent-session-close.ts +++ b/src/main/runtime/structured-agent-session-close.ts @@ -68,12 +68,31 @@ export async function closeStructuredAgentSessionChild( // started without a chat tab, or one the user had closed — which is a new side effect, not an undo. const restoreTabIfCloseFails = options.restoreTabOnUnprovenClose !== false ? readPersistedTabId(host, sessionId) : null + try { + return await closeHiddenStructuredAgentSessionChild( + host, + sessionId, + options, + restoreTabIfCloseFails + ) + } finally { + // After the outcome, not at the hide: a close that puts the tab back leaves its chat working. + host.notifySessionTabHidden?.(sessionId) + } +} + +async function closeHiddenStructuredAgentSessionChild( + host: StructuredAgentSessionHost, + sessionId: string, + options: StructuredAgentSessionCloseOptions, + restoreTabIfCloseFails: string | null +): Promise { // Set only once the close is actually issued: `setSessionTabVisibility` throwing first leaves a // running child, and a receipt that still said `closed_agent_terminal` for it would be the // close-that-never-happened this flag exists to rule out. let closeAttempted = false try { - await host.setSessionTabVisibility?.(sessionId, false) + await host.setSessionTabVisibility?.(sessionId, false, undefined, { deferHiddenNotice: true }) closeAttempted = true // A worktree teardown or an orchestration stop: not the user closing this chat. await host.close(sessionId, 'evict') diff --git a/src/main/runtime/structured-agent-session-runtime.ts b/src/main/runtime/structured-agent-session-runtime.ts index 46edfca9d47b..eb12f16935c1 100644 --- a/src/main/runtime/structured-agent-session-runtime.ts +++ b/src/main/runtime/structured-agent-session-runtime.ts @@ -131,6 +131,8 @@ export type StructuredAgentSessionRuntimeDeps = { statusSink?: StructuredAgentSessionHostDeps['statusSink'] /** See `StructuredAgentSessionHostDeps.hasOpenDispatch`. */ hasOpenDispatch?: StructuredAgentSessionHostDeps['hasOpenDispatch'] + /** See `StructuredAgentSessionHostDeps.onSessionTabHidden`. */ + onSessionTabHidden?: StructuredAgentSessionHostDeps['onSessionTabHidden'] /** Host-owned phone delivery and reconciliation from the current journal projection. */ attentionDelivery?: StructuredAttentionMobileDelivery /** The account home a structured launch would pin right now, for catalog @@ -299,6 +301,7 @@ async function installOnJournal( ...(deps.onSessionStatusChanged ? { onSessionStatusChanged: deps.onSessionStatusChanged } : {}), ...(deps.statusSink ? { statusSink: deps.statusSink } : {}), ...(deps.hasOpenDispatch ? { hasOpenDispatch: deps.hasOpenDispatch } : {}), + ...(deps.onSessionTabHidden ? { onSessionTabHidden: deps.onSessionTabHidden } : {}), ...(await modelCatalogHostDeps({ store, agents, deps, envResolvers })) }) if (deps.attentionDelivery) { diff --git a/src/main/runtime/structured-chat-coordinator-mail-rig.test-fixture.ts b/src/main/runtime/structured-chat-coordinator-mail-rig.test-fixture.ts index 8aca8f498038..6a9ac88504d7 100644 --- a/src/main/runtime/structured-chat-coordinator-mail-rig.test-fixture.ts +++ b/src/main/runtime/structured-chat-coordinator-mail-rig.test-fixture.ts @@ -261,7 +261,8 @@ beforeEach(async () => { openCodexConnection: codex.openConnection, readProcessStartTime: async () => 1_700_000_000_000, // The same calls the runtime's own host install makes. - onSessionStatusChanged: (summary) => runtime.onStructuredSessionStatusForMail(summary) + onSessionStatusChanged: (summary) => runtime.onStructuredSessionStatusForMail(summary), + onSessionTabHidden: (sessionId) => runtime.onStructuredSessionTabHidden(sessionId) }) dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) }) diff --git a/src/main/runtime/structured-chat-dispatch-worker-lifecycle.test.ts b/src/main/runtime/structured-chat-dispatch-worker-lifecycle.test.ts new file mode 100644 index 000000000000..76e9330b8f91 --- /dev/null +++ b/src/main/runtime/structured-chat-dispatch-worker-lifecycle.test.ts @@ -0,0 +1,234 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +// What happens to a chat assignee's Dispatch after the task is handed over, on the +// coordinator-mail rig: its reports, its liveness, a stop, and the chat being closed. + +import { describe, expect, it, vi } from 'vitest' +import { formatOrcaSessionAddress } from '../../shared/orca-session-address' +import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture' +import { idOf } from './rpc/orchestration-session-caller-test-fixture' +import { closeStructuredAgentSessionChild } from './structured-agent-session-close' +import type { FakeConnection } from './structured-chat-coordinator-fake-codex-fixture' +import { + COORDINATOR, + PEER_CHAT, + WAIT, + call, + clearChat, + coordinatorRunAndTask, + db, + dispatcher, + host, + openChat, + request, + sendUserMessage, + settleTurn, + startSuccessor +} from './structured-chat-coordinator-mail-rig.test-fixture' + +const WORKER = formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT)) + +/** A Dispatch of the coordinator's task to the peer chat, both chats open. */ +async function dispatchToChat(): Promise<{ + taskId: string + dispatchId: string + worker: FakeConnection +}> { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const { dispatch } = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER }, + { sessionId: COORDINATOR } + ) + return { taskId, dispatchId: idOf(dispatch), worker } +} + +function workerDone(taskId: string, dispatchId: string) { + return { + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({ taskId, dispatchId, outcome: 'succeeded' }) + } +} + +/** The chat is still open: its conversation attached and its tab listed. */ +function expectChatOpen(sessionId: string): void { + expect(host.hasSession(sessionId)).toBe(true) + expect(host.deps.store.getVisibleSessionTabIndex().sessionIds).toContain(sessionId) +} + +describe("a chat assignee's report", () => { + it('is admitted from the chat itself, and settles its Dispatch', async () => { + const { taskId, dispatchId } = await dispatchToChat() + + await call('orchestration.send', workerDone(taskId, dispatchId), { sessionId: PEER_CHAT }) + + expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed') + expect(db.getTask(taskId)?.status).toBe('completed') + }) + + it('is admitted from the session a /clear continued the chat in, which keeps its Dispatch', async () => { + const { taskId, dispatchId } = await dispatchToChat() + const successor = await clearChat(PEER_CHAT) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + await startSuccessor(successor) + + await call('orchestration.send', workerDone(taskId, dispatchId), { sessionId: successor }) + + expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed') + }) + + it('is refused from a terminal that names the chat as its caller', async () => { + const { taskId, dispatchId } = await dispatchToChat() + + const response = await dispatcher.dispatch( + request('orchestration.send', { ...workerDone(taskId, dispatchId), from: WORKER }) + ) + + expect(response).toMatchObject({ + ok: false, + error: { code: 'session_caller_chat_not_declarable' } + }) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + }) + + it("is refused from a terminal that carries the chat's Dispatch under its own handle", async () => { + const { taskId, dispatchId } = await dispatchToChat() + + const receipt = await call('orchestration.send', { + ...workerDone(taskId, dispatchId), + from: 'term_worker' + }) + + expect(receipt).toMatchObject({ lifecycle: { action: 'rejected' } }) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + expect(db.getTask(taskId)?.status).toBe('dispatched') + }) +}) + +describe("a chat assignee's liveness, stop and close", () => { + it('reads live at rest, since the next send starts its agent', async () => { + const { dispatchId } = await dispatchToChat() + await host.close(PEER_CHAT, 'evict') + expect(host.deps.store.getRecord(PEER_CHAT)?.lease.claimStatus).toBe('released') + + const shown = await call('orchestration.workerShow', { dispatch: dispatchId }) + + expect(shown).toMatchObject({ + observation: { status: 'live', exactWorker: true }, + projection: { liveness: { verdict: 'live', source: 'execution_host' } } + }) + }) + + it('is stopped and abandoned without closing or interrupting the chat', async () => { + const stopped = await dispatchToChat() + expect(await call('orchestration.workerStop', { dispatch: stopped.dispatchId })).toMatchObject({ + processAction: 'none' + }) + expectChatOpen(PEER_CHAT) + + const more = await call( + 'orchestration.taskCreate', + { spec: 'more' }, + { sessionId: COORDINATOR } + ) + const { dispatch } = await call( + 'orchestration.dispatch', + { task: idOf(more.task), to: WORKER }, + { sessionId: COORDINATOR } + ) + expect(await call('orchestration.workerAbandon', { dispatch: idOf(dispatch) })).toMatchObject({ + state: 'abandoned', + processAction: 'none' + }) + expectChatOpen(PEER_CHAT) + }) + + it('fails its Dispatch when the chat is closed, as a closed terminal does, and reads exited', async () => { + const { taskId, dispatchId } = await dispatchToChat() + + // As the close RPC does when the person closes the chat's tab. + await host.setSessionTabVisibility(PEER_CHAT, false) + + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + status: 'failed', + termination_reason: 'operator_close' + }) + expect(db.getTask(taskId)?.status).not.toBe('dispatched') + expect(await call('orchestration.workerShow', { dispatch: dispatchId })).toMatchObject({ + observation: { status: 'exited' } + }) + }) + + it('keeps its Dispatch while a close puts the tab back, and fails it once a close is proven', async () => { + const { dispatchId } = await dispatchToChat() + vi.spyOn(host, 'close').mockRejectedValueOnce(new Error('the agent would not stop')) + + // As a refusable worktree removal closes each chat in it. + const outcome = await closeStructuredAgentSessionChild(PEER_CHAT, { + restoreTabOnUnprovenClose: true + }) + + expect(outcome).toMatchObject({ stopped: false, closeAttempted: true }) + expectChatOpen(PEER_CHAT) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + + expect(await closeStructuredAgentSessionChild(PEER_CHAT)).toMatchObject({ stopped: true }) + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + status: 'failed', + termination_reason: 'operator_close' + }) + }) + + it("keeps its Dispatch when another chat's close lands while its own close is rolled back", async () => { + const { dispatchId } = await dispatchToChat() + let refuseClose: (error: Error) => void = () => undefined + const close = vi + .spyOn(host, 'close') + .mockImplementationOnce(() => new Promise((_resolve, reject) => (refuseClose = reject))) + const closing = closeStructuredAgentSessionChild(PEER_CHAT, { restoreTabOnUnprovenClose: true }) + await vi.waitFor(() => expect(close).toHaveBeenCalled(), WAIT) + + // Its tab is hidden pending the close when the other chat's notice re-derives. + await host.setSessionTabVisibility(COORDINATOR, false) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + refuseClose(new Error('the agent would not stop')) + + expect(await closing).toMatchObject({ stopped: false, closeAttempted: true }) + expectChatOpen(PEER_CHAT) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched') + }) + + it('reads unverifiable, never exited, when the session that continues the chat is unknown', async () => { + const { dispatchId } = await dispatchToChat() + const successor = await clearChat(PEER_CHAT) + const getRecord = host.deps.store.getRecord.bind(host.deps.store) + vi.spyOn(host.deps.store, 'getRecord').mockImplementation((sessionId) => + sessionId === successor ? null : getRecord(sessionId) + ) + + expect(await call('orchestration.workerShow', { dispatch: dispatchId })).toMatchObject({ + observation: { + status: 'unverifiable', + reason: expect.stringContaining('after a /clear cannot be verified') + }, + projection: { liveness: { verdict: 'unverifiable' } } + }) + expect(await call('orchestration.workerList', {})).toMatchObject({ + workers: [{ dispatchId, projection: { liveness: { verdict: 'unverifiable' } } }] + }) + }) + + it("reads the chat's transcript for worker-read", async () => { + const { dispatchId, worker } = await dispatchToChat() + expect(await sendUserMessage(PEER_CHAT, 'working on it')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(worker.turns).toHaveLength(1), WAIT) + await settleTurn(PEER_CHAT, 0) + + const read = await call('orchestration.workerRead', { dispatch: dispatchId }) + + expect(read).toMatchObject({ source: 'transcript', status: { liveness: 'live' } }) + expect(JSON.stringify(read.transcript)).toContain('working on it') + }) +}) diff --git a/src/main/runtime/structured-chat-dispatch-worker-mail.test.ts b/src/main/runtime/structured-chat-dispatch-worker-mail.test.ts new file mode 100644 index 000000000000..5460520db9d1 --- /dev/null +++ b/src/main/runtime/structured-chat-dispatch-worker-mail.test.ts @@ -0,0 +1,91 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +// A chat worker's coordinator mail lands in its Dispatch's mailbox. When the pointer to it was held +// (the provider died before taking it), the chat's next idle edge points it again, after a restart +// or in the session a /clear continued the chat in, as it does for a chat's own mail. + +import { describe, expect, it, vi } from 'vitest' +import { formatOrcaSessionAddress } from '../../shared/orca-session-address' +import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture' +import { idOf } from './rpc/orchestration-session-caller-test-fixture' +import { + COORDINATOR, + PEER_CHAT, + WAIT, + call, + clearChat, + codex, + connectionFor, + coordinatorRunAndTask, + db, + openChat, + restartRuntime, + runtime, + sendUserMessage, + settleTurn, + turnText +} from './structured-chat-coordinator-mail-rig.test-fixture' + +const WORKER = formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT)) +const POINTER = /orchestration message/ + +async function idleEdges(sessionId: string): Promise { + runtime.onStructuredSessionStatusForMail({ sessionId, status: null }) + runtime.onStructuredSessionStatusForMail({ sessionId, status: 'idle' }) + await new Promise((resolve) => setTimeout(resolve, 400)) +} + +/** Coordinator mail to the peer chat's Dispatch, its pointer lost with the provider. */ +async function heldDispatchMail(): Promise { + await openChat(COORDINATOR) + const chat = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const { dispatch } = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER }, + { sessionId: COORDINATOR } + ) + const mailbox = `dispatch:${idOf(dispatch)}` + await call('orchestration.send', { to: mailbox, subject: 'more' }, { sessionId: COORDINATOR }) + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + expect(turnText(chat.turns[0]!)).toMatch(POINTER) + chat.handlers.onExit?.(new Error('provider died before the echo')) + await idleEdges(PEER_CHAT) + expect(chat.turns).toHaveLength(1) + expect(db.getUndeliveredUnreadMessages(mailbox, undefined, {})).toHaveLength(1) + return mailbox +} + +/** The chat's next turn, the person's; resolves to the session's provider once its edges ran. */ +async function nextTurn(sessionId: string): Promise<{ turns: { text: string }[] }> { + const before = codex.connections.length + expect(await sendUserMessage(sessionId, 'again')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(codex.connections.length).toBe(before + 1), WAIT) + const revived = connectionFor(sessionId) + await vi.waitFor(() => expect(revived.turns).toHaveLength(1), WAIT) + await settleTurn(sessionId, 0) + await idleEdges(sessionId) + return revived +} + +describe("a chat worker's held Dispatch mail is pointed again", () => { + // Also covered by the restored-mail repoint a restart schedules; this pins the end result. + it('after a restart', async () => { + await heldDispatchMail() + restartRuntime() + + const revived = await nextTurn(PEER_CHAT) + + await vi.waitFor(() => expect(revived.turns).toHaveLength(2), WAIT) + expect(turnText(revived.turns[1]!)).toMatch(POINTER) + }) + + it('in the session a /clear continued the chat in', async () => { + await heldDispatchMail() + const successor = await clearChat(PEER_CHAT) + + const revived = await nextTurn(successor) + + await vi.waitFor(() => expect(revived.turns).toHaveLength(2), WAIT) + expect(turnText(revived.turns[1]!)).toMatch(POINTER) + }) +}) diff --git a/src/main/runtime/structured-chat-dispatch-worker.test.ts b/src/main/runtime/structured-chat-dispatch-worker.test.ts new file mode 100644 index 000000000000..ca7660d81632 --- /dev/null +++ b/src/main/runtime/structured-chat-dispatch-worker.test.ts @@ -0,0 +1,288 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +// A chat as the assignee of `orchestration dispatch`, end to end on the coordinator-mail rig: its +// task arrives as any agent's message into a chat does, and it then works the Dispatch as a +// terminal worker would. + +import { describe, expect, it, vi } from 'vitest' +import { formatOrcaSessionAddress } from '../../shared/orca-session-address' +import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture' +import { idOf } from './rpc/orchestration-session-caller-test-fixture' +import { sendChatTask } from './rpc/methods/orchestration/chat-task-delivery' +import { + providerFaults, + type FakeConnection +} from './structured-chat-coordinator-fake-codex-fixture' +import { + COORDINATOR, + PEER_CHAT, + WAIT, + call, + coordinatorRunAndTask, + db, + dispatcher, + host, + openChat, + queuedCardTexts, + request, + runtime, + sendUserMessage, + settleTurn, + turnText +} from './structured-chat-coordinator-mail-rig.test-fixture' + +const WORKER = formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT)) + +/** A person's turn in `sessionId`, started and still running; resolves to its end. */ +async function runningUserTurn( + sessionId: string, + chat: FakeConnection +): Promise<() => Promise> { + const index = chat.turns.length + expect(await sendUserMessage(sessionId, 'go')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(chat.turns).toHaveLength(index + 1), WAIT) + const notify = (method: string, params: unknown) => chat.handlers.onNotification?.(method, params) + notify('turn/started', { turn: { id: `user-${index}` } }) + notify('item/completed', { + item: { + type: 'userMessage', + id: `echo-user-${index}`, + clientId: chat.turns[index]!.clientUserMessageId, + content: [{ type: 'text', text: 'go' }] + } + }) + await host.flushStreamedEvents(sessionId) + return async () => { + notify('turn/completed', { turn: { id: `user-${index}` } }) + await host.flushStreamedEvents(sessionId) + } +} + +function queuedRows(sessionId: string) { + return host.collaboratorsForTests().sessions.get(sessionId)?.journal.queuedMessages.list() ?? [] +} + +async function refusal(method: string, params: Record, sessionId?: string) { + const response = await dispatcher.dispatch( + request(method, params, sessionId ? { sessionId } : {}) + ) + if (response.ok) { + throw new Error(`expected a refusal, got ${JSON.stringify(response)}`) + } + return response.error +} + +/** Who each agent message in the chat's journal says it is from. */ +async function taskSenders(sessionId: string): Promise { + return (await host.journalSnapshot(sessionId)).items.flatMap((item) => + item.body.kind === 'message' && item.body.from ? [item.body.from] : [] + ) +} + +function dispatchCount(): unknown { + return db.db.prepare('SELECT COUNT(*) AS n FROM dispatch_contexts').get() +} + +describe('a chat as the assignee of orchestration dispatch', () => { + it('records the chat by its Orca session ID and sends an idle chat its task as a turn', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { runId, taskId } = await coordinatorRunAndTask() + + const dispatched = call( + 'orchestration.dispatch', + { task: taskId, to: WORKER, inject: true }, + { sessionId: COORDINATOR } + ) + await vi.waitFor(() => expect(worker.turns).toHaveLength(1), WAIT) + await settleTurn(PEER_CHAT, 0) + const result = await dispatched + + expect(result).toMatchObject({ injected: true }) + const dispatchId = idOf(result.dispatch) + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + run_id: runId, + assignee_handle: WORKER, + assignee_orca_session_id: PEER_CHAT, + assignee_pane_key: null, + process_incarnation: null, + status: 'dispatched' + }) + const task = turnText(worker.turns[0]!) + expect(task).toContain('build it') + expect(task).toContain(`Your Orca session ID is: ${WORKER}`) + expect(task).not.toMatch(/address/i) + expect(task).toContain(dispatchId) + expect(await taskSenders(PEER_CHAT)).toMatchObject([ + { kind: 'agent', orchestration: { message: 'task', taskId, dispatchId } } + ]) + }) + + it("holds the task as a card in a busy chat's queue, naming the Dispatch, and sends it when the turn ends", async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { runId, taskId } = await coordinatorRunAndTask() + const endTurn = await runningUserTurn(PEER_CHAT, worker) + + const result = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER, inject: true }, + { sessionId: COORDINATOR } + ) + + expect(result).toMatchObject({ injected: true }) + const dispatchId = idOf(result.dispatch) + expect(worker.turns).toHaveLength(1) + const from = { + kind: 'agent', + senders: [ + { + party: { + address: formatOrcaSessionAddress(testOrcaSessionId(COORDINATOR)), + terminalHandle: null, + orcaSessionId: COORDINATOR + } + } + ], + orchestration: { message: 'task', runId, taskId, dispatchId } + } + // On the card's body: the turn the queue sends carries it, and the provider never sees it. + expect(queuedRows(PEER_CHAT).map((card) => card.body.from)).toMatchObject([from]) + + await endTurn() + await vi.waitFor(() => expect(worker.turns).toHaveLength(2), WAIT) + expect(turnText(worker.turns[1]!)).toContain(dispatchId) + expect(await queuedCardTexts(PEER_CHAT)).toEqual([]) + expect(await taskSenders(PEER_CHAT)).toMatchObject([from]) + }) + + it('queues one card for one Dispatch however often its task is sent', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const endTurn = await runningUserTurn(PEER_CHAT, worker) + const result = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER }, + { sessionId: COORDINATOR } + ) + const dispatch = db.getDispatchContextById(idOf(result.dispatch))! + + const from = { kind: 'agent' as const, senders: [], orchestration: null } + const send = () => sendChatTask({ db, dispatch, from, preamble: 'the task' }) + expect(await send()).toBe('queued') + expect(await send()).toBe('queued') + + expect(await queuedCardTexts(PEER_CHAT)).toEqual(['the task']) + await endTurn() + }) + + it('keeps the Dispatch open when the chat may have taken its task but never acknowledged it', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + providerFaults.dieBeforeEveryEcho = true + + const error = await refusal( + 'orchestration.dispatch', + { task: taskId, to: WORKER, inject: true }, + COORDINATOR + ) + + expect(error).toMatchObject({ code: 'operation_unknown' }) + expect(worker.turns.length).toBeGreaterThan(0) + expect(db.db.prepare('SELECT status FROM dispatch_contexts').all()).toEqual([ + { status: 'dispatched' } + ]) + expect(db.getTask(taskId)?.status).toBe('dispatched') + }) + + it('refuses a chat that injects a Dispatch into itself, naming no --from, with no row written', async () => { + await openChat(COORDINATOR) + const { taskId } = await coordinatorRunAndTask() + + const error = await refusal( + 'orchestration.dispatch', + { task: taskId, to: formatOrcaSessionAddress(testOrcaSessionId(COORDINATOR)), inject: true }, + COORDINATOR + ) + + expect(error).toMatchObject({ + code: 'terminal_is_coordinator', + message: expect.stringContaining("is this coordinator's own Orca session ID") + }) + expect(dispatchCount()).toEqual({ n: 0 }) + expect(db.getTask(taskId)?.status).toBe('ready') + }) + + it('refuses a chat that was closed, with no row written', async () => { + await openChat(COORDINATOR) + await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + await host.setSessionTabVisibility(PEER_CHAT, false) + + const error = await refusal( + 'orchestration.dispatch', + { task: taskId, to: WORKER, inject: true }, + COORDINATOR + ) + + expect(error).toMatchObject({ + code: 'session_caller_not_live', + message: `Agent session ${PEER_CHAT} has ended: its chat was closed. No effects were applied.` + }) + expect(dispatchCount()).toEqual({ n: 0 }) + }) + + it("points the chat at its coordinator's mail to the Dispatch, which its own check reads", async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const result = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER }, + { sessionId: COORDINATOR } + ) + const dispatchId = idOf(result.dispatch) + + await call( + 'orchestration.send', + { to: `dispatch:${dispatchId}`, subject: 'also check the tests' }, + { sessionId: COORDINATOR } + ) + + await vi.waitFor(() => expect(worker.turns).toHaveLength(1), WAIT) + expect(turnText(worker.turns[0]!)).toMatch( + /^You have 1 orchestration message\. Run `orca(-dev)? orchestration check`\.$/ + ) + expect(await queuedCardTexts(PEER_CHAT)).toEqual([]) + await settleTurn(PEER_CHAT, 0) + expect(await call('orchestration.check', {}, { sessionId: PEER_CHAT })).toMatchObject({ + messages: [expect.objectContaining({ subject: 'also check the tests' })] + }) + }) + + it("counts a chat worker's own Dispatch as depth 2, under the nesting cap", async () => { + await openChat(COORDINATOR) + await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const parent = await call( + 'orchestration.dispatch', + { task: taskId, to: WORKER }, + { sessionId: COORDINATOR } + ) + expect(parent.dispatch).toMatchObject({ depth: 1 }) + await call('orchestration.runCreate', { objective: 'sub' }, { sessionId: PEER_CHAT }) + const subTask = idOf( + (await call('orchestration.taskCreate', { spec: 'part' }, { sessionId: PEER_CHAT })).task + ) + + const subDispatch = { task: subTask, to: 'term_worker' } + + expect(await refusal('orchestration.dispatch', subDispatch, PEER_CHAT)).toMatchObject({ + code: 'nested_worker_depth_exceeded' + }) + vi.spyOn(runtime, 'getNestedWorkerMaxDepth').mockReturnValue(2) + const child = await call('orchestration.dispatch', subDispatch, { sessionId: PEER_CHAT }) + expect(child.dispatch).toMatchObject({ depth: 2, creator_dispatch_id: idOf(parent.dispatch) }) + }) +}) diff --git a/src/main/runtime/structured-chat-worker-start.test.ts b/src/main/runtime/structured-chat-worker-start.test.ts new file mode 100644 index 000000000000..54ef02409b1b --- /dev/null +++ b/src/main/runtime/structured-chat-worker-start.test.ts @@ -0,0 +1,241 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +// `worker-start --terminal orca_session_id:`: a chat adopted as a worker, on the +// coordinator-mail rig. Its task is a queued send, read the way the terminal path reads its own +// write; a stop or a failed start never touches the chat. + +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { formatOrcaSessionAddress } from '../../shared/orca-session-address' +import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture' +import type { FakeConnection } from './structured-chat-coordinator-fake-codex-fixture' +import { + COORDINATOR, + PEER_CHAT, + WAIT, + call, + coordinatorRunAndTask, + db, + dispatcher, + host, + openChat, + request, + runtime, + sendUserMessage, + settleTurn, + turnText +} from './structured-chat-coordinator-mail-rig.test-fixture' + +const WORKER = formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT)) + +beforeEach(() => { + vi.spyOn(runtime, 'showManagedTerminalWorkspace').mockImplementation(async (selector) => ({ + id: selector.replace(/^id:/, ''), + repoId: 'repo' + })) +}) + +/** A person's turn in the worker chat, started and still running; resolves to its end. */ +async function runningUserTurn(chat: FakeConnection): Promise<() => Promise> { + expect(await sendUserMessage(PEER_CHAT, 'go')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + const notify = (method: string, params: unknown) => chat.handlers.onNotification?.(method, params) + notify('turn/started', { turn: { id: 'user-turn' } }) + notify('item/completed', { + item: { + type: 'userMessage', + id: 'echo-go', + clientId: chat.turns[0]!.clientUserMessageId, + content: [{ type: 'text', text: 'go' }] + } + }) + await host.flushStreamedEvents(PEER_CHAT) + return async () => { + notify('turn/completed', { turn: { id: 'user-turn' } }) + await host.flushStreamedEvents(PEER_CHAT) + } +} + +function startWorker(params: Record = {}) { + return call( + 'orchestration.workerStart', + { task: params.task, terminal: WORKER, ...params }, + { sessionId: COORDINATOR } + ) +} + +async function refusal(params: Record) { + const response = await dispatcher.dispatch( + request('orchestration.workerStart', params, { sessionId: COORDINATOR }) + ) + if (response.ok) { + throw new Error(`expected a refusal, got ${JSON.stringify(response)}`) + } + return response.error +} + +function expectChatOpen(): void { + expect(host.hasSession(PEER_CHAT)).toBe(true) + expect(host.deps.store.getVisibleSessionTabIndex().sessionIds).toContain(PEER_CHAT) +} + +describe('worker-start --terminal names a chat', () => { + it('is ready once an idle chat takes its task, and records the chat as an external worker', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + + const started = startWorker({ task: taskId }) + await vi.waitFor(() => expect(worker.turns).toHaveLength(1), WAIT) + await settleTurn(PEER_CHAT, 0) + const receipt = await started + + expect(receipt).toMatchObject({ + state: 'ready', + turnStart: 'observed', + mode: { + detail: `--terminal names the chat ${WORKER}; its task goes to that chat and no agent is launched.` + } + }) + const dispatchId = String(receipt.dispatchId) + expect(turnText(worker.turns[0]!)).toContain(`Your Orca session ID is: ${WORKER}`) + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + assignee_handle: WORKER, + assignee_orca_session_id: PEER_CHAT, + assignee_pane_key: null, + process_incarnation: null + }) + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + terminal_handle: WORKER, + ownership_state: 'external' + }) + }) + + it('hands a busy chat its task as a queued card and reads the start as unknown, not failed', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const endTurn = await runningUserTurn(worker) + + const receipt = await startWorker({ task: taskId }) + + expect(receipt).toMatchObject({ + state: 'outcome_unknown', + turnStart: 'unobserved', + lastError: expect.stringMatching( + /waiting as a card in the chat's queue and is sent when the queue reaches it\..*worker-abandon .* does not remove the card/ + ), + effects: expect.arrayContaining([ + expect.objectContaining({ kind: 'dispatch_input', id: WORKER, state: 'accepted' }), + expect.objectContaining({ kind: 'dispatch_input', id: WORKER, state: 'turn_unobserved' }) + ]) + }) + expect(JSON.stringify(receipt.nextCommands)).not.toContain('terminal read') + expect(JSON.stringify(receipt.nextCommands)).toContain('worker-abandon') + const dispatchId = String(receipt.dispatchId) + expect(db.getWorkerDispatch(dispatchId)?.state).toBe('start_unknown') + expect(worker.turns).toHaveLength(1) + + await endTurn() + await vi.waitFor(() => expect(worker.turns).toHaveLength(2), WAIT) + expect(turnText(worker.turns[1]!)).toContain(dispatchId) + }) + + it("reads the start as unknown when the chat's agent has not taken the task within the wait", async () => { + await openChat(COORDINATOR) + await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + // The settlement wait running out, as for an agent still starting. + vi.spyOn(host, 'waitForSendSettlement').mockResolvedValue(undefined) + + const receipt = await startWorker({ task: taskId }) + + expect(receipt).toMatchObject({ + state: 'outcome_unknown', + turnStart: 'unobserved', + lastError: expect.stringContaining("the chat's agent had not started to take it") + }) + expect(db.getWorkerDispatch(String(receipt.dispatchId))?.state).toBe('start_unknown') + }) + + it('is stopped and abandoned without closing or interrupting the chat', async () => { + await openChat(COORDINATOR) + const worker = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const started = startWorker({ task: taskId }) + await vi.waitFor(() => expect(worker.turns).toHaveLength(1), WAIT) + await settleTurn(PEER_CHAT, 0) + const dispatchId = String((await started).dispatchId) + + expect(await call('orchestration.workerList', {})).toMatchObject({ + workers: [{ dispatchId, projection: { liveness: { verdict: 'live' } } }] + }) + // What the chat reports when the person types into it, a /clear included: a chat is no + // minted worker, so there is nothing to take over and its Dispatch is untouched. + expect(await call('orchestration.workerTerminalUserInput', { sessionId: PEER_CHAT })).toEqual({ + changed: 0 + }) + expect(db.getWorkerTerminalResourceByOwner(dispatchId)?.ownership_state).toBe('external') + expect(await call('orchestration.workerStop', { dispatch: dispatchId })).toMatchObject({ + state: 'stop_unknown', + processAction: 'none' + }) + expectChatOpen() + expect(await call('orchestration.workerAbandon', { dispatch: dispatchId })).toMatchObject({ + state: 'abandoned', + processAction: 'none' + }) + expectChatOpen() + }) +}) + +describe('worker-start --terminal refuses a chat before anything is created', () => { + async function expectNothingCreated(error: unknown, code: string) { + expect(error).toMatchObject({ code }) + expect(db.db.prepare('SELECT COUNT(*) AS n FROM dispatch_contexts').get()).toEqual({ n: 0 }) + } + + it('when it is the coordinator itself', async () => { + await openChat(COORDINATOR) + const { taskId } = await coordinatorRunAndTask() + const self = formatOrcaSessionAddress(testOrcaSessionId(COORDINATOR)) + + await expectNothingCreated( + await refusal({ task: taskId, terminal: self }), + 'terminal_is_coordinator' + ) + }) + + it('when it works in another worktree', async () => { + await openChat(COORDINATOR) + await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + + await expectNothingCreated( + await refusal({ task: taskId, terminal: WORKER, worktree: 'id:other-worktree' }), + 'terminal_worktree_mismatch' + ) + }) + + it('when it was closed', async () => { + await openChat(COORDINATOR) + await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + await host.setSessionTabVisibility(PEER_CHAT, false) + + await expectNothingCreated( + await refusal({ task: taskId, terminal: WORKER }), + 'session_caller_not_live' + ) + }) + + it('when it is named on another host, before any remote call', async () => { + await openChat(COORDINATOR) + const { taskId } = await coordinatorRunAndTask() + const remote = vi.spyOn(runtime, 'resolveOrchestrationWorkerServer') + + await expectNothingCreated( + await refusal({ task: taskId, terminal: WORKER, on: 'box' }), + 'session_caller_host_boundary' + ) + expect(remote).not.toHaveBeenCalled() + }) +}) diff --git a/src/renderer/src/lib/open-agent-message-sender.test.ts b/src/renderer/src/lib/open-agent-message-sender.test.ts index 7dffd495e96e..ccaac427ad0b 100644 --- a/src/renderer/src/lib/open-agent-message-sender.test.ts +++ b/src/renderer/src/lib/open-agent-message-sender.test.ts @@ -119,6 +119,21 @@ describe("opening a message's sender", () => { expect(mocks.focusRuntime).not.toHaveBeenCalled() }) + it("opens a task's coordinator by its address alone: a task carries no mail", async () => { + mocks.callRuntimeRpc.mockResolvedValue({ location: { kind: 'terminal', handle: 'term_now' } }) + mocks.focusRenderer.mockReturnValue(true) + const source: AgentMessageSource = { + kind: 'agent', + senders: [{ party: terminal, name: 'Coordinator' }], + orchestration: { message: 'task', runId: 'r1', taskId: 't1', dispatchId: 'd1' } + } + await openAgentMessageSender(source, source.senders[0]!, 'wt-chat') + expect(mocks.callRuntimeRpc).toHaveBeenCalledWith(HOST, 'orchestration.partyLocation', { + address: 'term_a' + }) + expect(mocks.focusRenderer).toHaveBeenCalledWith('term_now', 'env-host') + }) + it('says the pane is gone only when the terminal answer proves it', async () => { mocks.callRuntimeRpc.mockResolvedValue({ location: { kind: 'terminal', handle: 'term_a' } }) mocks.focusRenderer.mockReturnValue(false) diff --git a/src/renderer/src/lib/open-agent-message-sender.ts b/src/renderer/src/lib/open-agent-message-sender.ts index 72112ffca0ae..9cc1a526eee4 100644 --- a/src/renderer/src/lib/open-agent-message-sender.ts +++ b/src/renderer/src/lib/open-agent-message-sender.ts @@ -73,8 +73,10 @@ async function openSender( chatWorktreeId: string ): Promise { const environmentId = getRuntimeEnvironmentIdForWorktree(useAppStore.getState(), chatWorktreeId) - // The mail it carried from this sender, whose pane outlives a handle from an earlier run. - const messageIds = (source.orchestration?.messages ?? []) + // The mail it carried from this sender, whose pane outlives a handle from an earlier run. A task + // carries no mail, so its coordinator is found by address alone. + const mail = source.orchestration?.message === 'mail-notice' ? source.orchestration.messages : [] + const messageIds = mail .filter((message) => message.from === party.address) .map((message) => message.messageId) const found = await lookUpSender( diff --git a/src/shared/agent-session-message-source.test.ts b/src/shared/agent-session-message-source.test.ts index c4d3cd127908..0095831e88f3 100644 --- a/src/shared/agent-session-message-source.test.ts +++ b/src/shared/agent-session-message-source.test.ts @@ -38,8 +38,16 @@ describe("a message's sender, read through its one reader", () => { expect(readAgentMessageSource(structuredClone(FROM))).toEqual(FROM) }) + it("reads a Dispatch's task back whole", () => { + const task: AgentMessageSource = { + ...FROM, + orchestration: { message: 'task', runId: 'r1', taskId: 't1', dispatchId: 'ctx_1' } + } + expect(readAgentMessageSource(structuredClone(task))).toEqual(task) + }) + it('keeps an agent an agent when its orchestration kind is newer than this build', () => { - const newer = { ...FROM, orchestration: { message: 'task', taskId: 't1' } } + const newer = { ...FROM, orchestration: { message: 'review', dispatchId: 'ctx_1' } } expect(readAgentMessageSource(newer)).toEqual({ ...FROM, orchestration: null }) // A newer sender kind is still not the person. expect(readAgentMessageSource({ ...FROM, kind: 'automation' })).toMatchObject({ diff --git a/src/shared/agent-session-message-source.ts b/src/shared/agent-session-message-source.ts index d86d244df0a6..8b7254caf9cd 100644 --- a/src/shared/agent-session-message-source.ts +++ b/src/shared/agent-session-message-source.ts @@ -30,8 +30,16 @@ export type OrchestrationMailNotice = Readonly<{ messages: readonly OrchestrationMailMessage[] }> +/** A Dispatch's task, sent as the assignee's turn: the records it joins back to while they exist. */ +export type OrchestrationTaskMessage = Readonly<{ + message: 'task' + runId: string + taskId: string + dispatchId: string +}> + /** What Orca delivers for other agents, one shape per message kind. */ -export type OrchestrationAgentMessage = OrchestrationMailNotice +export type OrchestrationAgentMessage = OrchestrationMailNotice | OrchestrationTaskMessage export type AgentMessageSource = Readonly<{ kind: 'agent' @@ -66,6 +74,13 @@ const mailNoticeSchema = z.object({ messages: z.array(z.object({ messageId: z.string(), runId: z.string(), from: z.string() })) }) +const taskSchema = z.object({ + message: z.literal('task'), + runId: z.string(), + taskId: z.string(), + dispatchId: z.string() +}) + const senderSchema = z.object({ party: z.object({ address: z.string().min(1), @@ -82,7 +97,7 @@ const sourceSchema = z.object({ senders: salvagedField('senders', salvagingArray(senderSchema), () => []), orchestration: salvagedOptional( 'orchestration', - z.discriminatedUnion('message', [mailNoticeSchema]) + z.discriminatedUnion('message', [mailNoticeSchema, taskSchema]) ).transform((orchestration) => orchestration ?? null) }) diff --git a/src/shared/orchestration-session-caller-codes.ts b/src/shared/orchestration-session-caller-codes.ts index 7710ee95659f..749974b2e509 100644 --- a/src/shared/orchestration-session-caller-codes.ts +++ b/src/shared/orchestration-session-caller-codes.ts @@ -9,7 +9,5 @@ export const ORCHESTRATION_SESSION_CALLER_ERROR_CODES = { /** The session exists but has no live owner here: released, switching owners, or unreconciled. */ notLive: 'session_caller_not_live', /** A request with no session id named a chat's address as its caller. */ - chatNotDeclarable: 'session_caller_chat_not_declarable', - /** A Dispatch named a chat as its assignee, which a chat cannot be yet. */ - chatNotDispatchable: 'session_chat_not_dispatchable' + chatNotDeclarable: 'session_caller_chat_not_declarable' } as const