From 2c53a550ffce24041fda9c943cc2cca00a501b70 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:31:49 -0400 Subject: [PATCH] feat(connectors): pending prompts, answers and turn events behind experimentalConnectors (B3) (#32160) --- src/cli/handler-group-manifest.ts | 8 +- src/cli/handlers/connector.test.ts | 26 +- src/cli/handlers/connector.ts | 81 +++++- src/cli/specs/connector.ts | 33 +++ .../external-conversation-events.ts | 207 ++++++++++++++ .../external-conversation-prompts.test.ts | 264 ++++++++++++++++++ .../external-conversation-prompts.ts | 66 +++++ ...external-conversation-seam.test-fixture.ts | 16 +- .../external-conversation-seam.ts | 74 ++++- .../external-conversation-types.ts | 6 +- .../external-conversation.test-fixture.ts | 19 ++ .../external-conversations-install.test.ts | 4 +- .../external-conversations-install.ts | 6 +- src/main/runtime/rpc/methods/connector.ts | 62 +++- .../terminal-agent-prompt-answer.test.ts | 178 ++++++++++++ .../runtime/terminal-agent-prompt-answer.ts | 167 +++++++++++ .../runtime/terminal-agent-turn-news.test.ts | 80 ++++++ src/main/runtime/terminal-agent-turn-news.ts | 30 ++ src/shared/agent-pending-prompt.test.ts | 89 ++++++ src/shared/agent-pending-prompt.ts | 121 ++++++++ src/shared/rpc-contract/connector-params.ts | 22 ++ .../rpc-params-catalog.generated.ts | 11 +- 22 files changed, 1547 insertions(+), 23 deletions(-) create mode 100644 src/main/external-conversations/external-conversation-events.ts create mode 100644 src/main/external-conversations/external-conversation-prompts.test.ts create mode 100644 src/main/external-conversations/external-conversation-prompts.ts create mode 100644 src/main/runtime/terminal-agent-prompt-answer.test.ts create mode 100644 src/main/runtime/terminal-agent-prompt-answer.ts create mode 100644 src/main/runtime/terminal-agent-turn-news.test.ts create mode 100644 src/shared/agent-pending-prompt.test.ts create mode 100644 src/shared/agent-pending-prompt.ts diff --git a/src/cli/handler-group-manifest.ts b/src/cli/handler-group-manifest.ts index f9455dfc56d..ef333cb64e4 100644 --- a/src/cli/handler-group-manifest.ts +++ b/src/cli/handler-group-manifest.ts @@ -212,7 +212,13 @@ export const HANDLER_GROUPS: readonly HandlerGroup[] = [ }, { name: 'connector', - keys: ['connector start', 'connector send'], + keys: [ + 'connector start', + 'connector send', + 'connector prompt', + 'connector answer', + 'connector watch' + ], load: async () => (await import('./handlers/connector.js')).CONNECTOR_HANDLERS }, { diff --git a/src/cli/handlers/connector.test.ts b/src/cli/handlers/connector.test.ts index c503ffa1eee..ab0c81ac89b 100644 --- a/src/cli/handlers/connector.test.ts +++ b/src/cli/handlers/connector.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from 'vitest' import { RuntimeClientError } from '../runtime-client' -import { callConnector, formatConnectorStart, formatDelivery } from './connector' +import { + callConnector, + formatConnectorStart, + formatDelivery, + formatPendingPrompt +} from './connector' function refusingClient(code: string) { return { @@ -41,4 +46,23 @@ describe('orca connector', () => { 'refused: agent_asleep_in_window' ) }) + + it('prints a prompt with its id and options, and an unknown kind as one to answer in Orca', () => { + expect( + formatPendingPrompt({ + kind: 'approval', + promptId: 'p_1', + tool: 'Bash', + summary: 'ls', + options: [ + { id: 'allow_once', label: 'Allow once' }, + { id: 'reject_once', label: 'Reject' } + ] + }) + ).toBe('approval Bash: ls\n prompt id: p_1\n allow_once: Allow once\n reject_once: Reject') + expect(formatPendingPrompt({ kind: 'free-text', promptId: 'p_2' })).toBe( + 'unsupported (answer in Orca)\n prompt id: p_2' + ) + expect(formatPendingPrompt(null)).toBe('no pending prompt') + }) }) diff --git a/src/cli/handlers/connector.ts b/src/cli/handlers/connector.ts index 0bd4c4e0ddb..9890a2a0da4 100644 --- a/src/cli/handlers/connector.ts +++ b/src/cli/handlers/connector.ts @@ -1,5 +1,9 @@ import type { CommandHandler, HandlerContext } from '../dispatch' -import { getOptionalStringFlag, getRequiredStringFlag } from '../flags' +import { + getOptionalNonNegativeIntegerFlag, + getOptionalStringFlag, + getRequiredStringFlag +} from '../flags' import { printResult } from '../format' import { RuntimeClientError } from '../runtime-client' import { getRequiredWorktreeSelector } from '../selectors' @@ -26,10 +30,11 @@ export function formatConnectorStart(initial: ConnectorDelivery): string { export async function callConnector( client: Pick, method: string, - params: Record + params: Record, + options?: { timeoutMs: number } ) { try { - return await client.call(method, params) + return await client.call(method, params, options) } catch (error) { if (error instanceof RuntimeClientError && error.code === 'method_not_found') { throw new RuntimeClientError( @@ -41,12 +46,43 @@ export async function callConnector( } } -function conversationParams(flags: HandlerContext['flags']) { - return { - source: 'dev', - externalId: getRequiredStringFlag(flags, 'conversation'), - deliveryId: getRequiredStringFlag(flags, 'delivery-id') +type ConnectorPrompt = { + kind: string + promptId: string + tool?: string + summary?: string + question?: string + options?: { id: string; label: string }[] +} + +/** A kind this CLI does not know reads as one only Orca can answer, as the contract asks. */ +export function formatPendingPrompt(prompt: ConnectorPrompt | null): string { + if (!prompt) { + return 'no pending prompt' } + const known = prompt.kind === 'approval' || prompt.kind === 'question' + const head = + prompt.kind === 'approval' + ? `approval ${prompt.tool}${prompt.summary ? `: ${prompt.summary}` : ''}` + : prompt.kind === 'question' + ? `question: ${prompt.question}` + : 'unsupported (answer in Orca)' + const options = known ? (prompt.options ?? []) : [] + return [ + head, + ` prompt id: ${prompt.promptId}`, + ...options.map((option) => ` ${option.id}: ${option.label}`) + ].join('\n') +} + +const DEFAULT_WATCH_MS = 30_000 + +function conversation(flags: HandlerContext['flags']) { + return { source: 'dev', externalId: getRequiredStringFlag(flags, 'conversation') } +} + +function conversationParams(flags: HandlerContext['flags']) { + return { ...conversation(flags), deliveryId: getRequiredStringFlag(flags, 'delivery-id') } } export const CONNECTOR_HANDLERS: Record = { @@ -82,5 +118,34 @@ export const CONNECTOR_HANDLERS: Record = { text: getRequiredStringFlag(flags, 'text') }) printResult(result, json, formatDelivery) + }, + 'connector prompt': async ({ flags, client, json }) => { + const result = await callConnector<{ prompt: ConnectorPrompt | null }>( + client, + 'connector.pendingPrompt', + conversation(flags) + ) + printResult(result, json, (value) => formatPendingPrompt(value.prompt)) + }, + 'connector answer': async ({ flags, client, json }) => { + const result = await callConnector(client, 'connector.answer', { + ...conversation(flags), + promptId: getRequiredStringFlag(flags, 'prompt-id'), + optionId: getRequiredStringFlag(flags, 'option') + }) + printResult(result, json, formatDelivery) + }, + 'connector watch': async ({ flags, client, json }) => { + const forMs = getOptionalNonNegativeIntegerFlag(flags, 'for-ms') ?? DEFAULT_WATCH_MS + const result = await callConnector<{ snapshot: unknown; events: unknown[] }>( + client, + 'connector.watch', + { forMs }, + // The host holds the call open for the whole window. + { timeoutMs: forMs + 15_000 } + ) + printResult(result, json, (value) => + [value.snapshot, ...value.events].map((event) => JSON.stringify(event)).join('\n') + ) } } diff --git a/src/cli/specs/connector.ts b/src/cli/specs/connector.ts index abe1ac288ea..24fbb1d81c6 100644 --- a/src/cli/specs/connector.ts +++ b/src/cli/specs/connector.ts @@ -38,5 +38,38 @@ export const CONNECTOR_COMMAND_SPECS: CommandSpec[] = [ ...CONNECTOR_NOTES, 'Held while the agent is asking for permission, and sent once it is answered.' ] + }, + { + path: ['connector', 'prompt'], + hidden: true, + summary: "Show what a dev connector conversation's agent is waiting on", + usage: 'orca connector prompt --conversation [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'conversation'], + notes: [ + CONNECTOR_NOTES[0], + 'An approval or a one-question single choice can be answered here; anything else reads unsupported, to answer in Orca.' + ] + }, + { + path: ['connector', 'answer'], + hidden: true, + summary: "Answer a dev connector conversation's pending prompt", + usage: 'orca connector answer --conversation --prompt-id --option [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'conversation', 'prompt-id', 'option'], + notes: [ + CONNECTOR_NOTES[0], + 'Refused unless --prompt-id is the prompt still pending in the same agent process, so a late answer never types into whatever came next.' + ] + }, + { + path: ['connector', 'watch'], + hidden: true, + summary: 'Collect connector conversation events for a while', + usage: 'orca connector watch [--for-ms ] [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'for-ms'], + notes: [ + CONNECTOR_NOTES[0], + 'Prints a snapshot of every linked conversation, then each event (linked, prompt, turn-started, turn-ended) seen in the window; --for-ms defaults to 30000.' + ] } ] diff --git a/src/main/external-conversations/external-conversation-events.ts b/src/main/external-conversations/external-conversation-events.ts new file mode 100644 index 00000000000..a0ddc95fb26 --- /dev/null +++ b/src/main/external-conversations/external-conversation-events.ts @@ -0,0 +1,207 @@ +import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' +import { agentMainAgentVerdict } from '../../shared/agent-main-agent-verdict' +import type { PendingPrompt } from '../../shared/agent-pending-prompt' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import { hookRowAgentPhase } from '../../shared/agent-transitions' +import type { AgentTurnOutcome } from '../../shared/agent-turn-outcome' +import type { + ConnectorAgent, + ExternalConversationLink +} from '../../shared/external-conversation-link' +import { judgeIdleAgentTurn } from '../agent-turn-idle-verdict' +import { + TerminalAgentTurnNews, + type TerminalAgentTurnNewsItem +} from '../runtime/terminal-agent-turn-news' +import { + allLinks, + type ExternalConversationLinkStore, + type LinkedConversation +} from './external-conversation-links' +import type { ConversationKey } from './external-conversation-types' + +type ConversationPane = { conversation: ConversationKey; worktreeId: string; paneKey: string } + +export type ConversationEvent = + | (ConversationPane & { kind: 'linked'; agent: ConnectorAgent }) + | (ConversationPane & { kind: 'prompt'; prompt: PendingPrompt }) + | (ConversationPane & { kind: 'turn-started'; at: number }) + | (ConversationPane & { + kind: 'turn-ended' + /** The host's recorded verdict on the turn; null when it recorded none. */ + stopReason: AgentTurnOutcome | null + lastAssistantMessage?: string + at: number + }) + +export type ConversationSnapshot = { + kind: 'snapshot' + conversations: (ConversationPane & { + agent: ConnectorAgent + working: boolean + prompt: PendingPrompt | null + })[] +} + +export type ConversationWatchListener = (event: ConversationSnapshot | ConversationEvent) => void + +export type ConversationEventsDeps = { + store: ExternalConversationLinkStore + readHookRow: (paneKey: string) => AgentStatusIpcPayload | null + /** The detector's pane changes; returns the release. */ + onAgentPaneChange: (listener: (change: AgentPaneChange) => void) => () => void + pendingPrompt: (link: ExternalConversationLink) => PendingPrompt | null + setTimer: (run: () => void, delayMs: number) => { cancel: () => void } + now: () => number +} + +/** + * Linked agents' turns and prompts for connector watchers. Turn news is the phone push's own + * (`TerminalAgentTurnNews`), and a finish counts only when the idle verdict automations use agrees. + * Subscribed only while someone watches. + */ +export class ConversationEvents { + private readonly listeners = new Set() + private readonly turnStartedAt = new Map() + private release: (() => void) | null = null + + constructor(private readonly deps: ConversationEventsDeps) {} + + watch(listener: ConversationWatchListener): () => void { + if (!this.release) { + const news = new TerminalAgentTurnNews({ + readHookRow: (paneKey) => this.deps.readHookRow(paneKey), + prepare: (item) => item, + deliver: (item) => this.told(item), + setTimer: (run, delayMs) => this.deps.setTimer(run, delayMs), + started: (item) => this.started(item), + stopped: (item) => this.ended(item, agentMainAgentVerdict(item.row)) + }) + const unsubscribe = this.deps.onAgentPaneChange((change) => news.observe(change)) + this.release = () => { + unsubscribe() + news.dispose() + } + } + this.listeners.add(listener) + // After subscribing, so a change while the snapshot is read is not missed. + listener(this.snapshot()) + return () => { + this.listeners.delete(listener) + if (this.listeners.size === 0) { + this.dispose() + } + } + } + + linked({ workspace, link }: LinkedConversation): void { + this.emit({ + kind: 'linked', + ...pane(link, workspace.worktreeId), + agent: link.agent + }) + } + + dispose(): void { + this.release?.() + this.release = null + this.listeners.clear() + this.turnStartedAt.clear() + } + + private snapshot(): ConversationSnapshot { + return { + kind: 'snapshot', + conversations: [...allLinks(this.deps.store)].map(({ workspace, link }) => { + const row = this.deps.readHookRow(link.paneKey) + return { + ...pane(link, workspace.worktreeId), + agent: link.agent, + working: row ? hookRowAgentPhase(row) === 'working' : false, + prompt: this.deps.pendingPrompt(link) + } + }) + } + } + + private started(item: TerminalAgentTurnNewsItem): void { + const link = this.linkFor(item.paneKey) + if (!link) { + return + } + const at = item.row.turnStartedAt ?? item.row.stateStartedAt + this.turnStartedAt.set(item.paneKey, at) + this.emit({ kind: 'turn-started', ...pane(link, item.worktreeId), at }) + } + + private told(item: TerminalAgentTurnNewsItem): void { + const link = this.linkFor(item.paneKey) + if (!link) { + return + } + if (hookRowAgentPhase(item.row) === 'needs-you') { + const prompt = this.deps.pendingPrompt(link) + if (prompt) { + this.emit({ kind: 'prompt', ...pane(link, item.worktreeId), prompt }) + } + return + } + // A hook-reported `done` since this turn began; a reused pane's earlier one is not this turn's. + const verdict = judgeIdleAgentTurn( + { + rows: [item.row], + sentAt: this.turnStartedAt.get(item.paneKey) ?? 0, + tail: [], + readAgentCommands: () => [] + }, + this.deps.now() + ) + if (verdict.kind === 'completed') { + this.ended(item, agentMainAgentVerdict(item.row)) + } + } + + private ended(item: TerminalAgentTurnNewsItem, stopReason: AgentTurnOutcome | null): void { + const link = this.linkFor(item.paneKey) + if (!link) { + return + } + this.turnStartedAt.delete(item.paneKey) + const message = item.row.lastAssistantMessage + this.emit({ + kind: 'turn-ended', + ...pane(link, item.worktreeId), + stopReason, + ...(message ? { lastAssistantMessage: message } : {}), + at: item.row.stateStartedAt + }) + } + + private linkFor(paneKey: string): ExternalConversationLink | null { + for (const { link } of allLinks(this.deps.store)) { + if (link.paneKey === paneKey) { + return link + } + } + return null + } + + private emit(event: ConversationEvent): void { + for (const listener of this.listeners) { + // One watcher's failure must not cost the others the event. + try { + listener(event) + } catch (error) { + console.error('[connector] watch listener threw', error) + } + } + } +} + +function pane(link: ExternalConversationLink, worktreeId: string): ConversationPane { + return { + conversation: { source: link.source, externalId: link.externalId }, + worktreeId, + paneKey: link.paneKey + } +} diff --git a/src/main/external-conversations/external-conversation-prompts.test.ts b/src/main/external-conversations/external-conversation-prompts.test.ts new file mode 100644 index 00000000000..37529b8e1ea --- /dev/null +++ b/src/main/external-conversations/external-conversation-prompts.test.ts @@ -0,0 +1,264 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { AgentPhase, AgentTransition } from '../../shared/agent-transitions' +import type { ConversationEvent, ConversationSnapshot } from './external-conversation-events' +import type { ConversationKey } from './external-conversation-types' +import { + fakeLauncher, + fakeTerminal, + launchResult, + PANE, + WORKTREE +} from './external-conversation.test-fixture' +import { harness } from './external-conversation-seam.test-fixture' + +const KEY: ConversationKey = { source: 'dev', externalId: 'thread-1' } +const START = { + agent: 'claude', + target: { kind: 'existing' as const, worktree: `id:${WORKTREE}` }, + prompt: 'first' +} + +function hookRow(overrides: Partial): AgentStatusIpcPayload { + return { + state: 'working', + prompt: '', + agentType: 'claude', + paneKey: PANE, + connectionId: null, + receivedAt: 1, + stateStartedAt: 100, + ...overrides + } +} + +const APPROVAL = hookRow({ + state: 'waiting', + stateStartedAt: 200, + interactivePrompt: JSON.stringify({ approval: { tool: 'Bash', summary: 'ls' } }) +}) + +function moved( + from: AgentPhase, + to: AgentPhase, + transition: AgentTransition | null +): AgentPaneChange { + return { + kind: 'moved', + paneKey: PANE, + workspace: { worktreeId: WORKTREE, sshHostId: null }, + from, + to, + at: 1, + transition + } +} + +async function linked(settings = {}) { + const h = harness(settings) + const fake = fakeTerminal() + const host = { + launcher: fakeLauncher(async () => launchResult('handed-to-terminal')).launcher, + terminal: fake.terminal + } + await h.seam.start(host, KEY, 'd1', START) + return { ...h, fake, host } +} + +describe('ExternalConversationSeam prompts', () => { + it('reads the linked agent’s approval and answers it once, by its exact id', async () => { + const { seam, host, hookRows, fake } = await linked() + expect(seam.pendingPrompt(host, KEY)).toBeNull() + hookRows.set(PANE, APPROVAL) + const prompt = seam.pendingPrompt(host, KEY) + expect(prompt).toMatchObject({ kind: 'approval', tool: 'Bash', summary: 'ls' }) + await expect(seam.answer(host, KEY, prompt!.promptId, 'allow_once')).resolves.toEqual({ + outcome: 'accepted' + }) + expect(fake.keys).toEqual(['1']) + await expect(seam.answer(host, KEY, prompt!.promptId, 'allow_once')).resolves.toEqual({ + outcome: 'refused', + reason: 'prompt_not_pending' + }) + expect(fake.keys).toEqual(['1']) + }) + + it('refuses an answer for a stale prompt, a replaced process, or a shell in front', async () => { + const { seam, host, hookRows, fake } = await linked() + hookRows.set(PANE, APPROVAL) + const { promptId } = seam.pendingPrompt(host, KEY)! + fake.replace() + await expect(seam.answer(host, KEY, promptId, 'allow_once')).resolves.toMatchObject({ + reason: 'prompt_not_pending' + }) + const fresh = seam.pendingPrompt(host, KEY)!.promptId + fake.terminal.readLaunchedAgentForeground.mockResolvedValueOnce('shell') + await expect(seam.answer(host, KEY, fresh, 'reject_once')).resolves.toMatchObject({ + reason: 'agent_not_reachable' + }) + expect(fake.keys).toEqual([]) + }) + + it('reads a prompt as unsupported once the agent’s hooks are off', async () => { + const settings: { agentStatusHooksEnabled?: boolean } = {} + const { seam, host, hookRows } = await linked(settings) + hookRows.set(PANE, APPROVAL) + settings.agentStatusHooksEnabled = false + const prompt = seam.pendingPrompt(host, KEY)! + expect(prompt.kind).toBe('unsupported') + await expect(seam.answer(host, KEY, prompt.promptId, 'allow_once')).resolves.toMatchObject({ + reason: 'prompt_unsupported' + }) + }) + + it('throws for a conversation with no link, and refuses once connectors are off', async () => { + const { seam, host } = await linked() + const other = { source: 'dev' as const, externalId: 'nope' } + expect(() => seam.pendingPrompt(host, other)).toThrow('external_conversation_not_found') + seam.dispose('connectors_disabled') + await expect(seam.answer(host, KEY, 'p', 'allow_once')).resolves.toEqual({ + outcome: 'refused', + reason: 'connectors_disabled' + }) + expect(() => seam.watch(host, () => {})).toThrow('connectors_disabled') + }) +}) + +describe('ExternalConversationSeam watch', () => { + it('subscribes only while watched, and opens with a snapshot of every link', async () => { + const { seam, host, paneListeners, hookRows } = await linked() + expect(paneListeners.size).toBe(0) + hookRows.set(PANE, APPROVAL) + const seen: (ConversationSnapshot | ConversationEvent)[] = [] + const release = seam.watch(host, (event) => seen.push(event)) + expect(paneListeners.size).toBe(1) + expect(seen).toEqual([ + { + kind: 'snapshot', + conversations: [ + expect.objectContaining({ + conversation: KEY, + paneKey: PANE, + agent: 'claude', + working: false, + prompt: expect.objectContaining({ kind: 'approval' }) + }) + ] + } + ]) + release() + expect(paneListeners.size).toBe(0) + }) + + it('does not deliver a previous watcher’s queued finish to a new watcher', async () => { + const { seam, host, hookRows, paneChange } = await linked() + vi.useFakeTimers() + try { + const release = seam.watch(host, () => {}) + hookRows.set(PANE, hookRow({ state: 'working' })) + paneChange({ + kind: 'observed', + paneKey: PANE, + phase: 'working', + workspace: { worktreeId: WORKTREE, sshHostId: null }, + at: 100 + }) + hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 300 })) + paneChange(moved('working', 'done', 'turn-finished')) + release() + const seen: (ConversationSnapshot | ConversationEvent)[] = [] + const releaseNew = seam.watch(host, (event) => seen.push(event)) + await vi.runAllTimersAsync() + expect(seen.map((event) => event.kind)).toEqual(['snapshot']) + releaseNew() + } finally { + vi.useRealTimers() + } + }) + + it('reports no outcome when a completed hook turn records no verdict', async () => { + const { seam, host, hookRows, paneChange } = await linked() + const seen: (ConversationSnapshot | ConversationEvent)[] = [] + seam.watch(host, (event) => seen.push(event)) + hookRows.set(PANE, hookRow({ state: 'working' })) + paneChange(moved('rest', 'working', 'prompt-submitted')) + hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 300 })) + paneChange(moved('working', 'done', 'turn-finished')) + expect(seen.at(-1)).toMatchObject({ kind: 'turn-ended', stopReason: null }) + }) + + it('tells a linked turn’s start, its prompt, and its end with the host’s verdict', async () => { + const { seam, host, hookRows, paneChange } = await linked() + const seen: ConversationEvent[] = [] + seam.watch(host, (event) => { + if (event.kind !== 'snapshot') { + seen.push(event) + } + }) + hookRows.set(PANE, hookRow({ state: 'working', turnStartedAt: 100 })) + paneChange(moved('rest', 'working', 'prompt-submitted')) + hookRows.set(PANE, APPROVAL) + paneChange(moved('working', 'needs-you', 'needs-input')) + hookRows.set( + PANE, + hookRow({ + state: 'done', + stateStartedAt: 300, + lastAssistantMessage: 'Done.', + mainAgent: { state: 'done', stateStartedAt: 300, outcome: 'success' } + }) + ) + paneChange(moved('needs-you', 'done', 'turn-finished')) + expect(seen.map((event) => event.kind)).toEqual(['turn-started', 'prompt', 'turn-ended']) + expect(seen[0]).toMatchObject({ conversation: KEY, at: 100 }) + expect(seen[1]).toMatchObject({ prompt: { kind: 'approval' } }) + expect(seen[2]).toMatchObject({ stopReason: 'success', lastAssistantMessage: 'Done.', at: 300 }) + }) + + it('ends a turn the user stopped with its recorded verdict, never the caller’s', async () => { + const { seam, host, hookRows, paneChange } = await linked() + const seen: ConversationEvent[] = [] + seam.watch(host, (event) => { + if (event.kind !== 'snapshot') { + seen.push(event) + } + }) + hookRows.set(PANE, hookRow({ state: 'working' })) + paneChange(moved('rest', 'working', 'prompt-submitted')) + hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 150, interrupted: true })) + paneChange(moved('working', 'rest', null)) + expect(seen.at(-1)).toMatchObject({ kind: 'turn-ended', stopReason: 'cancellation' }) + }) + + it('tells nothing for a pane no conversation links, nor a done from before the turn', async () => { + const { seam, host, hookRows, paneChange } = await linked() + const seen: ConversationEvent[] = [] + seam.watch(host, (event) => { + if (event.kind !== 'snapshot') { + seen.push(event) + } + }) + hookRows.set(PANE, hookRow({ state: 'working', turnStartedAt: 500 })) + paneChange(moved('rest', 'working', 'prompt-submitted')) + hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 400, sessionBoundary: false })) + paneChange(moved('working', 'done', 'turn-finished')) + paneChange({ ...moved('rest', 'working', 'prompt-submitted'), paneKey: 'tab-9:other' }) + expect(seen.map((event) => event.kind)).toEqual(['turn-started']) + }) + + it('tells a watcher of a conversation linked while it watches', async () => { + const h = harness() + const host = { + launcher: fakeLauncher(async () => launchResult('handed-to-terminal')).launcher, + terminal: fakeTerminal().terminal + } + const seen: (ConversationSnapshot | ConversationEvent)[] = [] + h.seam.watch(host, (event) => seen.push(event)) + await h.seam.start(host, KEY, 'd1', START) + expect(seen).toEqual([ + { kind: 'snapshot', conversations: [] }, + { kind: 'linked', conversation: KEY, worktreeId: WORKTREE, paneKey: PANE, agent: 'claude' } + ]) + }) +}) diff --git a/src/main/external-conversations/external-conversation-prompts.ts b/src/main/external-conversations/external-conversation-prompts.ts new file mode 100644 index 00000000000..4d3364c0baa --- /dev/null +++ b/src/main/external-conversations/external-conversation-prompts.ts @@ -0,0 +1,66 @@ +import type { PendingPrompt } from '../../shared/agent-pending-prompt' +import type { AgentStatusHooksSettings } from '../../shared/agent-status-hooks-setting' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { ExternalConversationLink } from '../../shared/external-conversation-link' +import { + createLaunchedAgentWriteGuard, + type LaunchedAgentWriteGuardRuntime +} from '../runtime/launched-agent-write-guard' +import { + TerminalAgentPromptAnswers, + type PanePromptTarget, + type TerminalAgentPromptAnswerRuntime +} from '../runtime/terminal-agent-prompt-answer' +import { admitConnectorAgent } from './connector-agents' +import type { Delivery } from './external-conversation-types' + +export type PromptTerminal = TerminalAgentPromptAnswerRuntime & LaunchedAgentWriteGuardRuntime + +/** + * A linked agent's pending prompt and its answer, through the host's shared prompt answers. Not on + * the conversation's queue: a follow-up held behind an approval waits for exactly this answer. + */ +export class ConversationPrompts { + private readonly answers: TerminalAgentPromptAnswers + + constructor( + private readonly readSettings: () => AgentStatusHooksSettings, + readHookRow: (paneKey: string) => AgentStatusIpcPayload | null + ) { + this.answers = new TerminalAgentPromptAnswers(readHookRow) + } + + pending(terminal: PromptTerminal, link: ExternalConversationLink): PendingPrompt | null { + return this.answers.pending(terminal, this.target(link)) + } + + async answer( + terminal: PromptTerminal, + link: ExternalConversationLink, + answer: { promptId: string; optionId: string; signal: AbortSignal } + ): Promise { + // Keys typed at a shell that replaced the agent would run as a command. + const guard = createLaunchedAgentWriteGuard(terminal, link.agent, { + unprovableHost: 'write-unless-shell' + }) + try { + const outcome = await this.answers.answer(terminal, this.target(link), { + ...answer, + beforeWrite: (ptyId) => guard.beforeWrite(ptyId) + }) + return outcome.outcome === 'unknown' + ? { outcome: 'unknown', reason: 'writer_failed' } + : outcome + } finally { + guard.dispose() + } + } + + private target(link: ExternalConversationLink): PanePromptTarget { + return { + paneKey: link.paneKey, + agent: link.agent, + answerable: admitConnectorAgent(this.readSettings(), link.agent) + } + } +} diff --git a/src/main/external-conversations/external-conversation-seam.test-fixture.ts b/src/main/external-conversations/external-conversation-seam.test-fixture.ts index e36b35270ef..8925c288c18 100644 --- a/src/main/external-conversations/external-conversation-seam.test-fixture.ts +++ b/src/main/external-conversations/external-conversation-seam.test-fixture.ts @@ -1,4 +1,6 @@ import type { GlobalSettings } from '../../shared/global-settings-types' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' import { DeliveryReceipts } from './external-conversation-delivery-receipts' import { ExternalConversationSeam } from './external-conversation-seam' @@ -13,6 +15,8 @@ export function harness( sessionRows: SessionRows = { getAgentStatusRowsForPane: () => [] } ) { const removed = new Set<(worktreeId: string) => void>() + const hookRows = new Map() + const paneListeners = new Set<(change: AgentPaneChange) => void>() const sessionListeners = new Set< (sessions: readonly { paneKey: string; sessionId: string }[]) => void >() @@ -29,6 +33,11 @@ export function harness( onProviderSessionsChanged: (listener) => { sessionListeners.add(listener) return () => sessionListeners.delete(listener) + }, + readHookRow: (paneKey) => hookRows.get(paneKey) ?? null, + onAgentPaneChange: (listener) => { + paneListeners.add(listener) + return () => paneListeners.delete(listener) } }) const reportSessions = (sessions: { paneKey: string; sessionId: string }[]): void => @@ -43,7 +52,12 @@ export function harness( removed, removeWorkspace, sessionListeners, - reportSessions + reportSessions, + /** The hook store's newest row per pane, as `readHookRow` reads it. */ + hookRows, + paneListeners, + paneChange: (change: AgentPaneChange): void => + paneListeners.forEach((listener) => listener(change)) } } diff --git a/src/main/external-conversations/external-conversation-seam.ts b/src/main/external-conversations/external-conversation-seam.ts index 208704638c2..ecc5eed2b7b 100644 --- a/src/main/external-conversations/external-conversation-seam.ts +++ b/src/main/external-conversations/external-conversation-seam.ts @@ -1,8 +1,13 @@ +import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' +import type { PendingPrompt } from '../../shared/agent-pending-prompt' import type { AgentStatusHooksSettings } from '../../shared/agent-status-hooks-setting' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' import type { ExternalConversationLink } from '../../shared/external-conversation-link' import type { OrcaRuntimeService } from '../runtime/orca-runtime' import { admitConnectorAgent } from './connector-agents' import { ConversationChain } from './external-conversation-chain' +import { ConversationEvents, type ConversationWatchListener } from './external-conversation-events' +import { ConversationPrompts, type PromptTerminal } from './external-conversation-prompts' import type { DeliveryReceipts } from './external-conversation-delivery-receipts' import { findLink, @@ -35,7 +40,7 @@ export type ExternalConversationHost = { terminal: ConversationTerminal } -export type ConversationTerminal = FollowUpTerminal & ReopenTerminal +export type ConversationTerminal = FollowUpTerminal & ReopenTerminal & PromptTerminal export function runtimeConversationHost(runtime: OrcaRuntimeService): ExternalConversationHost { return { launcher: runtimeConversationLauncher(runtime), terminal: runtime } @@ -54,6 +59,10 @@ export type ExternalConversationSeamDeps = { onProviderSessionsChanged: ( listener: (sessions: readonly { paneKey: string; sessionId: string }[]) => void ) => () => void + /** The hook store's newest row for a pane. */ + readHookRow: (paneKey: string) => AgentStatusIpcPayload | null + /** The transition detector's pane changes; returns the release. */ + onAgentPaneChange: (listener: (change: AgentPaneChange) => void) => () => void } type StartInput = Omit & { agent: string } @@ -76,11 +85,15 @@ export class ExternalConversationSeam { */ private readonly admitted = new WeakMap() private readonly sessions: LinkedSessionCapture + private readonly prompts: ConversationPrompts + /** Built by the first watcher, so an unwatched seam subscribes to nothing. */ + private events: ConversationEvents | null = null private readonly releaseRemovals: () => void private readonly releaseSessions: () => void constructor(private readonly deps: ExternalConversationSeamDeps) { this.sessions = new LinkedSessionCapture(deps.store, deps.sessionRows) + this.prompts = new ConversationPrompts(deps.readSettings, deps.readHookRow) // A removed workspace may keep its PTY registered a while; its held and queued sends end now. // The event names no host, so a same-path twin's removal is told apart by the link surviving. this.releaseRemovals = deps.onWorkspaceRemoved((worktreeId) => { @@ -106,9 +119,58 @@ export class ExternalConversationSeam { dispose(reason: 'connectors_disabled'): void { this.releaseRemovals() this.releaseSessions() + this.events?.dispose() this.abort.abort(reason) } + pendingPrompt(host: ExternalConversationHost, key: ConversationKey): PendingPrompt | null { + return this.prompts.pending(host.terminal, this.requireLink(key).link) + } + + answer( + host: ExternalConversationHost, + key: ConversationKey, + promptId: string, + optionId: string + ): Promise { + if (this.abort.signal.aborted) { + return Promise.resolve(refused('connectors_disabled')) + } + return this.prompts.answer(host.terminal, this.requireLink(key).link, { + promptId, + optionId, + signal: this.abort.signal + }) + } + + /** A snapshot of every linked conversation first, then its events; returns the release. */ + watch(host: ExternalConversationHost, listener: ConversationWatchListener): () => void { + if (this.abort.signal.aborted) { + throw new Error('connectors_disabled') + } + this.events ??= new ConversationEvents({ + store: this.deps.store, + readHookRow: this.deps.readHookRow, + onAgentPaneChange: this.deps.onAgentPaneChange, + pendingPrompt: (link) => this.prompts.pending(host.terminal, link), + setTimer: (run, delayMs) => { + const timer = setTimeout(run, delayMs) + timer.unref?.() + return { cancel: () => clearTimeout(timer) } + }, + now: Date.now + }) + return this.events.watch(listener) + } + + private requireLink(key: ConversationKey): LinkedConversation { + const linked = findLink(this.deps.store, key) + if (!linked) { + throw new Error('external_conversation_not_found') + } + return linked + } + async start( host: ExternalConversationHost, key: ConversationKey, @@ -138,12 +200,9 @@ export class ExternalConversationSeam { { ...input, agent }, signal, (workspace, paneKey) => { - writeLink(this.deps.store, workspace, { - ...key, - paneKey, - agent, - createdAt: Date.now() - }) + const link = { ...key, paneKey, agent, createdAt: Date.now() } + writeLink(this.deps.store, workspace, link) + this.events?.linked({ workspace, link }) this.startedHere.add(conversationId(key)) this.sessions.capture(key, paneKey) }, @@ -234,6 +293,7 @@ export class ExternalConversationSeam { (paneKey) => { relinked.paneKey = paneKey writeLink(this.deps.store, workspace, relinked) + this.events?.linked({ workspace, link: relinked }) this.sessions.capture(key, paneKey) } ) diff --git a/src/main/external-conversations/external-conversation-types.ts b/src/main/external-conversations/external-conversation-types.ts index d5a8773a970..5de3d88dbd1 100644 --- a/src/main/external-conversations/external-conversation-types.ts +++ b/src/main/external-conversations/external-conversation-types.ts @@ -12,7 +12,11 @@ export const CONNECTOR_REFUSALS = [ // The three a closed agent's reopen ends on; each asks the user to open it in Orca. 'agent_asleep_in_window', 'no_session_to_reopen', - 'reopen_refused' + 'reopen_refused', + // An answer's: the prompt was answered or moved on, only Orca can answer it, or no such option. + 'prompt_not_pending', + 'prompt_unsupported', + 'option_not_offered' ] as const export type ConnectorRefusal = (typeof CONNECTOR_REFUSALS)[number] diff --git a/src/main/external-conversations/external-conversation.test-fixture.ts b/src/main/external-conversations/external-conversation.test-fixture.ts index 4a728144fb5..fc0ee9b8171 100644 --- a/src/main/external-conversations/external-conversation.test-fixture.ts +++ b/src/main/external-conversations/external-conversation.test-fixture.ts @@ -171,6 +171,8 @@ export function fakeTerminal() { const current = (): string => `inc-${state.ptyId}${state.incarnation > 1 ? `-${state.incarnation}` : ''}` const writes: string[] = [] + /** Raw keys written outside the prompt writer: prompt answers. */ + const keys: string[] = [] const waiters = new Set<(error?: Error) => void>() const hold = (signal: AbortSignal | undefined): Promise => new Promise((resolve, reject) => { @@ -210,6 +212,20 @@ export function fakeTerminal() { return { handle, accepted: true, bytesWritten: 1 } } ) + const sendTerminal = vi.fn( + async ( + handle: string, + action: { text?: string }, + options: { beforeWrite?: (ptyId: string) => unknown } + ): Promise => { + if (!state.ptyId) { + throw new Error('terminal_not_writable') + } + await options.beforeWrite?.(state.ptyId) + keys.push(action.text ?? '') + return { handle, accepted: true, bytesWritten: 1, writeSettlement: { outcome: 'accepted' } } + } + ) const resumeAgentSessionInWorkspace = vi.fn( async ( _worktreeId: string, @@ -234,6 +250,7 @@ export function fakeTerminal() { return { ptyId: state.ptyId, processIncarnation: current(), generation: 1 } }, sendTerminalAgentPrompt, + sendTerminal, readLaunchedAgentForeground: vi.fn( async (_ptyId: string): Promise => 'agent' ), @@ -248,6 +265,8 @@ export function fakeTerminal() { terminal, state, writes, + keys, + sendTerminal, sendTerminalAgentPrompt, resumeAgentSessionInWorkspace, waitForFreshWorkerComposer, diff --git a/src/main/external-conversations/external-conversations-install.test.ts b/src/main/external-conversations/external-conversations-install.test.ts index d511ff9f72b..9d93bf2928f 100644 --- a/src/main/external-conversations/external-conversations-install.test.ts +++ b/src/main/external-conversations/external-conversations-install.test.ts @@ -49,7 +49,8 @@ function runtime() { } }, subscribeAgentProviderSessionChangesFn: null, - getAgentStatusRowsForPane: (): AgentStatusIpcPayload[] => [] + getAgentStatusRowsForPane: (): AgentStatusIpcPayload[] => [], + agentTransitions: { latestRow: vi.fn(() => null), subscribe: vi.fn(() => () => {}) } } } @@ -74,6 +75,7 @@ describe('installExternalConversations', () => { expect(meta).not.toHaveBeenCalled() expect(folders).not.toHaveBeenCalled() expect(rows).not.toHaveBeenCalled() + expect(host.agentTransitions.subscribe).not.toHaveBeenCalled() }) it('builds the seam on first use when the setting is already on', () => { diff --git a/src/main/external-conversations/external-conversations-install.ts b/src/main/external-conversations/external-conversations-install.ts index e1094979706..fb78d93821e 100644 --- a/src/main/external-conversations/external-conversations-install.ts +++ b/src/main/external-conversations/external-conversations-install.ts @@ -1,4 +1,5 @@ import type { GlobalSettings } from '../../shared/global-settings-types' +import type { AgentTransitionDetector } from '../agent-hooks/agent-transition-detector' import type { RuntimeWorktreeLifecycleEvent } from '../runtime/orca-runtime-core' import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts' import type { ExternalConversationLinkStore } from './external-conversation-links' @@ -29,6 +30,7 @@ export function installExternalConversations( subscribeAgentProviderSessionChangesFn: | ExternalConversationSeamDeps['onProviderSessionsChanged'] | null + agentTransitions: Pick } & ExternalConversationSeamDeps['sessionRows'] ): { current(): ExternalConversationSeam | null } { if (!store) { @@ -55,7 +57,9 @@ export function installExternalConversations( sessionRows: runtime, // A host without a hook store reports no sessions; reopen then refuses for want of one. onProviderSessionsChanged: (listener) => - runtime.subscribeAgentProviderSessionChangesFn?.(listener) ?? (() => {}) + runtime.subscribeAgentProviderSessionChangesFn?.(listener) ?? (() => {}), + readHookRow: (paneKey) => runtime.agentTransitions.latestRow(paneKey), + onAgentPaneChange: (listener) => runtime.agentTransitions.subscribe(listener) }) } else if (!enabled && seam) { seam.dispose('connectors_disabled') diff --git a/src/main/runtime/rpc/methods/connector.ts b/src/main/runtime/rpc/methods/connector.ts index ed3a38b6547..b4a82f51e9c 100644 --- a/src/main/runtime/rpc/methods/connector.ts +++ b/src/main/runtime/rpc/methods/connector.ts @@ -4,7 +4,18 @@ * refused here whoever makes it. */ -import { ConnectorSend, ConnectorStart } from '../../../../shared/rpc-contract/connector-params' +import { delayUnlessAborted } from '../../../../shared/abort-aware-delay' +import { + ConnectorAnswer, + ConnectorPendingPrompt, + ConnectorSend, + ConnectorStart, + ConnectorWatch +} from '../../../../shared/rpc-contract/connector-params' +import type { + ConversationEvent, + ConversationSnapshot +} from '../../../external-conversations/external-conversation-events' import { runtimeConversationHost, type ExternalConversationSeam @@ -51,5 +62,54 @@ export const CONNECTOR_METHODS = [ deliveryId, text ) + }), + defineMethod({ + name: 'connector.pendingPrompt', + permission: 'workspace', + params: ConnectorPendingPrompt, + handler: async ({ source, externalId }, context) => ({ + prompt: requireSeam(context.runtime).pendingPrompt(runtimeConversationHost(context.runtime), { + source, + externalId + }) + }) + }), + defineMethod({ + name: 'connector.answer', + permission: 'workspace', + params: ConnectorAnswer, + handler: async ({ source, externalId, promptId, optionId }, context) => + requireSeam(context.runtime).answer( + runtimeConversationHost(context.runtime), + { source, externalId }, + promptId, + optionId + ) + }), + // Development only: the CLI has no stream transport, so a dev watch collects one window. + defineMethod({ + name: 'connector.watch', + permission: 'workspace', + params: ConnectorWatch, + handler: async ({ forMs }, context) => { + const seam = requireSeam(context.runtime) + let snapshot: ConversationSnapshot | null = null + const events: ConversationEvent[] = [] + const release = seam.watch(runtimeConversationHost(context.runtime), (event) => { + if (event.kind === 'snapshot') { + snapshot = event + } else { + events.push(event) + } + }) + // A closed connection or a disable ends the window early. + const ended = AbortSignal.any([seam.signal, ...(context.signal ? [context.signal] : [])]) + try { + await delayUnlessAborted(forMs, ended) + } finally { + release() + } + return { snapshot, events } + } }) ] diff --git a/src/main/runtime/terminal-agent-prompt-answer.test.ts b/src/main/runtime/terminal-agent-prompt-answer.test.ts new file mode 100644 index 00000000000..284c4ef0600 --- /dev/null +++ b/src/main/runtime/terminal-agent-prompt-answer.test.ts @@ -0,0 +1,178 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { WriteSettlement } from '../../shared/pty-write-settlement' +import type { RuntimeTerminalSend } from '../../shared/runtime-terminal-send-contract' +import { TerminalAgentPromptAnswers } from './terminal-agent-prompt-answer' + +const PANE = 'tab-1:11111111-1111-4111-8111-111111111111' +const target = { paneKey: PANE, agent: 'claude', answerable: true } + +function approvalRow(stateStartedAt = 10): AgentStatusIpcPayload { + return { + state: 'waiting', + prompt: '', + agentType: 'claude', + paneKey: PANE, + connectionId: null, + receivedAt: stateStartedAt, + stateStartedAt, + interactivePrompt: JSON.stringify({ approval: { tool: 'Bash', summary: 'ls' } }) + } +} + +function harness() { + const state: { + row: AgentStatusIpcPayload | null + incarnation: string + settlement: WriteSettlement + beforeWrite?: () => void + } = { row: approvalRow(), incarnation: 'inc-1', settlement: { outcome: 'accepted' } } + const keys: string[] = [] + const runtime = { + getTerminalHandleForPaneKey: (paneKey: string) => (paneKey === PANE ? 'term-1' : null), + getTerminalPromptRequestBinding: () => ({ + ptyId: 'pty-1', + processIncarnation: state.incarnation, + generation: 1 + }), + sendTerminal: vi.fn( + async ( + handle: string, + action: { text?: string; enter?: boolean }, + options: { beforeWrite?: (ptyId: string) => void | Promise } + ): Promise => { + state.beforeWrite?.() + await options.beforeWrite?.('pty-1') + expect(action.enter).toBe(false) + if (state.settlement.outcome === 'accepted') { + keys.push(action.text ?? '') + } + return { + handle, + accepted: state.settlement.outcome === 'accepted', + bytesWritten: 1, + writeSettlement: state.settlement + } + } + ) + } + const answers = new TerminalAgentPromptAnswers(() => state.row) + const signal = new AbortController().signal + const answer = (promptId: string, optionId = 'allow_once') => + answers.answer(runtime, target, { promptId, optionId, signal }) + return { state, keys, runtime, answers, answer } +} + +describe('TerminalAgentPromptAnswers', () => { + it('writes the chosen option’s key, without Enter, for the prompt still pending', async () => { + const h = harness() + const prompt = h.answers.pending(h.runtime, target) + expect(prompt?.kind).toBe('approval') + await expect(h.answer(prompt!.promptId, 'reject_once')).resolves.toEqual({ + outcome: 'accepted' + }) + expect(h.keys).toEqual(['\x1b']) + }) + + it('refuses a second answer to the same prompt before the agent reports moving on', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + await h.answer(promptId) + expect(h.answers.pending(h.runtime, target)).toBeNull() + await expect(h.answer(promptId)).resolves.toEqual({ + outcome: 'refused', + reason: 'prompt_not_pending' + }) + expect(h.keys).toEqual(['1']) + }) + + it('refuses a prompt the agent has moved on from, or a later one', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + h.state.row = approvalRow(20) + await expect(h.answer(promptId)).resolves.toMatchObject({ reason: 'prompt_not_pending' }) + h.state.row = { ...approvalRow(20), state: 'working' } + await expect(h.answer(promptId)).resolves.toMatchObject({ reason: 'prompt_not_pending' }) + expect(h.keys).toEqual([]) + }) + + it('refuses when the prompt or the process changes between the read and the write', async () => { + const h = harness() + const first = h.answers.pending(h.runtime, target)!.promptId + h.state.beforeWrite = () => { + h.state.incarnation = 'inc-2' + } + await expect(h.answer(first)).resolves.toMatchObject({ reason: 'prompt_not_pending' }) + h.state.incarnation = 'inc-1' + h.state.beforeWrite = () => { + h.state.row = approvalRow(30) + } + h.state.row = approvalRow() + await expect(h.answer(first)).resolves.toMatchObject({ reason: 'prompt_not_pending' }) + expect(h.keys).toEqual([]) + }) + + it('refuses an option the prompt does not offer, and a prompt only Orca can answer', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + await expect(h.answer(promptId, 'allow_always')).resolves.toMatchObject({ + reason: 'option_not_offered' + }) + const hookless = { ...target, answerable: false } + const unsupported = h.answers.pending(h.runtime, hookless)! + expect(unsupported.kind).toBe('unsupported') + await expect( + h.answers.answer(h.runtime, hookless, { + promptId: unsupported.promptId, + optionId: 'allow_once', + signal: new AbortController().signal + }) + ).resolves.toMatchObject({ reason: 'prompt_unsupported' }) + expect(h.keys).toEqual([]) + }) + + it('keeps a write the transport refused answerable, and an unprovable one answered', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + h.state.settlement = { outcome: 'refused', reason: 'endpoint_disconnected' } + await expect(h.answer(promptId)).resolves.toMatchObject({ reason: 'agent_not_reachable' }) + h.state.settlement = { + outcome: 'unverifiable', + reason: 'provider_threw_after_handoff', + bytesHandedToTransport: true + } + await expect(h.answer(promptId)).resolves.toEqual({ outcome: 'unknown' }) + await expect(h.answer(promptId)).resolves.toMatchObject({ reason: 'prompt_not_pending' }) + }) + + it('honors an older writer’s refusal without a settlement', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + h.runtime.sendTerminal.mockResolvedValueOnce({ + handle: 'term-1', + accepted: false, + bytesWritten: 0 + }) + await expect(h.answer(promptId)).resolves.toEqual({ + outcome: 'refused', + reason: 'agent_not_reachable' + }) + expect(h.answers.pending(h.runtime, target)?.promptId).toBe(promptId) + }) + + it('refuses before any key when the caller’s own check fails', async () => { + const h = harness() + const { promptId } = h.answers.pending(h.runtime, target)! + await expect( + h.answers.answer(h.runtime, target, { + promptId, + optionId: 'allow_once', + signal: new AbortController().signal, + beforeWrite: async () => { + throw new Error('agent_not_in_foreground') + } + }) + ).resolves.toEqual({ outcome: 'refused', reason: 'agent_not_reachable' }) + expect(h.keys).toEqual([]) + }) +}) diff --git a/src/main/runtime/terminal-agent-prompt-answer.ts b/src/main/runtime/terminal-agent-prompt-answer.ts new file mode 100644 index 00000000000..91a0e3f66df --- /dev/null +++ b/src/main/runtime/terminal-agent-prompt-answer.ts @@ -0,0 +1,167 @@ +// Answers are fenced to the prompt and process so late keys cannot reach a replacement shell. +import { createHash } from 'node:crypto' +import { readAgentPromptAnswers, type PendingPrompt } from '../../shared/agent-pending-prompt' +import { readTerminalSendAcknowledgment } from '../../shared/terminal-send-acknowledgment' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { OrcaRuntimeService } from './orca-runtime' +import type { AgentPromptExpectedBinding } from './runtime-terminal-contracts' + +export type TerminalAgentPromptAnswerRuntime = Pick< + OrcaRuntimeService, + 'getTerminalHandleForPaneKey' | 'getTerminalPromptRequestBinding' | 'sendTerminal' +> + +export type PanePromptTarget = { + paneKey: string + agent: string + /** The agent's status hooks are on, so its rows say what it waits on. */ + answerable: boolean +} + +type PanePrompt = { + prompt: PendingPrompt + keysByOption: ReadonlyMap + handle: string + binding: AgentPromptExpectedBinding +} + +export type PromptAnswerOutcome = + | { outcome: 'accepted' } + | { + outcome: 'refused' + reason: + | 'prompt_not_pending' + | 'prompt_unsupported' + | 'option_not_offered' + | 'agent_not_reachable' + } + | { outcome: 'unknown' } + +/** One prompt occurrence in one process: a later prompt, or the same one in a new process, differs. */ +function promptIdentity(binding: AgentPromptExpectedBinding, row: AgentStatusIpcPayload): string { + const facts = [ + binding.ptyId, + binding.processIncarnation, + row.stateStartedAt, + row.interactivePrompt ?? '', + row.toolName ?? '' + ] + return `p_${createHash('sha256').update(JSON.stringify(facts)).digest('hex').slice(0, 20)}` +} + +class PromptNoLongerPendingError extends Error { + constructor() { + super('prompt_not_pending') + } +} + +/** + * Answers prompts on this host. Remembers what it answered, so a second answer to one prompt + * refuses even while the agent has not yet reported moving on. + */ +export class TerminalAgentPromptAnswers { + private readonly answered = new Map() + + constructor(private readonly readHookRow: (paneKey: string) => AgentStatusIpcPayload | null) {} + + pending( + runtime: TerminalAgentPromptAnswerRuntime, + target: PanePromptTarget + ): PendingPrompt | null { + return this.read(runtime, target)?.prompt ?? null + } + + async answer( + runtime: TerminalAgentPromptAnswerRuntime, + target: PanePromptTarget, + answer: { + promptId: string + optionId: string + /** The caller's own check before the keys go in (its agent-in-front guard). */ + beforeWrite?: (ptyId: string) => Promise + signal: AbortSignal + } + ): Promise { + const asked = this.read(runtime, target) + if (!asked || asked.prompt.promptId !== answer.promptId) { + return { outcome: 'refused', reason: 'prompt_not_pending' } + } + if (asked.prompt.kind === 'unsupported') { + return { outcome: 'refused', reason: 'prompt_unsupported' } + } + const keys = asked.keysByOption.get(answer.optionId) + if (keys === undefined) { + return { outcome: 'refused', reason: 'option_not_offered' } + } + let claimed = false + try { + const sent = await runtime.sendTerminal( + asked.handle, + { text: keys, enter: false }, + { + inputKind: 'driving', + signal: answer.signal, + requireWriteSettlement: true, + beforeWrite: async (ptyId) => { + await answer.beforeWrite?.(ptyId) + // Last and synchronous, so nothing moves between this read and the write. + const now = this.read(runtime, target) + if ( + ptyId !== asked.binding.ptyId || + now?.prompt.promptId !== answer.promptId || + now.binding.processIncarnation !== asked.binding.processIncarnation + ) { + throw new PromptNoLongerPendingError() + } + this.answered.set(target.paneKey, answer.promptId) + claimed = true + } + } + ) + const settlement = readTerminalSendAcknowledgment({ send: sent }) + if (settlement === 'accepted') { + return { outcome: 'accepted' } + } + if (settlement === 'refused') { + if (this.answered.get(target.paneKey) === answer.promptId) { + this.answered.delete(target.paneKey) + } + return { outcome: 'refused', reason: 'agent_not_reachable' } + } + return { outcome: 'unknown' } + } catch (error) { + if (claimed) { + return { outcome: 'unknown' } + } + return { + outcome: 'refused', + reason: + error instanceof PromptNoLongerPendingError ? 'prompt_not_pending' : 'agent_not_reachable' + } + } + } + + private read( + runtime: TerminalAgentPromptAnswerRuntime, + target: PanePromptTarget + ): PanePrompt | null { + const handle = runtime.getTerminalHandleForPaneKey(target.paneKey) + const row = handle ? this.readHookRow(target.paneKey) : null + if (!handle || !row) { + return null + } + let binding: AgentPromptExpectedBinding + try { + const { ptyId, processIncarnation } = runtime.getTerminalPromptRequestBinding(handle) + binding = { ptyId, processIncarnation } + } catch { + return null + } + const promptId = promptIdentity(binding, row) + if (this.answered.get(target.paneKey) === promptId) { + return null + } + const read = readAgentPromptAnswers(row, target.agent, promptId, target.answerable) + return read ? { ...read, handle, binding } : null + } +} diff --git a/src/main/runtime/terminal-agent-turn-news.test.ts b/src/main/runtime/terminal-agent-turn-news.test.ts new file mode 100644 index 00000000000..89e990e449b --- /dev/null +++ b/src/main/runtime/terminal-agent-turn-news.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' +import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' +import type { AgentPhase, AgentTransition } from '../../shared/agent-transitions' +import { TerminalAgentTurnNews, type TerminalAgentTurnNewsDeps } from './terminal-agent-turn-news' + +const PANE = 'tab-1:11111111-1111-4111-8111-111111111111' +const WT = 'repo-1::/tmp/wt' + +function row(overrides: Partial): AgentStatusIpcPayload { + return { + state: 'working', + prompt: '', + agentType: 'claude', + paneKey: PANE, + connectionId: null, + receivedAt: 1, + stateStartedAt: 100, + ...overrides + } +} + +function moved( + from: AgentPhase, + to: AgentPhase, + transition: AgentTransition | null +): AgentPaneChange { + return { + kind: 'moved', + paneKey: PANE, + workspace: { worktreeId: WT, sshHostId: null }, + from, + to, + at: 1, + transition + } +} + +function news(current: { row: AgentStatusIpcPayload | null }, optIn: boolean) { + const readHookRow = vi.fn(() => current.row) + const deps: TerminalAgentTurnNewsDeps = { + readHookRow, + prepare: ({ row }) => row.state, + deliver: vi.fn(), + setTimer: () => ({ cancel: () => {} }), + ...(optIn ? { started: vi.fn(), stopped: vi.fn() } : {}) + } + return { turnNews: new TerminalAgentTurnNews(deps), deps, readHookRow } +} + +describe('TerminalAgentTurnNews opt-ins', () => { + it('tells a turn start, and a turn its user stopped, only to a listener that opted in', () => { + const current: { row: AgentStatusIpcPayload | null } = { row: row({ state: 'working' }) } + const { turnNews, deps } = news(current, true) + turnNews.observe(moved('rest', 'working', 'prompt-submitted')) + expect(deps.started).toHaveBeenCalledWith({ paneKey: PANE, worktreeId: WT, row: current.row }) + current.row = row({ state: 'done', interrupted: true }) + turnNews.observe(moved('working', 'rest', null)) + expect(deps.stopped).toHaveBeenCalledWith({ paneKey: PANE, worktreeId: WT, row: current.row }) + expect(deps.deliver).not.toHaveBeenCalled() + }) + + it('tells no stop for a pane that rested without a stopped turn', () => { + const current: { row: AgentStatusIpcPayload | null } = { row: null } + const { turnNews, deps } = news(current, true) + turnNews.observe(moved('working', 'rest', null)) + current.row = row({ state: 'done' }) + turnNews.observe(moved('done', 'rest', null)) + expect(deps.stopped).not.toHaveBeenCalled() + }) + + it('reads no extra row for a listener that did not opt in', () => { + const current: { row: AgentStatusIpcPayload | null } = { row: row({ state: 'working' }) } + const { turnNews, readHookRow } = news(current, false) + turnNews.observe(moved('rest', 'working', 'prompt-submitted')) + current.row = row({ state: 'done', interrupted: true }) + turnNews.observe(moved('working', 'rest', null)) + expect(readHookRow).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/terminal-agent-turn-news.ts b/src/main/runtime/terminal-agent-turn-news.ts index 75004eeac3b..36c2b6ddb93 100644 --- a/src/main/runtime/terminal-agent-turn-news.ts +++ b/src/main/runtime/terminal-agent-turn-news.ts @@ -2,6 +2,7 @@ // finished (or failed) and an agent that needs input, each with its hook row. The rules and timing // are the window's; the listener (the phone push, a connector) decides only how to word and send it. import type { AgentPaneChange } from '../agent-hooks/agent-transition-detector' +import { agentTurnEndedOnPurpose } from '../../shared/agent-main-agent-verdict' import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload' import { AGENT_DONE_QUIET_MS } from '../../shared/agent-transitions' import { isPiCompatibleAgentType } from '../../shared/pi-agent-kind' @@ -18,6 +19,10 @@ export type TerminalAgentTurnNewsDeps = { prepare: (news: TerminalAgentTurnNewsItem) => Prepared deliver: (prepared: Prepared) => void setTimer: (run: () => void, delayMs: number) => { cancel: () => void } + /** Opt-in: a turn began (a prompt went in), told at once. */ + started?: (news: TerminalAgentTurnNewsItem) => void + /** Opt-in: a turn its user stopped or a newer request replaced, which is otherwise no news. */ + stopped?: (news: TerminalAgentTurnNewsItem) => void } type PendingFinish = { @@ -51,6 +56,7 @@ export class TerminalAgentTurnNews { return } const { transition } = change + this.tellOptIn(change) const finished = transition === 'turn-finished' || transition === 'errored' const row = finished || transition === 'needs-input' ? this.deps.readHookRow(paneKey) : null // The all-clear of a turn announced when its lead agent finished: that finish still stands. @@ -81,6 +87,30 @@ export class TerminalAgentTurnNews { this.announce(paneKey, change.workspace.worktreeId, row, waitOutQuiet) } + dispose(): void { + for (const paneKey of this.pendingByPane.keys()) { + this.cancelPending(paneKey) + } + this.finishedTurnByPane.clear() + this.workingSeen.clear() + } + + private tellOptIn(change: Extract): void { + const { paneKey, workspace } = change + if (change.transition === 'prompt-submitted' && this.deps.started) { + const row = this.deps.readHookRow(paneKey) + if (row && !row.structuredHost) { + this.deps.started({ paneKey, worktreeId: workspace.worktreeId, row }) + } + } + if (change.to === 'rest' && change.from !== 'rest' && this.deps.stopped) { + const row = this.deps.readHookRow(paneKey) + if (row?.state === 'done' && !row.structuredHost && agentTurnEndedOnPurpose(row)) { + this.deps.stopped({ paneKey, worktreeId: workspace.worktreeId, row }) + } + } + } + private observeWorking(paneKey: string, worktreeId: string | undefined): void { const row = this.deps.readHookRow(paneKey) if (!row || row.state !== 'working' || !worktreeId) { diff --git a/src/shared/agent-pending-prompt.test.ts b/src/shared/agent-pending-prompt.test.ts new file mode 100644 index 00000000000..a8ffc8fc6bf --- /dev/null +++ b/src/shared/agent-pending-prompt.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from 'vitest' +import { readAgentPromptAnswers } from './agent-pending-prompt' + +const approval = JSON.stringify({ approval: { tool: 'Bash', summary: 'rm -rf build' } }) + +function question(overrides: Record = {}, count = 1): string { + const one = { + question: 'Which color?', + header: 'Color', + multiSelect: false, + options: [{ label: 'Red', description: 'warm' }, { label: 'Blue' }], + ...overrides + } + return JSON.stringify({ questions: Array.from({ length: count }, () => one) }) +} + +function waiting(interactivePrompt: string, toolName?: string) { + return { state: 'waiting' as const, interactivePrompt, ...(toolName ? { toolName } : {}) } +} + +describe('readAgentPromptAnswers', () => { + it('reads an approval with the cards’ own allow and deny keys', () => { + const read = readAgentPromptAnswers(waiting(approval), 'claude', 'p1', true) + expect(read?.prompt).toEqual({ + kind: 'approval', + promptId: 'p1', + tool: 'Bash', + summary: 'rm -rf build', + options: [ + { id: 'allow_once', label: 'Allow once' }, + { id: 'reject_once', label: 'Reject' } + ] + }) + expect(Object.fromEntries(read!.keysByOption)).toEqual({ + allow_once: '1', + reject_once: '\x1b' + }) + expect( + readAgentPromptAnswers(waiting(approval), 'codex', 'p1', true)?.keysByOption.get('allow_once') + ).toBe('1') + }) + + it('reads a one-question single choice with each option’s selector key', () => { + for (const agent of ['claude', 'codex']) { + const read = readAgentPromptAnswers(waiting(question(), 'AskUserQuestion'), agent, 'p2', true) + expect(read?.prompt).toEqual({ + kind: 'question', + promptId: 'p2', + question: 'Which color?', + header: 'Color', + options: [ + { id: '1', label: 'Red', description: 'warm' }, + { id: '2', label: 'Blue' } + ] + }) + expect(Object.fromEntries(read!.keysByOption)).toEqual({ '1': '1', '2': '2' }) + } + }) + + it('leaves to Orca what one keystroke cannot answer', () => { + const unsupported = { kind: 'unsupported', promptId: 'p' } + const cases = [ + waiting(question({}, 2)), + waiting(question({ multiSelect: true })), + waiting(question({ options: [] })), + waiting('not json'), + { state: 'blocked' as const } + ] + for (const row of cases) { + expect(readAgentPromptAnswers(row, 'claude', 'p', true)?.prompt).toEqual(unsupported) + } + }) + + it('leaves every prompt to Orca for an agent without hooks or known keys', () => { + expect(readAgentPromptAnswers(waiting(approval), 'claude', 'p', false)?.prompt.kind).toBe( + 'unsupported' + ) + expect(readAgentPromptAnswers(waiting(approval), 'aider', 'p', true)?.prompt.kind).toBe( + 'unsupported' + ) + }) + + it('reads nothing while the agent is not waiting on the user', () => { + expect( + readAgentPromptAnswers({ state: 'working', interactivePrompt: approval }, 'claude', 'p', true) + ).toBeNull() + expect(readAgentPromptAnswers({ state: 'done' }, 'claude', 'p', true)).toBeNull() + }) +}) diff --git a/src/shared/agent-pending-prompt.ts b/src/shared/agent-pending-prompt.ts new file mode 100644 index 00000000000..fa23c4836f7 --- /dev/null +++ b/src/shared/agent-pending-prompt.ts @@ -0,0 +1,121 @@ +import type { AgentStatusIpcPayload } from './agent-status-ipc-payload' +import { hookRowAgentPhase } from './agent-transitions' +import { parseAskFromStatus } from './native-chat-ask' +import { buildAskAnswerKeys, buildCodexAskAnswerKeys } from './native-chat-ask-answer' +import { parseNativeChatApprovalEnvelope } from './native-chat-approval-envelope' +import { + resolveNativeChatTranscriptAgent, + shouldStepNativeChatAskAnswer +} from './native-chat-agent-support' + +export type PendingPromptOption = { id: string; label: string; description?: string } + +/** + * What a hook-reported terminal agent is waiting on, as one outside the window can answer it. + * Readers must treat a kind they do not know as `unsupported`: more kinds (free text, several + * questions) may come. + */ +export type PendingPrompt = + | { + kind: 'approval' + promptId: string + tool: string + summary?: string + options: PendingPromptOption[] + } + | { + kind: 'question' + promptId: string + question: string + header?: string + options: PendingPromptOption[] + } + /** The agent is waiting on something only Orca's own views can answer. */ + | { kind: 'unsupported'; promptId: string } + +type PromptRow = Pick< + AgentStatusIpcPayload, + 'state' | 'interrupted' | 'mainAgent' | 'interactivePrompt' | 'toolName' +> + +/** The row's prompt and the bytes each offered option writes; null while the agent is not waiting. */ +export function readAgentPromptAnswers( + row: PromptRow, + agent: string, + promptId: string, + /** Whether this agent's status hooks and answer keys are known; without them only Orca can answer. */ + answerable: boolean +): { prompt: PendingPrompt; keysByOption: ReadonlyMap } | null { + if (hookRowAgentPhase(row) !== 'needs-you') { + return null + } + const unsupported = { + prompt: { kind: 'unsupported' as const, promptId }, + keysByOption: new Map() + } + if (!answerable || resolveNativeChatTranscriptAgent(agent) === null) { + return unsupported + } + const approval = parseNativeChatApprovalEnvelope(row.interactivePrompt, agent) + if (approval) { + return { + prompt: { + kind: 'approval', + promptId, + tool: approval.tool, + ...(approval.summary ? { summary: approval.summary } : {}), + options: [ + { id: 'allow_once', label: 'Allow once' }, + { id: 'reject_once', label: 'Reject' } + ] + }, + keysByOption: new Map([ + ['allow_once', approval.allowSend], + ['reject_once', approval.denySend] + ]) + } + } + const ask = parseAskFromStatus(row.interactivePrompt, row.toolName) + const question = ask?.questions.length === 1 ? ask.questions[0] : undefined + if (!ask || !question || question.multiSelect || ask.replyAsMessage) { + return unsupported + } + if (!shouldStepNativeChatAskAnswer(agent)) { + return unsupported + } + const keysByOption = new Map() + const options = question.options.flatMap((option, index): PendingPromptOption[] => { + const selections = [{ indices: [index] }] + const groups = + resolveNativeChatTranscriptAgent(agent) === 'codex' + ? buildCodexAskAnswerKeys(ask, selections) + : buildAskAnswerKeys(ask, selections) + // A pick that takes more than one keystroke needs the paced writer only Orca's views run. + const only = groups.length === 1 ? groups[0] : undefined + if (!only || !('raw' in only)) { + return [] + } + const id = String(index + 1) + keysByOption.set(id, only.raw) + return [ + { + id, + label: option.label, + ...(option.description ? { description: option.description } : {}) + } + ] + }) + if (options.length === 0 || options.length !== question.options.length) { + return unsupported + } + return { + prompt: { + kind: 'question', + promptId, + question: question.question, + ...(question.header ? { header: question.header } : {}), + options + }, + keysByOption + } +} diff --git a/src/shared/rpc-contract/connector-params.ts b/src/shared/rpc-contract/connector-params.ts index 57df28d2238..b1b07ecdf6c 100644 --- a/src/shared/rpc-contract/connector-params.ts +++ b/src/shared/rpc-contract/connector-params.ts @@ -21,3 +21,25 @@ export const ConnectorSend = z.object({ ...ConnectorDelivery, text: z.string().min(1, 'Missing text') }) + +const ConnectorConversation = { + source: ConnectorDelivery.source, + externalId: ConnectorDelivery.externalId +} + +export const ConnectorPendingPrompt = z.object(ConnectorConversation) + +export const ConnectorAnswer = z.object({ + ...ConnectorConversation, + /** The `promptId` the answer is for; any other prompt, or none, refuses. */ + promptId: z.string().min(1, 'Missing prompt id'), + optionId: z.string().min(1, 'Missing option id') +}) + +/** Longest a dev watch window may stay open. */ +export const CONNECTOR_WATCH_MAX_MS = 10 * 60 * 1000 + +/** A dev window over the events: the snapshot, then every event until `forMs` passes. */ +export const ConnectorWatch = z.object({ + forMs: z.number().int().min(0).max(CONNECTOR_WATCH_MAX_MS) +}) diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index 6caf287dc62..34994eef78a 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -177,7 +177,13 @@ import { SetValue, TypeText } from './computer-schemas-params' -import { ConnectorSend, ConnectorStart } from './connector-params' +import { + ConnectorAnswer, + ConnectorPendingPrompt, + ConnectorSend, + ConnectorStart, + ConnectorWatch +} from './connector-params' import { AttachParams as AttachParamsOfEmulatorParams, AxParams, @@ -894,8 +900,11 @@ export const RPC_PARAMS_BY_METHOD = { 'computer.scroll': ScrollOfComputerSchemasParams, 'computer.setValue': SetValue, 'computer.typeText': TypeText, + 'connector.answer': ConnectorAnswer, + 'connector.pendingPrompt': ConnectorPendingPrompt, 'connector.send': ConnectorSend, 'connector.start': ConnectorStart, + 'connector.watch': ConnectorWatch, 'desktopBrowserBridge.attach': DesktopBrowserBridgeAttachParams, 'desktopBrowserBridge.result': DesktopBrowserBridgeResultParams, 'diagnostics.memory': null,