From 634ca45fbbba08065757e2be55036912a2c25e28 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:47:12 -0700 Subject: [PATCH] feat(acp): hosts install a managed runtime over RPC, and mark agents whose program Orca provides (#29024) * feat(acp): hosts install a managed runtime over RPC, and mark agents whose program Orca provides managedRuntime.status / managedRuntime.install / managedRuntime.cancel, negotiated by managed-runtime.install.v1, act on the host that receives them, the one that runs the agent's chats, with no workspace needed. A status read naming the revision it shows is a long-poll on its own call-queue lane, so every client sees another's start, progress and outcome. agentSession.agents marks an agent whose program Orca provides on that host. * fix(runtime): install status says a copy outside Orca's runs only when a start would really run it * fix(runtime): install status is pushed to subscribers instead of long-polled, and Cancel answers with the outcome managedRuntime.subscribe {agent} sends a full snapshot on subscribe, then one snapshot per install state change, and closes with the transport. It replaces the managedRuntime.status long-poll and its slot registrations. --- .../acp/acp-structured-agent-definitions.ts | 2 + .../structured-agent-definition.ts | 2 + .../managed-runtime-installation.ts | 15 ++- src/main/runtime/revisioned-state.ts | 12 +- src/main/runtime/rpc/methods/index.ts | 2 + .../methods/managed-runtime-install.test.ts | 83 ++++++++++++ .../rpc/methods/managed-runtime-install.ts | 123 ++++++++++++++++++ .../structured-agent-session-agents.ts | 3 +- src/shared/agent-session-registered-agents.ts | 7 +- src/shared/host-operation-ownership.ts | 1 + src/shared/managed-runtime-install.ts | 3 + src/shared/protocol-version.ts | 2 + .../managed-runtime-install-params.ts | 9 ++ .../rpc-params-catalog.generated.ts | 7 + 14 files changed, 261 insertions(+), 10 deletions(-) create mode 100644 src/main/runtime/rpc/methods/managed-runtime-install.test.ts create mode 100644 src/main/runtime/rpc/methods/managed-runtime-install.ts create mode 100644 src/shared/rpc-contract/managed-runtime-install-params.ts diff --git a/src/main/acp/acp-structured-agent-definitions.ts b/src/main/acp/acp-structured-agent-definitions.ts index 1bdf6cb08dd..20a8f134613 100644 --- a/src/main/acp/acp-structured-agent-definitions.ts +++ b/src/main/acp/acp-structured-agent-definitions.ts @@ -19,6 +19,7 @@ export function isAcpStructuredOptionKey(key: string): boolean { /** What the host knows about an ACP agent before any of its sessions runs. */ export function acpStructuredAgentDefinition(spec: AcpLaunchSpec): StructuredAgentDefinition { + const { managedRuntime } = spec return { agent: spec.agent, handleTransport: ACP_HANDLE_TRANSPORT, @@ -35,6 +36,7 @@ export function acpStructuredAgentDefinition(spec: AcpLaunchSpec): StructuredAge // Orca answers every permission request the agent sends; the agent decides when it asks. approvalEnforcement: 'orca' }, + ...(managedRuntime && { managedRuntime: () => managedRuntime.pin() !== null }), restingOptions: { acceptsKey: isAcpStructuredOptionKey, // The agent lists its models over the protocol once it runs; Orca keeps no list of its own. diff --git a/src/main/native-chat/agent-session-wire/structured-agent-definition.ts b/src/main/native-chat/agent-session-wire/structured-agent-definition.ts index 5a1e43ce222..c50710e260b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-definition.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-definition.ts @@ -11,6 +11,8 @@ import type { AgentSessionModelOption } from '../../../shared/agent-session-wire /** `agent` names the Orca agent whose sessions this describes; the storage fields bound its records. */ export type StructuredAgentDefinition = AgentSessionStoredAgent & { capabilities: AgentSessionCapabilities + /** Whether Orca provides this agent's chat program on this host; absent when it never does. */ + managedRuntime?: () => boolean /** How a session's options read and change while no child runs. */ restingOptions: { /** Whether the agent takes a pick of this option key. */ diff --git a/src/main/node-runtime/managed-runtime-installation.ts b/src/main/node-runtime/managed-runtime-installation.ts index 3143a1e3e97..0fc7474618e 100644 --- a/src/main/node-runtime/managed-runtime-installation.ts +++ b/src/main/node-runtime/managed-runtime-installation.ts @@ -17,7 +17,6 @@ import { } from './managed-runtime-release' import { installFailureText, installRuntimeRelease } from './managed-runtime-release-install' -const STATUS_WAIT_MS = 25_000 const DEADLINE_MS = 45 * 60_000 export type ManagedRuntimeInstallationDeps = { @@ -46,14 +45,13 @@ export class ManagedRuntimeInstallation { this.mutation = this.reclaim() } - /** At once without `shown`, else once the revision passes it or after a bounded wait. Disk and - * PATH are reread on a first read and when the phase changes, never for byte progress alone. */ - async status(shown?: number, signal?: AbortSignal): Promise { - await this.state.waitForChange(shown, STATUS_WAIT_MS, signal) + /** Disk and PATH are reread on a `fresh` read and when the phase changes, never for byte + * progress alone. */ + async status(fresh = true): Promise { const { value: operation, revision } = this.state.get() const phase = `${operation?.id}:${operation?.phase}` let facts = this.facts - if (shown === undefined || facts?.phase !== phase) { + if (fresh || facts?.phase !== phase) { const read = this.readInstalledFacts() facts = { phase, read } this.facts = facts @@ -65,6 +63,11 @@ export class ManagedRuntimeInstallation { return { agent: this.agent, revision, ...(await facts.read), operation } } + /** Calls `onChange` after every state revision until the returned unsubscribe. */ + subscribe(onChange: () => void): () => void { + return this.state.subscribe(onChange) + } + private async readInstalledFacts(): Promise { const pin = this.runtime.pin() const { release, broken, unreadable } = resolveInstalledRuntimeRelease(this.root, pin) diff --git a/src/main/runtime/revisioned-state.ts b/src/main/runtime/revisioned-state.ts index acbfd7dd409..f815b33b94f 100644 --- a/src/main/runtime/revisioned-state.ts +++ b/src/main/runtime/revisioned-state.ts @@ -1,5 +1,5 @@ -// A value read by revision: a reader names the revision it shows and waits, bounded, for a newer -// one. Its owner keeps its own lifecycle policy; this only publishes and waits. +// A value read by revision: a reader subscribes to every change, or names the revision it shows and +// waits, bounded, for a newer one. Its owner keeps its own lifecycle policy; this only publishes. import { delayUnlessAborted } from '../../shared/abort-aware-delay' @@ -19,6 +19,14 @@ export class RevisionedState { this.changed.forEach((notify) => notify()) } + /** Calls `onChange` after every set until the returned unsubscribe. */ + subscribe(onChange: () => void): () => void { + this.changed.add(onChange) + return () => { + this.changed.delete(onChange) + } + } + /** Resolves once the revision differs from `shown`, after `waitMs`, or when `signal` aborts. */ async waitForChange(shown: number | undefined, waitMs: number, signal?: AbortSignal) { if (shown !== this.revision) { diff --git a/src/main/runtime/rpc/methods/index.ts b/src/main/runtime/rpc/methods/index.ts index 83bb67b49c3..c6fa10402e4 100644 --- a/src/main/runtime/rpc/methods/index.ts +++ b/src/main/runtime/rpc/methods/index.ts @@ -58,6 +58,7 @@ import { STRUCTURED_AGENT_SESSION_METHODS } from './structured-agent-session' import { STRUCTURED_AGENT_SESSION_ATTACHMENT_METHODS } from './structured-agent-session-attachments' import { STRUCTURED_AGENT_SESSION_VISUAL_METHODS } from './structured-agent-session-visual' import { STRUCTURED_AGENT_SESSION_AGENTS_METHODS } from './structured-agent-session-agents' +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' @@ -83,6 +84,7 @@ export const ALL_RPC_METHODS = [ ...STRUCTURED_AGENT_SESSION_ATTACHMENT_METHODS, ...STRUCTURED_AGENT_SESSION_VISUAL_METHODS, ...STRUCTURED_AGENT_SESSION_AGENTS_METHODS, + ...MANAGED_RUNTIME_INSTALL_METHODS, ...AGENT_LAUNCH_METHODS, ...TERMINAL_METHODS, ...TERMINAL_ORPHAN_METHODS, diff --git a/src/main/runtime/rpc/methods/managed-runtime-install.test.ts b/src/main/runtime/rpc/methods/managed-runtime-install.test.ts new file mode 100644 index 00000000000..6ea13e118b8 --- /dev/null +++ b/src/main/runtime/rpc/methods/managed-runtime-install.test.ts @@ -0,0 +1,83 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { expect, it, vi } from 'vitest' +import type * as LaunchSpecs from '../../../acp/acp-launch-specs' +import type { RpcContext } from '../core' +import { MANAGED_RUNTIME_INSTALL_METHODS } from './managed-runtime-install' + +const host = vi.hoisted(() => ({ dataDirectory: '' })) +vi.mock('../../../orca-profiles/profile-storage-paths', () => ({ + getProfileUserDataPath: () => host.dataDirectory +})) +vi.mock('../../structured-agent-launch-support', () => ({ + resolveHostStructuredAgentLaunchEnvironment: async () => ({ PATH: '' }) +})) +// The download never answers until its cancel. +vi.mock('../../../network/http-client', () => ({ + getMainHttpClient: () => ({ + fetch: (_url: string, { signal }: RequestInit) => + new Promise((_resolve, reject) => { + signal?.throwIfAborted() + signal?.addEventListener('abort', () => reject(signal.reason)) + }) + }) +})) +vi.mock('../../../acp/acp-launch-specs', async (importOriginal) => { + const actual = await importOriginal() + const pin = { + version: '1.0', + url: 'https://example.test/grok-1.0.tar.gz', + archiveSha256: 'a'.repeat(64), + archiveBytes: 10, + members: [{ path: 'server', sha256: 'b'.repeat(64), bytes: 6 }] + } + const managedRuntime = { label: 'Grok', root: (data: string) => `${data}/rt`, pin: () => pin } + return { + ...actual, + acpLaunchSpecFor: (agent: string) => { + const spec = actual.acpLaunchSpecFor(agent) + return spec && { ...spec, command: 'orca-rpc-managed-test', managedRuntime } + } + } +}) + +const [subscribeMethod, installMethod, cancelMethod] = MANAGED_RUNTIME_INSTALL_METHODS +// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the handlers read only the settings store and the stream registry. +const ctx = { + runtime: { + requireStore: () => ({ getSettings: () => ({}) }), + registerOwnedSubscriptionCleanup: (_id: string, cleanup: () => void) => ({ + releaseIfCurrent: cleanup + }) + } +} as unknown as RpcContext + +it('pushes a snapshot, then the install another client starts, until the transport closes', async () => { + host.dataDirectory = await mkdtemp(join(tmpdir(), 'orca-runtime-install-rpc-')) + const transport = new AbortController() + const frames: unknown[] = [] + const watcher = { ...ctx, signal: transport.signal } + await subscribeMethod.handler({ agent: 'grok' }, watcher, (frame) => frames.push(frame)) + expect(frames).toEqual([ + expect.objectContaining({ available: true, installedVersion: null, operation: null }) + ]) + const started = await installMethod.handler({ agent: 'grok' }, ctx) + const id = String(started.operation?.id) + await vi.waitFor(() => + expect(frames.at(-1)).toMatchObject({ operation: { id, phase: 'downloading' } }) + ) + // Cancel answers with the outcome once the run has unwound. + const outcome = await cancelMethod.handler({ agent: 'grok', operationId: id }, ctx) + expect(outcome).toMatchObject({ id, phase: 'cancelled' }) + await vi.waitFor(() => + expect(frames.at(-1)).toMatchObject({ operation: { phase: 'cancelled' } }) + ) + + transport.abort() + const seen = frames.length + const again = await installMethod.handler({ agent: 'grok' }, ctx) + await cancelMethod.handler({ agent: 'grok', operationId: String(again.operation?.id) }, ctx) + expect(frames).toHaveLength(seen) + await rm(host.dataDirectory, { recursive: true, force: true }) +}) diff --git a/src/main/runtime/rpc/methods/managed-runtime-install.ts b/src/main/runtime/rpc/methods/managed-runtime-install.ts new file mode 100644 index 00000000000..ec31cecee1b --- /dev/null +++ b/src/main/runtime/rpc/methods/managed-runtime-install.ts @@ -0,0 +1,123 @@ +// The runtime Orca installs for an agent, on the host that receives the call: the one that runs +// that agent's chats. Needs no workspace. Negotiated by `managed-runtime.install.v1`. Status is +// pushed: a snapshot on subscribe, then one per state revision. + +import type { OrcaRuntimeService } from '../../orca-runtime' +import { + ManagedRuntimeAgentParams, + ManagedRuntimeCancelParams +} from '../../../../shared/rpc-contract/managed-runtime-install-params' +import { acpLaunchSpecFor } from '../../../acp/acp-launch-specs' +import { resolveAcpLaunchCommand } from '../../../acp/acp-structured-launch-resolution' +import { ManagedRuntimeInstallation } from '../../../node-runtime/managed-runtime-installation' +import { isAgentSessionPreSpawnError } from '../../../native-chat/agent-session-wire/structured-agent-session-adapter' +import { getProfileUserDataPath } from '../../../orca-profiles/profile-storage-paths' +import { resolveHostStructuredAgentLaunchEnvironment } from '../../structured-agent-launch-support' +import { isPathInsideOrEqual } from '../../../../shared/cross-platform-path' +import { defineMethod, defineStreamingMethod } from '../core' +import { bindStructuredAgentSessionStream } from './structured-agent-session-status-stream' + +const installations = new Map() + +function installationFor(agent: string, runtime: OrcaRuntimeService): ManagedRuntimeInstallation { + const spec = acpLaunchSpecFor(agent) + if (!spec?.managedRuntime) { + throw new Error(`Orca installs no runtime for ${agent}.`) + } + const dataDirectory = getProfileUserDataPath() + const releases = spec.managedRuntime.root(dataDirectory) + // What a start would run, outside Orca's releases: the launch's own lookup decides. + const findExternal = async (): Promise => { + const settings = runtime.requireStore().getSettings() + const env = { ...(await resolveHostStructuredAgentLaunchEnvironment(agent, settings)), ...spec.env } + try { + const command = resolveAcpLaunchCommand(spec, env, { + commandSettings: settings, + dataDirectory + }) + return command !== spec.command && !isPathInsideOrEqual(releases, command) + } catch (error) { + // An unrunnable Command setting refuses the start itself; it is no copy. + if (isAgentSessionPreSpawnError(error)) { + return false + } + throw error + } + } + const installation = + installations.get(agent) ?? + // The chat host's own state root, where its launches look. + new ManagedRuntimeInstallation(agent, spec.managedRuntime, { + dataDirectory, + findExternal + }) + installations.set(agent, installation) + return installation +} + +export const MANAGED_RUNTIME_INSTALL_METHODS = [ + defineStreamingMethod({ + name: 'managedRuntime.subscribe', + permission: 'workspace', + params: ManagedRuntimeAgentParams, + handler: async ({ agent }, ctx, emit) => { + const installation = installationFor(agent, ctx.runtime) + // A failed first read fails the subscribe; a later one keeps the client's last snapshot. + let shown = await installation.status() + const frame = ctx.requestId ? `:${ctx.requestId}` : '' + const subscriptionId = `managedRuntime:${ctx.connectionId ?? 'local'}:${agent}${frame}` + let dispose = (): void => {} + const stream = bindStructuredAgentSessionStream(ctx, subscriptionId, () => dispose()) + if (stream.isClosed()) { + return + } + let reading = false + let again = false + // One read at a time; revisions during it coalesce into one more read. + const follow = async (): Promise => { + again = true + if (reading) { + return + } + reading = true + try { + while (again && !stream.isClosed()) { + again = false + const next = await installation.status(false) + if (next.revision !== shown.revision && !stream.isClosed()) { + shown = next + emit(next) + } + } + } catch (error) { + console.warn(`[managed-runtime] ${agent} status not read for a subscriber:`, error) + } finally { + reading = false + } + } + dispose = installation.subscribe(() => void follow()) + emit(shown) + // Catches a revision between the first read and the subscribe. + void follow() + } + }), + defineMethod({ + name: 'managedRuntime.install', + permission: 'host-admin', + params: ManagedRuntimeAgentParams, + handler: async ({ agent }, { runtime }) => { + const installation = installationFor(agent, runtime) + installation.install() + return installation.status() + } + }), + defineMethod({ + name: 'managedRuntime.cancel', + permission: 'host-admin', + params: ManagedRuntimeCancelParams, + handler: async ({ agent, operationId }, { runtime }) => { + // The outcome once the run has unwound: `cancelled`, or `succeeded` if it had committed. + return await installationFor(agent, runtime).cancel(operationId) + } + }) +] as const diff --git a/src/main/runtime/rpc/methods/structured-agent-session-agents.ts b/src/main/runtime/rpc/methods/structured-agent-session-agents.ts index 0bb7524d419..aa5169a2607 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-agents.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-agents.ts @@ -26,7 +26,8 @@ export const STRUCTURED_AGENT_SESSION_AGENTS_METHODS = [ clientReadsStructuredSessionAgent(ctx, definition.agent) ).map(({ definition }) => ({ agent: definition.agent, - capabilities: { ...definition.capabilities } + capabilities: { ...definition.capabilities }, + ...(definition.managedRuntime?.() ? { managedRuntime: true as const } : {}) })) } } diff --git a/src/shared/agent-session-registered-agents.ts b/src/shared/agent-session-registered-agents.ts index fc4b16f9001..c9aab9e2ee8 100644 --- a/src/shared/agent-session-registered-agents.ts +++ b/src/shared/agent-session-registered-agents.ts @@ -18,6 +18,9 @@ export const AGENT_SESSION_AGENTS_METHOD = 'agentSession.agents' export type AgentSessionRegisteredAgent = { agent: string capabilities: AgentSessionCapabilities + /** Orca provides this chat's program on this host, so it needs no installed CLI. Absent from + * older hosts, whose chats are offered only where the agent's CLI was detected. */ + managedRuntime?: true } export type AgentSessionAgentsResult = { @@ -46,7 +49,9 @@ const registeredAgentSchema = z.object({ approvalEnforcement: openEnum(['provider', 'orca'] as const, 'orca') .optional() .default('orca') - }) + }), + // A false from a host reads as absent, so the row keeps the detection-based offer. + managedRuntime: z.literal(true).optional().catch(undefined) }) /** The host's agents as this build reads them, or null when the reply is not an agent list. */ diff --git a/src/shared/host-operation-ownership.ts b/src/shared/host-operation-ownership.ts index 9d19ecd511e..b106126613c 100644 --- a/src/shared/host-operation-ownership.ts +++ b/src/shared/host-operation-ownership.ts @@ -36,6 +36,7 @@ export const HOST_OPERATION_OWNERSHIP = { jira: 'scope-picked', layout: 'host', linear: 'scope-picked', + managedRuntime: 'host', managedServer: 'host', markdown: 'host', mobileRelay: 'host', diff --git a/src/shared/managed-runtime-install.ts b/src/shared/managed-runtime-install.ts index 4d76016d9a7..cc5cf1f46ea 100644 --- a/src/shared/managed-runtime-install.ts +++ b/src/shared/managed-runtime-install.ts @@ -1,5 +1,8 @@ // A runtime Orca installs for an agent only when the person asks, as its host reports it. +// The host answers managedRuntime.subscribe, managedRuntime.install and managedRuntime.cancel. +export const MANAGED_RUNTIME_INSTALL_RUNTIME_CAPABILITY = 'managed-runtime.install.v1' as const + export type ManagedRuntimeOperation = { id: string kind: 'install' | 'remove' diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index c1190481642..e220e5540fe 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -62,6 +62,7 @@ export * from './agent-session-resume-runtime-capabilities' import { AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY } from './agent-session-background-task-child-views-capability' import { AGENT_SESSION_STOP_RUNTIME_CAPABILITIES } from './agent-session-stop-capabilities' import { AGENT_SESSION_CREATE_RUNTIME_CAPABILITIES } from './agent-session-create-capabilities' +import { MANAGED_RUNTIME_INSTALL_RUNTIME_CAPABILITY } from './managed-runtime-install' import { REMOTE_SERVER_UPDATE_CAPABILITY } from './remote-server-update' import { MOBILE_RELAY_RUNTIME_CAPABILITIES } from './mobile-desktop-relay-contract' import { SKILL_RUNTIME_CAPABILITIES } from './skill-install-capability' @@ -229,6 +230,7 @@ export const RUNTIME_CAPABILITIES = [ ...SESSION_TABS_RUNTIME_CAPABILITIES, AGENT_SESSION_BOUNDARY_RUNTIME_CAPABILITY, REMOTE_SERVER_UPDATE_CAPABILITY, + MANAGED_RUNTIME_INSTALL_RUNTIME_CAPABILITY, AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY, AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY, AGENT_SESSION_KEYBOARD_RUNTIME_CAPABILITY, diff --git a/src/shared/rpc-contract/managed-runtime-install-params.ts b/src/shared/rpc-contract/managed-runtime-install-params.ts new file mode 100644 index 00000000000..97bf0a99d19 --- /dev/null +++ b/src/shared/rpc-contract/managed-runtime-install-params.ts @@ -0,0 +1,9 @@ +import { z } from 'zod' +import { StructuredAgent } from './structured-agent-session-params' + +export const ManagedRuntimeAgentParams = z.strictObject({ agent: StructuredAgent }) + +export const ManagedRuntimeCancelParams = z.strictObject({ + agent: StructuredAgent, + operationId: z.string().min(1).max(64) +}) diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index a975737226e..848b21f728d 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -381,6 +381,10 @@ import { WorkspaceSelection } from './linear-params' import { CreateProject } from './linear-project-create-params' +import { + ManagedRuntimeAgentParams, + ManagedRuntimeCancelParams +} from './managed-runtime-install-params' import { ManagedServerRecover, ManagedServerSelector, @@ -1083,6 +1087,9 @@ export const RPC_PARAMS_BY_METHOD = { 'linear.teamStates': TeamId, 'linear.testConnection': WorkspaceSelection, 'linear.updateIssue': IssueUpdateOfLinearParams, + 'managedRuntime.cancel': ManagedRuntimeCancelParams, + 'managedRuntime.install': ManagedRuntimeAgentParams, + 'managedRuntime.subscribe': ManagedRuntimeAgentParams, 'managedServer.cancelStop': ManagedServerSelector, 'managedServer.forget': ManagedServerSelector, 'managedServer.recover': ManagedServerRecover,