mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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<ManagedRuntimeInstallState> {
|
||||
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<ManagedRuntimeInstallState> {
|
||||
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<InstalledFacts> {
|
||||
const pin = this.runtime.pin()
|
||||
const { release, broken, unreadable } = resolveInstalledRuntimeRelease(this.root, pin)
|
||||
|
||||
@@ -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<S> {
|
||||
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) {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof LaunchSpecs>()
|
||||
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 })
|
||||
})
|
||||
@@ -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<string, ManagedRuntimeInstallation>()
|
||||
|
||||
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<boolean> => {
|
||||
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<void> => {
|
||||
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
|
||||
@@ -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 } : {})
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user