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 + })[] + }