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
This commit is contained in:
Jinwoo Hong
2026-10-22 17:40:40 -04:00
committed by GitHub
parent aee882325f
commit 57b5082404
35 changed files with 1363 additions and 22 deletions
+5
View File
@@ -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'],
+35
View File
@@ -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<never> => {
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'
)
})
})
+76
View File
@@ -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<TResult>(
client: Pick<HandlerContext['client'], 'call'>,
method: string,
params: Record<string, unknown>
) {
try {
return await client.call<TResult>(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<string, CommandHandler> = {
'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))
}
}
+31
View File
@@ -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 <id> --delivery-id <id> --agent claude|codex --prompt <text> (--worktree <selector> | --project-group <id> [--name <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.'
]
}
]
+2
View File
@@ -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,
@@ -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)
)
}
@@ -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<string, Promise<unknown>>()
constructor(private readonly signal: AbortSignal) {}
enqueue(id: string, run: () => Promise<Delivery>): Promise<Delivery> {
const previous = this.tails.get(id) ?? Promise.resolve()
const { signal } = this
const turn = new Promise<void>((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
}
}
@@ -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<string, Promise<Delivery>>()
constructor(private readonly db: () => DeliveryReceiptDb) {}
run(
key: ConversationKey,
deliveryId: string,
method: 'connector.start' | 'connector.send',
payload: unknown,
execute: () => Promise<Delivery>
): Promise<Delivery> {
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<DeliveryReceiptDb['beginMutationReceipt']>
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
}
}
@@ -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<string, LinkRecord & { hostId?: ExecutionHostId }>
getAllWorktreeMetaForHost?: (
hostId: ExecutionHostId
) => Record<string, LinkRecord & { hostId?: ExecutionHostId }>
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<ExecutionHostId>([
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
}
@@ -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<GlobalSettings> = {},
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<T>() {
let resolve!: (value: T) => void
const promise = new Promise<T>((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<void>()
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<void>()
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()
})
})
@@ -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<ConversationLaunchInput, 'agent'> & { 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<ConversationStart> {
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 }
}
}
@@ -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<AgentLaunchResult>
/** The launched workspace as hook rows name it, qualified by the host that runs it. */
paneWorkspace(worktreeId: string): Promise<AgentPaneWorkspace>
}
/**
* `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<Delivery> {
// Settled to its error, so a failed write waits for the launch instead of going unhandled.
const surface: { published: boolean; linked: Promise<unknown> | 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' }
}
@@ -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])
}
@@ -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<ExecutionHostId, Record<string, LinkRow>>()
const legacy: Record<string, LinkRow> = {}
const folders: (LinkRow & { id: string })[] = []
const writes: HostWorktreeTarget[] = []
const rows = (hostId: ExecutionHostId): Record<string, LinkRow> => {
const existing = byHost.get(hostId)
if (existing) {
return existing
}
const created: Record<string, LinkRow> = {}
byHost.set(hostId, created)
return created
}
const put = (id: string, hostId: ExecutionHostId, row: Omit<LinkRow, 'hostId'>): 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<AgentLaunchResult>,
workspace: AgentPaneWorkspace = { worktreeId: WORKTREE, sshHostId: null },
publish = true
) {
const launch = vi.fn(
async (
_input: unknown,
onSurfacePublished: (surface: AgentLaunchPublishedSurface) => void,
_signal?: AbortSignal
): Promise<AgentLaunchResult> => {
if (publish) {
onSurfacePublished(launchResult(null, workspace.worktreeId))
}
return settle()
}
)
const launcher: ConversationLauncher = { launch, paneWorkspace: async () => workspace }
return { launcher, launch }
}
@@ -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<GlobalSettings>) => void
function fakeStore(initial: Partial<GlobalSettings>) {
let settings = { ...initial }
const listeners = new Set<SettingsListener>()
const getSettings = vi.fn(() => settings)
const store = {
...fakeLinkStore().store,
getSettings,
onSettingsChanged: (listener: SettingsListener) => {
listeners.add(listener)
return () => listeners.delete(listener)
}
}
const update = (updates: Partial<GlobalSettings>): 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()
})
})
@@ -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<GlobalSettings, 'experimentalConnectors' | 'agentStatusHooksEnabled' | 'disabledTuiAgents'>
>
onSettingsChanged?: (listener: (updates: Partial<GlobalSettings>) => 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
}
}
}
@@ -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')
@@ -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<HostQualifiedWorktreeMetaStore, 'getAllWorktreeMeta' | 'getAllWorktreeMetaForHost'>,
export function readAllWorktreeMetaForHost<
TMeta extends { hostId?: ExecutionHostId } = WorktreeMeta
>(
store: {
getAllWorktreeMeta: () => Record<string, TMeta>
getAllWorktreeMetaForHost?: (executionHostId: ExecutionHostId) => Record<string, TMeta>
},
executionHostId: ExecutionHostId
): Record<string, WorktreeMeta> {
): Record<string, TMeta> {
const qualified = store.getAllWorktreeMetaForHost?.(executionHostId)
if (qualified) {
return qualified
}
const projected: Record<string, WorktreeMeta> = {}
const projected: Record<string, TMeta> = {}
for (const [worktreeId, meta] of Object.entries(store.getAllWorktreeMeta())) {
if (!meta.hostId || meta.hostId === executionHostId) {
projected[worktreeId] = meta
@@ -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
+18 -17
View File
@@ -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<typeof installStructuredAgentAvailability>
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)
+1
View File
@@ -101,6 +101,7 @@ const RUNTIME_PASSTHROUGH_CODES: ReadonlySet<string> = new Set([
'remote_update_manual_required',
'remote_update_not_available',
'remote_update_not_downloaded',
'connectors_disabled',
...AGENT_SESSION_RPC_ERROR_CODES
])
@@ -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'
)
})
})
+43
View File
@@ -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<OrcaRuntimeService, 'getExternalConversationSeam'>
): 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
)
}
})
]
+2
View File
@@ -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,
@@ -53,6 +53,7 @@ export type RuntimeHostDisplayLabelOverrides = Partial<
export type RuntimeClientSettingsUpdate = Pick<
Partial<GlobalSettings>,
| 'agentStatusHooksEnabled'
| 'experimentalConnectors'
| 'defaultTuiAgent'
| 'disabledTuiAgents'
| 'agentDefaultArgs'
@@ -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<typeof listReferenceWorkspaces>[0],
listing: {
cached: () => readonly Worktree[] | undefined
scan: () => Promise<readonly Worktree[]>
},
selector: string,
cwd?: string
): Promise<ReferenceWorkspace> {
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,
+1
View File
@@ -14,6 +14,7 @@ export const CLI_COMMAND_NAMES = [
'click',
'clipboard',
'computer',
'connector',
'console',
'cookie',
'dblclick',
+15
View File
@@ -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
}
+3
View File
@@ -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
+5 -1
View File
@@ -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(
+1
View File
@@ -23,6 +23,7 @@ export const HOST_OPERATION_OWNERSHIP = {
browser: 'host',
clipboard: 'host',
computer: 'host',
connector: 'host',
desktopBrowserBridge: 'host',
diagnostics: 'host',
emulator: 'host',
@@ -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(),
@@ -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')
})
+2
View File
@@ -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,
+3
View File
@@ -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[]
}