From 57b50824044965cf4ff3bb53c359eca4eb66c284 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 22 Oct 2026 17:40:40 -0400 Subject: [PATCH] feat(connectors): connector start and conversation links behind the off-by-default experimentalConnectors flag (B2b-1) (#31805) * feat(connectors): connector seam lifecycle, conversation links and start behind experimentalConnectors (B2b-1) * feat(agent-write): opt-in launch cancel, approval wait and binding fence on the shared writer executeAgentLaunch takes an optional signal; sendTerminalAgentPrompt (and sendAgentTurn) take optional onPermission: 'wait' and expectedBinding. No existing caller sets them, so every current path is unchanged. * fix(agent-write): keep the default writer path main's; fence cancel and real submit boundary Only opted-in writes take the submit layer, so default timing and the leaf re-read are unchanged. Cancellation is rechecked before a structured-refusal terminal fallback, and an opted-in launch learns Enter from an onSubmitWritten callback at the write itself, OMP's combined frame included. * fix(connectors): pass the seam's signal to the launch, read links from each host's rows, and keep delivery ids on orchestration's durable receipts (B2b-1) * fix(agent-launch): a cancel after a failed startup spawn recovers the created workspace * fix(connectors): keep an unknown receipt once the launch published, and restore local-only metadata ownership (B2b-1) * fix(connectors): classify connector RPCs as host-owned; CLI start output claims a start only when definite --- src/cli/handler-group-manifest.ts | 5 + src/cli/handlers/connector.test.ts | 35 +++ src/cli/handlers/connector.ts | 76 +++++ src/cli/specs/connector.ts | 31 ++ src/cli/specs/index.ts | 2 + .../connector-agents.ts | 21 ++ .../external-conversation-chain.ts | 33 ++ ...external-conversation-delivery-receipts.ts | 82 +++++ .../external-conversation-links.ts | 112 +++++++ .../external-conversation-seam.test.ts | 291 ++++++++++++++++++ .../external-conversation-seam.ts | 99 ++++++ .../external-conversation-start.ts | 120 ++++++++ .../external-conversation-types.ts | 33 ++ .../external-conversation.test-fixture.ts | 116 +++++++ .../external-conversations-install.test.ts | 79 +++++ .../external-conversations-install.ts | 56 ++++ ...rsistence-local-only-worktree-meta.test.ts | 9 + .../host-qualified-worktree-meta.ts | 13 +- .../folder-workspace-operations.ts | 4 + src/main/runtime/orca-runtime.ts | 35 ++- src/main/runtime/rpc/errors.ts | 1 + .../runtime/rpc/methods/connector.test.ts | 10 + src/main/runtime/rpc/methods/connector.ts | 43 +++ src/main/runtime/rpc/methods/index.ts | 2 + .../runtime/runtime-client-settings-shape.ts | 1 + src/main/runtime/runtime-reference-catalog.ts | 25 ++ src/main/startup/cli-command-names.ts | 1 + src/shared/external-conversation-link.ts | 15 + src/shared/folder-workspace-types.ts | 3 + src/shared/folder-workspaces.ts | 6 +- src/shared/host-operation-ownership.ts | 1 + .../rpc-contract/client-settings-params.ts | 2 + src/shared/rpc-contract/connector-params.ts | 18 ++ .../rpc-params-catalog.generated.ts | 2 + src/shared/worktree/meta-types.ts | 3 + 35 files changed, 1363 insertions(+), 22 deletions(-) create mode 100644 src/cli/handlers/connector.test.ts create mode 100644 src/cli/handlers/connector.ts create mode 100644 src/cli/specs/connector.ts create mode 100644 src/main/external-conversations/connector-agents.ts create mode 100644 src/main/external-conversations/external-conversation-chain.ts create mode 100644 src/main/external-conversations/external-conversation-delivery-receipts.ts create mode 100644 src/main/external-conversations/external-conversation-links.ts create mode 100644 src/main/external-conversations/external-conversation-seam.test.ts create mode 100644 src/main/external-conversations/external-conversation-seam.ts create mode 100644 src/main/external-conversations/external-conversation-start.ts create mode 100644 src/main/external-conversations/external-conversation-types.ts create mode 100644 src/main/external-conversations/external-conversation.test-fixture.ts create mode 100644 src/main/external-conversations/external-conversations-install.test.ts create mode 100644 src/main/external-conversations/external-conversations-install.ts create mode 100644 src/main/runtime/rpc/methods/connector.test.ts create mode 100644 src/main/runtime/rpc/methods/connector.ts create mode 100644 src/shared/external-conversation-link.ts create mode 100644 src/shared/rpc-contract/connector-params.ts diff --git a/src/cli/handler-group-manifest.ts b/src/cli/handler-group-manifest.ts index 63be9a63f7e..b43c983347f 100644 --- a/src/cli/handler-group-manifest.ts +++ b/src/cli/handler-group-manifest.ts @@ -212,6 +212,11 @@ export const HANDLER_GROUPS: readonly HandlerGroup[] = [ keys: ['agent launch'], load: async () => (await import('./handlers/agent-launch.js')).AGENT_LAUNCH_HANDLERS }, + { + name: 'connector', + keys: ['connector start'], + load: async () => (await import('./handlers/connector.js')).CONNECTOR_HANDLERS + }, { name: 'profile-state', keys: ['profile state exports', 'profile state rollback'], diff --git a/src/cli/handlers/connector.test.ts b/src/cli/handlers/connector.test.ts new file mode 100644 index 00000000000..da7e3e8a3fa --- /dev/null +++ b/src/cli/handlers/connector.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest' +import { RuntimeClientError } from '../runtime-client' +import { callConnector, formatConnectorStart } from './connector' + +function refusingClient(code: string) { + return { + call: async (): Promise => { + throw new RuntimeClientError(code, code) + } + } +} + +describe('orca connector', () => { + it('passes the host’s connectors_disabled refusal through, with no local flag check', async () => { + await expect( + callConnector(refusingClient('connectors_disabled'), 'connector.start', {}) + ).rejects.toMatchObject({ code: 'connectors_disabled' }) + }) + + it('names the update instead of surfacing method_not_found from an older host', async () => { + await expect( + callConnector(refusingClient('method_not_found'), 'connector.start', {}) + ).rejects.toThrow('Update Orca on the host') + }) + + it('claims a start only for a definite answer', () => { + expect(formatConnectorStart({ outcome: 'accepted' })).toBe('started; prompt accepted') + expect(formatConnectorStart({ outcome: 'refused', reason: 'unsupported_agent' })).toBe( + 'not started; prompt refused: unsupported_agent' + ) + expect(formatConnectorStart({ outcome: 'unknown', reason: 'replayed_after_restart' })).toBe( + 'prompt unknown: replayed_after_restart' + ) + }) +}) diff --git a/src/cli/handlers/connector.ts b/src/cli/handlers/connector.ts new file mode 100644 index 00000000000..9fcbf1f33f1 --- /dev/null +++ b/src/cli/handlers/connector.ts @@ -0,0 +1,76 @@ +import type { CommandHandler, HandlerContext } from '../dispatch' +import { getOptionalStringFlag, getRequiredStringFlag } from '../flags' +import { printResult } from '../format' +import { RuntimeClientError } from '../runtime-client' +import { getRequiredWorktreeSelector } from '../selectors' + +type ConnectorDelivery = { outcome: string; reason?: string } + +function formatDelivery(delivery: ConnectorDelivery): string { + return delivery.reason ? `${delivery.outcome}: ${delivery.reason}` : delivery.outcome +} + +/** `unknown` cannot say whether this call started the agent, so only the definite answers claim it. */ +export function formatConnectorStart(initial: ConnectorDelivery): string { + const prompt = `prompt ${formatDelivery(initial)}` + if (initial.outcome === 'accepted') { + return `started; ${prompt}` + } + return initial.outcome === 'refused' ? `not started; ${prompt}` : prompt +} + +// The host decides whether connectors are on; this only names an older host's refusal. +export async function callConnector( + client: Pick, + method: string, + params: Record +) { + try { + return await client.call(method, params) + } catch (error) { + if (error instanceof RuntimeClientError && error.code === 'method_not_found') { + throw new RuntimeClientError( + 'incompatible_runtime', + 'This Orca host has no connectors. Update Orca on the host and try again.' + ) + } + throw error + } +} + +function conversationParams(flags: HandlerContext['flags']) { + return { + source: 'dev', + externalId: getRequiredStringFlag(flags, 'conversation'), + deliveryId: getRequiredStringFlag(flags, 'delivery-id') + } +} + +export const CONNECTOR_HANDLERS: Record = { + 'connector start': async ({ flags, client, cwd, json }) => { + const projectGroupId = getOptionalStringFlag(flags, 'project-group') + if ((projectGroupId !== undefined) === flags.has('worktree')) { + throw new RuntimeClientError( + 'invalid_argument', + 'Pass exactly one of --worktree or --project-group.' + ) + } + const name = getOptionalStringFlag(flags, 'name') + const target = projectGroupId + ? { + kind: 'create-folder-workspace', + create: { projectGroupId, ...(name ? { name } : {}) } + } + : { + kind: 'existing', + worktree: await getRequiredWorktreeSelector(flags, 'worktree', cwd, client) + } + const result = await callConnector<{ initial: ConnectorDelivery }>(client, 'connector.start', { + ...conversationParams(flags), + agent: getRequiredStringFlag(flags, 'agent'), + prompt: getRequiredStringFlag(flags, 'prompt'), + target + }) + printResult(result, json, (value) => formatConnectorStart(value.initial)) + } +} diff --git a/src/cli/specs/connector.ts b/src/cli/specs/connector.ts new file mode 100644 index 00000000000..f528d339abc --- /dev/null +++ b/src/cli/specs/connector.ts @@ -0,0 +1,31 @@ +import type { CommandSpec } from '../args' +import { GLOBAL_FLAGS } from '../args' + +const CONNECTOR_NOTES = [ + 'Development only: drives the connector seam as a `dev` conversation, without Slack. The host refuses it unless its experimentalConnectors setting is on.', + 'A retry with the same --delivery-id and payload returns the first answer instead of delivering again.' +] + +export const CONNECTOR_COMMAND_SPECS: CommandSpec[] = [ + { + path: ['connector', 'start'], + hidden: true, + summary: 'Start a dev connector conversation with a background terminal agent', + usage: + 'orca connector start --conversation --delivery-id --agent claude|codex --prompt (--worktree | --project-group [--name ]) [--json]', + allowedFlags: [ + ...GLOBAL_FLAGS, + 'conversation', + 'delivery-id', + 'agent', + 'prompt', + 'worktree', + 'project-group', + 'name' + ], + notes: [ + ...CONNECTOR_NOTES, + '--worktree starts in an existing workspace; --project-group creates a folder workspace in that folder project.' + ] + } +] diff --git a/src/cli/specs/index.ts b/src/cli/specs/index.ts index 7b66b87a03e..ceb4d5834b2 100644 --- a/src/cli/specs/index.ts +++ b/src/cli/specs/index.ts @@ -12,6 +12,7 @@ import { ENVIRONMENT_COMMAND_SPECS } from './environment' import { MANAGED_SERVER_COMMAND_SPECS } from './managed-server' import { AGENT_HOOK_COMMAND_SPECS } from './agent-hooks' import { AGENT_LAUNCH_COMMAND_SPECS } from './agent-launch' +import { CONNECTOR_COMMAND_SPECS } from './connector' import { DIAGNOSTICS_COMMAND_SPECS } from './diagnostics' import { SPEECH_COMMAND_SPECS } from './speech' import { TERMINAL_DAEMON_COMMAND_SPECS } from './terminal-daemon' @@ -40,6 +41,7 @@ export const COMMAND_SPECS: CommandSpec[] = [ ...COMPUTER_COMMAND_SPECS, ...AGENT_HOOK_COMMAND_SPECS, ...AGENT_LAUNCH_COMMAND_SPECS, + ...CONNECTOR_COMMAND_SPECS, ...DIAGNOSTICS_COMMAND_SPECS, ...SPEECH_COMMAND_SPECS, ...TERMINAL_DAEMON_COMMAND_SPECS, diff --git a/src/main/external-conversations/connector-agents.ts b/src/main/external-conversations/connector-agents.ts new file mode 100644 index 00000000000..33b283e07c1 --- /dev/null +++ b/src/main/external-conversations/connector-agents.ts @@ -0,0 +1,21 @@ +import type { ConnectorAgent } from '../../shared/external-conversation-link' +import { + isAgentStatusHooksEnabledForAgent, + type AgentStatusHooksSettings +} from '../../shared/agent-status-hooks-setting' + +/** + * The one list of agents a connector may start. Each needs hook status reporting (checked below) + * and a terminal that takes input mid-turn; no capability data records the latter for terminals + * yet, so membership here is where that fact lives. + */ +export const CONNECTOR_AGENTS = ['claude', 'codex'] as const satisfies readonly ConnectorAgent[] + +export function admitConnectorAgent( + settings: AgentStatusHooksSettings, + agent: string +): agent is ConnectorAgent { + return CONNECTOR_AGENTS.some( + (candidate) => candidate === agent && isAgentStatusHooksEnabledForAgent(settings, candidate) + ) +} diff --git a/src/main/external-conversations/external-conversation-chain.ts b/src/main/external-conversations/external-conversation-chain.ts new file mode 100644 index 00000000000..1e407892203 --- /dev/null +++ b/src/main/external-conversations/external-conversation-chain.ts @@ -0,0 +1,33 @@ +import { refused, type Delivery } from './external-conversation-types' + +/** + * Ordering per conversation: one active head, the rest behind it in arrival order. Disposal + * settles everything still waiting at once; the head keeps the outcome of what it already did. + */ +export class ConversationChain { + private readonly tails = new Map>() + + constructor(private readonly signal: AbortSignal) {} + + enqueue(id: string, run: () => Promise): Promise { + const previous = this.tails.get(id) ?? Promise.resolve() + const { signal } = this + const turn = new Promise((resolve) => { + const onAbort = (): void => resolve() + signal.addEventListener('abort', onAbort, { once: true }) + void previous.then(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }) + }) + const result = turn.then(() => (signal.aborted ? refused('connectors_disabled') : run())) + const tail = result.catch(() => undefined) + this.tails.set(id, tail) + void tail.then(() => { + if (this.tails.get(id) === tail) { + this.tails.delete(id) + } + }) + return result + } +} diff --git a/src/main/external-conversations/external-conversation-delivery-receipts.ts b/src/main/external-conversations/external-conversation-delivery-receipts.ts new file mode 100644 index 00000000000..1858b88a219 --- /dev/null +++ b/src/main/external-conversations/external-conversation-delivery-receipts.ts @@ -0,0 +1,82 @@ +import { z } from 'zod' +import type { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' +import { OrchestrationError } from '../runtime/orchestration/orchestration-error' +import { hashCanonical } from '../runtime/rpc/orchestration-mutation-receipt' +import { refused, type ConversationKey, type Delivery } from './external-conversation-types' + +export type DeliveryReceiptDb = OrchestrationDb + +const RecordedDelivery = z.discriminatedUnion('outcome', [ + z.object({ outcome: z.literal('accepted') }), + z.object({ + outcome: z.literal('unknown'), + reason: z.enum(['launch_outcome_unknown', 'writer_failed', 'replayed_after_restart']) + }) +]) + +const LOST: Delivery = { outcome: 'unknown', reason: 'replayed_after_restart' } + +/** + * A connector's delivery ids on orchestration's durable mutation receipts, under one caller + * identity per source: the same id and payload share one answer, across restarts too, and another + * payload is refused. A duplicate of one still running waits for it; a running one a restart + * interrupted reads `unknown`. Refusals and throws wrote nothing, so they leave no receipt. + */ +export class DeliveryReceipts { + private readonly inFlight = new Map>() + + constructor(private readonly db: () => DeliveryReceiptDb) {} + + run( + key: ConversationKey, + deliveryId: string, + method: 'connector.start' | 'connector.send', + payload: unknown, + execute: () => Promise + ): Promise { + const identity = { + callerFingerprint: `connector:${key.source}`, + requestId: JSON.stringify([key.externalId, deliveryId]), + method, + payloadHash: hashCanonical(payload) + } + const flightKey = JSON.stringify([identity.callerFingerprint, identity.requestId]) + const db = this.db() + let begun: ReturnType + try { + begun = db.beginMutationReceipt(identity) + } catch (error) { + if (error instanceof OrchestrationError && error.code === 'request_mismatch') { + return Promise.resolve(refused('delivery_id_reused')) + } + throw error + } + if (begun.disposition === 'completed') { + const recorded = RecordedDelivery.safeParse(JSON.parse(begun.row.receipt ?? 'null')) + return Promise.resolve(recorded.success ? recorded.data : LOST) + } + if (begun.disposition === 'pending') { + return this.inFlight.get(flightKey) ?? Promise.resolve(LOST) + } + const delivery = execute().then( + (result) => { + if (result.outcome === 'refused') { + db.discardPendingMutationReceipt(identity.callerFingerprint, identity.requestId) + } else { + db.completeMutationReceipt({ ...identity, receipt: JSON.stringify(result) }) + } + return result + }, + (error: unknown) => { + db.discardPendingMutationReceipt(identity.callerFingerprint, identity.requestId) + throw error + } + ) + this.inFlight.set(flightKey, delivery) + const forget = (): void => { + this.inFlight.delete(flightKey) + } + delivery.then(forget, forget) + return delivery + } +} diff --git a/src/main/external-conversations/external-conversation-links.ts b/src/main/external-conversations/external-conversation-links.ts new file mode 100644 index 00000000000..4e09a049fc6 --- /dev/null +++ b/src/main/external-conversations/external-conversation-links.ts @@ -0,0 +1,112 @@ +import { z } from 'zod' +import { + CONNECTOR_SOURCES, + type ExternalConversationLink +} from '../../shared/external-conversation-link' +import { + getRepoExecutionHostId, + LOCAL_EXECUTION_HOST_ID, + type ExecutionHostId +} from '../../shared/execution-host' +import { readAllWorktreeMetaForHost } from '../persistence/host-qualified-worktree-meta' +import { folderWorkspaceKey } from '../../shared/workspace-scope' +import type { AgentPaneWorkspace } from '../agent-hooks/agent-transition-detector' +import { + readWorkspaceRecord, + writeWorkspaceRecord, + type WorkspaceRecordStore +} from '../runtime/sidebar-inputs/sort-stamp-records' +import { CONNECTOR_AGENTS } from './connector-agents' +import type { ConversationKey } from './external-conversation-types' + +type LinkRecord = { externalConversations?: unknown } + +/** The workspace records links live on; the runtime's store satisfies it. */ +export type ExternalConversationLinkStore = WorkspaceRecordStore< + LinkRecord, + { externalConversations: ExternalConversationLink[] } +> & { + getAllWorktreeMeta: () => Record + getAllWorktreeMetaForHost?: ( + hostId: ExecutionHostId + ) => Record + getFolderWorkspaces?: () => readonly (LinkRecord & { id: string })[] +} + +export type LinkedConversation = { workspace: AgentPaneWorkspace; link: ExternalConversationLink } + +const ExternalConversationLinkSchema = z.object({ + source: z.enum(CONNECTOR_SOURCES), + externalId: z.string(), + paneKey: z.string(), + agent: z.enum(CONNECTOR_AGENTS), + createdAt: z.number() +}) + +/** The workspace record's links; malformed entries (hand edits, a newer host's shape) are skipped. */ +export function readLinks( + store: ExternalConversationLinkStore, + workspace: AgentPaneWorkspace +): ExternalConversationLink[] { + const record = readWorkspaceRecord(store, workspace) + return linksOf(record?.externalConversations) +} + +function linksOf(value: unknown): ExternalConversationLink[] { + const entries: unknown[] = Array.isArray(value) ? value : [] + return entries.flatMap((entry) => { + const parsed = ExternalConversationLinkSchema.safeParse(entry) + return parsed.success ? [parsed.data] : [] + }) +} + +function sameConversation(link: ExternalConversationLink, key: ConversationKey): boolean { + return link.source === key.source && link.externalId === key.externalId +} + +/** Replaces this conversation's link on the workspace record, host-qualified like any meta write. */ +export function writeLink( + store: ExternalConversationLinkStore, + workspace: AgentPaneWorkspace, + link: ExternalConversationLink +): void { + const others = readLinks(store, workspace).filter((existing) => !sameConversation(existing, link)) + writeWorkspaceRecord(store, workspace, { externalConversations: [...others, link] }) + // The record writer skips a workspace whose host it cannot name rather than guess. + if (!readLinks(store, workspace).some((current) => current.paneKey === link.paneKey)) { + throw new Error('external_conversation_link_unwritten') + } +} + +/** + * The conversation's link, read from every host's canonical rows each time: the locator-keyed map + * keeps only one host's row per id, and a removed, renamed or re-adopted workspace is never + * answered from a cache. + */ +export function findLink( + store: ExternalConversationLinkStore, + key: ConversationKey +): LinkedConversation | null { + const matches = (links: unknown): ExternalConversationLink | undefined => + linksOf(links).find((link) => sameConversation(link, key)) + const hosts = new Set([ + LOCAL_EXECUTION_HOST_ID, + ...store.getRepos().map(getRepoExecutionHostId) + ]) + for (const hostId of hosts) { + for (const [worktreeId, meta] of Object.entries(readAllWorktreeMetaForHost(store, hostId))) { + const link = matches(meta.externalConversations) + if (link) { + const sshHostId = hostId === LOCAL_EXECUTION_HOST_ID ? null : hostId + return { workspace: { worktreeId, sshHostId }, link } + } + } + } + for (const folder of store.getFolderWorkspaces?.() ?? []) { + const link = matches(folder.externalConversations) + if (link) { + return { workspace: { worktreeId: folderWorkspaceKey(folder.id), sshHostId: null }, link } + } + } + return null +} diff --git a/src/main/external-conversations/external-conversation-seam.test.ts b/src/main/external-conversations/external-conversation-seam.test.ts new file mode 100644 index 00000000000..8208c2e592f --- /dev/null +++ b/src/main/external-conversations/external-conversation-seam.test.ts @@ -0,0 +1,291 @@ +import { describe, expect, it, vi } from 'vitest' +import type { GlobalSettings } from '../../shared/global-settings-types' +import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' +import { hashCanonical } from '../runtime/rpc/orchestration-mutation-receipt' +import { ExternalConversationSeam } from './external-conversation-seam' +import type { ConversationKey } from './external-conversation-types' +import { + fakeLauncher, + fakeLinkStore, + launchResult, + LOCAL_REPO, + PANE, + WORKTREE +} from './external-conversation.test-fixture' + +const KEY: ConversationKey = { source: 'dev', externalId: 'thread-1' } +const TARGET = { kind: 'existing' as const, worktree: `id:${WORKTREE}` } +const START = { agent: 'claude', target: TARGET, prompt: 'first' } + +function harness( + settings: Partial = {}, + records = fakeLinkStore(), + db = new OrchestrationDb(':memory:') +) { + const seam = (): ExternalConversationSeam => + new ExternalConversationSeam({ + store: records.store, + readSettings: () => settings, + receiptDb: () => db + }) + return { ...records, seam: seam(), restart: seam, db } +} + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((done) => { + resolve = done + }) + return { promise, resolve } +} + +const UNKNOWN = { outcome: 'unknown', reason: 'launch_outcome_unknown' } + +describe('ExternalConversationSeam start', () => { + it.each([ + ['handed-to-terminal', { outcome: 'accepted' }], + ['not-delivered', UNKNOWN], + ['unconfirmed', UNKNOWN] + ] as const)('maps a %s launch prompt to %o and links the pane', async (prompt, initial) => { + const { seam, row } = harness() + const { launcher } = fakeLauncher(async () => launchResult(prompt)) + expect(await seam.start({ launcher }, KEY, 'd1', START)).toEqual({ + conversation: KEY, + initial + }) + expect(row(WORKTREE)?.externalConversations).toEqual([ + expect.objectContaining({ source: 'dev', externalId: 'thread-1', paneKey: PANE }) + ]) + }) + + it('answers a launch that fails after publishing as linked but unknown', async () => { + const { seam, row } = harness() + const { launcher } = fakeLauncher(async () => { + throw new Error('prompt delivery crashed') + }) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual(UNKNOWN) + expect(row(WORKTREE)?.externalConversations).toHaveLength(1) + }) + + it('throws a launch that fails before publishing, links nothing, and lets the id retry', async () => { + const { seam, row } = harness() + const failing = fakeLauncher( + async () => { + throw new Error('selector_not_found') + }, + undefined, + false + ) + await expect(seam.start({ launcher: failing.launcher }, KEY, 'd1', START)).rejects.toThrow( + 'selector_not_found' + ) + expect(row(WORKTREE)).toBeUndefined() + const { launcher } = fakeLauncher(async () => launchResult('handed-to-terminal')) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual({ + outcome: 'accepted' + }) + }) + + it('passes its signal to the launch', async () => { + const { seam } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + await seam.start({ launcher }, KEY, 'd1', START) + expect(launch.mock.calls[0]?.[2]).toBe(seam.signal) + }) + + it('shares one creation between concurrent distinct starts', async () => { + const { seam } = harness() + const launched = deferred() + const { launcher, launch } = fakeLauncher(async () => { + await launched.promise + return launchResult('handed-to-terminal') + }) + const first = seam.start({ launcher }, KEY, 'd1', START) + const second = seam.start({ launcher }, KEY, 'd2', { ...START, prompt: 'second' }) + launched.resolve() + expect((await first).initial).toEqual({ outcome: 'accepted' }) + expect((await second).initial).toEqual(UNKNOWN) + expect(launch).toHaveBeenCalledTimes(1) + }) + + it('answers a start rebuilt from a link alone as unknown, without launching', async () => { + const { seam, put } = harness() + put(WORKTREE, 'local', { + externalConversations: [{ ...KEY, paneKey: PANE, agent: 'claude', createdAt: 1 }] + }) + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual(UNKNOWN) + expect(launch).not.toHaveBeenCalled() + }) + + it('links a created folder workspace on its folder record', async () => { + const { seam, folders } = harness() + folders.push({ id: 'f-1' }) + const { launcher } = fakeLauncher( + async () => launchResult('handed-to-terminal', 'folder:f-1'), + { + worktreeId: 'folder:f-1', + sshHostId: null + } + ) + await seam.start({ launcher }, KEY, 'd1', { + ...START, + target: { kind: 'create-folder-workspace', create: { projectGroupId: 'g-1' } } + }) + expect(folders[0].externalConversations).toEqual([expect.objectContaining({ paneKey: PANE })]) + }) + + it.each([ + ['an agent off the list', 'opencode', {}], + ['an agent without status hooks', 'claude', { agentStatusHooksEnabled: false }] + ] as const)('refuses %s', async (_label, agent, settings) => { + const { seam } = harness(settings) + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + expect((await seam.start({ launcher }, KEY, 'd1', { ...START, agent })).initial).toEqual({ + outcome: 'refused', + reason: 'unsupported_agent' + }) + expect(launch).not.toHaveBeenCalled() + }) +}) + +describe('ExternalConversationSeam delivery ids', () => { + it.each([ + ['its workspace cannot be resolved', 'resolve'], + ['its link cannot be written', 'write'] + ] as const)( + 'keeps an unknown answer, across a restart too, when %s after the launch', + async (_label, failing) => { + const { seam, restart, store } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + if (failing === 'resolve') { + launcher.paneWorkspace = async () => { + throw new Error('scope lost') + } + } else { + store.setWorktreeMeta = () => { + throw new Error('disk full') + } + } + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual(UNKNOWN) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual(UNKNOWN) + expect((await restart().start({ launcher }, KEY, 'd1', START)).initial).toEqual(UNKNOWN) + expect(launch).toHaveBeenCalledTimes(1) + warn.mockRestore() + } + ) + + it('shares one answer for a repeated id and payload, and refuses another payload', async () => { + const { seam } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + const first = seam.start({ launcher }, KEY, 'd1', START) + const replay = seam.start({ launcher }, KEY, 'd1', START) + expect((await first).initial).toEqual({ outcome: 'accepted' }) + expect((await replay).initial).toEqual({ outcome: 'accepted' }) + expect(launch).toHaveBeenCalledTimes(1) + expect((await seam.start({ launcher }, KEY, 'd1', { ...START, prompt: 'x' })).initial).toEqual({ + outcome: 'refused', + reason: 'delivery_id_reused' + }) + }) + + it('keeps its answer across a restart, and reads an interrupted one as unknown', async () => { + const { seam, restart, db } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + await seam.start({ launcher }, KEY, 'd1', START) + expect((await restart().start({ launcher }, KEY, 'd1', START)).initial).toEqual({ + outcome: 'accepted' + }) + // A start the host died during: its receipt is still pending. + db.beginMutationReceipt({ + callerFingerprint: 'connector:dev', + requestId: JSON.stringify(['thread-9', 'd1']), + method: 'connector.start', + payloadHash: hashCanonical([START.agent, START.target, START.prompt]) + }) + const lost = await restart().start( + { launcher }, + { ...KEY, externalId: 'thread-9' }, + 'd1', + START + ) + expect(lost.initial).toEqual({ outcome: 'unknown', reason: 'replayed_after_restart' }) + expect(launch).toHaveBeenCalledTimes(1) + }) + + it('scopes delivery ids by conversation', async () => { + const { seam } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + await seam.start({ launcher }, KEY, 'd1', START) + await seam.start({ launcher }, { ...KEY, externalId: 'thread-2' }, 'd1', START) + expect(launch).toHaveBeenCalledTimes(2) + }) +}) + +describe('ExternalConversationSeam links', () => { + it('forgets a removed workspace, and follows a renamed one', async () => { + const { seam, row, put, remove } = harness() + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal')) + await seam.start({ launcher }, KEY, 'd1', START) + put('repo-1::/work/renamed', 'local', { + externalConversations: row(WORKTREE)?.externalConversations + }) + remove(WORKTREE) + expect((await seam.start({ launcher }, KEY, 'd2', START)).initial).toEqual(UNKNOWN) + remove('repo-1::/work/renamed') + await seam.start({ launcher }, KEY, 'd3', START) + expect(launch).toHaveBeenCalledTimes(2) + }) + + it('finds an SSH twin’s link that only its host’s row holds, after a restart too', async () => { + const records = fakeLinkStore([LOCAL_REPO, { ...LOCAL_REPO, connectionId: 'box' }]) + // The local twin wrote this locator first, so the locator-keyed map keeps its row. + records.put(WORKTREE, 'local', {}) + const { seam, restart, writes } = harness({}, records) + const { launcher, launch } = fakeLauncher(async () => launchResult('handed-to-terminal'), { + worktreeId: WORKTREE, + sshHostId: 'ssh:box' + }) + await seam.start({ launcher }, KEY, 'd1', START) + expect(writes.at(-1)).toEqual({ id: WORKTREE, hostId: 'ssh:box' }) + expect(records.row(WORKTREE)?.externalConversations).toBeUndefined() + expect((await restart().start({ launcher }, KEY, 'd2', START)).initial).toEqual(UNKNOWN) + expect(launch).toHaveBeenCalledTimes(1) + }) +}) + +describe('ExternalConversationSeam disposal', () => { + it('settles a queued start at once while the active launch keeps its real outcome', async () => { + const { seam } = harness() + const launched = deferred() + const { launcher, launch } = fakeLauncher(async () => { + await launched.promise + return launchResult('handed-to-terminal') + }) + const first = seam.start({ launcher }, KEY, 'd1', START) + const queued = seam.start({ launcher }, KEY, 'd2', { ...START, prompt: 'second' }) + await vi.waitFor(() => expect(launch).toHaveBeenCalledTimes(1)) + seam.dispose('connectors_disabled') + expect((await queued).initial).toEqual({ outcome: 'refused', reason: 'connectors_disabled' }) + launched.resolve() + expect((await first).initial).toEqual({ outcome: 'accepted' }) + }) + + it('refuses a launch the disable stopped before the agent existed', async () => { + const { seam, row } = harness() + const { launcher } = fakeLauncher( + async () => { + seam.dispose('connectors_disabled') + throw new Error('request_aborted') + }, + undefined, + false + ) + expect((await seam.start({ launcher }, KEY, 'd1', START)).initial).toEqual({ + outcome: 'refused', + reason: 'connectors_disabled' + }) + expect(row(WORKTREE)).toBeUndefined() + }) +}) diff --git a/src/main/external-conversations/external-conversation-seam.ts b/src/main/external-conversations/external-conversation-seam.ts new file mode 100644 index 00000000000..58e57230a6d --- /dev/null +++ b/src/main/external-conversations/external-conversation-seam.ts @@ -0,0 +1,99 @@ +import type { AgentStatusHooksSettings } from '../../shared/agent-status-hooks-setting' +import type { OrcaRuntimeService } from '../runtime/orca-runtime' +import { admitConnectorAgent } from './connector-agents' +import { ConversationChain } from './external-conversation-chain' +import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts' +import { + findLink, + writeLink, + type ExternalConversationLinkStore +} from './external-conversation-links' +import { + launchConversation, + runtimeConversationLauncher, + type ConversationLaunchInput, + type ConversationLauncher +} from './external-conversation-start' +import { + conversationId, + refused, + type ConversationKey, + type ConversationStart +} from './external-conversation-types' + +/** What the seam drives on its host. */ +export type ExternalConversationHost = { launcher: ConversationLauncher } + +export function runtimeConversationHost(runtime: OrcaRuntimeService): ExternalConversationHost { + return { launcher: runtimeConversationLauncher(runtime) } +} + +export type ExternalConversationSeamDeps = { + store: ExternalConversationLinkStore + readSettings: () => AgentStatusHooksSettings + receiptDb: () => DeliveryReceiptDb +} + +type StartInput = Omit & { agent: string } + +/** + * The connector seam for one enabled period. Turning connectors off disposes it: queued work + * settles `connectors_disabled` at once and running work stops at its next boundary. Turning them + * on builds a fresh one. + */ +export class ExternalConversationSeam { + private readonly abort = new AbortController() + private readonly chain = new ConversationChain(this.abort.signal) + private readonly receipts: DeliveryReceipts + + constructor(private readonly deps: ExternalConversationSeamDeps) { + this.receipts = new DeliveryReceipts(deps.receiptDb) + } + + /** Deferred work passes this to every effect, so nothing starts or writes after a disable. */ + get signal(): AbortSignal { + return this.abort.signal + } + + dispose(reason: 'connectors_disabled'): void { + this.abort.abort(reason) + } + + async start( + host: ExternalConversationHost, + key: ConversationKey, + deliveryId: string, + input: StartInput + ): Promise { + const initial = await this.receipts.run( + key, + deliveryId, + 'connector.start', + [input.agent, input.target, input.prompt], + () => + this.chain.enqueue(conversationId(key), async () => { + const { agent } = input + if (!admitConnectorAgent(this.deps.readSettings(), agent)) { + return refused('unsupported_agent') + } + if (findLink(this.deps.store, key)) { + // Concurrent starts share the creation; only the first carried its prompt. + return { outcome: 'unknown', reason: 'launch_outcome_unknown' } + } + return launchConversation( + host.launcher, + { ...input, agent }, + this.signal, + (workspace, paneKey) => + writeLink(this.deps.store, workspace, { + ...key, + paneKey, + agent, + createdAt: Date.now() + }) + ) + }) + ) + return { conversation: key, initial } + } +} diff --git a/src/main/external-conversations/external-conversation-start.ts b/src/main/external-conversations/external-conversation-start.ts new file mode 100644 index 00000000000..d3c9b9441a9 --- /dev/null +++ b/src/main/external-conversations/external-conversation-start.ts @@ -0,0 +1,120 @@ +import type { AgentLaunchResult } from '../../shared/agent-launch-intent' +import type { ConnectorAgent } from '../../shared/external-conversation-link' +import type { AgentLaunchParams } from '../../shared/rpc-contract/agent-launch-params' +import { executeAgentLaunch } from '../agent-launch/agent-launch-executor' +import type { AgentLaunchPublishedSurface } from '../agent-launch/agent-launch-execution' +import { + agentPaneWorkspace, + type AgentPaneWorkspace +} from '../agent-hooks/agent-transition-detector' +import type { OrcaRuntimeService } from '../runtime/orca-runtime' +import { resolveUnlaunchedIntent } from '../runtime/rpc/methods/agent-launch-intent-resolution' +import { agentLaunchSurfaceFactory } from '../runtime/rpc/methods/agent-launch-surfaces' +import { agentLaunchWorkspaceFactory } from '../runtime/rpc/methods/agent-launch-worktree-creation' +import { refused, type Delivery } from './external-conversation-types' + +export type ConversationLaunchInput = { + agent: ConnectorAgent + target: AgentLaunchParams['target'] + prompt: string +} + +export type ConversationLauncher = { + launch( + input: ConversationLaunchInput, + onSurfacePublished: (surface: AgentLaunchPublishedSurface) => void, + signal: AbortSignal + ): Promise + /** The launched workspace as hook rows name it, qualified by the host that runs it. */ + paneWorkspace(worktreeId: string): Promise +} + +/** + * `agent.launch`'s own executor and factories with no paired caller: a background terminal agent + * that takes the prompt on argv, in a workspace the target names or creates (worktree or folder). + */ +export function runtimeConversationLauncher(runtime: OrcaRuntimeService): ConversationLauncher { + const context = { runtime } + return { + launch: async ({ agent, target, prompt }, onSurfacePublished, signal) => { + const intent = await resolveUnlaunchedIntent( + { agent, target, prompt: { text: prompt, delivery: 'submit' } }, + runtime, + null + ) + return executeAgentLaunch({ + runtime, + intent, + terminalOnly: true, + surfaces: agentLaunchSurfaceFactory(context, undefined, undefined, false, undefined, { + early: null, + presentation: 'background' + }), + workspaces: agentLaunchWorkspaceFactory(context, agent), + onSurfacePublished, + signal + }) + }, + paneWorkspace: async (worktreeId) => { + const scope = await runtime.showTerminalWorkspaceLaunchScope(`id:${worktreeId}`) + return agentPaneWorkspace(scope.id, scope.connectionId) + } + } +} + +/** + * Launches the conversation's agent and links its pane as soon as the surface is published. Only + * a failure before anything was published throws, so its delivery id may retry; after that an + * agent may exist, so any failure, the link's included, is `unknown` and never launches again. + */ +export async function launchConversation( + launcher: ConversationLauncher, + input: ConversationLaunchInput, + signal: AbortSignal, + writeLink: (workspace: AgentPaneWorkspace, paneKey: string) => void +): Promise { + // Settled to its error, so a failed write waits for the launch instead of going unhandled. + const surface: { published: boolean; linked: Promise | null } = { + published: false, + linked: null + } + const onSurfacePublished = (published: AgentLaunchPublishedSurface): void => { + surface.published = true + const { outcome } = published + if (outcome.kind === 'terminal' && outcome.paneKey) { + const paneKey = outcome.paneKey + surface.linked = launcher + .paneWorkspace(published.worktreeId) + .then((workspace) => writeLink(workspace, paneKey)) + .then( + () => null, + (error: unknown) => error ?? new Error('connector_link_failed') + ) + } + } + let result: AgentLaunchResult | null = null + try { + result = await launcher.launch(input, onSurfacePublished, signal) + } catch (error) { + if (!surface.published) { + // Turned off before the agent existed: the launch stopped before any effect. + if (signal.aborted) { + return refused('connectors_disabled') + } + throw error + } + } + const linkError = surface.linked + ? await surface.linked + : new Error('connector_launch_without_pane') + if (linkError) { + console.warn( + '[connectors] the agent launched but its conversation link was not written', + linkError + ) + return { outcome: 'unknown', reason: 'launch_outcome_unknown' } + } + return result?.prompt?.outcome === 'handed-to-terminal' + ? { outcome: 'accepted' } + : { outcome: 'unknown', reason: 'launch_outcome_unknown' } +} diff --git a/src/main/external-conversations/external-conversation-types.ts b/src/main/external-conversations/external-conversation-types.ts new file mode 100644 index 00000000000..aac290a2d4c --- /dev/null +++ b/src/main/external-conversations/external-conversation-types.ts @@ -0,0 +1,33 @@ +import type { ConnectorSource } from '../../shared/external-conversation-link' + +export type ConversationKey = { source: ConnectorSource; externalId: string } + +/** Why the seam declined before any byte left it. */ +export type ConnectorRefusal = + | 'connectors_disabled' + | 'agent_not_reachable' + | 'unsupported_agent' + | 'delivery_id_reused' + +/** + * The connector's own answer, weaker than a `WriteSettlement`: `accepted` means the writer + * returned, never that the agent read it. + */ +export type Delivery = + | { outcome: 'accepted' } + | { outcome: 'refused'; reason: ConnectorRefusal } + | { + outcome: 'unknown' + reason: 'launch_outcome_unknown' | 'writer_failed' | 'replayed_after_restart' + } + +export type ConversationStart = { conversation: ConversationKey; initial: Delivery } + +export function refused(reason: ConnectorRefusal): Delivery { + return { outcome: 'refused', reason } +} + +/** One string per conversation, unambiguous for any id text. */ +export function conversationId(key: ConversationKey): string { + return JSON.stringify([key.source, key.externalId]) +} diff --git a/src/main/external-conversations/external-conversation.test-fixture.ts b/src/main/external-conversations/external-conversation.test-fixture.ts new file mode 100644 index 00000000000..1bf73f1ecc9 --- /dev/null +++ b/src/main/external-conversations/external-conversation.test-fixture.ts @@ -0,0 +1,116 @@ +import { vi } from 'vitest' +import type { AgentLaunchResult } from '../../shared/agent-launch-intent' +import type { ExecutionHostId } from '../../shared/execution-host' +import type { Repo } from '../../shared/repo-types' +import type { HostWorktreeTarget } from '../../shared/worktree/host-qualified-identity' +import type { AgentPaneWorkspace } from '../agent-hooks/agent-transition-detector' +import type { AgentLaunchPublishedSurface } from '../agent-launch/agent-launch-execution' +import type { ExternalConversationLinkStore } from './external-conversation-links' +import type { ConversationLauncher } from './external-conversation-start' + +export const WORKTREE = 'repo-1::/work/a' +export const PANE = 'tab-1:11111111-1111-4111-8111-111111111111' +export const LOCAL_REPO: Repo = { + id: 'repo-1', + path: '/work', + displayName: 'repo', + badgeColor: '#000000', + addedAt: 0 +} + +type LinkRow = { externalConversations?: unknown; hostId?: ExecutionHostId } + +/** + * The real store's two projections: each host's canonical rows, and the locator-keyed map that + * keeps only the first host to write an id. + */ +export function fakeLinkStore(repos: Repo[] = [LOCAL_REPO]) { + const byHost = new Map>() + const legacy: Record = {} + const folders: (LinkRow & { id: string })[] = [] + const writes: HostWorktreeTarget[] = [] + const rows = (hostId: ExecutionHostId): Record => { + const existing = byHost.get(hostId) + if (existing) { + return existing + } + const created: Record = {} + byHost.set(hostId, created) + return created + } + const put = (id: string, hostId: ExecutionHostId, row: Omit): void => { + const next = { ...rows(hostId)[id], ...row, hostId } + rows(hostId)[id] = next + if (!legacy[id] || legacy[id].hostId === hostId) { + legacy[id] = next + } + } + const remove = (id: string, hostId: ExecutionHostId = 'local'): void => { + delete rows(hostId)[id] + if (legacy[id]?.hostId === hostId) { + delete legacy[id] + } + } + const store = { + getRepos: () => repos, + getAllWorktreeMeta: () => legacy, + getAllWorktreeMetaForHost: (hostId: ExecutionHostId) => rows(hostId), + getWorktreeMeta: (id: string) => legacy[id], + getWorktreeMetaForHost: (id: string, hostId: ExecutionHostId) => rows(hostId)[id], + setWorktreeMeta: (target: HostWorktreeTarget, updates: { externalConversations: unknown }) => { + writes.push(target) + put(target.id, target.hostId, updates) + }, + getFolderWorkspace: (id: string) => folders.find((folder) => folder.id === id), + getFolderWorkspaces: () => folders, + updateFolderWorkspace: (id: string, updates: { externalConversations: unknown }) => { + const folder = folders.find((candidate) => candidate.id === id) + if (folder) { + folder.externalConversations = updates.externalConversations + } + } + } satisfies ExternalConversationLinkStore + const row = (id: string, hostId: ExecutionHostId = 'local'): LinkRow | undefined => + rows(hostId)[id] + return { store, row, put, remove, folders, writes } +} + +export function launchResult( + prompt: 'handed-to-terminal' | 'not-delivered' | 'unconfirmed' | null, + worktreeId = WORKTREE, + paneKey = PANE +): AgentLaunchResult { + return { + outcome: { kind: 'terminal', handle: 'term-1', paneKey }, + worktreeId, + receipt: { + mode: 'terminal', + preferred: 'terminal', + reason: 'user_default', + detail: 'terminal' + }, + ...(prompt ? { prompt: { delivery: 'submit', outcome: prompt } } : {}) + } +} + +/** Publishes, then settles as `settle` says; `workspace` is the pane's host-qualified workspace. */ +export function fakeLauncher( + settle: () => Promise, + workspace: AgentPaneWorkspace = { worktreeId: WORKTREE, sshHostId: null }, + publish = true +) { + const launch = vi.fn( + async ( + _input: unknown, + onSurfacePublished: (surface: AgentLaunchPublishedSurface) => void, + _signal?: AbortSignal + ): Promise => { + if (publish) { + onSurfacePublished(launchResult(null, workspace.worktreeId)) + } + return settle() + } + ) + const launcher: ConversationLauncher = { launch, paneWorkspace: async () => workspace } + return { launcher, launch } +} diff --git a/src/main/external-conversations/external-conversations-install.test.ts b/src/main/external-conversations/external-conversations-install.test.ts new file mode 100644 index 00000000000..a426210bd2b --- /dev/null +++ b/src/main/external-conversations/external-conversations-install.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it, vi } from 'vitest' +import type { GlobalSettings } from '../../shared/global-settings-types' +import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' +import { installExternalConversations } from './external-conversations-install' +import { fakeLinkStore } from './external-conversation.test-fixture' + +type SettingsListener = (updates: Partial) => void + +function fakeStore(initial: Partial) { + let settings = { ...initial } + const listeners = new Set() + const getSettings = vi.fn(() => settings) + const store = { + ...fakeLinkStore().store, + getSettings, + onSettingsChanged: (listener: SettingsListener) => { + listeners.add(listener) + return () => listeners.delete(listener) + } + } + const update = (updates: Partial): void => { + settings = { ...settings, ...updates } + listeners.forEach((listener) => listener(updates)) + } + return { store, listeners, getSettings, update } +} + +function runtime() { + const db = new OrchestrationDb(':memory:') + return { getOrchestrationDb: vi.fn(() => db) } +} + +describe('installExternalConversations', () => { + it('builds nothing and opens nothing while connectors are off', () => { + const { store, getSettings, update } = fakeStore({}) + const host = runtime() + const connectors = installExternalConversations(store, host) + // The flag is read on the first connector call, never at startup. + expect(getSettings).not.toHaveBeenCalled() + expect(connectors.current()).toBeNull() + update({ experimentalConnectors: false }) + expect(connectors.current()).toBeNull() + expect(host.getOrchestrationDb).not.toHaveBeenCalled() + }) + + it('builds the seam on first use when the setting is already on', () => { + const { store } = fakeStore({ experimentalConnectors: true }) + const seam = installExternalConversations(store, runtime()).current() + expect(seam?.signal.aborted).toBe(false) + }) + + it('disposes on disable without another call, and builds one fresh seam on re-enable', () => { + const { store, listeners, update } = fakeStore({ experimentalConnectors: true }) + const connectors = installExternalConversations(store, runtime()) + const first = connectors.current() + + update({ experimentalConnectors: false }) + expect(first?.signal.reason).toBe('connectors_disabled') + expect(connectors.current()).toBeNull() + + update({ experimentalConnectors: true }) + const second = connectors.current() + expect(second).not.toBe(first) + expect(second?.signal.aborted).toBe(false) + expect(listeners.size).toBe(1) + }) + + it('keeps the seam through unrelated settings changes', () => { + const { store, update } = fakeStore({ experimentalConnectors: true }) + const connectors = installExternalConversations(store, runtime()) + const seam = connectors.current() + update({ editorAutoSave: true }) + expect(connectors.current()).toBe(seam) + }) + + it('stays off without a store', () => { + expect(installExternalConversations(null, runtime()).current()).toBeNull() + }) +}) diff --git a/src/main/external-conversations/external-conversations-install.ts b/src/main/external-conversations/external-conversations-install.ts new file mode 100644 index 00000000000..038567d41a4 --- /dev/null +++ b/src/main/external-conversations/external-conversations-install.ts @@ -0,0 +1,56 @@ +import type { GlobalSettings } from '../../shared/global-settings-types' +import type { DeliveryReceiptDb } from './external-conversation-delivery-receipts' +import type { ExternalConversationLinkStore } from './external-conversation-links' +import { ExternalConversationSeam } from './external-conversation-seam' + +export type ExternalConversationStore = ExternalConversationLinkStore & { + getSettings: () => Partial< + Pick + > + onSettingsChanged?: (listener: (updates: Partial) => void) => unknown +} + +/** + * Owns the seam's lifetime, on the settings-change pattern `installStructuredAgentAvailability` + * uses: one exists exactly while `experimentalConnectors` is on. The flag is + * first read on the first connector call, so a host that never sees one never reads it; a change + * after that disposes or rebuilds the seam without waiting for another call. The host is the + * authority, so every connector entry point asks it rather than its caller's profile. + */ +export function installExternalConversations( + store: ExternalConversationStore | null, + runtime: { getOrchestrationDb: () => DeliveryReceiptDb } +): { current(): ExternalConversationSeam | null } { + if (!store) { + return { current: () => null } + } + let seam: ExternalConversationSeam | null = null + let reconciled = false + const reconcile = (): void => { + reconciled = true + const enabled = store.getSettings().experimentalConnectors === true + if (enabled && !seam) { + seam = new ExternalConversationSeam({ + store, + readSettings: () => store.getSettings(), + receiptDb: () => runtime.getOrchestrationDb() + }) + } else if (!enabled && seam) { + seam.dispose('connectors_disabled') + seam = null + } + } + store.onSettingsChanged?.((updates) => { + if ('experimentalConnectors' in updates) { + reconcile() + } + }) + return { + current: () => { + if (!reconciled) { + reconcile() + } + return seam + } + } +} diff --git a/src/main/persistence-local-only-worktree-meta.test.ts b/src/main/persistence-local-only-worktree-meta.test.ts index 810648ad32f..e0c884c8261 100644 --- a/src/main/persistence-local-only-worktree-meta.test.ts +++ b/src/main/persistence-local-only-worktree-meta.test.ts @@ -126,6 +126,15 @@ describe('local-only workspace metadata', () => { expect(reloaded.getAllWorktreeMetaForHost(SSH_HOST)).toEqual({}) }) + it('owns a floating terminal row on this machine before any row exists', () => { + expect( + resolveWorktreeMetaOwner( + { getRepos: () => [], getWorktreeMeta: () => undefined }, + FLOATING_TERMINAL_WORKTREE_ID + )?.hostId + ).toBe('local') + }) + it('lands every floating terminal write on this machine, whichever host the caller named', () => { const store = createUpgradedStore() expect(resolveWorktreeMetaOwner(store, FLOATING_TERMINAL_WORKTREE_ID)?.hostId).toBe('local') diff --git a/src/main/persistence/host-qualified-worktree-meta.ts b/src/main/persistence/host-qualified-worktree-meta.ts index b65f9e699f8..2b923309d62 100644 --- a/src/main/persistence/host-qualified-worktree-meta.ts +++ b/src/main/persistence/host-qualified-worktree-meta.ts @@ -28,15 +28,20 @@ export type HostQualifiedWorktreeMetaStore = { } /** Canonical host snapshot, with a filtered legacy fallback for partial test/compatibility stores. */ -export function readAllWorktreeMetaForHost( - store: Pick, +export function readAllWorktreeMetaForHost< + TMeta extends { hostId?: ExecutionHostId } = WorktreeMeta +>( + store: { + getAllWorktreeMeta: () => Record + getAllWorktreeMetaForHost?: (executionHostId: ExecutionHostId) => Record + }, executionHostId: ExecutionHostId -): Record { +): Record { const qualified = store.getAllWorktreeMetaForHost?.(executionHostId) if (qualified) { return qualified } - const projected: Record = {} + const projected: Record = {} for (const [worktreeId, meta] of Object.entries(store.getAllWorktreeMeta())) { if (!meta.hostId || meta.hostId === executionHostId) { projected[worktreeId] = meta diff --git a/src/main/persistence/restoring-sessions/folder-workspace-operations.ts b/src/main/persistence/restoring-sessions/folder-workspace-operations.ts index 6413ccb305f..8f830eec7c7 100644 --- a/src/main/persistence/restoring-sessions/folder-workspace-operations.ts +++ b/src/main/persistence/restoring-sessions/folder-workspace-operations.ts @@ -150,6 +150,7 @@ export class FolderWorkspacePersistenceOperations { | 'smartAt' | 'lastTerminalUseAt' | 'diffComments' + | 'externalConversations' > > & WorkspaceAttachmentMutation @@ -261,6 +262,9 @@ export class FolderWorkspacePersistenceOperations { if (updates.diffComments !== undefined) { workspace.diffComments = updates.diffComments } + if (updates.externalConversations !== undefined) { + workspace.externalConversations = updates.externalConversations + } workspace.updatedAt = Date.now() this.scheduleSave() return workspace diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 8369fb2b16b..91196389278 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -26,7 +26,7 @@ import type { RemoteLayoutSnapshot } from '../../shared/remote-layout-snapshot' import { registerDetectedWorktreeScanInvalidation } from '../ipc/worktrees/listing/register-detected-worktree-scan-invalidation' import { listReferenceWorkspaces, - selectReferenceWorkspace, + selectWarmedReferenceWorkspace, listWorkspaceReferences } from './runtime-reference-catalog' import { findWorkspaceReferences } from './runtime-reference-find' @@ -52,6 +52,8 @@ import { sidebarInputsFeedDomains } from './sidebar-inputs/sidebar-inputs-feed-d import { requireSwitchedOn } from './sidebar-inputs/require-switched-on' import { HEADLESS_RUNTIME_WINDOW_ID } from '../../shared/runtime-session-contracts' import { getHostDisplayLabelOverrides } from '../../shared/host-setting-overrides' +import { installExternalConversations } from '../external-conversations/external-conversations-install' +import type { ExternalConversationSeam } from '../external-conversations/external-conversation-seam' // Switch point: a window-fed build keeps no sidebar facts layer. const OrcaRuntimeSidebarBase: typeof OrcaRuntimeWithAgentTransitions = ORCA_SIDEBAR_INPUTS_IS_MAIN @@ -120,24 +122,16 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { ) : null - private async selectReferenceCatalogWorkspace(selector: string, cwd?: string) { - const catalog = listReferenceWorkspaces( + private selectReferenceCatalogWorkspace(selector: string, cwd?: string) { + return selectWarmedReferenceWorkspace( this.requireStore(), - this.resolvedWorktrees.peek()?.worktrees + { + cached: () => this.resolvedWorktrees.peek()?.worktrees, + scan: () => this.listResolvedWorktrees() + }, + selector, + cwd ) - try { - return selectReferenceWorkspace(catalog, selector, cwd) - } catch (error) { - if (!(error instanceof Error) || error.message !== 'selector_not_found') { - throw error - } - // Why: unsaved worktrees and cold-cache branches only resolve after a worktree scan. - const warmed = listReferenceWorkspaces( - this.requireStore(), - await this.listResolvedWorktrees() - ) - return selectReferenceWorkspace(warmed, selector, cwd) - } } async listWorkspaceReferences(selector: string, cwd?: string) { @@ -185,6 +179,13 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { private structuredAgentAvailability?: ReturnType + private readonly externalConversations = installExternalConversations(this.store, this) + + /** The connector seam while `experimentalConnectors` is on; null refuses every connector call. */ + getExternalConversationSeam(): ExternalConversationSeam | null { + return this.externalConversations.current() + } + /** This host's agent health snapshot and every replacement, from first use; returns the release. */ subscribeStructuredAgentAvailability(read: StructuredAgentAvailabilityReader): () => void { this.structuredAgentAvailability ??= installStructuredAgentAvailability(this.store, this) diff --git a/src/main/runtime/rpc/errors.ts b/src/main/runtime/rpc/errors.ts index 435b6fca3e0..fa527981c58 100644 --- a/src/main/runtime/rpc/errors.ts +++ b/src/main/runtime/rpc/errors.ts @@ -101,6 +101,7 @@ const RUNTIME_PASSTHROUGH_CODES: ReadonlySet = new Set([ 'remote_update_manual_required', 'remote_update_not_available', 'remote_update_not_downloaded', + 'connectors_disabled', ...AGENT_SESSION_RPC_ERROR_CODES ]) diff --git a/src/main/runtime/rpc/methods/connector.test.ts b/src/main/runtime/rpc/methods/connector.test.ts new file mode 100644 index 00000000000..8e366c8e049 --- /dev/null +++ b/src/main/runtime/rpc/methods/connector.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, it } from 'vitest' +import { requireSeam } from './connector' + +describe('connector RPC methods', () => { + it('are refused while the host has connectors off', () => { + expect(() => requireSeam({ getExternalConversationSeam: () => null })).toThrow( + 'connectors_disabled' + ) + }) +}) diff --git a/src/main/runtime/rpc/methods/connector.ts b/src/main/runtime/rpc/methods/connector.ts new file mode 100644 index 00000000000..6ff40ebe321 --- /dev/null +++ b/src/main/runtime/rpc/methods/connector.ts @@ -0,0 +1,43 @@ +/** + * `connector.*`: the external-conversation seam (Slack and other threads) behind + * `experimentalConnectors`. The host's setting is the only gate, so a call while it is off is + * refused here whoever makes it. + */ + +import { ConnectorStart } from '../../../../shared/rpc-contract/connector-params' +import { + runtimeConversationHost, + type ExternalConversationSeam +} from '../../../external-conversations/external-conversation-seam' +import type { OrcaRuntimeService } from '../../orca-runtime' +import { defineMethod } from '../core' +import { assertAgentLaunchTargetAuthorized } from './agent-launch-target-authorization' + +/** Every `connector.*` handler's first step. */ +export function requireSeam( + runtime: Pick +): ExternalConversationSeam { + const seam = runtime.getExternalConversationSeam() + if (!seam) { + throw new Error('connectors_disabled') + } + return seam +} + +export const CONNECTOR_METHODS = [ + defineMethod({ + name: 'connector.start', + permission: 'workspace', + params: ConnectorStart, + handler: async ({ source, externalId, deliveryId, ...input }, context) => { + const seam = requireSeam(context.runtime) + assertAgentLaunchTargetAuthorized(input.target, context) + return seam.start( + runtimeConversationHost(context.runtime), + { source, externalId }, + deliveryId, + input + ) + } + }) +] diff --git a/src/main/runtime/rpc/methods/index.ts b/src/main/runtime/rpc/methods/index.ts index 8ac079962b0..531a2f12210 100644 --- a/src/main/runtime/rpc/methods/index.ts +++ b/src/main/runtime/rpc/methods/index.ts @@ -64,6 +64,7 @@ import { MANAGED_RUNTIME_INSTALL_METHODS } from './managed-runtime-install' import { ARTIFACT_METHODS } from './artifacts' import { AGENT_HOOK_METHODS } from './agent-hooks' import { AGENT_LAUNCH_METHODS } from './agent-launch' +import { CONNECTOR_METHODS } from './connector' import { LAYOUT_APPLY_METHODS, LAYOUT_VIEW_REPORT_METHODS, @@ -93,6 +94,7 @@ export const ALL_RPC_METHODS = [ ...STRUCTURED_AGENT_SESSION_AGENTS_METHODS, ...MANAGED_RUNTIME_INSTALL_METHODS, ...AGENT_LAUNCH_METHODS, + ...CONNECTOR_METHODS, ...TERMINAL_METHODS, ...TERMINAL_ORPHAN_METHODS, ...BROWSER_CORE_METHODS, diff --git a/src/main/runtime/runtime-client-settings-shape.ts b/src/main/runtime/runtime-client-settings-shape.ts index 7868c9f3a8e..d49e9d172be 100644 --- a/src/main/runtime/runtime-client-settings-shape.ts +++ b/src/main/runtime/runtime-client-settings-shape.ts @@ -53,6 +53,7 @@ export type RuntimeHostDisplayLabelOverrides = Partial< export type RuntimeClientSettingsUpdate = Pick< Partial, | 'agentStatusHooksEnabled' + | 'experimentalConnectors' | 'defaultTuiAgent' | 'disabledTuiAgents' | 'agentDefaultArgs' diff --git a/src/main/runtime/runtime-reference-catalog.ts b/src/main/runtime/runtime-reference-catalog.ts index 2fd4f3ed21f..a092f93ff3f 100644 --- a/src/main/runtime/runtime-reference-catalog.ts +++ b/src/main/runtime/runtime-reference-catalog.ts @@ -135,6 +135,31 @@ export function listWorkspaceReferences(worktree: ReferenceWorkspace): RuntimeRe } } +/** Selects from the cached listing, then once from a fresh scan: unsaved worktrees and cold-cache + * branches only resolve after one. */ +export async function selectWarmedReferenceWorkspace( + store: Parameters[0], + listing: { + cached: () => readonly Worktree[] | undefined + scan: () => Promise + }, + selector: string, + cwd?: string +): Promise { + try { + return selectReferenceWorkspace(listReferenceWorkspaces(store, listing.cached()), selector, cwd) + } catch (error) { + if (!(error instanceof Error) || error.message !== 'selector_not_found') { + throw error + } + return selectReferenceWorkspace( + listReferenceWorkspaces(store, await listing.scan()), + selector, + cwd + ) + } +} + export function selectReferenceWorkspace( workspaces: readonly ReferenceWorkspace[], selector: string, diff --git a/src/main/startup/cli-command-names.ts b/src/main/startup/cli-command-names.ts index 8344f08f219..7b820a3ccc7 100644 --- a/src/main/startup/cli-command-names.ts +++ b/src/main/startup/cli-command-names.ts @@ -14,6 +14,7 @@ export const CLI_COMMAND_NAMES = [ 'click', 'clipboard', 'computer', + 'connector', 'console', 'cookie', 'dblclick', diff --git a/src/shared/external-conversation-link.ts b/src/shared/external-conversation-link.ts new file mode 100644 index 00000000000..23edb0efd81 --- /dev/null +++ b/src/shared/external-conversation-link.ts @@ -0,0 +1,15 @@ +/** Where a connector conversation comes from; `dev` is the CLI that drives one without a service. */ +export const CONNECTOR_SOURCES = ['slack', 'github', 'linear', 'discord', 'dev'] as const +export type ConnectorSource = (typeof CONNECTOR_SOURCES)[number] + +/** Terminal agents a connector may start; see `CONNECTOR_AGENTS` for why only these. */ +export type ConnectorAgent = 'claude' | 'codex' + +/** An outside thread's agent pane, persisted on the workspace it runs in (system-owned). */ +export type ExternalConversationLink = { + source: ConnectorSource + externalId: string + paneKey: string + agent: ConnectorAgent + createdAt: number +} diff --git a/src/shared/folder-workspace-types.ts b/src/shared/folder-workspace-types.ts index 7318ba88609..b3651117073 100644 --- a/src/shared/folder-workspace-types.ts +++ b/src/shared/folder-workspace-types.ts @@ -3,6 +3,7 @@ import type { TaskSourceContext } from './task-source-context' import type { TuiAgent } from './tui-agent' import type { DiffComment } from './diff-comment-types' import type { SmartTier } from './sidebar-list/sort-stamps' +import type { ExternalConversationLink } from './external-conversation-link' import type { WorkspaceCreatorProvenance, WorkspaceLinkedItem, @@ -48,6 +49,8 @@ export type FolderWorkspace = { createdAt: number updatedAt: number diffComments?: DiffComment[] + /** See {@link WorktreeMeta.externalConversations}. */ + externalConversations?: ExternalConversationLink[] } export type FolderWorkspaceLinkedTask = WorkspaceLinkedItem diff --git a/src/shared/folder-workspaces.ts b/src/shared/folder-workspaces.ts index 91359db44b7..3a0f2388de7 100644 --- a/src/shared/folder-workspaces.ts +++ b/src/shared/folder-workspaces.ts @@ -129,7 +129,11 @@ export function normalizeFolderWorkspaces( typeof raw.updatedAt === 'number' && Number.isFinite(raw.updatedAt) ? raw.updatedAt : now, // Legacy read: unreleased #14112 builds wrote notes inline. Canonical home is // PersistedState.folderWorkspaceDiffComments. - ...(Array.isArray(raw.diffComments) ? { diffComments: raw.diffComments } : {}) + ...(Array.isArray(raw.diffComments) ? { diffComments: raw.diffComments } : {}), + // Entries are validated where they are read (external-conversation-links.ts). + ...(Array.isArray(raw.externalConversations) + ? { externalConversations: raw.externalConversations } + : {}) }) } return workspaces.sort( diff --git a/src/shared/host-operation-ownership.ts b/src/shared/host-operation-ownership.ts index dc04daf09d5..2a6360e8af5 100644 --- a/src/shared/host-operation-ownership.ts +++ b/src/shared/host-operation-ownership.ts @@ -23,6 +23,7 @@ export const HOST_OPERATION_OWNERSHIP = { browser: 'host', clipboard: 'host', computer: 'host', + connector: 'host', desktopBrowserBridge: 'host', diagnostics: 'host', emulator: 'host', diff --git a/src/shared/rpc-contract/client-settings-params.ts b/src/shared/rpc-contract/client-settings-params.ts index bebab8d31bb..66d89f3d58e 100644 --- a/src/shared/rpc-contract/client-settings-params.ts +++ b/src/shared/rpc-contract/client-settings-params.ts @@ -112,6 +112,8 @@ export const SettingsUpdate = z .enum(['issues', 'my-issues', 'prs', 'my-prs', 'review', 'all']) .optional(), experimentalNewWorktreeCardStyle: z.boolean().optional(), + /** No UI: the host-side switch for connectors, so a headless server can be turned on. */ + experimentalConnectors: z.boolean().optional(), agentStatusHooksEnabled: z.boolean().optional(), defaultRepoSelection: z.array(z.string()).nullable().optional(), defaultLinearTeamSelection: z.array(z.string()).nullable().optional(), diff --git a/src/shared/rpc-contract/connector-params.ts b/src/shared/rpc-contract/connector-params.ts new file mode 100644 index 00000000000..357a296ce06 --- /dev/null +++ b/src/shared/rpc-contract/connector-params.ts @@ -0,0 +1,18 @@ +import { z } from 'zod' +import { CONNECTOR_SOURCES } from '../external-conversation-link' +import { AgentLaunchFields } from './agent-launch-params' + +const ConnectorDelivery = { + source: z.enum(CONNECTOR_SOURCES), + externalId: z.string().min(1, 'Missing external conversation id'), + /** The caller's id for this delivery; a retry with the same id and payload shares its answer. */ + deliveryId: z.string().min(1, 'Missing delivery id') +} + +export const ConnectorStart = z.object({ + ...ConnectorDelivery, + /** Admitted by the host's connector agent list, so an unsupported one is a refusal, not invalid. */ + agent: z.string().min(1, 'Missing agent'), + target: AgentLaunchFields.shape.target, + prompt: z.string().min(1, 'Missing prompt') +}) diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index 0213f941799..ed4ece7c2ca 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -173,6 +173,7 @@ import { SetValue, TypeText } from './computer-schemas-params' +import { ConnectorStart } from './connector-params' import { AttachParams as AttachParamsOfEmulatorParams, AxParams, @@ -888,6 +889,7 @@ export const RPC_PARAMS_BY_METHOD = { 'computer.scroll': ScrollOfComputerSchemasParams, 'computer.setValue': SetValue, 'computer.typeText': TypeText, + 'connector.start': ConnectorStart, 'desktopBrowserBridge.attach': DesktopBrowserBridgeAttachParams, 'desktopBrowserBridge.result': DesktopBrowserBridgeResultParams, 'diagnostics.memory': null, diff --git a/src/shared/worktree/meta-types.ts b/src/shared/worktree/meta-types.ts index 56872b977b2..8746330ac6f 100644 --- a/src/shared/worktree/meta-types.ts +++ b/src/shared/worktree/meta-types.ts @@ -14,6 +14,7 @@ import type { TuiAgent } from '../tui-agent' import type { OrcaWorkspaceLayout } from '../global-settings-types' import type { DiffComment, MobileDiffReviewState } from '../diff-comment-types' import type { SmartTier } from '../sidebar-list/sort-stamps' +import type { ExternalConversationLink } from '../external-conversation-link' // ─── Worktree metadata (persisted user-authored fields only) ───────── export type WorktreeMeta = { @@ -101,4 +102,6 @@ export type WorktreeMeta = { automationProvenance?: AutomationWorkspaceProvenance /** System-owned provenance for workspaces created via `orca worktree create`. */ cliProvenance?: CliWorkspaceProvenance + /** System-owned connector threads whose agents run here; never published on `Worktree`. */ + externalConversations?: ExternalConversationLink[] }