Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ export function setStructuredAgentSessionTabVisibility(
tabId?: string
) => Promise<void>
}
onSessionTabHidden?: (sessionId: string) => void
}
restartResume: { dismiss: (sessionIds: readonly string[]) => Promise<number> }
},
Expand Down Expand Up @@ -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<void> => {
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) {
Expand All @@ -113,3 +120,19 @@ export function createStructuredAgentSessionTabSurface(
}
}
}

function notifyTabHidden(
host: Parameters<typeof setStructuredAgentSessionTabVisibility>[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
})
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,16 @@ 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
} from './orchestration/cli-command'
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 {
Expand Down Expand Up @@ -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
Expand Down
28 changes: 27 additions & 1 deletion src/main/runtime/orca-runtime-get-pty-record-for-pane-key.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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
}
Expand Down
2 changes: 2 additions & 0 deletions src/main/runtime/orca-runtime-get-worktree-ps.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) =>
Expand Down
4 changes: 3 additions & 1 deletion src/main/runtime/orca-runtime-stop-requested-pty-ids.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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)
Expand Down
17 changes: 10 additions & 7 deletions src/main/runtime/orchestration/agent-facing-parity.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrcaRuntimeService['orchestrationSenderNames'], 'nameOf'> } =
{
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
}
Expand All @@ -101,6 +103,7 @@ async function renderPreamble(worker: 'chat' | 'terminal'): Promise<string> {
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',
Expand Down
66 changes: 66 additions & 0 deletions src/main/runtime/orchestration/chat-assignee.test.ts
Original file line number Diff line number Diff line change
@@ -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'
})
})
})
109 changes: 109 additions & 0 deletions src/main/runtime/orchestration/chat-assignee.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
/**
* A chat as a Dispatch assignee. Its Dispatch row names it by its `/clear` root address,
* `orca_session_id:<root>`, 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'
)
})
}
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
Loading
Loading