From 98690db2ab1d53d7e67eae296dc061c000c2ea2f Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:08:31 -0400 Subject: [PATCH] feat(connector): the desktop holds the Orca Cloud connector poll (#33852) With the experimentalConnectors setting on and the desktop signed in to Orca, the desktop keeps one long-poll request open to the Orca Cloud connector and hands each delivery to the existing connector code that starts or talks to an agent. Each delivery is confirmed twice (received, then its result), and the agent's turn and prompt events are reported back. The desktop authenticates with a 5-minute connector token from the Orca sign-in service, follows the relay's start/stop lifecycle, and stops on sign-out. Its hello lists Slack as the one source it accepts. Off by default with no settings UI. --- .../connector-cloud-install.ts | 52 ++ .../connector-cloud-request.ts | 97 +++ .../connector-cloud/connector-cloud.test.ts | 168 +++++ src/main/connector-cloud/connector-cloud.ts | 103 +++ .../connector-cloud/connector-deliveries.ts | 84 +++ .../connector-cloud/connector-hello.test.ts | 59 ++ src/main/connector-cloud/connector-hello.ts | 49 ++ src/main/connector-cloud/connector-machine.ts | 21 + src/main/connector-cloud/connector-origin.ts | 7 + .../connector-cloud/connector-outbox.test.ts | 155 ++++ src/main/connector-cloud/connector-outbox.ts | 213 ++++++ .../connector-cloud/connector-places.test.ts | 125 ++++ src/main/connector-cloud/connector-places.ts | 184 +++++ .../connector-poll-client.test.ts | 677 ++++++++++++++++++ .../connector-cloud/connector-poll-client.ts | 322 +++++++++ .../connector-snapshot.test.ts | 90 +++ .../connector-cloud/connector-snapshot.ts | 66 ++ .../connector-token-mint.test.ts | 84 +++ .../connector-cloud/connector-token-mint.ts | 79 ++ .../connector-cloud/connector-transport.ts | 85 +++ .../connector-turn-events.test.ts | 138 ++++ .../connector-cloud/connector-turn-events.ts | 231 ++++++ .../fake-connector-cloud.test-fixture.ts | 200 ++++++ .../fake-connector-seam.test-fixture.ts | 74 ++ ...nal-conversation-delivery-receipts.test.ts | 36 + ...external-conversation-delivery-receipts.ts | 33 +- .../external-conversation-events.ts | 27 +- .../external-conversation-prompts.test.ts | 7 +- .../external-conversation-seam.ts | 5 + src/main/ipc/orca-profile-id-arg.ts | 10 + src/main/ipc/orca-profiles.ts | 15 +- .../register-core-handlers.ts | 4 +- .../profile-cloud-auth-config.test.ts | 2 + .../profile-cloud-auth-config.ts | 4 + .../profile-cloud-client.test.ts | 1 + .../profile-cloud-org-members-client.test.ts | 1 + .../orca-profiles/profile-cloud-pkce.test.ts | 1 + .../startup/main-process-connector-startup.ts | 22 + src/main/startup/main-process-quit.ts | 1 + ...-process-runtime-launch-activation.test.ts | 4 + .../startup/main-process-runtime-launch.ts | 8 + src/main/startup/main-process-state.ts | 3 + src/main/startup/main-window-core-services.ts | 6 +- src/shared/connector-cloud-contract.ts | 113 ++- 44 files changed, 3640 insertions(+), 26 deletions(-) create mode 100644 src/main/connector-cloud/connector-cloud-install.ts create mode 100644 src/main/connector-cloud/connector-cloud-request.ts create mode 100644 src/main/connector-cloud/connector-cloud.test.ts create mode 100644 src/main/connector-cloud/connector-cloud.ts create mode 100644 src/main/connector-cloud/connector-deliveries.ts create mode 100644 src/main/connector-cloud/connector-hello.test.ts create mode 100644 src/main/connector-cloud/connector-hello.ts create mode 100644 src/main/connector-cloud/connector-machine.ts create mode 100644 src/main/connector-cloud/connector-origin.ts create mode 100644 src/main/connector-cloud/connector-outbox.test.ts create mode 100644 src/main/connector-cloud/connector-outbox.ts create mode 100644 src/main/connector-cloud/connector-places.test.ts create mode 100644 src/main/connector-cloud/connector-places.ts create mode 100644 src/main/connector-cloud/connector-poll-client.test.ts create mode 100644 src/main/connector-cloud/connector-poll-client.ts create mode 100644 src/main/connector-cloud/connector-snapshot.test.ts create mode 100644 src/main/connector-cloud/connector-snapshot.ts create mode 100644 src/main/connector-cloud/connector-token-mint.test.ts create mode 100644 src/main/connector-cloud/connector-token-mint.ts create mode 100644 src/main/connector-cloud/connector-transport.ts create mode 100644 src/main/connector-cloud/connector-turn-events.test.ts create mode 100644 src/main/connector-cloud/connector-turn-events.ts create mode 100644 src/main/connector-cloud/fake-connector-cloud.test-fixture.ts create mode 100644 src/main/connector-cloud/fake-connector-seam.test-fixture.ts create mode 100644 src/main/ipc/orca-profile-id-arg.ts create mode 100644 src/main/startup/main-process-connector-startup.ts diff --git a/src/main/connector-cloud/connector-cloud-install.ts b/src/main/connector-cloud/connector-cloud-install.ts new file mode 100644 index 00000000000..a4ded4bb7eb --- /dev/null +++ b/src/main/connector-cloud/connector-cloud-install.ts @@ -0,0 +1,52 @@ +import { hostname } from 'node:os' +import { getAppEnvironment } from '../../shared/app-environment' +import { admitConnectorAgent, CONNECTOR_AGENTS } from '../external-conversations/connector-agents' +import { runtimeConversationHost } from '../external-conversations/external-conversation-seam' +import { getOrcaCloudAuthConfig } from '../orca-profiles/profile-cloud-auth-config' +import type { OrcaRuntimeService } from '../runtime/orca-runtime' +import type { RuntimeStore } from '../runtime/runtime-store-contract' +import { ConnectorCloud } from './connector-cloud' +import { resolveConnectorOrigin } from './connector-origin' +import { connectorProjects, type ConnectorPlacesStore } from './connector-places' +import { connectorAccount, desktopConnectorTokenMint } from './connector-token-mint' + +export type ConnectorCloudStore = ConnectorPlacesStore & + Pick + +/** Desktop only: the connector acts as the desktop's Orca account; `orca serve` has no sign-in yet. */ +export function createDesktopConnectorCloud( + runtime: OrcaRuntimeService, + store: ConnectorCloudStore, + userDataPath: string, + isFenced: () => boolean +): ConnectorCloud | null { + const auth = getOrcaCloudAuthConfig() + if (!auth.configured) { + return null + } + const environment = getAppEnvironment() + return new ConnectorCloud({ + userDataPath, + origin: resolveConnectorOrigin(process.env, environment.isPackaged()), + isFenced, + account: () => connectorAccount(userDataPath), + mintFor: (account) => desktopConnectorTokenMint(auth.config, userDataPath, account), + machine: { + runtime: 'desktop', + orcaVersion: environment.getVersion(), + hostName: hostname(), + seam: () => runtime.getExternalConversationSeam(), + conversationHost: () => runtimeConversationHost(runtime), + agents: () => + CONNECTOR_AGENTS.filter((agent) => admitConnectorAgent(store.getSettings(), agent)), + projects: () => connectorProjects(store), + onConnectorsToggled: (listener) => { + store.onSettingsChanged?.((updates) => { + if ('experimentalConnectors' in updates) { + listener() + } + }) + } + } + }) +} diff --git a/src/main/connector-cloud/connector-cloud-request.ts b/src/main/connector-cloud/connector-cloud-request.ts new file mode 100644 index 00000000000..78fb05ed569 --- /dev/null +++ b/src/main/connector-cloud/connector-cloud-request.ts @@ -0,0 +1,97 @@ +import type { z } from 'zod' +import { ConnectorErrorResponse } from '../../shared/connector-cloud-contract' +import { cancelUnreadResponseBody } from '../lib/unread-response-body' + +const DEFAULT_REQUEST_TIMEOUT_MS = 15_000 + +export type ConnectorFailure = { + ok: false + /** null when no readable HTTP answer arrived: network loss, timeout, or a body that is not the contract. */ + status: number | null + /** The cloud's error code, when it sent one. */ + error: string | null + retryAfterMs: number | null +} +/** An answer that is not the contract reads like no answer at all. */ +export const UNREADABLE_ANSWER: ConnectorFailure = { + ok: false, + status: null, + error: null, + retryAfterMs: null +} + +export type ConnectorResult = { ok: true; value: T } | ConnectorFailure + +export type ConnectorRequest = { + fetchImpl: typeof globalThis.fetch + url: string + method?: 'POST' | 'PUT' + body: unknown + token?: string + timeoutMs?: number + signal?: AbortSignal +} + +function failure( + status: number | null, + error: string | null = null, + retryAfterMs: number | null = null +): ConnectorFailure { + return { ok: false, status, error, retryAfterMs } +} + +function readRetryAfterMs(response: Response): number | null { + const seconds = Number(response.headers.get('retry-after')) + return Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : null +} + +async function readJson(response: Response): Promise { + try { + return await response.json() + } catch { + await cancelUnreadResponseBody(response) + return undefined + } +} + +export async function requestConnector( + request: ConnectorRequest, + schema: TSchema +): Promise>> { + const timeout = AbortSignal.timeout(request.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS) + let response: Response + try { + response = await request.fetchImpl(request.url, { + method: request.method ?? 'POST', + headers: { + 'content-type': 'application/json', + ...(request.token ? { authorization: `Bearer ${request.token}` } : {}) + }, + // A redirect would replay the token to whatever origin it named. + redirect: 'error', + signal: request.signal ? AbortSignal.any([timeout, request.signal]) : timeout, + body: JSON.stringify(request.body) + }) + } catch { + return failure(null) + } + if (!response.ok) { + const error = ConnectorErrorResponse.safeParse(await readJson(response)) + return failure( + response.status, + error.success ? error.data.error : null, + readRetryAfterMs(response) + ) + } + // Only the contract's success counts: 204 with no body for a no-content reader, 200 with its JSON. + if (response.status !== 200 && response.status !== 204) { + await cancelUnreadResponseBody(response) + return failure(null) + } + const payload = response.status === 204 ? undefined : await readJson(response) + if (response.status === 200 && payload === undefined) { + return failure(null) + } + const parsed = schema.safeParse(payload) + return parsed.success ? { ok: true, value: parsed.data } : failure(null) +} diff --git a/src/main/connector-cloud/connector-cloud.test.ts b/src/main/connector-cloud/connector-cloud.test.ts new file mode 100644 index 00000000000..3081faf5880 --- /dev/null +++ b/src/main/connector-cloud/connector-cloud.test.ts @@ -0,0 +1,168 @@ +import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { CONNECTOR_PATHS } from '../../shared/connector-cloud-contract' +import { ConnectorCloud } from './connector-cloud' +import { FakeConnectorCloud } from './fake-connector-cloud.test-fixture' +import { fakeMachine, fakeSeam, type FakeSeam } from './fake-connector-seam.test-fixture' + +let userDataPath: string +let outbox: string +let fake: FakeConnectorCloud +let enabled: boolean +let account: string | null +let quitting: boolean +let seam: FakeSeam +let machine: ReturnType + +function connectorCloud(): ConnectorCloud { + return new ConnectorCloud({ + userDataPath, + origin: fake.origin, + machine, + account: () => account, + isFenced: () => quitting, + mintFor: (who) => fake.mintFor(who), + fetchImpl: fake.fetch + }) +} + +const turnEnd = { + kind: 'turn-ended', + conversation: { source: 'slack', externalId: 'T1:C9:1792.9:U5' }, + paneKey: 'tab:1', + turnId: 1, + at: 2, + stopReason: null +} + +beforeEach(() => { + vi.useFakeTimers() + userDataPath = mkdtempSync(join(tmpdir(), 'orca-connector-cloud-')) + outbox = join(userDataPath, 'connector-turn-outbox.json') + fake = new FakeConnectorCloud() + enabled = true + account = 'ada' + quitting = false + seam = fakeSeam() + machine = fakeMachine(() => (enabled ? seam : null)) +}) + +afterEach(() => { + vi.useRealTimers() + rmSync(userDataPath, { recursive: true, force: true }) +}) + +describe('ConnectorCloud', () => { + it('holds the poll only while signed in and on, and stops on disable and sign-out', async () => { + account = null + const cloud = connectorCloud() + cloud.connect() + await vi.advanceTimersByTimeAsync(0) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(0) + + account = 'ada' + cloud.accountChanged() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(1) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(1) + + enabled = false + seam.disable() + machine.toggle() + enabled = true + seam = fakeSeam() + machine.toggle() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(2) + + cloud.signOutStarted() + fake.answerPoll({ status: 200, body: { v: 1, serverNow: 1, holdMs: 0, deliveries: [] } }) + await vi.advanceTimersByTimeAsync(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(2) + }) + + it('stays stopped while a sign-out runs, whatever refreshes or toggles happen meanwhile', async () => { + const cloud = connectorCloud() + cloud.connect() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(1) + + // The account being signed out is still readable until the sign-out completes. + cloud.signOutStarted() + machine.toggle() + cloud.accountChanged() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(1) + + account = null + cloud.signOutFinished() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(1) + + account = 'ada' + cloud.accountChanged() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(2) + }) + + it('resumes when the sign-out ends with a sign-in that finished meanwhile', async () => { + account = null + const cloud = connectorCloud() + cloud.connect() + cloud.signOutStarted() + account = 'ada' + cloud.accountChanged() + cloud.signOutFinished() + await vi.advanceTimersByTimeAsync(10) + expect(fake.requestsTo(CONNECTOR_PATHS.hello)).toHaveLength(1) + }) + + it('keeps pending turn ends for the same account, and drops them on a switch or sign-out', async () => { + writeFileSync(outbox, JSON.stringify({ account: 'ada', turnEnds: [turnEnd] })) + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + const cloud = connectorCloud() + cloud.connect() + await vi.advanceTimersByTimeAsync(10) + cloud.accountChanged() + expect(existsSync(outbox)).toBe(true) + + fake.signedIn.add('grace') + account = 'grace' + cloud.accountChanged() + expect(existsSync(outbox)).toBe(false) + await vi.advanceTimersByTimeAsync(10) + expect(fake.minted.at(-1)?.account).toBe('grace') + + writeFileSync(outbox, JSON.stringify({ account: 'grace', turnEnds: [turnEnd] })) + const restarted = connectorCloud() + restarted.signOutStarted() + expect(existsSync(outbox)).toBe(false) + }) + + it('never sends turn ends another account left before a relaunch', async () => { + writeFileSync(outbox, JSON.stringify({ account: 'grace', turnEnds: [turnEnd] })) + connectorCloud().connect() + await vi.advanceTimersByTimeAsync(10) + expect(fake.events).toEqual([{ kind: 'snapshot', conversations: [] }]) + }) + + it('stops when Orca quits or relaunches into another profile, keeping pending turn ends', async () => { + writeFileSync(outbox, JSON.stringify({ account: 'ada', turnEnds: [turnEnd] })) + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + const cloud = connectorCloud() + cloud.connect() + await vi.advanceTimersByTimeAsync(10) + quitting = true + account = 'grace' + cloud.fence() + machine.toggle() + cloud.accountChanged() + fake.answerPoll({ status: 200, body: { v: 1, serverNow: 1, holdMs: 0, deliveries: [] } }) + await vi.advanceTimersByTimeAsync(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(1) + expect(fake.minted.every((minted) => minted.account === 'ada')).toBe(true) + expect(existsSync(outbox)).toBe(true) + }) +}) diff --git a/src/main/connector-cloud/connector-cloud.ts b/src/main/connector-cloud/connector-cloud.ts new file mode 100644 index 00000000000..35168a7dae4 --- /dev/null +++ b/src/main/connector-cloud/connector-cloud.ts @@ -0,0 +1,103 @@ +// This desktop's link to the Orca Cloud connector, as whichever Orca account it is signed in to. +import { ConnectorPollClient } from './connector-poll-client' +import type { ConnectorMachine } from './connector-machine' +import { ConnectorAccountTransport, type ConnectorTokenMint } from './connector-transport' +import { ConnectorTurnEvents } from './connector-turn-events' + +export type ConnectorCloudOptions = { + userDataPath: string + origin: string + machine: ConnectorMachine + /** The signed-in Orca account, or null while signed out. */ + account: () => string | null + /** Quitting or relaunching into another profile: nothing new starts, as for the relay. */ + isFenced: () => boolean + /** Mints connector tokens for one account and no other. */ + mintFor: (account: string) => ConnectorTokenMint + fetchImpl?: typeof globalThis.fetch +} + +export class ConnectorCloud { + private connected = false + private client: ConnectorPollClient | null = null + // The signed-in account's outbox: one owner per process, replaced only when another account signs in. + private events: ConnectorTurnEvents | null = null + // While a sign-out runs: the account it removes is still readable, and a refresh may even renew it. + private signingOut = false + + constructor(private readonly options: ConnectorCloudOptions) {} + + /** From now on, hold the poll whenever connectors are on and an Orca account is signed in. */ + connect(): void { + if (this.connected) { + return + } + this.connected = true + this.options.machine.onConnectorsToggled(() => this.reconcile()) + this.reconcile() + } + + /** The desktop signed in, switched accounts or refreshed its sign-in. */ + accountChanged(): void { + this.reconcile() + } + + /** Just before the desktop signs out: stop now, and drop turn ends no other account can report. */ + signOutStarted(): void { + this.signingOut = true + this.stopClient() + if (this.events) { + this.events.clear() + this.events = null + } else { + ConnectorTurnEvents.clearStored(this.options.userDataPath) + } + } + + /** Quit or relaunch: stop now; pending turn ends stay on disk for the next launch of this account. */ + fence(): void { + this.reconcile() + } + + /** The sign-out ended, either way; a sign-in that finished meanwhile resumes here. */ + signOutFinished(): void { + this.signingOut = false + this.reconcile() + } + + private reconcile(): void { + if (!this.connected) { + return + } + const account = this.signingOut || this.options.isFenced() ? null : this.options.account() + const seam = this.options.machine.seam() + if (account && this.events?.account !== account) { + this.stopClient() + // Pending turn ends belong to the previous account's threads; no other account may report them. + this.events?.clear() + this.events = new ConnectorTurnEvents(this.options.userDataPath, account) + } + if (this.client && (this.client.seam !== seam || !account)) { + this.stopClient() + } + const { events } = this + if (!this.client && seam && account && events) { + this.client = new ConnectorPollClient({ + transport: new ConnectorAccountTransport( + this.options.origin, + this.options.mintFor(account), + this.options.fetchImpl + ), + events, + seam, + machine: this.options.machine + }) + this.client.start() + } + } + + private stopClient(): void { + this.client?.stop() + this.client = null + } +} diff --git a/src/main/connector-cloud/connector-deliveries.ts b/src/main/connector-cloud/connector-deliveries.ts new file mode 100644 index 00000000000..470edb402aa --- /dev/null +++ b/src/main/connector-cloud/connector-deliveries.ts @@ -0,0 +1,84 @@ +// Each cloud delivery mapped onto the existing connector seam call. +import { + CONNECTOR_CLIENT_LIMITS, + ConnectorDelivery, + ConnectorDeliveryId, + type ConnectorDeliveryResult +} from '../../shared/connector-cloud-contract' +import type { + ExternalConversationHost, + ExternalConversationSeam +} from '../external-conversations/external-conversation-seam' +import type { Delivery } from '../external-conversations/external-conversation-types' +import { placeLaunchTarget } from './connector-places' + +export type DispatchedDelivery = { + deliveryId: string + /** False when this Orca refused it without calling the seam; only done is sent then. */ + received: boolean + done: Promise +} + +/** A thrown seam call (e.g. a conversation that no longer exists) is a refusal, not a lost delivery. */ +function settle(call: () => Promise): Promise { + let running: Promise + try { + running = call() + } catch (error) { + running = Promise.reject(error) + } + return running.catch((error: unknown): ConnectorDeliveryResult => ({ + outcome: 'refused', + reason: error instanceof Error && error.message ? error.message : 'failed' + })) +} + +/** + * Calls the seam synchronously, so its receipt exists before `received` is sent. + * `msLeft` is what remains of the delivery's deadline, measured from when the poll was sent. + */ +export function dispatchDelivery( + seam: ExternalConversationSeam, + host: ExternalConversationHost, + raw: unknown, + msLeft: (expiresAt: number) => number +): DispatchedDelivery | null { + const parsed = ConnectorDelivery.safeParse(raw) + if (!parsed.success) { + const id = ConnectorDeliveryId.safeParse(raw) + return id.success ? refuse(id.data.deliveryId, { outcome: 'unsupported' }) : null + } + const delivery = parsed.data + const { deliveryId, conversation: key } = delivery + if (msLeft(delivery.expiresAt) < CONNECTOR_CLIENT_LIMITS.minDeadlineLeftMs) { + return refuse(deliveryId, { outcome: 'expired' }) + } + let done: Promise + switch (delivery.kind) { + case 'start': { + const target = placeLaunchTarget(delivery.placeId, deliveryId) + if (!target) { + return refuse(deliveryId, { outcome: 'unsupported' }) + } + const { agent, prompt } = delivery + done = settle( + async () => (await seam.start(host, key, deliveryId, { agent, target, prompt })).initial + ) + break + } + case 'send': + done = settle(() => seam.send(host, key, deliveryId, delivery.text)) + break + case 'answer': + done = settle(() => seam.answer(host, key, delivery.promptId, delivery.optionId)) + break + case 'cancel': + done = settle(() => seam.cancel(host, key, deliveryId)) + break + } + return { deliveryId, received: true, done } +} + +function refuse(deliveryId: string, result: ConnectorDeliveryResult): DispatchedDelivery { + return { deliveryId, received: false, done: Promise.resolve(result) } +} diff --git a/src/main/connector-cloud/connector-hello.test.ts b/src/main/connector-cloud/connector-hello.test.ts new file mode 100644 index 00000000000..8c6bd409989 --- /dev/null +++ b/src/main/connector-cloud/connector-hello.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from 'vitest' +import { z } from 'zod' +import { + CONNECTOR_CLIENT_LIMITS, + type ConnectorProject +} from '../../shared/connector-cloud-contract' +import { connectorHello } from './connector-hello' +import { fakeMachine } from './fake-connector-seam.test-fixture' + +const longest = (unit: string): string => + unit.repeat(CONNECTOR_CLIENT_LIMITS.maxDisplayTextLength / unit.length) + +// Every string at its cap, in characters JSON escapes to six bytes each. +function largestProject(index: number): ConnectorProject { + return { + placeId: String(index).padStart(512, 'p'), + name: longest('\u0001'), + remote: longest('\u0001'), + kind: 'git', + where: { kind: 'ssh', label: longest('\u0001') } + } +} + +describe('connectorHello', () => { + it('always fits the request cap, keeping the most recently used projects first', () => { + const projects = Array.from({ length: CONNECTOR_CLIENT_LIMITS.maxHelloProjects }, (_, index) => + largestProject(index) + ) + const { body, trimmed } = connectorHello( + fakeMachine(() => null), + { + agents: ['claude', 'codex'], + projects + } + ) + const json = JSON.stringify(body) + expect(Buffer.byteLength(json)).toBeLessThanOrEqual(CONNECTOR_CLIENT_LIMITS.maxRequestBytes) + expect(trimmed).toBe(true) + const sent = z.object({ projects: z.array(z.unknown()) }).parse(JSON.parse(json)).projects + expect(sent.length).toBeGreaterThan(0) + expect(sent).toEqual(projects.slice(0, sent.length)) + }) + + it('sends every project when they fit', () => { + const project: ConnectorProject = { + placeId: 'pl_1', + name: longest('a'), + kind: 'folder', + where: { kind: 'local' } + } + const projects = Array.from({ length: CONNECTOR_CLIENT_LIMITS.maxHelloProjects }, () => project) + const { body, trimmed } = connectorHello( + fakeMachine(() => null), + { agents: [], projects } + ) + expect(trimmed).toBe(false) + expect(body).toMatchObject({ projects }) + }) +}) diff --git a/src/main/connector-cloud/connector-hello.ts b/src/main/connector-cloud/connector-hello.ts new file mode 100644 index 00000000000..57098892c11 --- /dev/null +++ b/src/main/connector-cloud/connector-hello.ts @@ -0,0 +1,49 @@ +// What a hello says about this machine, always within the cloud's request cap. +import { + CONNECTOR_CLIENT_LIMITS, + CONNECTOR_CONTRACT_VERSION, + type ConnectorProject +} from '../../shared/connector-cloud-contract' +import type { ConnectorMachine } from './connector-machine' +import { jsonBytes } from './connector-snapshot' + +// `reply` is advertised once `orca slack reply` ships. +const FEATURES = ['answer', 'cancel'] +const PROMPT_KINDS = ['approval', 'question'] +// Only Slack conversations reach this machine's seam so far. +const SOURCES = ['slack'] + +export type ConnectorHelloSaid = { agents: string[]; projects: ConnectorProject[] } + +/** The hello body; projects come most recently used first, so the least recent go when it would not fit. */ +export function connectorHello( + machine: ConnectorMachine, + said: ConnectorHelloSaid +): { body: unknown; trimmed: boolean } { + const envelope = { + v: 1, + contract: CONNECTOR_CONTRACT_VERSION, + orcaVersion: machine.orcaVersion, + platform: process.platform, + runtime: machine.runtime, + hostName: machine.hostName, + agents: said.agents, + promptKinds: PROMPT_KINDS, + features: FEATURES, + sources: SOURCES + } + let bytes = jsonBytes({ ...envelope, projects: [] }) + let count = 0 + for (const project of said.projects) { + // Each project adds its JSON and a comma. + bytes += jsonBytes(project) + 1 + if (bytes > CONNECTOR_CLIENT_LIMITS.maxRequestBytes) { + break + } + count++ + } + return { + body: { ...envelope, projects: said.projects.slice(0, count) }, + trimmed: count < said.projects.length + } +} diff --git a/src/main/connector-cloud/connector-machine.ts b/src/main/connector-cloud/connector-machine.ts new file mode 100644 index 00000000000..922e6ea1832 --- /dev/null +++ b/src/main/connector-cloud/connector-machine.ts @@ -0,0 +1,21 @@ +import type { ConnectorProject } from '../../shared/connector-cloud-contract' +import type { + ExternalConversationHost, + ExternalConversationSeam +} from '../external-conversations/external-conversation-seam' +import type { ConnectorAccountTransport } from './connector-transport' + +/** What the connector needs from this Orca; the seam exists exactly while connectors are on. */ +export type ConnectorMachine = { + runtime: 'desktop' + orcaVersion: string + hostName: string + seam: () => ExternalConversationSeam | null + conversationHost: () => ExternalConversationHost + agents: () => string[] + projects: () => ConnectorProject[] + onConnectorsToggled: (listener: () => void) => void +} + +/** How the client reaches the cloud as this machine; the credential behind it can change. */ +export type ConnectorTransport = Pick diff --git a/src/main/connector-cloud/connector-origin.ts b/src/main/connector-cloud/connector-origin.ts new file mode 100644 index 00000000000..84684643eca --- /dev/null +++ b/src/main/connector-cloud/connector-origin.ts @@ -0,0 +1,7 @@ +import { cleanCloudServiceOrigin } from '../../shared/cloud-service-url' + +export function resolveConnectorOrigin(env: NodeJS.ProcessEnv, packaged: boolean): string { + return ( + cleanCloudServiceOrigin(env.ORCA_CONNECTOR_URL, !packaged) ?? 'https://connector.onorca.dev' + ) +} diff --git a/src/main/connector-cloud/connector-outbox.test.ts b/src/main/connector-cloud/connector-outbox.test.ts new file mode 100644 index 00000000000..e1e42441710 --- /dev/null +++ b/src/main/connector-cloud/connector-outbox.test.ts @@ -0,0 +1,155 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { ConnectorFailure } from './connector-cloud-request' +import { ConnectorOutbox, ConnectorOutboxSender } from './connector-outbox' + +const TOO_MANY: ConnectorFailure = { ok: false, status: 429, error: null, retryAfterMs: 300_000 } + +function recordingSender( + outbox: ConnectorOutbox, + helloRev: () => number | null, + answer: () => Promise +) { + const sent: { batch: string[]; helloRev: number; seq: number }[] = [] + const sender = new ConnectorOutboxSender( + outbox, + (batch, rev, seq) => { + sent.push({ batch, helloRev: rev, seq }) + return answer() + }, + helloRev, + () => 0.5 + ) + return { sender, sent } +} + +beforeEach(() => { + vi.useFakeTimers() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('ConnectorOutbox', () => { + it('sends nothing without an accepted hello', async () => { + const outbox = new ConnectorOutbox({ headSize: () => 10 }) + const { sender, sent } = recordingSender( + outbox, + () => null, + async () => null + ) + outbox.push('a') + sender.flush() + await vi.advanceTimersByTimeAsync(0) + expect(sent).toEqual([]) + }) + + it('never lets a later item overtake a refused one: the retry resends the same batch and number', async () => { + const outbox = new ConnectorOutbox({ headSize: () => 2 }) + const answers: (ConnectorFailure | null)[] = [TOO_MANY, null, null] + const { sender, sent } = recordingSender( + outbox, + () => 1, + async () => answers.shift() ?? null + ) + outbox.push('old prompt') + sender.flush() + await vi.advanceTimersByTimeAsync(0) + outbox.push('linked', 'new prompt') + sender.flush() + await vi.advanceTimersByTimeAsync(300_000) + expect(sent.map(({ batch, seq }) => [batch, seq])).toEqual([ + [['old prompt'], 0], + [['old prompt'], 0], + [['linked', 'new prompt'], 1] + ]) + }) + + it('keeps an item whose 204 lands after a newer hello went out, and sends it under that hello', async () => { + const outbox = new ConnectorOutbox({ headSize: () => 10 }) + let rev: number | null = 1 + const held = Promise.withResolvers() + const answers = [held.promise] + const { sender, sent } = recordingSender( + outbox, + () => rev, + () => answers.shift() ?? Promise.resolve(null) + ) + outbox.push('done') + sender.flush() + rev = null + held.resolve(null) + await vi.advanceTimersByTimeAsync(0) + rev = 2 + sender.resume() + await vi.advanceTimersByTimeAsync(0) + expect(sent.map(({ batch, helloRev, seq }) => [batch, helloRev, seq])).toEqual([ + [['done'], 1, 0], + [['done'], 2, 0] + ]) + }) + + it('a new hello sends at once, past the last one’s backoff', async () => { + const outbox = new ConnectorOutbox({ headSize: () => 10 }) + let rev = 1 + const answers: (ConnectorFailure | null)[] = [TOO_MANY] + const { sender, sent } = recordingSender( + outbox, + () => rev, + async () => answers.shift() ?? null + ) + outbox.push('a') + sender.flush() + await vi.advanceTimersByTimeAsync(0) + rev = 2 + sender.resume() + await vi.advanceTimersByTimeAsync(0) + expect(sent.map(({ helloRev }) => helloRev)).toEqual([1, 2]) + }) + + it('restarts numbering with its first items in front, keeping only what is still wanted', () => { + const outbox = new ConnectorOutbox({ headSize: () => 10 }) + outbox.push('turn end', 'prompt') + expect(outbox.next(1)?.stamp.seq).toBe(0) + outbox.restart(['snapshot'], (item) => item === 'turn end') + const head = outbox.next(2) + expect(head?.batch).toEqual(['snapshot', 'turn end']) + expect(head?.stamp.seq).toBe(0) + }) + + it('ignores a superseded stamp', () => { + const outbox = new ConnectorOutbox({ headSize: () => 10 }) + outbox.push('a') + const old = outbox.next(1) + outbox.restart([], () => true) + if (old) { + outbox.accepted(old.stamp) + } + expect(outbox.next(2)?.batch).toEqual(['a']) + }) + + it('splits a head the cloud refuses as too large, same number, and drops a lone item once', async () => { + const outbox = new ConnectorOutbox({ headSize: () => 4 }) + const TOO_LARGE: ConnectorFailure = { ok: false, status: 413, error: null, retryAfterMs: null } + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const { sender, sent } = recordingSender( + outbox, + () => 1, + async () => (sent.at(-1)?.batch.includes('huge') ? TOO_LARGE : null) + ) + outbox.push('a', 'huge', 'b', 'c') + sender.flush() + await vi.advanceTimersByTimeAsync(0) + expect(sent.map(({ batch, seq }) => [batch, seq])).toEqual([ + [['a', 'huge', 'b', 'c'], 0], + [['a', 'huge'], 0], + [['a'], 0], + [['huge', 'b', 'c'], 1], + [['huge', 'b'], 1], + [['huge'], 1], + [['b', 'c'], 2] + ]) + expect(warn).toHaveBeenCalledTimes(1) + warn.mockRestore() + }) +}) diff --git a/src/main/connector-cloud/connector-outbox.ts b/src/main/connector-cloud/connector-outbox.ts new file mode 100644 index 00000000000..c9c24dce10b --- /dev/null +++ b/src/main/connector-cloud/connector-outbox.ts @@ -0,0 +1,213 @@ +import type { ConnectorFailure } from './connector-cloud-request' + +const RETRY_BASE_MS = 1_000 +const RETRY_MAX_MS = 30_000 + +/** Random doubling from 1 s to 30 s; a cloud's Retry-After wins when longer. */ +export function connectorRetryDelayMs( + failures: number, + failure: ConnectorFailure, + random: () => number +): number { + const ceiling = Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** failures) + return Math.max(failure.retryAfterMs ?? 0, ceiling * (0.5 + random() / 2)) +} + +/** The head as last numbered: `size` items sent as `seq` under `helloRev`; 0 once taken. */ +export type ConnectorStamp = { + readonly helloRev: number + readonly seq: number + readonly size: number +} + +export type ConnectorOutboxHead = { stamp: ConnectorStamp; batch: T[] } + +/** + * Items owed to the cloud, in order. One leaves only through `accepted` with the stamp it went out under, so + * no send, failure or hello change can lose it. + */ +export class ConnectorOutbox { + private items: T[] + private stamp: ConnectorStamp | null = null + + constructor( + private readonly options: { + /** How many of the first items one request carries; at least one. */ + headSize: (items: readonly T[]) => number + initial?: T[] + /** Trims what is owed, never the first `sent` items, so everything owed stays within bounds. */ + bound?: (items: T[], sent: number) => T[] + /** Called after every change with everything still owed, e.g. to persist it. */ + onChange?: (items: readonly T[]) => void + onAccepted?: (batch: readonly T[]) => void + } + ) { + this.items = options.initial ?? [] + } + + push(...items: T[]): void { + this.replace([...this.items, ...items]) + } + + /** Numbering starts again from 0 with `first` in front, keeping what `keep` accepts. */ + restart(first: T[], keep: (item: T) => boolean): void { + this.stamp = null + this.replace([...first, ...this.items.filter(keep)]) + } + + /** + * What to send under `helloRev`. The cloud skips a number it has seen, so an untaken head goes again with + * the same number and items; only a taken one, or a new hello, numbers the next. + */ + next(helloRev: number): ConnectorOutboxHead | null { + let stamp = this.stamp + if (stamp?.helloRev !== helloRev || stamp.size === 0) { + if (this.items.length === 0) { + return null + } + stamp = this.stamp = { + helloRev, + seq: stamp?.helloRev === helloRev ? stamp.seq + 1 : 0, + size: Math.max(1, Math.min(this.options.headSize(this.items), this.items.length)) + } + } + return { stamp, batch: this.items.slice(0, stamp.size) } + } + + /** The cloud took `stamp`'s head under the hello still current; a superseded stamp changes nothing. */ + accepted(stamp: ConnectorStamp): void { + if (this.stamp !== stamp) { + return + } + const batch = this.items.splice(0, stamp.size) + this.stamp = { ...stamp, size: 0 } + this.notify() + this.options.onAccepted?.(batch) + } + + /** + * The cloud refused `stamp`'s head as too large or malformed, which no retry fixes: it goes again as its + * first half, same number. A single item is dropped instead; returns it so the caller can say so. + */ + refused(stamp: ConnectorStamp): T | null { + if (this.stamp !== stamp) { + return null + } + if (stamp.size > 1) { + this.stamp = { ...stamp, size: Math.ceil(stamp.size / 2) } + return null + } + const [dropped] = this.items.splice(0, 1) + this.stamp = { ...stamp, size: 0 } + this.notify() + return dropped ?? null + } + + private replace(items: T[]): void { + this.items = this.options.bound ? this.options.bound(items, this.stamp?.size ?? 0) : items + this.notify() + } + + private notify(): void { + this.options.onChange?.(this.items) + } +} + +// Too large, or not the contract: the same request can never succeed. +const isRefusedAsSent = (failure: ConnectorFailure): boolean => + failure.status === 413 || failure.status === 400 + +/** Sends a head; null when the cloud answered 204. */ +export type ConnectorBatchSender = ( + batch: T[], + helloRev: number, + seq: number +) => Promise + +/** + * Drains an outbox one request at a time under the accepted hello, and does nothing without one. An answer + * counts only if that hello is still accepted when it lands, so a retired or superseded one consumes nothing. + */ +export class ConnectorOutboxSender { + private sending = false + private warned = false + private failures = 0 + private retry: ReturnType | null = null + + constructor( + private readonly outbox: Pick, 'next' | 'accepted' | 'refused'>, + private readonly send: ConnectorBatchSender, + private readonly helloRev: () => number | null, + private readonly random: () => number + ) {} + + /** Sends what is owed, unless a backoff is pending. */ + flush(): void { + if (!this.retry) { + void this.drain() + } + } + + /** A new hello: send now, whatever the last one's backoff. */ + resume(): void { + this.failures = 0 + this.stop() + void this.drain() + } + + stop(): void { + if (this.retry) { + clearTimeout(this.retry) + this.retry = null + } + } + + private async drain(): Promise { + if (this.sending) { + return + } + this.sending = true + try { + for (let helloRev = this.helloRev(); helloRev !== null; helloRev = this.helloRev()) { + const head = this.outbox.next(helloRev) + if (!head) { + return + } + const failure = await this.send(head.batch, helloRev, head.stamp.seq) + if (this.helloRev() !== helloRev) { + continue + } + if (failure && isRefusedAsSent(failure)) { + this.dropped(this.outbox.refused(head.stamp)) + continue + } + if (failure) { + this.scheduleRetry(failure) + return + } + this.failures = 0 + this.outbox.accepted(head.stamp) + } + } finally { + this.sending = false + } + } + + private dropped(item: T | null): void { + if (item !== null && !this.warned) { + this.warned = true + console.warn('[connector] Orca Cloud refused an item that cannot be sent; dropped it') + } + } + + private scheduleRetry(failure: ConnectorFailure): void { + this.retry = setTimeout( + () => { + this.retry = null + void this.drain() + }, + connectorRetryDelayMs(this.failures++, failure, this.random) + ) + this.retry.unref?.() + } +} diff --git a/src/main/connector-cloud/connector-places.test.ts b/src/main/connector-cloud/connector-places.test.ts new file mode 100644 index 00000000000..6dc55fc126d --- /dev/null +++ b/src/main/connector-cloud/connector-places.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from 'vitest' +import { CONNECTOR_CLIENT_LIMITS } from '../../shared/connector-cloud-contract' +import type { ProjectGroup } from '../../shared/project-group-types' +import type { Repo } from '../../shared/repo-types' +import { connectorProjects, placeLaunchTarget, type ConnectorPlacesStore } from './connector-places' + +function repo(fields: Partial & Pick): Repo { + return { + path: `/src/${fields.id}`, + displayName: fields.id, + badgeColor: '#000', + addedAt: 1, + ...fields + } +} + +function store(repos: Repo[], groups: Partial[] = []): ConnectorPlacesStore { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: connectorProjects reads only these fields of each group. + const projectGroups = groups as ProjectGroup[] + return { + getRepos: () => repos, + getProjectGroups: () => projectGroups, + getFolderWorkspaces: () => [], + getSshTargets: () => [ + { id: 'box', label: 'openclaw', host: 'openclaw', port: 22, username: 'me' } + ], + getProjectHostSetups: () => [], + getAllWorktreeMeta: () => ({}) + } +} + +describe('connector places', () => { + it('names where each repo runs without paths, and keeps twins on other hosts apart', () => { + const projects = connectorProjects( + store([ + repo({ + id: 'orca', + displayName: 'orca', + gitRemoteIdentity: { + canonicalKey: 'github.com/stablyai/orca', + remoteName: 'origin', + remoteUrl: 'git@github.com:stablyai/orca.git' + } + }), + repo({ id: 'orca', connectionId: 'box', executionHostId: 'ssh:box' }), + repo({ id: 'linux', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\linux' }), + repo({ id: 'served', executionHostId: 'runtime:server-1' }), + repo({ id: 'notes', kind: 'folder' }) + ]) + ) + expect(projects.map(({ name, remote, where }) => ({ name, remote, where }))).toEqual([ + { name: 'orca', remote: 'stablyai/orca', where: { kind: 'local' } }, + { name: 'orca', remote: undefined, where: { kind: 'ssh', label: 'openclaw' } }, + { name: 'linux', remote: undefined, where: { kind: 'wsl', label: 'Ubuntu' } } + ]) + expect(JSON.stringify(projects)).not.toContain('/src/') + const [local, remote] = projects + expect(placeLaunchTarget(local!.placeId, 'T1:C9:1')).toEqual({ + kind: 'create-worktree', + create: { + repo: 'id:orca', + repoHostId: 'local', + name: expect.stringMatching(/^slack-[0-9a-f]{8}$/), + nameWasGenerated: true + } + }) + expect(placeLaunchTarget(remote!.placeId, 'T1:C9:1')).toMatchObject({ + create: { repo: 'id:orca', repoHostId: 'ssh:box' } + }) + }) + + it('starts folder projects as fresh folder workspaces', () => { + const [folder] = connectorProjects( + store( + [], + [ + { id: 'g1', name: 'Docs', parentPath: '/docs' }, + { id: 'g2', name: 'Group', parentPath: null } + ] + ) + ) + expect(folder).toMatchObject({ name: 'Docs', kind: 'folder', where: { kind: 'local' } }) + expect(placeLaunchTarget(folder!.placeId, 'd1')).toEqual({ + kind: 'create-folder-workspace', + create: { projectGroupId: 'g1', name: expect.stringMatching(/^slack-[0-9a-f]{8}$/) } + }) + }) + + it('derives the same worktree name for a resent start, and refuses a place it cannot read', () => { + const [project] = connectorProjects(store([repo({ id: 'orca' })])) + expect(placeLaunchTarget(project!.placeId, 'd1')).toEqual( + placeLaunchTarget(project!.placeId, 'd1') + ) + expect(placeLaunchTarget(project!.placeId, 'd1')).not.toEqual( + placeLaunchTarget(project!.placeId, 'd2') + ) + expect(placeLaunchTarget('not-a-place', 'd1')).toBeNull() + }) + + it('lists the most recently used projects first, at most maxHelloProjects, with short names', () => { + const repos = Array.from({ length: CONNECTOR_CLIENT_LIMITS.maxHelloProjects + 1 }, (_, index) => + repo({ id: `r${index}`, addedAt: index, displayName: 'n'.repeat(300) }) + ) + const projects = connectorProjects({ + ...store(repos), + getAllWorktreeMeta: () => ({ + 'r0::/src/r0': { + displayName: 'r0', + comment: '', + linkedIssue: null, + linkedPR: null, + linkedLinearIssue: null, + isArchived: false, + isUnread: false, + isPinned: false, + sortOrder: 0, + lastActivityAt: 10_000 + } + }) + }) + expect(projects).toHaveLength(CONNECTOR_CLIENT_LIMITS.maxHelloProjects) + expect(projects[0]?.name).toHaveLength(CONNECTOR_CLIENT_LIMITS.maxDisplayTextLength) + expect(projects[0]?.placeId).toBe(connectorProjects(store([repos[0]!]))[0]?.placeId) + }) +}) diff --git a/src/main/connector-cloud/connector-places.ts b/src/main/connector-cloud/connector-places.ts new file mode 100644 index 00000000000..d378c5a2e4d --- /dev/null +++ b/src/main/connector-cloud/connector-places.ts @@ -0,0 +1,184 @@ +// A place is where a Slack start may run; its id rebuilds the launch target with no lookup, so twins on other hosts stay apart. +import { createHash } from 'node:crypto' +import { z } from 'zod' +import { + CONNECTOR_CLIENT_LIMITS, + type ConnectorProject +} from '../../shared/connector-cloud-contract' +import { + getRepoExecutionHostId, + parseExecutionHostId, + type ExecutionHostId +} from '../../shared/execution-host' +import { splitGitRemoteKey } from '../../shared/git-remote-identity' +import { getRepoIdFromWorktreeId } from '../../shared/worktree/id' +import { isFolderRepo } from '../../shared/repo-kind' +import type { AgentLaunchParams } from '../../shared/rpc-contract/agent-launch-params' +import { parseWslUncPath } from '../../shared/wsl-paths' +import { readAllWorktreeMetaForHost } from '../persistence/host-qualified-worktree-meta' +import type { RuntimeStore } from '../runtime/runtime-store-contract' +import { + runtimeVisibleProjectGroups, + runtimeVisibleRepos +} from '../runtime/runtime-visible-catalog' + +const Place = z.union([ + z.tuple([z.literal('git'), z.string().min(1), z.string().min(1)]), + z.tuple([z.literal('folder'), z.string().min(1)]) +]) +type Place = z.infer + +export type ConnectorPlacesStore = Pick< + RuntimeStore, + | 'getRepos' + | 'getProjectGroups' + | 'getFolderWorkspaces' + | 'getSshTargets' + | 'getProjectHostSetups' + | 'getAllWorktreeMeta' + | 'getAllWorktreeMetaForHost' +> + +function encodePlace(place: Place): string { + return Buffer.from(JSON.stringify(place)).toString('base64url') +} + +function decodePlace(placeId: string): Place | null { + try { + const parsed = Place.safeParse(JSON.parse(Buffer.from(placeId, 'base64url').toString('utf8'))) + return parsed.success ? parsed.data : null + } catch { + return null + } +} + +/** A pure function of the delivery, so a resent start hashes to the same receipt. */ +export function placeLaunchTarget( + placeId: string, + deliveryId: string +): AgentLaunchParams['target'] | null { + const place = decodePlace(placeId) + if (!place) { + return null + } + const name = `slack-${createHash('sha256').update(deliveryId).digest('hex').slice(0, 8)}` + if (place[0] === 'folder') { + return { kind: 'create-folder-workspace', create: { projectGroupId: place[1], name } } + } + const [, repoHostId, repoId] = place + const hostId = parseExecutionHostId(repoHostId)?.id + return hostId + ? { + kind: 'create-worktree', + create: { repo: `id:${repoId}`, repoHostId: hostId, name, nameWasGenerated: true } + } + : null +} + +/** Never paths: where a place runs, by host kind and label only. Null for a paired server's place, which it lists itself. */ +function placeWhere( + hostId: ExecutionHostId, + path: string | null, + store: ConnectorPlacesStore +): ConnectorProject['where'] | null { + const host = parseExecutionHostId(hostId) + if (host?.kind === 'ssh') { + const label = store.getSshTargets?.().find((target) => target.id === host.targetId)?.label + return { kind: 'ssh', ...(label ? { label: displayText(label) } : {}) } + } + if (host?.kind !== 'local') { + return null + } + const wsl = path ? parseWslUncPath(path) : null + return wsl ? { kind: 'wsl', label: displayText(wsl.distro) } : { kind: 'local' } +} + +/** The latest time per key, in one pass. */ +function latestBy( + items: Iterable, + key: (item: T) => string, + at: (item: T) => number +): Map { + const latest = new Map() + for (const item of items) { + latest.set(key(item), Math.max(latest.get(key(item)) ?? 0, at(item))) + } + return latest +} + +/** At most the contract's display length, never splitting a surrogate pair. */ +function displayText(text: string): string { + const max = CONNECTOR_CLIENT_LIMITS.maxDisplayTextLength + if (text.length <= max) { + return text + } + const cut = text.slice(0, max) + return /[\uD800-\uDBFF]$/.test(cut) ? cut.slice(0, -1) : cut +} + +/** + * Git repos (each a fresh worktree) and folder projects (each a fresh folder workspace), most recently used + * first by their workspaces' activity, at most maxHelloProjects. + */ +export function connectorProjects(store: ConnectorPlacesStore): ConnectorProject[] { + const projects: { project: ConnectorProject; usedAt: number }[] = [] + // Each host's worktree activity, read once and keyed by repo id; same-id repos on other hosts stay apart. + const activityByHost = new Map>() + const repoActivity = (hostId: ExecutionHostId): Map => { + let activity = activityByHost.get(hostId) + if (!activity) { + activity = latestBy( + Object.entries(readAllWorktreeMetaForHost(store, hostId)), + ([worktreeId]) => getRepoIdFromWorktreeId(worktreeId), + ([, meta]) => meta.lastActivityAt + ) + activityByHost.set(hostId, activity) + } + return activity + } + for (const repo of runtimeVisibleRepos(store)) { + const hostId = getRepoExecutionHostId(repo) + const where = isFolderRepo(repo) ? null : placeWhere(hostId, repo.path, store) + if (!where) { + continue + } + const remote = splitGitRemoteKey(repo.gitRemoteIdentity?.canonicalKey, (host) => host)?.tail + const usedAt = Math.max(repo.addedAt, repoActivity(hostId).get(repo.id) ?? 0) + projects.push({ + project: { + placeId: encodePlace(['git', hostId, repo.id]), + name: displayText(repo.displayName), + ...(remote ? { remote: displayText(remote) } : {}), + kind: 'git', + where + }, + usedAt + }) + } + const folderActivity = latestBy( + store.getFolderWorkspaces?.() ?? [], + (folder) => folder.projectGroupId, + (folder) => folder.lastActivityAt + ) + for (const group of runtimeVisibleProjectGroups(store)) { + const where = group.parentPath + ? placeWhere(getRepoExecutionHostId(group), group.parentPath, store) + : null + if (where) { + const usedAt = Math.max(group.updatedAt, folderActivity.get(group.id) ?? 0) + projects.push({ + project: { + placeId: encodePlace(['folder', group.id]), + name: displayText(group.name), + kind: 'folder', + where + }, + usedAt + }) + } + } + return projects + .sort((a, b) => b.usedAt - a.usedAt) + .slice(0, CONNECTOR_CLIENT_LIMITS.maxHelloProjects) + .map(({ project }) => project) +} diff --git a/src/main/connector-cloud/connector-poll-client.test.ts b/src/main/connector-cloud/connector-poll-client.test.ts new file mode 100644 index 00000000000..41316458f47 --- /dev/null +++ b/src/main/connector-cloud/connector-poll-client.test.ts @@ -0,0 +1,677 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { CONNECTOR_PATHS } from '../../shared/connector-cloud-contract' +import type { Delivery } from '../external-conversations/external-conversation-types' +import { connectorProjects } from './connector-places' +import { ConnectorPollClient } from './connector-poll-client' +import { ConnectorAccountTransport } from './connector-transport' +import { ConnectorTurnEvents } from './connector-turn-events' +import { FakeConnectorCloud } from './fake-connector-cloud.test-fixture' +import { fakeMachine, fakeSeam } from './fake-connector-seam.test-fixture' + +const SERVER_NOW = 1_800_000_000_000 +const conversation = { source: 'slack', externalId: 'T1:C9:1792.9:U5' } as const +const [project] = connectorProjects({ + getRepos: () => [ + { id: 'orca', path: '/src/orca', displayName: 'orca', badgeColor: '#000', addedAt: 1 } + ], + getProjectGroups: () => [], + getFolderWorkspaces: () => [], + getSshTargets: () => [], + getProjectHostSetups: () => [], + getAllWorktreeMeta: () => ({}) +}) + +let userDataPath: string +let fake: FakeConnectorCloud + +/** `events` stands for one process's outbox; a new one models a restart. */ +function startClient(seam = fakeSeam(), events = new ConnectorTurnEvents(userDataPath, 'ada')) { + const machine = fakeMachine(() => seam, [project!]) + const client = new ConnectorPollClient({ + transport: new ConnectorAccountTransport(fake.origin, fake.mintFor('ada'), fake.fetch), + events, + seam: seam.seam, + machine, + random: () => 0.5 + }) + client.start() + return { client, machine, ...seam } +} + +// Each step lets a chain of fetches and seam answers run to its next timer. +const settle = async (ms = 0): Promise => { + for (let step = 0; step < 50; step++) { + await vi.advanceTimersByTimeAsync(ms / 50) + } +} + +function poll(deliveries: unknown[], extra: Record = {}) { + fake.answerPoll({ + status: 200, + body: { v: 1, serverNow: SERVER_NOW, holdMs: 50_000, deliveries, ...extra } + }) +} + +const base = { seq: 1, expiresAt: SERVER_NOW + 300_000, conversation } + +beforeEach(() => { + vi.useFakeTimers({ + toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'], + now: SERVER_NOW + }) + userDataPath = mkdtempSync(join(tmpdir(), 'orca-connector-client-')) + fake = new FakeConnectorCloud() +}) + +afterEach(() => { + vi.useRealTimers() + rmSync(userDataPath, { recursive: true, force: true }) +}) + +describe('ConnectorPollClient', () => { + it('says hello with capabilities and projects, then sends the snapshot', async () => { + const { client } = startClient() + await settle() + expect(fake.hellos[0]).toMatchObject({ + v: 1, + contract: 1, + runtime: 'desktop', + hostName: 'box', + agents: ['claude', 'codex'], + promptKinds: ['approval', 'question'], + features: ['answer', 'cancel'], + sources: ['slack'], + projects: [{ placeId: project!.placeId, name: 'orca', kind: 'git' }] + }) + expect(fake.events).toEqual([{ kind: 'snapshot', conversations: [] }]) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(1) + client.stop() + }) + + it('confirms received right after the seam call, then done with its result', async () => { + const seam = fakeSeam() + const sent = Promise.withResolvers() + seam.outcomes.set('d2', sent.promise) + const { client, calls } = startClient(seam) + await settle() + poll([ + { + ...base, + deliveryId: 'd1', + kind: 'start', + agent: 'claude', + placeId: project!.placeId, + prompt: 'fix it' + }, + { ...base, deliveryId: 'd2', kind: 'send', text: 'more' } + ]) + await settle() + expect(calls.start).toHaveBeenCalledWith(expect.anything(), conversation, 'd1', { + agent: 'claude', + prompt: 'fix it', + target: expect.objectContaining({ kind: 'create-worktree' }) + }) + expect(fake.acks).toEqual([ + { deliveryId: 'd1', step: 'received' }, + { deliveryId: 'd2', step: 'received' }, + { deliveryId: 'd1', step: 'done', result: { outcome: 'accepted' } } + ]) + sent.resolve({ outcome: 'refused', reason: 'agent_not_reachable' }) + await settle() + expect(fake.acks.at(-1)).toEqual({ + deliveryId: 'd2', + step: 'done', + result: { outcome: 'refused', reason: 'agent_not_reachable' } + }) + client.stop() + }) + + it('refuses stale and unreadable deliveries without the seam, and still runs the rest', async () => { + const { client, calls } = startClient() + await settle() + poll([ + { ...base, deliveryId: 'old', kind: 'send', text: 'hi', expiresAt: SERVER_NOW + 9_000 }, + { ...base, deliveryId: 'new-kind', kind: 'rewind' }, + { kind: 'no-id' }, + { + ...base, + deliveryId: 'nowhere', + kind: 'start', + agent: 'claude', + placeId: '???', + prompt: 'x' + }, + { ...base, deliveryId: 'gone', kind: 'answer', promptId: 'p1', optionId: 'allow_once' }, + { ...base, deliveryId: 'c1', kind: 'cancel' } + ]) + await settle() + expect(calls.send).not.toHaveBeenCalled() + expect(calls.start).not.toHaveBeenCalled() + expect(fake.acks).toEqual( + expect.arrayContaining([ + { deliveryId: 'old', step: 'done', result: { outcome: 'expired' } }, + { deliveryId: 'new-kind', step: 'done', result: { outcome: 'unsupported' } }, + { deliveryId: 'nowhere', step: 'done', result: { outcome: 'unsupported' } }, + { deliveryId: 'gone', step: 'received' }, + { + deliveryId: 'gone', + step: 'done', + result: { outcome: 'refused', reason: 'external_conversation_not_found' } + }, + { deliveryId: 'c1', step: 'done', result: { outcome: 'accepted' } } + ]) + ) + expect(fake.acks).not.toContainEqual({ deliveryId: 'old', step: 'received' }) + client.stop() + }) + + it('measures the deadline from when the poll was sent, so a slow answer only shortens it', async () => { + const { client, calls } = startClient() + await settle() + // 15 s were left when the poll went out; the answer lands 6 s later, leaving 9 s. + fake.pollReplyDelayMs = 6_000 + poll([ + { ...base, deliveryId: 'slow', kind: 'send', text: 'hi', expiresAt: SERVER_NOW + 15_000 } + ]) + await settle(7_000) + expect(calls.send).not.toHaveBeenCalled() + expect(fake.acks).toContainEqual({ + deliveryId: 'slow', + step: 'done', + result: { outcome: 'expired' } + }) + client.stop() + }) + + it('recovers deliveries an earlier process received from their receipts, never re-running them', async () => { + const seam = fakeSeam() + seam.recorded.set('d1', { outcome: 'accepted' }) + const { client, calls } = startClient(seam) + await settle() + poll([], { + unsettled: [ + { deliveryId: 'd1', conversation }, + { deliveryId: 'd2', conversation } + ] + }) + await settle() + expect(fake.acks).toEqual([ + { deliveryId: 'd1', step: 'done', result: { outcome: 'accepted' } }, + { deliveryId: 'd2', step: 'done', result: { outcome: 'unknown', reason: 'no_local_record' } } + ]) + expect(calls.send).not.toHaveBeenCalled() + expect(calls.start).not.toHaveBeenCalled() + client.stop() + }) + + it('answers a resend after a lost received without calling the seam twice', async () => { + const seam = fakeSeam() + seam.outcomes.set('d1', new Promise(() => {})) + const { client, calls } = startClient(seam) + await settle() + const send = { ...base, deliveryId: 'd1', kind: 'send', text: 'hi' } + poll([send]) + await settle() + poll([send], { unsettled: [{ deliveryId: 'd1', conversation }] }) + await settle() + expect(calls.send).toHaveBeenCalledTimes(1) + expect(seam.calls.recorded).not.toHaveBeenCalled() + expect(fake.acks).toEqual([ + { deliveryId: 'd1', step: 'received' }, + { deliveryId: 'd1', step: 'received' } + ]) + client.stop() + }) + + it('says hello again, with a fresh snapshot, when the cloud lost the hello', async () => { + const { client } = startClient() + await settle() + fake.forgetHello() + poll([]) + await settle() + expect(fake.hellos).toHaveLength(2) + expect(fake.events).toEqual([ + { kind: 'snapshot', conversations: [] }, + { kind: 'snapshot', conversations: [] } + ]) + client.stop() + }) + + it('keeps a turn end on disk until the cloud accepts it, across a restart', async () => { + const outboxFile = join(userDataPath, 'connector-turn-outbox.json') + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + const first = startClient() + await settle() + first.emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 100, + stopReason: 'success', + lastAssistantMessage: `x${'é'.repeat(5_000)}`, + at: 300 + }) + first.emit({ + kind: 'turn-started', + conversation: { source: 'dev', externalId: 'x' }, + worktreeId: 'wt', + paneKey: 'tab:2', + at: 5 + }) + await settle() + const stored = JSON.parse(readFileSync(outboxFile, 'utf8')).turnEnds + expect(stored).toHaveLength(1) + // 8 KB would split a two-byte character after the leading 'x', so the cut steps back one byte. + expect(stored[0].lastAssistantMessage).toBe(`x${'é'.repeat(4_095)}`) + first.client.stop() + + fake.forced.delete(CONNECTOR_PATHS.events) + const second = startClient() + await settle() + expect(fake.events).toContainEqual( + expect.objectContaining({ kind: 'turn-ended', turnId: 100, at: 300 }) + ) + expect(fake.events).not.toContainEqual(expect.objectContaining({ paneKey: 'tab:2' })) + expect(existsSync(outboxFile)).toBe(false) + second.client.stop() + }) + + it('stops when the cloud says so, e.g. for a too-old client', async () => { + startClient() + await settle() + poll([], { control: { stop: 'client_too_old' } }) + await settle() + const polls = fake.requestsTo(CONNECTOR_PATHS.poll).length + await settle(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(polls) + }) + + it('mints a new token when one is refused, and waits out a lapsed sign-in without stopping', async () => { + const { client } = startClient() + await settle() + fake.refuseTokens() + poll([]) + await settle() + expect(fake.minted).toHaveLength(2) + expect(fake.requestsTo(CONNECTOR_PATHS.poll).at(-1)?.token).toBe(fake.minted[1]?.token) + + fake.signedIn.delete('ada') + fake.refuseTokens() + poll([]) + await settle(10_000) + const polls = fake.requestsTo(CONNECTOR_PATHS.poll).length + fake.signedIn.add('ada') + await settle(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll).length).toBeGreaterThan(polls) + client.stop() + }) + + it('backs off through an outage and flushes confirmations once it is back', async () => { + const { client } = startClient() + await settle() + fake.online = false + poll([]) + await settle() + const offlineAttempts = fake.requests.length + await settle(3_000) + expect(fake.requests.length).toBe(offlineAttempts) + fake.online = true + await settle(30_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll).length).toBeGreaterThan(1) + client.stop() + }) + + it('stops when connectors are turned off', async () => { + const { client, disable } = startClient() + await settle() + disable() + await settle() + const polls = fake.requestsTo(CONNECTOR_PATHS.poll).length + poll([]) + await settle(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(polls) + client.stop() + }) + + it('admits nothing once stopped: not a late poll answer, not a late result', async () => { + const seam = fakeSeam() + const late = Promise.withResolvers() + seam.outcomes.set('d1', late.promise) + const { client, calls } = startClient(seam) + await settle() + poll([{ ...base, deliveryId: 'd1', kind: 'send', text: 'hi' }]) + await settle() + fake.pollReplyDelayMs = 1_000 + poll([{ ...base, deliveryId: 'd2', kind: 'send', text: 'late' }]) + await settle(10) + client.stop() + late.resolve({ outcome: 'accepted' }) + await settle(2_000) + expect(calls.send).toHaveBeenCalledTimes(1) + expect(fake.acks).toEqual([{ deliveryId: 'd1', step: 'received' }]) + }) + + it('waits out Retry-After before sending confirmations again, however many are queued', async () => { + const seam = fakeSeam() + const done = Promise.withResolvers() + seam.outcomes.set('d1', done.promise) + const { client } = startClient(seam) + await settle() + fake.forced.set(CONNECTOR_PATHS.acks, { status: 429, headers: { 'retry-after': '300' } }) + poll([{ ...base, deliveryId: 'd1', kind: 'send', text: 'hi' }]) + await settle() + done.resolve({ outcome: 'accepted' }) + await settle(60_000) + expect(fake.requestsTo(CONNECTOR_PATHS.acks)).toHaveLength(1) + fake.forced.delete(CONNECTOR_PATHS.acks) + await settle(250_000) + expect(fake.acks).toEqual([ + { deliveryId: 'd1', step: 'received' }, + { deliveryId: 'd1', step: 'done', result: { outcome: 'accepted' } } + ]) + client.stop() + }) + + it('says hello again when the agents it can run change', async () => { + const { client, machine } = startClient() + await settle() + machine.agentList = ['claude'] + poll([]) + await settle() + expect(fake.hellos.at(-1)).toMatchObject({ agents: ['claude'] }) + expect(fake.hellos).toHaveLength(2) + client.stop() + }) + + it('keeps at most 200 turn ends, in memory and on disk alike', async () => { + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + const { client, emit } = startClient() + await settle() + for (let turn = 1; turn <= 250; turn++) { + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: turn, + stopReason: null, + at: turn + }) + } + const stored = JSON.parse( + readFileSync(join(userDataPath, 'connector-turn-outbox.json'), 'utf8') + ).turnEnds + expect(stored.map((event: { turnId: number }) => event.turnId)).toEqual( + Array.from({ length: 200 }, (_, index) => index + 51) + ) + client.stop() + }) + + it('opens each hello with the snapshot as batch 0 and keeps later events in order behind a retry', async () => { + const { client, emit } = startClient() + await settle() + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + emit({ + kind: 'prompt', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + prompt: { kind: 'unsupported', promptId: 'old' } + }) + await settle() + emit({ kind: 'turn-started', conversation, worktreeId: 'wt', paneKey: 'tab:1', at: 7 }) + fake.forced.delete(CONNECTOR_PATHS.events) + await settle(5_000) + expect(fake.requestsTo(CONNECTOR_PATHS.events).map(({ body }) => body)).toMatchObject([ + { seq: 0, events: [{ kind: 'snapshot' }] }, + { seq: 1, events: [{ kind: 'prompt' }] }, + { seq: 1, events: [{ kind: 'prompt' }] }, + { seq: 2, events: [{ kind: 'turn-started' }] } + ]) + expect(fake.events).toMatchObject([ + { kind: 'snapshot' }, + { kind: 'prompt' }, + { kind: 'turn-started' } + ]) + + // Each hello names the revision the cloud gave it. + expect(fake.requestsTo(CONNECTOR_PATHS.events)[0]?.body).toMatchObject({ helloRev: 1 }) + fake.forgetHello() + poll([]) + await settle() + const renewed = fake.requestsTo(CONNECTOR_PATHS.events).at(-1)?.body + expect(renewed).toMatchObject({ + helloRev: 2, + seq: 0, + events: [{ kind: 'snapshot' }] + }) + client.stop() + }) + + it('says hello again with a snapshot once the caps dropped a live event', async () => { + const { client, emit } = startClient() + await settle() + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + for (let index = 0; index <= 550; index++) { + emit({ + kind: 'prompt', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + prompt: { kind: 'unsupported', promptId: String(index) } + }) + } + fake.forced.delete(CONNECTOR_PATHS.events) + poll([]) + await settle(40_000) + expect(fake.hellos).toHaveLength(2) + expect( + fake.events.filter((event) => isRecord(event) && event.kind === 'snapshot') + ).toHaveLength(2) + client.stop() + }) + + it('settles a recovered delivery whose earlier run fails as unknown, never running it again', async () => { + const seam = fakeSeam() + seam.calls.recorded.mockRejectedValueOnce(new Error('launch_failed')) + const { client, calls } = startClient(seam) + await settle() + poll([], { unsettled: [{ deliveryId: 'd1', conversation }] }) + await settle() + expect(fake.acks).toEqual([ + { deliveryId: 'd1', step: 'done', result: { outcome: 'unknown', reason: 'launch_failed' } } + ]) + expect(calls.send).not.toHaveBeenCalled() + client.stop() + }) + + it('keeps turn ends that finish before the first hello, and sends them after its snapshot', async () => { + fake.online = false + const { client, emit } = startClient() + await settle() + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 4, + stopReason: 'success', + at: 9 + }) + expect(existsSync(join(userDataPath, 'connector-turn-outbox.json'))).toBe(true) + fake.online = true + await settle(30_000) + expect(fake.events).toMatchObject([{ kind: 'snapshot' }, { kind: 'turn-ended', turnId: 4 }]) + client.stop() + }) + + it('sends a done again under a new hello when the old one was still in flight', async () => { + const seam = fakeSeam() + const sent = Promise.withResolvers() + seam.outcomes.set('d1', sent.promise) + const { client, machine } = startClient(seam) + await settle() + poll([{ ...base, deliveryId: 'd1', kind: 'send', text: 'hi' }]) + await settle() + fake.holdBefore.add(CONNECTOR_PATHS.acks) + sent.resolve({ outcome: 'accepted' }) + await settle() + machine.agentList = ['claude'] + poll([], { unsettled: [{ deliveryId: 'd1', conversation }] }) + await settle() + expect(fake.hellos).toHaveLength(2) + fake.release(CONNECTOR_PATHS.acks) + await settle(5_000) + expect(fake.acks).toContainEqual({ + deliveryId: 'd1', + step: 'done', + result: { outcome: 'accepted' } + }) + client.stop() + }) + + it('keeps a turn end whose request the cloud ignored while switching to a new hello', async () => { + const { client, machine, emit } = startClient() + await settle() + fake.holdBefore.add(CONNECTOR_PATHS.events) + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 4, + stopReason: 'success', + at: 9 + }) + await settle() + // The cloud takes the new hello but its answer is late; the old request lands first, ignored. + fake.holdAfter.add(CONNECTOR_PATHS.hello) + machine.agentList = ['claude'] + poll([]) + await settle() + fake.release(CONNECTOR_PATHS.events) + await settle() + fake.release(CONNECTOR_PATHS.hello) + await settle(5_000) + expect(fake.events).toContainEqual(expect.objectContaining({ kind: 'turn-ended', turnId: 4 })) + client.stop() + }) + + it('says hello again after a failed one, even when what it says went back to the last accepted hello', async () => { + const { client, machine, emit } = startClient() + await settle() + fake.forced.set(CONNECTOR_PATHS.hello, { status: 503 }) + machine.agentList = ['claude'] + poll([]) + await settle() + machine.agentList = ['claude', 'codex'] + fake.forced.delete(CONNECTOR_PATHS.hello) + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 4, + stopReason: 'success', + at: 9 + }) + await settle(30_000) + expect(fake.events).toContainEqual(expect.objectContaining({ kind: 'turn-ended', turnId: 4 })) + client.stop() + }) + + it('says hello again after a poll fails, so the cloud lists unsettled deliveries again', async () => { + const { client } = startClient() + await settle() + fake.answerPoll({ status: 503 }) + await settle(5_000) + expect(fake.hellos).toHaveLength(2) + client.stop() + }) + + it('counts only the contract’s success: a 200 that is not one keeps the turn end', async () => { + const { client, emit } = startClient() + await settle() + fake.forced.set(CONNECTOR_PATHS.events, { status: 200, body: 'proxy' }) + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 4, + stopReason: 'success', + at: 9 + }) + await settle() + fake.forced.delete(CONNECTOR_PATHS.events) + await settle(5_000) + expect(fake.events).toContainEqual(expect.objectContaining({ kind: 'turn-ended', turnId: 4 })) + client.stop() + }) + + it('never takes a hello whose answer is not the contract', async () => { + const { client } = startClient() + fake.forced.set(CONNECTOR_PATHS.hello, { status: 204 }) + await settle(5_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll)).toHaveLength(0) + fake.forced.delete(CONNECTOR_PATHS.hello) + await settle(30_000) + expect(fake.requestsTo(CONNECTOR_PATHS.poll).length).toBeGreaterThan(0) + client.stop() + }) + + it('fills event requests by bytes, so a backlog of long turn ends never meets a 413', async () => { + const { client, emit } = startClient() + fake.forced.set(CONNECTOR_PATHS.events, { status: 503 }) + for (let turnId = 1; turnId <= 60; turnId++) { + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId, + stopReason: 'success', + lastAssistantMessage: 'x'.repeat(8 * 1024), + at: turnId + }) + } + await settle() + fake.forced.delete(CONNECTOR_PATHS.events) + await settle(30_000) + expect( + fake.events.filter((event) => isRecord(event) && event.kind === 'turn-ended') + ).toHaveLength(60) + expect(fake.requestsTo(CONNECTOR_PATHS.events).length).toBeGreaterThan(2) + client.stop() + }) + + it('keeps a turn end refused under a hello a retired client replaced, and resends it after hello', async () => { + const { client, emit } = startClient() + await settle() + fake.helloFromElsewhere() + emit({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId: 4, + stopReason: 'success', + at: 9 + }) + await settle() + // The cloud wakes the held poll for the newer hello, as the real one does. + fake.answerPoll({ status: 409, body: { error: 'hello_required' } }) + await settle(5_000) + expect(fake.hellos).toHaveLength(2) + expect(fake.events).toContainEqual(expect.objectContaining({ kind: 'turn-ended', turnId: 4 })) + client.stop() + }) +}) + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null +} diff --git a/src/main/connector-cloud/connector-poll-client.ts b/src/main/connector-cloud/connector-poll-client.ts new file mode 100644 index 00000000000..cfe159a0706 --- /dev/null +++ b/src/main/connector-cloud/connector-poll-client.ts @@ -0,0 +1,322 @@ +// The machine side of the connector: one held poll, two confirmations per delivery, and turn events back. +import { z } from 'zod' +import { delayUnlessAborted } from '../../shared/abort-aware-delay' +import { + CONNECTOR_CLIENT_LIMITS, + CONNECTOR_PATHS, + ConnectorDeliveryId, + ConnectorHelloResponse, + ConnectorPollControl, + ConnectorPollResponse, + type ConnectorAck, + type ConnectorConversation, + type ConnectorDeliveryResult, + type ConnectorEvent +} from '../../shared/connector-cloud-contract' +import type { + ExternalConversationHost, + ExternalConversationSeam +} from '../external-conversations/external-conversation-seam' +import { UNREADABLE_ANSWER, type ConnectorFailure } from './connector-cloud-request' +import { dispatchDelivery } from './connector-deliveries' +import { connectorHello, type ConnectorHelloSaid } from './connector-hello' +import type { ConnectorMachine, ConnectorTransport } from './connector-machine' +import { connectorRetryDelayMs, ConnectorOutbox, ConnectorOutboxSender } from './connector-outbox' +import type { ConnectorTurnEvents } from './connector-turn-events' + +const RECONNECT_SPREAD_MS = 2_000 +const MAX_ACKS_PER_REQUEST = 100 + +export type ConnectorPollClientOptions = { + transport: ConnectorTransport + /** Owned by the caller across clients; this client only sends through it while running. */ + events: ConnectorTurnEvents + seam: ExternalConversationSeam + machine: ConnectorMachine + random?: () => number +} + +/** Why the cloud stopped this client, e.g. `client_too_old`; it stays stopped until replaced. */ +type Stop = string + +/** A refusal no retry can fix; any other failure, a lapsed sign-in included, is retried after a backoff. */ +function stopFor(failure: ConnectorFailure): Stop | null { + return failure.status === 426 ? 'client_too_old' : null +} + +export class ConnectorPollClient { + readonly seam: ExternalConversationSeam + private readonly host: ExternalConversationHost + private readonly transport: ConnectorTransport + private readonly stopped = new AbortController() + private readonly acks: ConnectorOutbox + private readonly ackSender: ConnectorOutboxSender + private readonly eventSender: ConnectorOutboxSender + // Received in this process and not yet confirmed done. + private readonly active = new Set() + private readonly random: () => number + private releaseWatch: (() => void) | null = null + // The hello the cloud took, which scopes the poll and everything sent; null until then, once the cloud + // says it is no longer current, and once stopped. A new hello only replaces it. The cloud refuses + // anything under another hello with 409, so a 204 always means applied. + private helloRev: number | null = null + // What that hello said. + private helloSaid = '' + private failures = 0 + private warnedTrimmed = false + private stopReason: Stop | null = null + + constructor(private readonly options: ConnectorPollClientOptions) { + this.seam = options.seam + this.transport = options.transport + this.random = options.random ?? Math.random + this.host = options.machine.conversationHost() + this.acks = new ConnectorOutbox({ + headSize: () => MAX_ACKS_PER_REQUEST, + onAccepted: (sent) => { + for (const ack of sent) { + if (ack.step === 'done') { + this.active.delete(ack.deliveryId) + } + } + } + }) + const helloRev = (): number | null => this.helloRev + this.ackSender = new ConnectorOutboxSender( + this.acks, + (acks, rev) => this.post(CONNECTOR_PATHS.acks, rev, { v: 1, helloRev: rev, acks }), + helloRev, + this.random + ) + // Every hello restarts the numbering with a snapshot. + this.eventSender = new ConnectorOutboxSender( + options.events, + (events, rev, seq) => + this.post(CONNECTOR_PATHS.events, rev, { v: 1, helloRev: rev, seq, events }), + helloRev, + this.random + ) + } + + start(): void { + // A disable disposes the seam; nothing here may outlive it. + this.seam.signal.addEventListener('abort', () => this.stop(), { + once: true, + signal: this.stopped.signal + }) + // Turn ends that finish before the first hello still reach the outbox; only sending waits for it. + this.watch() + void this.run() + } + + stop(): void { + this.stopped.abort() + // Confirmations still owed are listed as unsettled after the next hello and settled from receipts. + this.helloRev = null + this.ackSender.stop() + this.eventSender.stop() + this.releaseWatch?.() + this.releaseWatch = null + } + + private async run(): Promise { + while (!this.stopped.signal.aborted) { + let stop: Stop | null + try { + stop = await this.cycle() + } catch (error) { + console.error('[connector] poll cycle failed', error) + stop = await this.backoff(UNREADABLE_ANSWER) + } + if (stop) { + this.halt(stop) + } + } + if (this.stopReason) { + console.warn(`[connector] Orca Cloud stopped this connection: ${this.stopReason}`) + } + } + + /** One hello-if-needed and one held poll; returns why to stop, or null to go on. */ + private async cycle(): Promise { + const { machine } = this.options + const said = { agents: machine.agents(), projects: machine.projects() } + // A hello restarts the numbered events with a snapshot, which also restates anything the caps dropped. + if ( + this.helloRev === null || + JSON.stringify(said) !== this.helloSaid || + this.options.events.needsSnapshot + ) { + const refused = await this.hello(said) + if (refused) { + return stopFor(refused) ?? this.backoff(refused) + } + } + const { helloRev } = this + // Stopped during the hello. + if (helloRev === null) { + return null + } + const sentAt = performance.now() + const polled = await this.transport.call( + CONNECTOR_PATHS.poll, + { v: 1, helloRev }, + ConnectorPollResponse, + { timeoutMs: CONNECTOR_CLIENT_LIMITS.pollTimeoutMs, signal: this.stopped.signal } + ) + // A stopped client admits nothing, even from an answer that arrived anyway. + if (this.stopped.signal.aborted) { + return null + } + if (!polled.ok) { + // A lost answer may have carried this hello's one list of unsettled deliveries; a new hello lists + // them again. A 409 means the cloud wants one anyway. + this.helloRev = null + return polled.status === 409 ? null : (stopFor(polled) ?? this.backoff(polled)) + } + this.failures = 0 + const response = polled.value + for (const unsettled of response.unsettled ?? []) { + void this.reconcile(unsettled.deliveryId, unsettled.conversation) + } + // Network delay only shortens the deadline: it runs from when this poll was sent. + const msLeft = (expiresAt: number): number => + expiresAt - response.serverNow - (performance.now() - sentAt) + for (const raw of response.deliveries) { + this.deliver(raw, msLeft) + } + return this.obey(response.control) + } + + /** Null once the cloud took it; otherwise why not. Each hello also gets a fresh watch and snapshot. */ + private async hello(said: ConnectorHelloSaid): Promise { + const { body, trimmed } = connectorHello(this.options.machine, said) + if (trimmed && !this.warnedTrimmed) { + this.warnedTrimmed = true + console.warn('[connector] the hello left out the least recently used projects to fit its cap') + } + const answer = await this.transport.call(CONNECTOR_PATHS.hello, body, ConnectorHelloResponse, { + method: 'PUT', + signal: this.stopped.signal + }) + if (!answer.ok) { + return answer + } + if (this.stopped.signal.aborted) { + return null + } + this.helloSaid = JSON.stringify(said) + // A fresh watch's first event is the snapshot, which opens this hello's numbered events. + this.watch() + this.helloRev = answer.value.helloRev + this.eventSender.resume() + this.ackSender.resume() + return null + } + + private watch(): void { + const { events } = this.options + const previous = this.releaseWatch + this.releaseWatch = this.seam.watch(this.host, (event) => { + events.observe(event) + this.eventSender.flush() + }) + previous?.() + } + + private deliver(raw: unknown, msLeft: (expiresAt: number) => number): void { + const id = ConnectorDeliveryId.safeParse(raw) + // A lost `received` resends the delivery; this process already holds it. + if (id.success && this.active.has(id.data.deliveryId)) { + this.pushAck({ deliveryId: id.data.deliveryId, step: 'received' }) + return + } + const dispatched = dispatchDelivery(this.seam, this.host, raw, msLeft) + if (!dispatched) { + return + } + const { deliveryId, received, done } = dispatched + this.active.add(deliveryId) + if (received) { + this.pushAck({ deliveryId, step: 'received' }) + } + void done.then((result) => this.pushAck({ deliveryId, step: 'done', result })) + } + + /** A delivery an earlier process received: replay its receipt, never run it again. */ + private async reconcile(deliveryId: string, conversation: ConnectorConversation): Promise { + if (this.active.has(deliveryId)) { + return + } + this.active.add(deliveryId) + // A receipt still running can fail; its outcome is then unknown, and it is still never run again. + const result: ConnectorDeliveryResult = await this.seam.recorded(conversation, deliveryId).then( + (recorded) => recorded ?? { outcome: 'unknown', reason: 'no_local_record' }, + (error: unknown) => ({ + outcome: 'unknown', + reason: error instanceof Error && error.message ? error.message : 'failed' + }) + ) + if (this.stopped.signal.aborted) { + return + } + this.pushAck({ deliveryId, step: 'done', result }) + } + + private pushAck(ack: ConnectorAck): void { + this.acks.push(ack) + this.ackSender.flush() + } + + private async obey(raw: unknown): Promise { + const control = ConnectorPollControl.safeParse(raw) + if (!control.success) { + return null + } + if ('stop' in control.data) { + return control.data.stop + } + const waitMs = + 'pausedForMs' in control.data ? control.data.pausedForMs : this.random() * RECONNECT_SPREAD_MS + await delayUnlessAborted(waitMs, this.stopped.signal) + return null + } + + private async backoff(failure: ConnectorFailure): Promise { + await delayUnlessAborted( + connectorRetryDelayMs(this.failures++, failure, this.random), + this.stopped.signal + ) + return null + } + + private async post( + path: string, + helloRev: number, + body: unknown + ): Promise { + const sent = await this.transport.call(path, body, z.undefined(), { + signal: this.stopped.signal + }) + if (sent.ok) { + return null + } + // That hello is no longer current, e.g. a retired client's late one landed after it: say hello again. + if (sent.status === 409 && this.helloRev === helloRev) { + this.helloRev = null + } + const stop = stopFor(sent) + if (stop) { + this.halt(stop) + } + return sent + } + + /** Only a running client may stop for a reason; a retired one's late answers mean nothing. */ + private halt(reason: Stop): void { + if (!this.stopped.signal.aborted) { + this.stopReason = reason + this.stop() + } + } +} diff --git a/src/main/connector-cloud/connector-snapshot.test.ts b/src/main/connector-cloud/connector-snapshot.test.ts new file mode 100644 index 00000000000..16ec84ce240 --- /dev/null +++ b/src/main/connector-cloud/connector-snapshot.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from 'vitest' +import { CONNECTOR_CLIENT_LIMITS } from '../../shared/connector-cloud-contract' +import type { ConversationSnapshot } from '../external-conversations/external-conversation-events' +import { connectorSnapshot } from './connector-snapshot' + +const NOW = 1_800_000_000_000 +const DAY = 24 * 60 * 60 * 1000 + +type Entry = ConversationSnapshot['conversations'][number] + +function entry(id: string, activeAt: number, fields: Partial = {}): Entry { + return { + conversation: { source: 'slack', externalId: `T1:C1:${id}` }, + worktreeId: 'wt', + paneKey: `pane-${id}`, + agent: 'claude', + working: false, + prompt: null, + turnId: null, + activeAt, + ...fields + } +} + +const ids = (event: ConversationSnapshot) => + connectorSnapshot(event, NOW).snapshot.conversations.map((item) => item.paneKey) + +describe('connectorSnapshot', () => { + it('lists conversations active in the last 7 days, newest first; working counts as now', () => { + const conversations = [ + entry('old', NOW - 8 * DAY), + entry('recent', NOW - DAY), + entry('newer', NOW - 1000), + entry('busy', NOW - 30 * DAY, { working: true }) + ] + expect(ids({ kind: 'snapshot', conversations })).toEqual([ + 'pane-busy', + 'pane-newer', + 'pane-recent' + ]) + expect(connectorSnapshot({ kind: 'snapshot', conversations }, NOW).trimmed).toBe(false) + }) + + it('keeps the newest maxSnapshotConversations and says it left some out', () => { + const conversations = Array.from( + { length: CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations + 5 }, + (_, index) => entry(String(index), NOW - index) + ) + const { snapshot, trimmed } = connectorSnapshot({ kind: 'snapshot', conversations }, NOW) + expect(trimmed).toBe(true) + expect(snapshot.conversations).toHaveLength(CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations) + expect(snapshot.conversations[0]?.paneKey).toBe('pane-0') + }) + + it('fits the event cap: prompts go unsupported first, then the least recent go', () => { + const question = 'x'.repeat(4 * 1024) + const conversations = Array.from( + { length: CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations }, + (_, index) => + entry(String(index), NOW - index, { + paneKey: `${index}`.padStart(128, 'p'), + prompt: { kind: 'question', promptId: `p${index}`, question, options: [] } + }) + ) + const { snapshot } = connectorSnapshot({ kind: 'snapshot', conversations }, NOW) + expect(Buffer.byteLength(JSON.stringify(snapshot))).toBeLessThanOrEqual( + CONNECTOR_CLIENT_LIMITS.maxEventBytes + ) + expect(snapshot.conversations).toHaveLength(CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations) + expect(snapshot.conversations[0]?.prompt).toEqual({ kind: 'unsupported', promptId: 'p0' }) + }) + + it('leaves out the least recent when even bare entries pass the event cap', () => { + const escaped = '\u0001'.repeat(128) + const conversations = Array.from( + { length: CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations }, + (_, index) => + entry(String(index), NOW - index, { + conversation: { source: 'slack', externalId: `${index}${escaped}` }, + paneKey: `${index}${escaped}` + }) + ) + const { snapshot, trimmed } = connectorSnapshot({ kind: 'snapshot', conversations }, NOW) + expect(trimmed).toBe(true) + expect(Buffer.byteLength(JSON.stringify(snapshot))).toBeLessThanOrEqual( + CONNECTOR_CLIENT_LIMITS.maxEventBytes + ) + expect(snapshot.conversations[0]?.paneKey).toBe(`0${escaped}`) + }) +}) diff --git a/src/main/connector-cloud/connector-snapshot.ts b/src/main/connector-cloud/connector-snapshot.ts new file mode 100644 index 00000000000..7da83e6ce3e --- /dev/null +++ b/src/main/connector-cloud/connector-snapshot.ts @@ -0,0 +1,66 @@ +// What a hello's snapshot tells the cloud: recently active Slack conversations, within the contract's caps. +import type { PendingPrompt } from '../../shared/agent-pending-prompt' +import { + CONNECTOR_CLIENT_LIMITS, + type ConnectorConversation, + type ConnectorEvent +} from '../../shared/connector-cloud-contract' +import type { ConversationSnapshot } from '../external-conversations/external-conversation-events' +import type { ConversationKey } from '../external-conversations/external-conversation-types' + +type Snapshot = Extract + +export const jsonBytes = (value: unknown): number => Buffer.byteLength(JSON.stringify(value)) + +/** A prompt too large to carry, offered as one only Orca can answer. */ +export const unsupported = (prompt: PendingPrompt): PendingPrompt => ({ + kind: 'unsupported', + promptId: prompt.promptId +}) + +export function slackConversation(key: ConversationKey): ConnectorConversation | null { + return key.source === 'slack' ? { source: 'slack', externalId: key.externalId } : null +} + +/** + * Linked conversations active within the window, newest first, at most maxSnapshotConversations. Working + * or prompting counts as active now. Past the event cap, prompts go unsupported, then the least recent go. + * `trimmed` says whether a recently active conversation was left out. + */ +export function connectorSnapshot( + event: ConversationSnapshot, + now: number +): { snapshot: Snapshot; trimmed: boolean } { + const since = now - CONNECTOR_CLIENT_LIMITS.snapshotActiveWindowMs + const recent = event.conversations + .flatMap((entry) => { + const conversation = slackConversation(entry.conversation) + const activeAt = entry.working || entry.prompt ? now : entry.activeAt + return conversation && activeAt >= since ? [{ ...entry, conversation, activeAt }] : [] + }) + .sort((a, b) => b.activeAt - a.activeAt) + let conversations: Snapshot['conversations'] = recent + .slice(0, CONNECTOR_CLIENT_LIMITS.maxSnapshotConversations) + .map(({ conversation, paneKey, agent, working, prompt, turnId }) => ({ + conversation, + paneKey, + agent, + working, + prompt, + ...(turnId === null ? {} : { turnId }) + })) + const fits = (): boolean => + jsonBytes({ kind: 'snapshot', conversations }) <= CONNECTOR_CLIENT_LIMITS.maxEventBytes + if (!fits()) { + conversations = conversations.map((entry) => + entry.prompt ? { ...entry, prompt: unsupported(entry.prompt) } : entry + ) + } + while (conversations.length > 0 && !fits()) { + conversations = conversations.slice(0, -1) + } + return { + snapshot: { kind: 'snapshot', conversations }, + trimmed: conversations.length < recent.length + } +} diff --git a/src/main/connector-cloud/connector-token-mint.test.ts b/src/main/connector-cloud/connector-token-mint.test.ts new file mode 100644 index 00000000000..9083e447742 --- /dev/null +++ b/src/main/connector-cloud/connector-token-mint.test.ts @@ -0,0 +1,84 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { OrcaCloudAuthConfig } from '../orca-profiles/profile-cloud-auth-config' +import { desktopConnectorTokenMint } from './connector-token-mint' + +const profile = vi.hoisted(() => { + const state: { cloud?: { userId: string } } = { cloud: { userId: 'ada' } } + return state +}) +const runFresh = vi.hoisted(() => vi.fn()) + +vi.mock('../orca-profiles/profile-index-store', () => ({ + ensureActiveOrcaProfile: () => ({ profile: { id: 'p1', cloud: profile.cloud } }) +})) +vi.mock('../orca-profiles/profile-cloud-session-refresh', () => ({ + runWithFreshOrcaCloudSession: runFresh +})) +vi.mock('../runtime/host-descriptor', () => ({ + loadOrCreateHostInstallationId: async () => '6b1f2c1e-0d7a-4a57-9b38-6f1c2e3d4a5b' +})) + +// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mint reads only this endpoint. +const authConfig = { + connectorTokenEndpoint: 'https://login.test/v1/desktop/auth/connector-token' +} as OrcaCloudAuthConfig + +let fetchImpl: ReturnType> + +beforeEach(() => { + profile.cloud = { userId: 'ada' } + runFresh.mockReset() + runFresh.mockImplementation(async (_config, _active, _path, operation) => ({ + status: 'ok', + value: await operation({ accessToken: 'access-1' }) + })) + fetchImpl = vi.fn( + async () => + new Response(JSON.stringify({ connectorToken: 'ct-1', expiresAt: 1_800_000_300_000 }), { + status: 200 + }) + ) +}) + +describe('desktopConnectorTokenMint', () => { + it('asks the auth service with the Orca access token and this install id', async () => { + const minted = await desktopConnectorTokenMint(authConfig, '/data', 'ada', fetchImpl)() + expect(minted).toEqual({ + ok: true, + value: { connectorToken: 'ct-1', expiresAt: 1_800_000_300_000 } + }) + const [url, init] = fetchImpl.mock.calls[0]! + expect(url).toBe(authConfig.connectorTokenEndpoint) + expect(new Headers(init?.headers).get('authorization')).toBe('Bearer access-1') + expect(JSON.parse(String(init?.body))).toEqual({ + machineId: '6b1f2c1e-0d7a-4a57-9b38-6f1c2e3d4a5b' + }) + }) + + it('mints nothing for an account that is no longer the signed-in one', async () => { + profile.cloud = { userId: 'grace' } + const minted = await desktopConnectorTokenMint(authConfig, '/data', 'ada', fetchImpl)() + expect(minted).toMatchObject({ ok: false, status: 401, error: 'signed_out' }) + expect(fetchImpl).not.toHaveBeenCalled() + }) + + it('reads a sign-in the refresh could not keep as signed out', async () => { + runFresh.mockResolvedValue({ status: 'reconnect-required' }) + const minted = await desktopConnectorTokenMint(authConfig, '/data', 'ada', fetchImpl)() + expect(minted).toMatchObject({ ok: false, status: 401, error: 'signed_out' }) + }) + + it('hands a 401 to the session refresh, and keeps Retry-After on other refusals', async () => { + fetchImpl.mockResolvedValueOnce( + new Response(JSON.stringify({ error: 'invalid_token' }), { status: 401 }) + ) + const refused = await desktopConnectorTokenMint(authConfig, '/data', 'ada', fetchImpl)() + expect(refused).toMatchObject({ ok: false, status: 401 }) + + fetchImpl.mockResolvedValueOnce( + new Response(null, { status: 429, headers: { 'retry-after': '30' } }) + ) + const paced = await desktopConnectorTokenMint(authConfig, '/data', 'ada', fetchImpl)() + expect(paced).toMatchObject({ ok: false, status: 429, retryAfterMs: 30_000 }) + }) +}) diff --git a/src/main/connector-cloud/connector-token-mint.ts b/src/main/connector-cloud/connector-token-mint.ts new file mode 100644 index 00000000000..3cad1375b51 --- /dev/null +++ b/src/main/connector-cloud/connector-token-mint.ts @@ -0,0 +1,79 @@ +// Connector tokens for the desktop's signed-in Orca account, fetched the way relay tokens are. +import { z } from 'zod' +import type { OrcaCloudAuthConfig } from '../orca-profiles/profile-cloud-auth-config' +import { OrcaCloudRequestError } from '../orca-profiles/profile-cloud-client' +import { runWithFreshOrcaCloudSession } from '../orca-profiles/profile-cloud-session-refresh' +import { ensureActiveOrcaProfile } from '../orca-profiles/profile-index-store' +import { loadOrCreateHostInstallationId } from '../runtime/host-descriptor' +import { requestConnector, type ConnectorFailure } from './connector-cloud-request' +import type { ConnectorTokenMint } from './connector-transport' + +const ConnectorTokenResponse = z.object({ + connectorToken: z + .string() + .min(1) + .max(8 * 1024), + expiresAt: z.number().int().positive() +}) + +const SIGNED_OUT: ConnectorFailure = { + ok: false, + status: 401, + error: 'signed_out', + retryAfterMs: null +} + +/** The account the connector serves, read like the relay's: the active profile's cloud user. */ +export function connectorAccount(userDataPath: string): string | null { + return ensureActiveOrcaProfile(userDataPath).profile.cloud?.userId ?? null +} + +/** Mints only for `account`; after a switch or sign-out the old client gets no token for the new one. */ +export function desktopConnectorTokenMint( + authConfig: OrcaCloudAuthConfig, + userDataPath: string, + account: string, + fetchImpl: typeof globalThis.fetch = globalThis.fetch +): ConnectorTokenMint { + return async () => { + if (connectorAccount(userDataPath) !== account) { + return SIGNED_OUT + } + const active = ensureActiveOrcaProfile(userDataPath) + try { + // The per-install id the host descriptor already keeps; the cloud keys this machine on it. + const machineId = await loadOrCreateHostInstallationId(userDataPath) + const minted = await runWithFreshOrcaCloudSession( + authConfig, + active, + userDataPath, + async (session) => { + const answer = await requestConnector( + { + fetchImpl, + url: authConfig.connectorTokenEndpoint, + body: { machineId }, + token: session.accessToken + }, + ConnectorTokenResponse + ) + // Refreshes the Orca session once, as every desktop cloud call does. + if (!answer.ok && (answer.status === 401 || answer.status === 403)) { + throw new OrcaCloudRequestError(answer.status, answer.error ?? undefined) + } + return answer + } + ) + return minted.status === 'ok' && connectorAccount(userDataPath) === account + ? minted.value + : SIGNED_OUT + } catch (error) { + return { + ok: false, + status: error instanceof OrcaCloudRequestError ? error.statusCode : null, + error: null, + retryAfterMs: null + } + } + } +} diff --git a/src/main/connector-cloud/connector-transport.ts b/src/main/connector-cloud/connector-transport.ts new file mode 100644 index 00000000000..1919efa7431 --- /dev/null +++ b/src/main/connector-cloud/connector-transport.ts @@ -0,0 +1,85 @@ +import type { z } from 'zod' +import { + requestConnector, + type ConnectorRequest, + type ConnectorResult +} from './connector-cloud-request' + +// Renew early so a token cannot lapse between the check and the request. +const TOKEN_RENEWAL_MARGIN_MS = 60_000 + +export type ConnectorToken = { connectorToken: string; expiresAt: number } + +/** Asks Orca's auth service for a short-lived connector token as this machine's signed-in account. */ +export type ConnectorTokenMint = () => Promise> + +export type ConnectorCallOptions = Pick + +/** Calls to the connector with a connector token, the way the relay is reached with relay tokens. */ +export class ConnectorAccountTransport { + private token: ConnectorToken | null = null + private pending: Promise> | null = null + + constructor( + private readonly origin: string, + private readonly mint: ConnectorTokenMint, + private readonly fetchImpl: typeof globalThis.fetch = globalThis.fetch, + private readonly now: () => number = Date.now + ) {} + + /** An authorized call; a 401 mints one new token and tries once more. */ + async call( + path: string, + body: unknown, + schema: TSchema, + options: ConnectorCallOptions = {} + ): Promise>> { + const token = await this.ensure() + if (!token.ok) { + return token + } + const result = await this.send(token.value, path, body, schema, options) + if (result.ok || result.status !== 401) { + return result + } + // Drop only the refused token; a concurrent call may already hold its replacement. + if (this.token === token.value) { + this.token = null + } + const renewed = await this.ensure() + return renewed.ok ? await this.send(renewed.value, path, body, schema, options) : renewed + } + + private send( + token: ConnectorToken, + path: string, + body: unknown, + schema: TSchema, + options: ConnectorCallOptions + ): Promise>> { + const url = `${this.origin}${path}` + return requestConnector( + { ...options, fetchImpl: this.fetchImpl, url, body, token: token.connectorToken }, + schema + ) + } + + private async ensure(): Promise> { + const cached = this.token + if (cached && cached.expiresAt - TOKEN_RENEWAL_MARGIN_MS > this.now()) { + return { ok: true, value: cached } + } + // Concurrent calls share one mint. + this.pending ??= this.mint() + .then((minted) => { + if (minted.ok) { + this.token = minted.value + } + return minted + }) + .finally(() => { + this.pending = null + }) + return await this.pending + } +} diff --git a/src/main/connector-cloud/connector-turn-events.test.ts b/src/main/connector-cloud/connector-turn-events.test.ts new file mode 100644 index 00000000000..04739cf0824 --- /dev/null +++ b/src/main/connector-cloud/connector-turn-events.test.ts @@ -0,0 +1,138 @@ +import { mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { CONNECTOR_CLIENT_LIMITS } from '../../shared/connector-cloud-contract' +import { ConnectorTurnEvents } from './connector-turn-events' + +const conversation = { source: 'slack', externalId: 'T1:C9:1' } as const +let userDataPath: string + +const prompt = (promptId: string) => + ({ + kind: 'prompt', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + prompt: { kind: 'unsupported', promptId } + }) as const +const turnEnded = (turnId: number) => + ({ + kind: 'turn-ended', + conversation, + worktreeId: 'wt', + paneKey: 'tab:1', + turnId, + stopReason: null, + at: turnId + }) as const + +beforeEach(() => { + userDataPath = mkdtempSync(join(tmpdir(), 'orca-connector-events-')) +}) + +afterEach(() => { + vi.useRealTimers() + rmSync(userDataPath, { recursive: true, force: true }) +}) + +describe('ConnectorTurnEvents', () => { + it('never drops a new hello’s snapshot, even behind a head still in flight', () => { + const events = new ConnectorTurnEvents(userDataPath, 'ada') + events.observe(prompt('old')) + expect(events.next(1)?.batch).toHaveLength(1) + events.observe({ kind: 'snapshot', conversations: [] }) + for (let index = 0; index < 600; index++) { + events.observe(prompt(String(index))) + } + const head = events.next(2) + expect(head?.stamp.seq).toBe(0) + expect(head?.batch[0]).toEqual({ kind: 'snapshot', conversations: [] }) + }) + + it('keeps 200 turn ends in all, never trimming the head in flight', () => { + const events = new ConnectorTurnEvents(userDataPath, 'ada') + for (let turn = 1; turn <= 50; turn++) { + events.observe(turnEnded(turn)) + } + expect(events.next(1)?.batch).toHaveLength(50) + for (let turn = 51; turn <= 300; turn++) { + events.observe(turnEnded(turn)) + } + const { account, turnEnds: stored } = z + .object({ account: z.string(), turnEnds: z.array(z.unknown()) }) + .parse(JSON.parse(readFileSync(join(userDataPath, 'connector-turn-outbox.json'), 'utf8'))) + expect(account).toBe('ada') + expect(stored).toHaveLength(200) + expect(stored[0]).toMatchObject({ turnId: 1 }) + expect(stored.at(-1)).toMatchObject({ turnId: 300 }) + }) + + it('asks for a snapshot once the caps drop an event, until the next snapshot arrives', () => { + const events = new ConnectorTurnEvents(userDataPath, 'ada') + for (let index = 0; index < 500; index++) { + events.observe(prompt(String(index))) + } + expect(events.needsSnapshot).toBe(false) + events.observe(prompt('one too many')) + expect(events.needsSnapshot).toBe(true) + events.observe({ kind: 'snapshot', conversations: [] }) + expect(events.needsSnapshot).toBe(false) + }) + + it('restores only the turn ends its own account left on disk', () => { + new ConnectorTurnEvents(userDataPath, 'ada').observe(turnEnded(1)) + expect(new ConnectorTurnEvents(userDataPath, 'grace').next(1)).toBeNull() + expect(new ConnectorTurnEvents(userDataPath, 'ada').next(1)?.batch).toEqual([ + expect.objectContaining({ kind: 'turn-ended', turnId: 1 }) + ]) + }) + + it('offers a prompt too large for one event as one only Orca can answer', () => { + const events = new ConnectorTurnEvents(userDataPath, 'ada') + const question = 'x'.repeat(CONNECTOR_CLIENT_LIMITS.maxEventBytes) + events.observe({ + ...prompt('big'), + prompt: { kind: 'question', promptId: 'big', question, options: [] } + }) + expect(events.next(1)?.batch).toEqual([ + expect.objectContaining({ prompt: { kind: 'unsupported', promptId: 'big' } }) + ]) + }) + + it('names each conversation’s latest turn in the snapshot', () => { + const now = 1_800_000_000_000 + vi.useFakeTimers({ toFake: ['Date'], now }) + const events = new ConnectorTurnEvents(userDataPath, 'ada') + const entry = { conversation, worktreeId: 'wt', paneKey: 'tab:1', agent: 'claude' } as const + events.observe({ + kind: 'snapshot', + conversations: [ + { ...entry, working: true, prompt: null, turnId: 200, activeAt: 0 }, + { + ...entry, + paneKey: 'tab:2', + working: false, + prompt: null, + turnId: null, + activeAt: now + } + ] + }) + expect(events.next(1)?.batch[0]).toEqual({ + kind: 'snapshot', + conversations: [ + { + conversation, + paneKey: 'tab:1', + agent: 'claude', + working: true, + prompt: null, + turnId: 200 + }, + { conversation, paneKey: 'tab:2', agent: 'claude', working: false, prompt: null } + ] + }) + }) +}) diff --git a/src/main/connector-cloud/connector-turn-events.ts b/src/main/connector-cloud/connector-turn-events.ts new file mode 100644 index 00000000000..932fdb73327 --- /dev/null +++ b/src/main/connector-cloud/connector-turn-events.ts @@ -0,0 +1,231 @@ +// The seam's watch, as the cloud reads it. Turn ends wait in an owner-only file so a restart still reports them. +import { readFileSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { z } from 'zod' +import { CONNECTOR_CLIENT_LIMITS, type ConnectorEvent } from '../../shared/connector-cloud-contract' +import { writeSecureJsonFile } from '../../shared/secure-file' +import type { + ConversationEvent, + ConversationSnapshot +} from '../external-conversations/external-conversation-events' +import { isMissingFileError } from '../runtime/device-registry-corruption' +import { ConnectorOutbox, type ConnectorOutboxHead, type ConnectorStamp } from './connector-outbox' +import { connectorSnapshot, jsonBytes, slackConversation, unsupported } from './connector-snapshot' + +const TURN_OUTBOX_FILE = 'connector-turn-outbox.json' +const MAX_TURN_ENDS = 200 +// Live events past this are dropped oldest first; a fresh hello and snapshot then restate them. +const MAX_LIVE_EVENTS = 500 +// Room for the request's own fields around its events. +const EVENTS_BUDGET_BYTES = CONNECTOR_CLIENT_LIMITS.maxRequestBytes - 1024 + +type TurnEnded = Extract + +// Tagged with the account that produced them, so another account never sends them after a relaunch. +const StoredOutbox = z.object({ + account: z.string().min(1), + turnEnds: z.array( + z.object({ + kind: z.literal('turn-ended'), + conversation: z.object({ source: z.literal('slack'), externalId: z.string().min(1) }), + paneKey: z.string().min(1), + turnId: z.number(), + at: z.number(), + stopReason: z.string().min(1).nullable(), + lastAssistantMessage: z.string().optional() + }) + ) +}) + +const isTurnEnded = (event: ConnectorEvent): event is TurnEnded => event.kind === 'turn-ended' +const isLive = (event: ConnectorEvent): boolean => + event.kind !== 'turn-ended' && event.kind !== 'snapshot' + +/** Cut on a character boundary so the cloud's byte limit always holds. */ +function capUtf8(text: string, maxBytes: number): string { + const bytes = Buffer.from(text) + if (bytes.byteLength <= maxBytes) { + return text + } + let end = maxBytes + // Step back over continuation bytes (10xxxxxx) to the start of the cut character. + while (end > 0 && ((bytes[end] ?? 0) & 0xc0) === 0x80) { + end-- + } + return bytes.subarray(0, end).toString('utf8') +} + +/** As many leading events as fit in one request, and at least one. */ +function eventsHeadSize(events: readonly ConnectorEvent[]): number { + let bytes = 0 + for (const [index, event] of events.entries()) { + bytes += jsonBytes(event) + 1 + if (bytes > EVENTS_BUDGET_BYTES) { + return Math.max(1, index) + } + } + return events.length +} + +/** Within the per-event cap. */ +function promptEvent(event: Extract): ConnectorEvent { + return jsonBytes(event) <= CONNECTOR_CLIENT_LIMITS.maxEventBytes + ? event + : { ...event, prompt: unsupported(event.prompt) } +} + +export function toConnectorEvent(event: ConversationEvent): ConnectorEvent | null { + const conversation = slackConversation(event.conversation) + if (!conversation) { + return null + } + const { paneKey } = event + switch (event.kind) { + case 'linked': + return { kind: 'linked', conversation, paneKey, agent: event.agent } + case 'prompt': + return promptEvent({ kind: 'prompt', conversation, paneKey, prompt: event.prompt }) + case 'turn-started': + return { kind: 'turn-started', conversation, paneKey, at: event.at } + case 'turn-ended': + // Without its start the cloud cannot place the turn; the next snapshot marks it unverifiable. + return event.turnId === null + ? null + : { + kind: 'turn-ended', + conversation, + paneKey, + turnId: event.turnId, + at: event.at, + stopReason: event.stopReason, + ...(event.lastAssistantMessage + ? { + lastAssistantMessage: capUtf8( + event.lastAssistantMessage, + CONNECTOR_CLIENT_LIMITS.maxLastAssistantMessageBytes + ) + } + : {}) + } + } +} + +/** + * Keeps the newest turn ends and live events within their caps, never trimming the first `sent`, so memory + * and disk hold the same set. A snapshot is never dropped: it opens its hello. + */ +function boundEvents(events: ConnectorEvent[], sent: number): ConnectorEvent[] { + let turnEndsOver = events.filter(isTurnEnded).length - MAX_TURN_ENDS + let liveOver = events.filter(isLive).length - MAX_LIVE_EVENTS + if (turnEndsOver <= 0 && liveOver <= 0) { + return events + } + return events.filter( + (event, index) => + index < sent || (isTurnEnded(event) ? turnEndsOver-- <= 0 : !isLive(event) || liveOver-- <= 0) + ) +} + +/** + * One account's events for the cloud, in order, owned once per process so a replacement client never + * races a retired one over the file. Turn ends also wait on disk until the cloud accepts them. + */ +export class ConnectorTurnEvents { + private readonly outbox: ConnectorOutbox + private readonly filePath: string + // Something owed was dropped to stay within the caps; only a fresh snapshot restates it. + private dropped = false + private warnedTrimmed = false + + constructor( + userDataPath: string, + readonly account: string + ) { + this.filePath = join(userDataPath, TURN_OUTBOX_FILE) + this.outbox = new ConnectorOutbox({ + headSize: eventsHeadSize, + initial: this.readStored(), + bound: (events, sent) => { + const kept = boundEvents(events, sent) + this.dropped ||= kept.length < events.length + return kept + }, + onChange: (events) => this.store(events) + }) + } + + /** True until the next snapshot once the caps dropped an event. */ + get needsSnapshot(): boolean { + return this.dropped + } + + /** + * A snapshot opens a hello's events: it goes first, numbered 0, and replaces every live event still + * owed; only turn ends carry over. Everything else follows in order. + */ + observe(event: ConversationSnapshot | ConversationEvent): void { + if (event.kind === 'snapshot') { + const { snapshot, trimmed } = connectorSnapshot(event, Date.now()) + if (trimmed && !this.warnedTrimmed) { + this.warnedTrimmed = true + console.warn('[connector] a snapshot left out some active conversations to fit its caps') + } + this.dropped = false + this.outbox.restart([snapshot], isTurnEnded) + return + } + const wire = toConnectorEvent(event) + if (wire) { + this.outbox.push(wire) + } + } + + next(helloRev: number): ConnectorOutboxHead | null { + return this.outbox.next(helloRev) + } + + accepted(stamp: ConnectorStamp): void { + this.outbox.accepted(stamp) + } + + refused(stamp: ConnectorStamp): ConnectorEvent | null { + return this.outbox.refused(stamp) + } + + /** Drops everything, file included: no other account can report these turns. */ + clear(): void { + this.outbox.restart([], () => false) + } + + /** The same, before any owner exists in this process. */ + static clearStored(userDataPath: string): void { + rmSync(join(userDataPath, TURN_OUTBOX_FILE), { force: true }) + } + + private readStored(): ConnectorEvent[] { + let raw: string + try { + raw = readFileSync(this.filePath, 'utf8') + } catch (error) { + if (isMissingFileError(error)) { + return [] + } + throw error + } + try { + const parsed = StoredOutbox.safeParse(JSON.parse(raw)) + return parsed.success && parsed.data.account === this.account ? parsed.data.turnEnds : [] + } catch { + return [] + } + } + + private store(events: readonly ConnectorEvent[]): void { + const turnEnds = events.filter(isTurnEnded) + if (turnEnds.length === 0) { + rmSync(this.filePath, { force: true }) + } else { + writeSecureJsonFile(this.filePath, { account: this.account, turnEnds }) + } + } +} diff --git a/src/main/connector-cloud/fake-connector-cloud.test-fixture.ts b/src/main/connector-cloud/fake-connector-cloud.test-fixture.ts new file mode 100644 index 00000000000..d8cfbfddf89 --- /dev/null +++ b/src/main/connector-cloud/fake-connector-cloud.test-fixture.ts @@ -0,0 +1,200 @@ +// An in-process Orca Cloud connector speaking the real wire shapes, with tokens minted per account. +import { randomBytes } from 'node:crypto' +import { CONNECTOR_CLIENT_LIMITS, CONNECTOR_PATHS } from '../../shared/connector-cloud-contract' +import type { ConnectorResult } from './connector-cloud-request' +import type { ConnectorToken, ConnectorTokenMint } from './connector-transport' + +export type Reply = { status: number; body?: unknown; headers?: Record } + +const TOKEN_TTL_MS = 5 * 60_000 + +export class FakeConnectorCloud { + readonly origin = 'https://connector.test' + readonly requests: { path: string; body: unknown; token: string | null }[] = [] + /** Forces a status on every request to these paths, e.g. 404 from an older cloud. */ + readonly forced = new Map() + online = true + /** How long a poll answer takes to arrive once it is ready. */ + pollReplyDelayMs = 0 + /** Requests never answer, as on a stalled network. */ + hang = false + /** Requests to these paths wait for `release(path)`, then are judged against the cloud's state then. */ + readonly holdBefore = new Set() + /** Requests to these paths take effect at once; their answers wait for `release(path)`. */ + readonly holdAfter = new Set() + private readonly held = new Map void)[]>() + /** Accounts the auth service still mints for; removing one is a sign-out elsewhere. */ + readonly signedIn = new Set(['ada']) + /** Every token minted, in order. */ + readonly minted: { account: string; token: string }[] = [] + private readonly tokens = new Map() + // The hello the cloud keeps, and the highest event batch applied under it. + private hello: { rev: number; seq: number } | null = null + // The last revision handed out; it outlives a lost hello, as the host row does. + private lastRev = 0 + /** Every hello, ack and event the machine sent, in order. */ + readonly hellos: unknown[] = [] + readonly acks: unknown[] = [] + readonly events: unknown[] = [] + /** Poll answers in order; a poll with none queued is held until one is queued. */ + private readonly pollReplies: Reply[] = [] + private heldPoll: (() => void) | null = null + + readonly fetch: typeof globalThis.fetch = async (input, init) => { + if (!this.online) { + throw new TypeError('fetch failed') + } + if (this.hang) { + return new Promise(() => {}) + } + const url = new URL(String(input)) + const path = url.pathname + if (Buffer.byteLength(String(init?.body ?? '')) > CONNECTOR_CLIENT_LIMITS.maxRequestBytes) { + return Response.json({ error: 'request_too_large' }, { status: 413 }) + } + const body: unknown = init?.body ? JSON.parse(String(init.body)) : undefined + const auth = new Headers(init?.headers).get('authorization') + const token = auth?.startsWith('Bearer ') ? auth.slice(7) : null + this.requests.push({ path, body, token }) + if (this.holdBefore.has(path)) { + await this.waitForRelease(path) + } + const reply = this.forced.get(path) ?? (await this.route(path, body, token, init?.signal)) + if (this.holdAfter.has(path)) { + await this.waitForRelease(path) + } + return new Response(reply.body === undefined ? null : JSON.stringify(reply.body), { + status: reply.status, + headers: { 'content-type': 'application/json', ...reply.headers } + }) + } + + /** The auth service's connector-token route for one account, minus the HTTP. */ + mintFor(account: string): ConnectorTokenMint { + return async (): Promise> => { + if (!this.signedIn.has(account)) { + return { ok: false, status: 401, error: 'signed_out', retryAfterMs: null } + } + const connectorToken = randomBytes(16).toString('base64url') + const expiresAt = Date.now() + TOKEN_TTL_MS + this.tokens.set(connectorToken, { account, expiresAt }) + this.minted.push({ account, token: connectorToken }) + return { ok: true, value: { connectorToken, expiresAt } } + } + } + + requestsTo(path: string): { body: unknown; token: string | null }[] { + return this.requests.filter((request) => request.path === path) + } + + /** Answers the next poll; `status` other than 200 is an error reply. */ + answerPoll(reply: Reply): void { + this.pollReplies.push(reply) + this.heldPoll?.() + } + + /** Lets every held request or answer for `path` go on, and stops holding new ones. */ + release(path: string): void { + this.holdBefore.delete(path) + this.holdAfter.delete(path) + for (const resume of this.held.get(path)?.splice(0) ?? []) { + resume() + } + } + + private waitForRelease(path: string): Promise { + return new Promise((resolve) => { + const waiting = this.held.get(path) ?? [] + waiting.push(resolve) + this.held.set(path, waiting) + }) + } + + /** Refuses every token issued so far, as a key rotation would. */ + refuseTokens(): void { + this.tokens.clear() + } + + /** Another hello for this machine lands, as a retired client's late one would. */ + helloFromElsewhere(): void { + this.hello = { rev: ++this.lastRev, seq: -1 } + } + + /** Loses the hello, as a cloud restart would; the machine must say it again. */ + forgetHello(): void { + this.hello = null + } + + private async route( + path: string, + body: unknown, + token: string | null, + signal: AbortSignal | null | undefined + ): Promise { + switch (path) { + case CONNECTOR_PATHS.hello: + case CONNECTOR_PATHS.poll: + case CONNECTOR_PATHS.acks: + case CONNECTOR_PATHS.events: + return this.machineCall(path, body, token, signal) + default: + return { status: 404, body: { error: 'not_found' } } + } + } + + private async machineCall( + path: string, + body: unknown, + token: string | null, + signal: AbortSignal | null | undefined + ): Promise { + const issued = token ? this.tokens.get(token) : undefined + if (!issued || issued.expiresAt <= Date.now()) { + return { status: 401, body: { error: 'invalid_token' } } + } + const fields = isRecord(body) ? body : {} + const rev = typeof fields.helloRev === 'number' ? fields.helloRev : -1 + if (path === CONNECTOR_PATHS.hello) { + this.hellos.push(body) + // Every hello gets the next revision and a fresh event mark. + this.hello = { rev: ++this.lastRev, seq: -1 } + return { status: 200, body: { helloRev: this.hello.rev } } + } + const { hello } = this + // Anything under another hello is refused and applies nothing. + if (!hello || rev !== hello.rev) { + return { status: 409, body: { error: 'hello_required' } } + } + if (path === CONNECTOR_PATHS.acks) { + this.acks.push(...(Array.isArray(fields.acks) ? fields.acks : [])) + return { status: 204 } + } + if (path === CONNECTOR_PATHS.events) { + // Applied only above this hello's high-water mark; a late or repeated batch is dropped. + const seq = typeof fields.seq === 'number' ? fields.seq : -1 + if (seq > hello.seq) { + hello.seq = seq + this.events.push(...(Array.isArray(fields.events) ? fields.events : [])) + } + return { status: 204 } + } + while (this.pollReplies.length === 0) { + await new Promise((resolve, reject) => { + this.heldPoll = resolve + signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')), { + once: true + }) + }) + this.heldPoll = null + } + const reply = this.pollReplies.shift() ?? { status: 500 } + if (this.pollReplyDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, this.pollReplyDelayMs)) + } + return reply + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null +} diff --git a/src/main/connector-cloud/fake-connector-seam.test-fixture.ts b/src/main/connector-cloud/fake-connector-seam.test-fixture.ts new file mode 100644 index 00000000000..fc6bb4af6b5 --- /dev/null +++ b/src/main/connector-cloud/fake-connector-seam.test-fixture.ts @@ -0,0 +1,74 @@ +// A stand-in for the connector seam and the machine around it, recording each call a client makes. +import { vi } from 'vitest' +import type { ConnectorProject } from '../../shared/connector-cloud-contract' +import type { + ConversationEvent, + ConversationWatchListener +} from '../external-conversations/external-conversation-events' +import type { ExternalConversationSeam } from '../external-conversations/external-conversation-seam' +import type { Delivery } from '../external-conversations/external-conversation-types' +import type { ConnectorMachine } from './connector-machine' + +export type FakeSeam = ReturnType + +/** The seam calls a client makes, answered from `outcomes` (accepted by default). */ +export function fakeSeam() { + const disabled = new AbortController() + const watchers = new Set() + const outcomes = new Map>() + const recorded = new Map() + const outcome = async (id: string): Promise => + (await outcomes.get(id)) ?? { outcome: 'accepted' } + const seam = { + signal: disabled.signal, + start: vi.fn(async (_host: unknown, key: unknown, id: string) => ({ + conversation: key, + initial: await outcome(id) + })), + send: vi.fn((_host: unknown, _key: unknown, id: string) => outcome(id)), + answer: vi.fn((): Promise => { + throw new Error('external_conversation_not_found') + }), + cancel: vi.fn((_host: unknown, _key: unknown, id: string) => outcome(id)), + recorded: vi.fn(async (_key: unknown, id: string) => recorded.get(id) ?? null), + watch: vi.fn((_host: unknown, listener: ConversationWatchListener) => { + listener({ kind: 'snapshot', conversations: [] }) + watchers.add(listener) + return () => watchers.delete(listener) + }) + } + return { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the client calls only these seam members. + seam: seam as unknown as ExternalConversationSeam, + calls: seam, + outcomes, + recorded, + emit: (event: ConversationEvent) => watchers.forEach((listener) => listener(event)), + disable: () => disabled.abort('connectors_disabled') + } +} + +/** `current` is the seam while connectors are on, null while they are off. */ +export function fakeMachine( + current: () => FakeSeam | null, + projects: ConnectorProject[] = [] +): ConnectorMachine & { toggle: () => void; agentList: string[] } { + const toggled = new Set<() => void>() + const machine: ConnectorMachine & { toggle: () => void; agentList: string[] } = { + /** What `agents()` reports; tests change it to model a settings change. */ + agentList: ['claude', 'codex'], + runtime: 'desktop', + hostName: 'box', + orcaVersion: '1.4.250', + seam: () => current()?.seam ?? null, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the fake seam never reads its host. + conversationHost: () => ({}) as ReturnType, + agents: () => machine.agentList, + projects: () => projects, + onConnectorsToggled: (listener) => { + toggled.add(listener) + }, + toggle: () => toggled.forEach((listener) => listener()) + } + return machine +} diff --git a/src/main/external-conversations/external-conversation-delivery-receipts.test.ts b/src/main/external-conversations/external-conversation-delivery-receipts.test.ts index 787d3fa631d..6faee10c156 100644 --- a/src/main/external-conversations/external-conversation-delivery-receipts.test.ts +++ b/src/main/external-conversations/external-conversation-delivery-receipts.test.ts @@ -138,3 +138,39 @@ describe('DeliveryReceipts ledger capacity', () => { db.close() }) }) + +describe('DeliveryReceipts.recorded', () => { + it('replays a result, reads an interrupted delivery as unknown, and runs nothing', async () => { + const { path, db } = fileDb() + const receipts = new DeliveryReceipts(() => db) + await receipts.run(KEY, 'done', 'connector.send', ['hi'], async () => ({ outcome: 'accepted' })) + void receipts.run(KEY, 'interrupted', 'connector.send', ['hi'], never) + await receipts.run(KEY, 'refused', 'connector.send', ['hi'], async () => ({ + outcome: 'refused', + reason: 'agent_not_reachable' + })) + db.close() + + const reopened = new OrchestrationDb(path) + const restarted = new DeliveryReceipts(() => reopened) + await expect(restarted.recorded(KEY, 'done')).resolves.toEqual({ outcome: 'accepted' }) + await expect(restarted.recorded(KEY, 'interrupted')).resolves.toEqual({ + outcome: 'unknown', + reason: 'replayed_after_restart' + }) + await expect(restarted.recorded(KEY, 'refused')).resolves.toBeNull() + await expect(restarted.recorded(KEY, 'never-received')).resolves.toBeNull() + reopened.close() + }) + + it('answers a delivery still running in this process with its own result', async () => { + const { db } = fileDb() + const receipts = new DeliveryReceipts(() => db) + const { promise, resolve } = Promise.withResolvers() + void receipts.run(KEY, 'running', 'connector.send', ['hi'], () => promise) + const recorded = receipts.recorded(KEY, 'running') + resolve({ outcome: 'accepted' }) + await expect(recorded).resolves.toEqual({ outcome: 'accepted' }) + db.close() + }) +}) diff --git a/src/main/external-conversations/external-conversation-delivery-receipts.ts b/src/main/external-conversations/external-conversation-delivery-receipts.ts index 15cf3a1c18b..035624ffa84 100644 --- a/src/main/external-conversations/external-conversation-delivery-receipts.ts +++ b/src/main/external-conversations/external-conversation-delivery-receipts.ts @@ -47,12 +47,7 @@ export class DeliveryReceipts { payloadHash: hashCanonical(payload) } const flightKey = JSON.stringify([identity.callerFingerprint, identity.requestId]) - const db = this.db() - if (!this.leftoversSettled) { - // Settle old pending rows as unknown; deleting them could repeat effects they already wrote. - db.settlePendingMutationReceipts(CONNECTOR_FINGERPRINTS, JSON.stringify(LOST)) - this.leftoversSettled = true - } + const db = this.settledDb() if ( !db.getMutationReceipt(identity.callerFingerprint, identity.requestId) && db.countPendingMutationReceipts(CONNECTOR_FINGERPRINTS) >= @@ -98,4 +93,30 @@ export class DeliveryReceipts { delivery.then(forget, forget) return delivery } + + /** Read-only: what a received delivery came to, or null when nothing was recorded (refusals and answers leave none). */ + recorded(key: ConversationKey, deliveryId: string): Promise { + const requestId = JSON.stringify([key.externalId, deliveryId]) + const fingerprint = callerFingerprint(key.source) + const running = this.inFlight.get(JSON.stringify([fingerprint, requestId])) + if (running) { + return running + } + const row = this.settledDb().getMutationReceipt(fingerprint, requestId) + if (row?.state !== 'completed') { + return Promise.resolve(null) + } + const recorded = RecordedDelivery.safeParse(JSON.parse(row.receipt ?? 'null')) + return Promise.resolve(recorded.success ? recorded.data : LOST) + } + + private settledDb(): DeliveryReceiptDb { + const db = this.db() + if (!this.leftoversSettled) { + // Settle old pending rows as unknown; deleting them could repeat effects they already wrote. + db.settlePendingMutationReceipts(CONNECTOR_FINGERPRINTS, JSON.stringify(LOST)) + this.leftoversSettled = true + } + return db + } } diff --git a/src/main/external-conversations/external-conversation-events.ts b/src/main/external-conversations/external-conversation-events.ts index 3b745a2f41a..978b367a5b3 100644 --- a/src/main/external-conversations/external-conversation-events.ts +++ b/src/main/external-conversations/external-conversation-events.ts @@ -30,6 +30,8 @@ export type ConversationEvent = | (ConversationPane & { kind: 'turn-started'; at: number }) | (ConversationPane & { kind: 'turn-ended' + /** The ended turn's turn-started `at`; null when its start predates this watch and the row kept none. */ + turnId: number | null /** Host verdict; null means none was recorded. */ stopReason: AgentTurnOutcome | null lastAssistantMessage?: string @@ -42,6 +44,10 @@ export type ConversationSnapshot = { agent: ConnectorAgent working: boolean prompt: PendingPrompt | null + /** The pane's latest turn, as its turn events name it; null before any. */ + turnId: number | null + /** When the conversation last showed activity: its link, or its agent's latest hook report. */ + activeAt: number })[] } @@ -184,10 +190,10 @@ export class ConversationEvents { return } this.rejected.delete(paneKey) - this.turnStartedAt.delete(paneKey) this.emit({ kind: 'turn-ended', ...pane(linked.link, linked.workspace.worktreeId), + turnId: this.takeTurnId(paneKey, row), stopReason: 'cancellation', at: activity.observedAt }) @@ -198,11 +204,18 @@ export class ConversationEvents { kind: 'snapshot', conversations: [...allLinks(this.deps.store)].map(({ workspace, link }) => { const row = this.deps.readHookRow(link.paneKey) + const working = row ? hookRowAgentPhase(row) === 'working' : false return { ...pane(link, workspace.worktreeId), agent: link.agent, - working: row ? hookRowAgentPhase(row) === 'working' : false, - prompt: this.deps.pendingPrompt(link) + working, + prompt: this.deps.pendingPrompt(link), + // The same id started() and takeTurnId() give this turn. + turnId: + this.turnStartedAt.get(link.paneKey) ?? + row?.turnStartedAt ?? + (row && working ? row.stateStartedAt : null), + activeAt: Math.max(link.createdAt, row?.receivedAt ?? 0) } }) } @@ -250,17 +263,23 @@ export class ConversationEvents { if (!link) { return } - this.turnStartedAt.delete(item.paneKey) const message = item.row.lastAssistantMessage this.emit({ kind: 'turn-ended', ...pane(link, item.worktreeId), + turnId: this.takeTurnId(item.paneKey, item.row), stopReason, ...(message ? { lastAssistantMessage: message } : {}), at: item.row.stateStartedAt }) } + private takeTurnId(paneKey: string, row: AgentStatusIpcPayload): number | null { + const turnId = this.turnStartedAt.get(paneKey) ?? row.turnStartedAt ?? null + this.turnStartedAt.delete(paneKey) + return turnId + } + private linkedFor(paneKey: string): LinkedConversation | null { for (const linked of allLinks(this.deps.store)) { if (linked.link.paneKey === paneKey) { diff --git a/src/main/external-conversations/external-conversation-prompts.test.ts b/src/main/external-conversations/external-conversation-prompts.test.ts index 37529b8e1ea..1e513ea11fd 100644 --- a/src/main/external-conversations/external-conversation-prompts.test.ts +++ b/src/main/external-conversations/external-conversation-prompts.test.ts @@ -213,7 +213,12 @@ describe('ExternalConversationSeam watch', () => { expect(seen.map((event) => event.kind)).toEqual(['turn-started', 'prompt', 'turn-ended']) expect(seen[0]).toMatchObject({ conversation: KEY, at: 100 }) expect(seen[1]).toMatchObject({ prompt: { kind: 'approval' } }) - expect(seen[2]).toMatchObject({ stopReason: 'success', lastAssistantMessage: 'Done.', at: 300 }) + expect(seen[2]).toMatchObject({ + stopReason: 'success', + lastAssistantMessage: 'Done.', + at: 300, + turnId: 100 + }) }) it('ends a turn the user stopped with its recorded verdict, never the caller’s', async () => { diff --git a/src/main/external-conversations/external-conversation-seam.ts b/src/main/external-conversations/external-conversation-seam.ts index 9a63f3ee3c4..2076a33b38e 100644 --- a/src/main/external-conversations/external-conversation-seam.ts +++ b/src/main/external-conversations/external-conversation-seam.ts @@ -150,6 +150,11 @@ export class ExternalConversationSeam { ) } + /** What an earlier process's received delivery came to, from its receipt; never runs it again. */ + recorded(key: ConversationKey, deliveryId: string): Promise { + return this.deps.receipts.recorded(key, deliveryId) + } + /** A snapshot of every linked conversation first, then its events; returns the release. */ watch(host: ExternalConversationHost, listener: ConversationWatchListener): () => void { if (this.abort.signal.aborted) { diff --git a/src/main/ipc/orca-profile-id-arg.ts b/src/main/ipc/orca-profile-id-arg.ts new file mode 100644 index 00000000000..7eaa2c1e9c0 --- /dev/null +++ b/src/main/ipc/orca-profile-id-arg.ts @@ -0,0 +1,10 @@ +export function profileIdFromArgs(args: unknown): string { + const profileId = + args && typeof args === 'object' && 'profileId' in args && typeof args.profileId === 'string' + ? args.profileId.trim() + : '' + if (!profileId) { + throw new Error('invalid_orca_profile_id') + } + return profileId +} diff --git a/src/main/ipc/orca-profiles.ts b/src/main/ipc/orca-profiles.ts index ad69c8d9af6..a69f23470c7 100644 --- a/src/main/ipc/orca-profiles.ts +++ b/src/main/ipc/orca-profiles.ts @@ -57,23 +57,14 @@ import { getOrcaCloudSignInProgress, onOrcaCloudSignInProgressChanged } from '../orca-profiles/profile-cloud-sign-in-progress' +import { profileIdFromArgs } from './orca-profile-id-arg' import { transferProjectArgsFromUnknown } from './orca-profile-project-transfer-args' type RegisterOrcaProfileHandlersOptions = { onBeforeRelaunch?: () => void | Promise onAuthMutation?: () => void onBeforeSignOut?: () => void -} - -function profileIdFromArgs(args: unknown): string { - const profileId = - args && typeof args === 'object' && 'profileId' in args && typeof args.profileId === 'string' - ? args.profileId.trim() - : '' - if (!profileId) { - throw new Error('invalid_orca_profile_id') - } - return profileId + onAfterSignOut?: () => void } function findProjectsByPathArgsFromUnknown(args: unknown): FindOrcaProfileProjectsByPathArgs { @@ -309,7 +300,7 @@ export function registerOrcaProfileHandlers( 'orcaProfiles:signOutCurrent', async (): Promise => { options.onBeforeSignOut?.() - return signOutCurrentOrcaProfile(getProfileUserDataPath()) + return signOutCurrentOrcaProfile(getProfileUserDataPath()).finally(options.onAfterSignOut) } ) diff --git a/src/main/ipc/register-core-handlers/register-core-handlers.ts b/src/main/ipc/register-core-handlers/register-core-handlers.ts index 19df2875807..c5ea208ea6e 100644 --- a/src/main/ipc/register-core-handlers/register-core-handlers.ts +++ b/src/main/ipc/register-core-handlers/register-core-handlers.ts @@ -130,6 +130,7 @@ type CoreHandlerLifecycleOptions = { onBeforeRelaunch?: () => void | Promise onOrcaProfileAuthMutation?: () => void onBeforeOrcaProfileSignOut?: () => void + onAfterOrcaProfileSignOut?: () => void getAdditionalAiVaultCodexHomePaths?: () => readonly string[] prepareAiVaultSessionResume?: ( args: AiVaultPrepareSessionResumeArgs @@ -235,7 +236,8 @@ export function registerCoreHandlers( registerOrcaProfileHandlers(store, { onBeforeRelaunch: lifecycleOptions.onBeforeRelaunch, onAuthMutation: lifecycleOptions.onOrcaProfileAuthMutation, - onBeforeSignOut: lifecycleOptions.onBeforeOrcaProfileSignOut + onBeforeSignOut: lifecycleOptions.onBeforeOrcaProfileSignOut, + onAfterSignOut: lifecycleOptions.onAfterOrcaProfileSignOut }) registerBrowserHandlers() registerShellHandlers(store) diff --git a/src/main/orca-profiles/profile-cloud-auth-config.test.ts b/src/main/orca-profiles/profile-cloud-auth-config.test.ts index 424d8da53a9..8f58d81d0e0 100644 --- a/src/main/orca-profiles/profile-cloud-auth-config.test.ts +++ b/src/main/orca-profiles/profile-cloud-auth-config.test.ts @@ -37,6 +37,7 @@ describe('Orca cloud auth config', () => { orgEndpoint: 'https://orca-cloud.example/v1/desktop/auth/org', logoutEndpoint: 'https://orca-cloud.example/v1/desktop/auth/logout', relayTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/relay-token', + connectorTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/connector-token', relayDirectorUrl: 'https://relay.onorca.dev', clientId: 'desktop-client', scope: 'openid profile email offline_access' @@ -57,6 +58,7 @@ describe('Orca cloud auth config', () => { orgEndpoint: 'https://login.onorca.dev/v1/desktop/auth/org', logoutEndpoint: 'https://login.onorca.dev/v1/desktop/auth/logout', relayTokenEndpoint: 'https://login.onorca.dev/v1/desktop/auth/relay-token', + connectorTokenEndpoint: 'https://login.onorca.dev/v1/desktop/auth/connector-token', relayDirectorUrl: 'https://relay.onorca.dev', clientId: 'orca-desktop', scope: 'openid profile email offline_access' diff --git a/src/main/orca-profiles/profile-cloud-auth-config.ts b/src/main/orca-profiles/profile-cloud-auth-config.ts index 3706436146d..55fbcc4e350 100644 --- a/src/main/orca-profiles/profile-cloud-auth-config.ts +++ b/src/main/orca-profiles/profile-cloud-auth-config.ts @@ -15,6 +15,7 @@ export type OrcaCloudAuthConfig = { orgEndpoint: string logoutEndpoint: string relayTokenEndpoint: string + connectorTokenEndpoint: string relayDirectorUrl: string clientId: string scope: string @@ -89,6 +90,9 @@ export function getOrcaCloudAuthConfig( relayTokenEndpoint: cleanEndpointUrl(env.ORCA_CLOUD_RELAY_TOKEN_URL) ?? endpoint(apiBaseUrl, '/v1/desktop/auth/relay-token'), + connectorTokenEndpoint: + cleanEndpointUrl(env.ORCA_CLOUD_CONNECTOR_TOKEN_URL) ?? + endpoint(apiBaseUrl, '/v1/desktop/auth/connector-token'), relayDirectorUrl: cleanOrigin(env.ORCA_RELAY_URL, allowLoopbackHttp) ?? PRODUCTION_RELAY_DIRECTOR_URL, clientId, diff --git a/src/main/orca-profiles/profile-cloud-client.test.ts b/src/main/orca-profiles/profile-cloud-client.test.ts index 5ff6f2e13c7..9f347110642 100644 --- a/src/main/orca-profiles/profile-cloud-client.test.ts +++ b/src/main/orca-profiles/profile-cloud-client.test.ts @@ -22,6 +22,7 @@ const config: OrcaCloudAuthConfig = { orgEndpoint: 'https://orca-cloud.example/v1/desktop/auth/org', logoutEndpoint: 'https://orca-cloud.example/v1/desktop/auth/logout', relayTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/relay-token', + connectorTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/connector-token', relayDirectorUrl: 'https://relay.example', clientId: 'desktop-client', scope: 'openid profile email offline_access' diff --git a/src/main/orca-profiles/profile-cloud-org-members-client.test.ts b/src/main/orca-profiles/profile-cloud-org-members-client.test.ts index 637d5f308bf..a7971c77bad 100644 --- a/src/main/orca-profiles/profile-cloud-org-members-client.test.ts +++ b/src/main/orca-profiles/profile-cloud-org-members-client.test.ts @@ -22,6 +22,7 @@ const config: OrcaCloudAuthConfig = { orgEndpoint: 'https://orca-cloud.example/v1/desktop/auth/org', logoutEndpoint: 'https://orca-cloud.example/v1/desktop/auth/logout', relayTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/relay-token', + connectorTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/connector-token', relayDirectorUrl: 'https://relay.example', clientId: 'desktop-client', scope: 'openid profile email offline_access' diff --git a/src/main/orca-profiles/profile-cloud-pkce.test.ts b/src/main/orca-profiles/profile-cloud-pkce.test.ts index 3f7e8f29fa9..a8a14a1d7bb 100644 --- a/src/main/orca-profiles/profile-cloud-pkce.test.ts +++ b/src/main/orca-profiles/profile-cloud-pkce.test.ts @@ -31,6 +31,7 @@ const config: OrcaCloudAuthConfig = { orgEndpoint: 'https://orca-cloud.example/v1/desktop/auth/org', logoutEndpoint: 'https://orca-cloud.example/v1/desktop/auth/logout', relayTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/relay-token', + connectorTokenEndpoint: 'https://orca-cloud.example/v1/desktop/auth/connector-token', relayDirectorUrl: 'https://relay.example', clientId: 'desktop-client', scope: 'openid profile email offline_access' diff --git a/src/main/startup/main-process-connector-startup.ts b/src/main/startup/main-process-connector-startup.ts new file mode 100644 index 00000000000..8479a3a61e7 --- /dev/null +++ b/src/main/startup/main-process-connector-startup.ts @@ -0,0 +1,22 @@ +import { createDesktopConnectorCloud } from '../connector-cloud/connector-cloud-install' +import { getProfileUserDataPath } from '../orca-profiles/profile-storage-paths' +import { mainProcessState as state } from './main-process-state' + +// Desktop only: the connector acts as the signed-in Orca account, which `orca serve` does not have yet. +// Created before the window exists, so its sign-in and sign-out hooks always find it. +export function installDesktopConnectorCloud(): void { + if (!state.runtime || !state.store) { + return + } + state.connectorCloud = createDesktopConnectorCloud( + state.runtime, + state.store, + getProfileUserDataPath(), + () => state.isQuitting + ) +} + +// Registered with the relay; like it, the first request waits for the persisted proxy. +export function connectDesktopConnectorCloud(): void { + void state.initialProxyApplicationReady.then(() => state.connectorCloud?.connect()) +} diff --git a/src/main/startup/main-process-quit.ts b/src/main/startup/main-process-quit.ts index e852d6f19fa..63e525b2249 100644 --- a/src/main/startup/main-process-quit.ts +++ b/src/main/startup/main-process-quit.ts @@ -81,6 +81,7 @@ function installBeforeQuitHandler(): void { state.isQuitting = true flushUnsavedCloudSessions() state.desktopRelayService?.fenceAndCloseNow() + state.connectorCloud?.fence() state.runtimeRpc?.setMobileRelayPairingProvider(null) state.unsubscribeAgentAwakeStatusChanges?.() state.unsubscribeAgentAwakeStatusChanges = null diff --git a/src/main/startup/main-process-runtime-launch-activation.test.ts b/src/main/startup/main-process-runtime-launch-activation.test.ts index 9597db442c5..09e6448ea02 100644 --- a/src/main/startup/main-process-runtime-launch-activation.test.ts +++ b/src/main/startup/main-process-runtime-launch-activation.test.ts @@ -88,6 +88,10 @@ vi.mock('../ipc/startup-notification-registration', () => ({ triggerStartupNotificationRegistration: vi.fn() })) vi.mock('./main-process-push-startup', () => ({ startDesktopPushService: vi.fn() })) +vi.mock('./main-process-connector-startup', () => ({ + installDesktopConnectorCloud: vi.fn(), + connectDesktopConnectorCloud: vi.fn() +})) vi.mock('./startup-diagnostics', () => ({ logStartupMilestone: vi.fn() })) const setupHeadlessServeAutoUpdater = vi.hoisted(() => vi.fn()) vi.mock('../window/main-window-updater', () => ({ setupHeadlessServeAutoUpdater })) diff --git a/src/main/startup/main-process-runtime-launch.ts b/src/main/startup/main-process-runtime-launch.ts index 2d4320047e9..c513f48d3cb 100644 --- a/src/main/startup/main-process-runtime-launch.ts +++ b/src/main/startup/main-process-runtime-launch.ts @@ -40,6 +40,10 @@ import { startAppImageCopyPreExtraction } from '../cli/appimage-copy-pre-extract import { scheduleAllPendingHistoryTreeRemovals } from '../terminal-history-deletion' import { triggerStartupNotificationRegistration } from '../ipc/startup-notification-registration' import { startDesktopPushService } from './main-process-push-startup' +import { + connectDesktopConnectorCloud, + installDesktopConnectorCloud +} from './main-process-connector-startup' import { mainProcessState as state } from './main-process-state' import { logStartupMilestone } from './startup-diagnostics' import { setupHeadlessServeAutoUpdater } from '../window/main-window-updater' @@ -228,6 +232,7 @@ async function launchDesktopMode( } // Register before proxy await so concurrent token mints wait rather than fail provider_missing; installer requests are already guarded. startDesktopRelayService(runtimeRpc) + connectDesktopConnectorCloud() // Create the window before awaiting proxy readiness; the request guard already fences fetches. await state.initialProxyApplicationReady // Apply persisted proxy before the push gateway's first request. @@ -261,6 +266,9 @@ export async function initializeMainProcessRuntimeLaunch( state.serveOptions = serveOptions installCloudSessionShutdownFlush() const runtimeRpc = installRuntimeRpc(runtime, serveOptions) + if (!serveOptions) { + installDesktopConnectorCloud() + } const shellPathReady = shellPathHydration.whenReady() // Publish the same shell-PATH generation to renderer Git and terminal startup barriers. state.shellPathReady = shellPathReady diff --git a/src/main/startup/main-process-state.ts b/src/main/startup/main-process-state.ts index 29bf038f868..32dcff25a08 100644 --- a/src/main/startup/main-process-state.ts +++ b/src/main/startup/main-process-state.ts @@ -14,6 +14,7 @@ import type { OrcaRuntimeService } from '../runtime/orca-runtime' import type { RateLimitService } from '../rate-limits/service' import type { OrcaRuntimeRpcServer } from '../runtime/runtime-rpc' import type { DesktopRelayService } from '../runtime/mobile-cloud-relay/desktop-relay-service' +import type { ConnectorCloud } from '../connector-cloud/connector-cloud' import type { DesktopRelayServiceInstaller } from '../runtime/mobile-cloud-relay/desktop-relay-service-installer' import type { DesktopPushService } from '../runtime/push/desktop-push-service' import type { StarNagService } from '../star-nag/service' @@ -91,6 +92,8 @@ export const mainProcessState = { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: widens the null slot to the installer assigned by main-process-relay-startup. desktopRelayInstaller: null as DesktopRelayServiceInstaller | null, desktopPushService: null as DesktopPushService | null, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: widens the null slot to the connector assigned by main-process-connector-startup. + connectorCloud: null as ConnectorCloud | null, desktopRelayStatus: 'offline' as RelayBrokerStatus, desktopRelayCellUrl: undefined as string | undefined, pendingUnpairedDeviceAuthFailure: false, diff --git a/src/main/startup/main-window-core-services.ts b/src/main/startup/main-window-core-services.ts index aa65f98d103..f220cf52ebe 100644 --- a/src/main/startup/main-window-core-services.ts +++ b/src/main/startup/main-window-core-services.ts @@ -97,18 +97,22 @@ export function attachMainWindowCoreServices( onBeforeRelaunch: async () => { state.isQuitting = true state.desktopRelayService?.fenceAndCloseNow() + state.connectorCloud?.fence() await preserveAgentAuthBeforeRestart({ codexRuntimeHome, store }) }, onOrcaProfileAuthMutation: () => { // An auth change retries relay installation after launch-time failure. void state.desktopRelayInstaller?.authChanged() state.desktopRelayService?.authMutated() + state.connectorCloud?.accountChanged() }, // Only sign-out fences paired phones; reasonless quit/relaunch must not look like logout. onBeforeOrcaProfileSignOut: () => { state.desktopRelayInstaller?.suspend() state.desktopRelayService?.fenceAndCloseNow(RELAY_HOST_CLOSE_REASON.SIGNED_OUT) - } + state.connectorCloud?.signOutStarted() + }, + onAfterOrcaProfileSignOut: () => state.connectorCloud?.signOutFinished() }, state.pluginService ?? undefined, state.pluginMarketplaceService && state.pluginMarketplaceInstaller diff --git a/src/shared/connector-cloud-contract.ts b/src/shared/connector-cloud-contract.ts index 389dc26bdbd..5873ce5dc61 100644 --- a/src/shared/connector-cloud-contract.ts +++ b/src/shared/connector-cloud-contract.ts @@ -1,4 +1,6 @@ // Mirror of cloud/packages/connector-contract/src for the machine side of the Orca Cloud connector. +import { z } from 'zod' +import type { PendingPrompt } from './agent-pending-prompt' export const CONNECTOR_CONTRACT_VERSION = 1 @@ -16,5 +18,114 @@ export const CONNECTOR_FEATURES = ['answer', 'cancel', 'reply'] as const export const CONNECTOR_CLIENT_LIMITS = { pollTimeoutMs: 70_000, minDeadlineLeftMs: 10_000, - maxLastAssistantMessageBytes: 8 * 1024 + maxLastAssistantMessageBytes: 8 * 1024, + // The cloud refuses a larger request body with 413. + maxRequestBytes: 256 * 1024, + // One event's JSON, so any single event fits in an events request. + maxEventBytes: 192 * 1024, + // A snapshot lists linked conversations active in the last 7 days, newest first, up to this many. + maxSnapshotConversations: 200, + snapshotActiveWindowMs: 7 * 24 * 60 * 60 * 1000, + // A hello lists the most recently used projects, up to this many, and always fits maxRequestBytes. + maxHelloProjects: 500, + maxDisplayTextLength: 200 } as const + +// Machine-side readers: they keep what this Orca uses and ignore fields a newer cloud adds. +export const ConnectorErrorResponse = z.object({ error: z.string().min(1) }) + +// Each hello gets a revision above every earlier one for this machine; polls, acks and events carry it. +export const ConnectorHelloResponse = z.object({ helloRev: z.number().int().positive() }) + +const ConnectorConversation = z.object({ + source: z.literal('slack'), + externalId: z.string().min(1) +}) +export type ConnectorConversation = z.infer + +const DeliveryBase = { + deliveryId: z.string().min(1), + expiresAt: z.number().int().positive(), + conversation: ConnectorConversation +} + +export const ConnectorDelivery = z.discriminatedUnion('kind', [ + z.object({ + ...DeliveryBase, + kind: z.literal('start'), + agent: z.string().min(1), + placeId: z.string().min(1), + prompt: z.string().min(1) + }), + z.object({ ...DeliveryBase, kind: z.literal('send'), text: z.string().min(1) }), + z.object({ + ...DeliveryBase, + kind: z.literal('answer'), + promptId: z.string().min(1), + optionId: z.string().min(1) + }), + z.object({ ...DeliveryBase, kind: z.literal('cancel') }) +]) +export type ConnectorDelivery = z.infer + +/** Enough of a delivery this Orca cannot read to settle it as unsupported. */ +export const ConnectorDeliveryId = z.object({ deliveryId: z.string().min(1) }) + +// Each delivery and the control are read one by one, so one this Orca cannot read never drops the rest. +export const ConnectorPollResponse = z.object({ + serverNow: z.number().int().positive(), + holdMs: z.number().int().nonnegative(), + unsettled: z + .array(z.object({ deliveryId: z.string().min(1), conversation: ConnectorConversation })) + .optional(), + deliveries: z.array(z.unknown()), + control: z.unknown().optional() +}) + +export const ConnectorPollControl = z.union([ + z.object({ stop: z.string().min(1) }), + z.object({ pausedForMs: z.number().int().positive() }), + z.object({ reconnect: z.literal(true) }) +]) +export type ConnectorPollControl = z.infer + +export type ConnectorProject = { + placeId: string + name: string + remote?: string + kind: 'git' | 'folder' + where: { kind: 'local' | 'ssh' | 'wsl'; label?: string } +} + +export type ConnectorDeliveryResult = + | { outcome: 'accepted'; reopened?: true } + | { outcome: 'refused' | 'unknown'; reason: string } + | { outcome: 'expired' | 'unsupported' } + +export type ConnectorAck = + | { deliveryId: string; step: 'received' } + | { deliveryId: string; step: 'done'; result: ConnectorDeliveryResult } + +type PaneEvent = { conversation: ConnectorConversation; paneKey: string } + +export type ConnectorEvent = + | (PaneEvent & { kind: 'linked'; agent: string }) + | (PaneEvent & { kind: 'prompt'; prompt: PendingPrompt }) + | (PaneEvent & { kind: 'turn-started'; at: number }) + | (PaneEvent & { + kind: 'turn-ended' + turnId: number + at: number + stopReason: string | null + lastAssistantMessage?: string + }) + | { + kind: 'snapshot' + conversations: (PaneEvent & { + agent: string + working: boolean + prompt: PendingPrompt | null + // The pane's latest turn; a turn end older than it changes nothing. + turnId?: number + })[] + }