diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/summary.md b/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/summary.md new file mode 100644 index 000000000..b04d0f9c8 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/summary.md @@ -0,0 +1,23 @@ +# Trajectory: Publish fleet and attach subpaths and forward sandbox readonlyPaths + +> **Status:** ❌ Abandoned +> **Started:** September 18, 2026 at 03:11 PM +> **Completed:** September 18, 2026 at 03:21 PM + +--- + +## Key Decisions + +### Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics +- **Chose:** Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics +- **Reasoning:** Reviewed plan identifies Cloud as the consumer import target, no remote exit-code protocol, and no server mount handler in this repository. D4 running-agent orchestration remains pending user scope clarification. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics: Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics +- Abandoned: Attach extraction and readonlyPaths client implementation ready; full spawnFleetSandbox orchestration awaits reviewed-plan.md D4 scope decision. Server chmod enforcement is outside this repository. diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/trajectory.json b/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/trajectory.json new file mode 100644 index 000000000..7b06f1256 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_qdtkdvki0xoh/trajectory.json @@ -0,0 +1,54 @@ +{ + "id": "traj_qdtkdvki0xoh", + "version": 1, + "task": { + "title": "Publish fleet and attach subpaths and forward sandbox readonlyPaths" + }, + "status": "abandoned", + "startedAt": "2026-09-18T15:11:05.520Z", + "completedAt": "2026-09-18T15:21:12.959Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-09-18T15:20:29.731Z" + } + ], + "chapters": [ + { + "id": "chap_jnv5kvm9t42d", + "title": "Work", + "agentName": "default", + "startedAt": "2026-09-18T15:20:29.731Z", + "endedAt": "2026-09-18T15:21:12.959Z", + "events": [ + { + "ts": 1789744829731, + "type": "decision", + "content": "Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics: Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics", + "raw": { + "question": "Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics", + "chosen": "Implement shared attach transport in Cloud with SDK re-exports; preserve ESM exports and inferred completion semantics", + "alternatives": [], + "reasoning": "Reviewed plan identifies Cloud as the consumer import target, no remote exit-code protocol, and no server mount handler in this repository. D4 running-agent orchestration remains pending user scope clarification." + }, + "significance": "high" + }, + { + "ts": 1789744872960, + "type": "note", + "content": "Abandoned: Attach extraction and readonlyPaths client implementation ready; full spawnFleetSandbox orchestration awaits reviewed-plan.md D4 scope decision. Server chmod enforcement is outside this repository.", + "significance": "high" + } + ] + } + ], + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "c97830275c0849f8ae1ed7fc1f347a5198124b11", + "endRef": "c97830275c0849f8ae1ed7fc1f347a5198124b11" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index a0c4a641a..6fe86ce50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,13 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Minor] + +### Added + +- `@agent-relay/cloud/attach` and `@agent-relay/sdk/attach` expose fleet terminal attachment with a private local stdio socket and completion promise. +- `@agent-relay/cloud/fleet` and `@agent-relay/sdk/fleet` expose existing sandbox ensure and deletion primitives. +- Fleet sandbox ensure and `fleet spawn --sandbox-readonly-path` forward explicit read-only Relayfile subtree requests to Cloud for server-side enforcement. ## [12.2.4] - 2026-09-18 diff --git a/README.md b/README.md index 489ad5157..129efe2fa 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,7 @@

TypeScript Tests Last commit npm version Downloads License

# Infrastructure for coding agents + Tired of copy/pasting from Claude Code into Slack so your coworker can paste it into their agent? Ever put an important rule in a skill or AGENTS.md, only for the agent to completely ignore it? @@ -13,34 +14,40 @@ Agent Relay is an open-source toolkit for problems like these. It gives engineer Use the pieces you need, or combine them to build workflows across agents, tools, people, and machines. ### Messaging + Claude/Codex/etc can talk directly through shared channels, threads, DMs, files, search, and real-time events. Agents can run on different machines and still coordinate in the same workspace. [Read the docs](https://agentrelay.com/docs/introduction) ### Integrations + GitHub, Linear, Notion, Slack, and other tools are exposed as a virtual filesystem. Agents use ls, cat, grep, and ordinary file writes to work with them. [Peep the open source repo](https://github.com/agentworkforce/relayfile) ### Shared Sessions + Capture coding agent sessions so your team and their agents can search previous work, decisions, and context. [How we capture sessions](https://github.com/agentworkforce/relayhistory)
[How we capture decisions](https://github.com/agentworkforce/trajectories) ### Flows -Turn instructions you hope an agent follows into workflows you can enforce. + +Turn instructions you hope an agent follows into workflows you can enforce. Define multi-step workflows in TypeScript with deterministic checks, required steps, and human gates. Put the rules that matter in code instead of relying on a skill or prompt to be remembered and followed. [Learn how write a flow](https://github.com/agentworkforce/flows) (or lets be honest, show your agent how) ## Getting Started -The easiest way to get started is to use [Agent Relay Cloud](https://agentrelay.com/flows). + +The easiest way to get started is to use [Agent Relay Cloud](https://agentrelay.com/flows). You don't need a credit card and you can explore all the pieces without setting up any infrastructure. ### Self Hosting + Agent Relay has self hosting options for each primitive. We're happy to help you set up the whole system on your environment, just reach out to our team hi(at)agentrelay.com and we'll walk you through it. ## License diff --git a/package-lock.json b/package-lock.json index e2264cad8..e986031aa 100644 --- a/package-lock.json +++ b/package-lock.json @@ -14080,12 +14080,14 @@ "@agent-relay/config": "12.2.3", "@aws-sdk/client-s3": "3.1020.0", "ignore": "^7.0.5", - "tar": "^7.5.21" + "tar": "^7.5.21", + "ws": "^8.18.3" }, "devDependencies": { "@types/node": "^22.19.3", "@types/ssh2": "^1.15.5", - "vitest": "^4.1.0" + "vitest": "^4.1.0", + "@types/ws": "^8.18.1" }, "engines": { "node": ">=22.0.0" @@ -14205,7 +14207,8 @@ "dependencies": { "@relaycast/sdk": "^8.0.7", "@relaycast/types": "^8.0.7", - "zod": "^4.4.3" + "zod": "^4.4.3", + "@agent-relay/cloud": "12.2.4" }, "devDependencies": { "@types/node": "^22.13.10" diff --git a/packages/cli/README.md b/packages/cli/README.md index 44e019ca0..73cfae0e6 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -395,6 +395,19 @@ agent-relay fleet spawn claude \ --task 'Review the live draft under /workspace/live-review/run-123' ``` +Use `--sandbox-readonly-path '/reference/**'` to request a read-only Relayfile +subtree. It requires `--sandbox` with Relayfile enabled and accepts explicit +`/path/**` subtrees (no traversal or other wildcards). This client forwards the +request; the deployed Cloud ensure handler must implement chmod enforcement. + +The ESM `@agent-relay/cloud/attach` and `@agent-relay/sdk/attach` entries expose +`startFleetNodeAttachProxy({ nodeId, mode })`. An omitted agent is discovered +only when the node has exactly one agent. `socketPath` accepts one raw stdio +connection; `close()` removes it. `finished` reports an inferred status (0 for +terminal closure or detach, 1 for transport failure), because Relaycast does +not currently transmit the remote harness exit code. Existing CLI broker +transport fields remain available for compatibility. + `--session-ref` is a real CLI resume, not a logical collaboration label. Pass the actual Claude session ID or Codex thread ID and target its origin node. `--cwd` must name an absolute directory that exists on the selected node. The diff --git a/packages/cli/src/cli/commands/fleet.test.ts b/packages/cli/src/cli/commands/fleet.test.ts index e1414923d..27ac2882b 100644 --- a/packages/cli/src/cli/commands/fleet.test.ts +++ b/packages/cli/src/cli/commands/fleet.test.ts @@ -983,6 +983,8 @@ describe('fleet command support', () => { REPLAY_SANDBOX_NAME, '--sandbox-relayfile-path', '/live-review/run-123/**', + '--sandbox-readonly-path', + '/live-review/run-123/reference/**', '--name', 'sandbox-worker', '--task', @@ -999,6 +1001,7 @@ describe('fleet command support', () => { maxAgents: 1, mountRelayfile: true, relayfilePaths: ['/live-review/run-123/**'], + readonlyPaths: ['/live-review/run-123/reference/**'], sandboxId: REPLAY_SANDBOX_ID, forceProvision: true, providerId: 'agent37', @@ -3962,3 +3965,38 @@ describe('fleet command support', () => { expect(errors.join('\n')).toMatch(/relay cloud enroll/); }); }); + +describe('sandbox read-only path guards', () => { + it.each([[], ['--sandbox', '--no-sandbox-relayfile']])( + 'rejects unavailable mounts (%j)', + async (...flags) => { + const ensureCloudFleetSandbox = vi.fn(); + const error = vi.fn(); + const program = new Command(); + program.exitOverride(); + registerFleetCommands(program, { + ensureCloudFleetSandbox, + sdk: { error, exit: vi.fn() as never }, + }); + await program.parseAsync( + [ + 'fleet', + 'spawn', + 'claude', + ...flags, + '--sandbox-readonly-path', + '/reference/**', + '--name', + 'worker', + '--task', + 'review', + ], + { from: 'user' } + ); + expect(error).toHaveBeenCalledWith( + expect.stringContaining('--sandbox-readonly-path requires --sandbox') + ); + expect(ensureCloudFleetSandbox).not.toHaveBeenCalled(); + } + ); +}); diff --git a/packages/cli/src/cli/commands/fleet.ts b/packages/cli/src/cli/commands/fleet.ts index 50ab4f2f3..c0eb0f554 100644 --- a/packages/cli/src/cli/commands/fleet.ts +++ b/packages/cli/src/cli/commands/fleet.ts @@ -6,13 +6,13 @@ import { findProjectRoot } from '@agent-relay/config'; import { CloudFleetSandboxProvisionError, deleteCloudFleetSandbox, - ensureCloudFleetSandbox, materializeCloudRelayfileRepository, resolveWorkspaceByKey, type CloudFleetSandboxProviderId, type CloudRelayfileRepositoryMaterialization, type EnsureCloudFleetSandboxResult, } from '@agent-relay/cloud'; +import { ensureCloudFleetSandbox } from '@agent-relay/sdk/fleet'; import { HarnessDriverClient } from '@agent-relay/harness-driver'; import { createWorkspaceClient, @@ -331,6 +331,10 @@ export function registerFleetCommands( '--sandbox-relayfile-path ', 'Mount only these Relayfile subtrees (each path must end in /**)' ) + .option( + '--sandbox-readonly-path ', + 'Read-only Relayfile subtrees (/path/**); requires Cloud enforcement' + ) .option('--no-sandbox-relayfile', 'Provision the sandbox without mounting Relayfile') .option('--channel ', 'Channel for the worker to join') .option('--persona ', 'Worker persona (automatic placement)') @@ -397,6 +401,12 @@ export function registerFleetCommands( throw new Error('--sandbox-provider must be daytona, e2b, or agent37.'); } const mountSandboxRelayfile = options.sandboxRelayfile !== false; + const sandboxReadonlyPaths = optionalTextList(options.sandboxReadonlyPath, 'Sandbox read-only path'); + if (sandboxReadonlyPaths && (!useSandbox || !mountSandboxRelayfile)) { + throw new Error( + '--sandbox-readonly-path requires --sandbox and cannot be combined with --no-sandbox-relayfile.' + ); + } const sandboxRelayfilePaths = optionalTextList(options.sandboxRelayfilePath, 'Sandbox Relayfile path'); if (useSandbox && targetNode) { throw new Error('--sandbox cannot be combined with --node or --target-node.'); @@ -591,6 +601,7 @@ export function registerFleetCommands( requiredCapability: `spawn:${cli}`, maxAgents: 1, mountRelayfile: mountSandboxRelayfile, + ...(sandboxReadonlyPaths === undefined ? {} : { readonlyPaths: sandboxReadonlyPaths }), ...(liveRepository ? { relayfilePaths: liveRelayfileMountPaths(liveRepository, sandboxRelayfilePaths) } : sandboxRelayfilePaths === undefined diff --git a/packages/cli/src/cli/lib/attach-fleet-node.ts b/packages/cli/src/cli/lib/attach-fleet-node.ts index 634b9aaae..71c417be1 100644 --- a/packages/cli/src/cli/lib/attach-fleet-node.ts +++ b/packages/cli/src/cli/lib/attach-fleet-node.ts @@ -1,1222 +1 @@ -/** - * Ticketed fleet-node attach adapter. - * - * The established attach clients intentionally continue to speak the local - * broker HTTP/WebSocket contract. This short-lived loopback adapter maps that - * contract onto Relaycast's authenticated terminal session, so view/drive and - * passthrough retain their behaviour without exposing a remote broker listener - * or copying a broker API key off a physical or Daytona node. - */ - -import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'; -import { once } from 'node:events'; -import { Buffer } from 'node:buffer'; -import { randomBytes } from 'node:crypto'; - -import WebSocket, { WebSocketServer } from 'ws'; - -import { AGENT37_RELAYCAST_ORIGIN, CANONICAL_RELAYCAST_ORIGIN } from '@agent-relay/cloud'; -import type { AttachMode } from './attach-mode.js'; -import { collectWithRetry } from './collect-with-retry.js'; -import { resolveWorkspaceTransport } from './sdk-client.js'; - -const MAX_BUFFERED_BYTES = 1024 * 1024; -const MAX_WEBSOCKET_CLOSE_REASON_BYTES = 123; -// Use the same finite 30s request window as the broker's Relaycast HTTP calls. -// Terminal-session creation can exceed the shorter startup-handshake latency -// under load, and this POST cannot be replayed safely after an ambiguous client -// timeout because allocation may have completed server-side. -const SESSION_REQUEST_TIMEOUT_MS = 30_000; -// Immediate structured reachability failures still receive the complete -// five-attempt/31.2s retry schedule, while slow responses cannot multiply the -// per-attempt timeout into a roughly three-minute CLI hang. -const SESSION_REQUEST_TOTAL_TIMEOUT_MS = 90_000; -const SESSION_REQUEST_RETRIES = 4; -// The fleet node publishes liveness every 12s. With four retries, the helper's -// deterministic delays are 6s, 7.2s, 8.4s, and 9.6s: 31.2s total. That spans -// more than two heartbeat intervals and matches the established terminal -// transport's bounded recovery window instead of exhausting every retry inside -// the same stale control-plane read. -const SESSION_REQUEST_RETRY_DELAY_MS = 6_000; -const TERMINAL_CONNECT_TIMEOUT_MS = 10_000; -const INITIAL_RECONNECT_DELAY_MS = 500; -const MAX_RECONNECT_DELAY_MS = 30_000; -// Six delays (0.5s + 1s + 2s + 4s + 8s + 16s) cover the node terminal -// transport's independent 30s maximum reconnect backoff without leaving this -// client unbounded. Five attempts previously stopped after only 15.5s. -const MAX_RECONNECT_ATTEMPTS = 6; -// Bounds the acknowledgement for a delivery-mode command after readiness. -// The command is not replayable across reconnects, so this is intentionally a -// separate post-readiness phase rather than another full reconnect window. -const DELIVERY_MODE_TIMEOUT_MS = 10_000; -// The caller starts its HTTP deadline before the loopback handler starts its -// readiness timer. Leave a small response-delivery margin so the proxy's -// actionable readiness error wins that race. -const LOOPBACK_REQUEST_TIMEOUT_MARGIN_MS = 1_000; - -type FleetSessionResponse = { - ok?: boolean; - data?: { - session_id?: string; - terminal_url?: string; - resume_token?: string; - }; - error?: { code?: string; message?: string }; -}; - -type TerminalFrame = Record & { type?: string; session_id?: string }; - -type TerminalReadiness = { - generation: number; - settled: boolean; - promise: Promise; - resolve: () => void; - reject: (error: Error) => void; -}; - -export interface FleetNodeAttachOptions { - agent: string; - node: string; - mode: AttachMode; - env?: NodeJS.ProcessEnv; - baseUrl?: string; - workspaceKey?: string; - fetch?: typeof globalThis.fetch; - /** Deterministic test seam for the bounded session-request retry delay. */ - sessionRequest?: { - timeoutMs?: number; - totalTimeoutMs?: number; - sleep?: (ms: number) => Promise; - }; - /** Deterministic test seam for established-session reconnect timing. */ - reconnectDelay?: { - initialMs?: number; - maxMs?: number; - handshakeTimeoutMs?: number; - readyTimeoutMs?: number; - beforeReadyTimeoutTerminate?: (socket: WebSocket) => void; - }; -} - -export interface FleetNodeAttachProxy { - brokerUrl: string; - apiKey: string; - requestTimeoutMs: number; - close(): Promise; -} - -export class FleetNodeAttachError extends Error { - constructor( - message: string, - readonly code?: string - ) { - super(message); - this.name = 'FleetNodeAttachError'; - } -} - -const TRUSTED_RELAYCAST_ORIGINS = new Set([CANONICAL_RELAYCAST_ORIGIN, AGENT37_RELAYCAST_ORIGIN]); - -export function validateFleetAttachBaseUrl(value: string): string { - let parsed: URL; - try { - parsed = new URL(value); - } catch { - throw new FleetNodeAttachError('Fleet node attach requires a trusted Relaycast origin.'); - } - if ( - parsed.protocol !== 'https:' || - parsed.username || - parsed.password || - parsed.port || - parsed.search || - parsed.hash || - (parsed.pathname !== '' && parsed.pathname !== '/') || - !TRUSTED_RELAYCAST_ORIGINS.has(parsed.origin) - ) { - throw new FleetNodeAttachError('Fleet node attach requires a trusted Relaycast origin.'); - } - return parsed.origin; -} - -class TerminalSessionAttemptError extends FleetNodeAttachError { - constructor( - message: string, - code: string | undefined, - readonly status: number | undefined, - readonly retryable: boolean, - readonly completionUnknown: boolean - ) { - super(message, code); - } -} - -/** - * Canonical HTTP status for a terminal failure code. - * - * `agent_not_found` must stay 404: {@link switchInboundDeliveryModeOrAbort} - * only emits the "no agent named X" message and the cross-node placement hint - * on a 404, so collapsing it into 503 replaces actionable guidance with an - * opaque unreachable-node error. `unsupported_runtime` stays 409. Everything - * else is a transport-level failure and reports 503. - */ -function terminalErrorStatus(code: string | undefined): number { - if (code === 'agent_not_found') return 404; - if (code === 'unsupported_runtime') return 409; - return 503; -} - -function json(response: ServerResponse, status: number, payload: unknown): void { - response.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); - response.end(JSON.stringify(payload)); -} - -function readBody(request: IncomingMessage): Promise> { - return new Promise((resolve) => { - let settled = false; - const finish = (body: Record) => { - if (settled) return; - settled = true; - resolve(body); - }; - let body = ''; - request.setEncoding('utf8'); - request.on('data', (chunk: string) => { - body += chunk; - }); - request.on('end', () => { - try { - const parsed = JSON.parse(body) as unknown; - finish( - parsed && typeof parsed === 'object' && !Array.isArray(parsed) - ? (parsed as Record) - : {} - ); - } catch { - finish({}); - } - }); - request.on('error', () => finish({})); - request.on('aborted', () => finish({})); - }); -} - -function asWsUrl(value: string): string { - const lower = value.toLowerCase(); - if (lower.startsWith('https://')) return 'wss://' + value.slice(8); - if (lower.startsWith('http://')) return 'ws://' + value.slice(7); - return value; -} - -function safeNodePath(node: string): string { - const trimmed = node.trim().replace(/^#/, ''); - if (!trimmed) throw new FleetNodeAttachError('Error: --node requires a node name or id.', 'invalid_node'); - return encodeURIComponent(trimmed); -} - -function parseFrame(data: WebSocket.RawData): TerminalFrame | null { - try { - const parsed = JSON.parse(rawDataToString(data)) as unknown; - return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as TerminalFrame) : null; - } catch { - return null; - } -} - -function rawDataToString(data: WebSocket.RawData): string { - if (Buffer.isBuffer(data)) return data.toString('utf8'); - if (data instanceof ArrayBuffer) return Buffer.from(data).toString('utf8'); - if (Array.isArray(data)) return Buffer.concat(data).toString('utf8'); - return String(data); -} - -function diagnosticEndpoint(value: string): string { - try { - const url = new URL(value); - url.username = ''; - url.password = ''; - url.search = ''; - url.hash = ''; - return url.toString(); - } catch { - return '(invalid endpoint)'; - } -} - -function diagnosticEndpointOrigin(value: string): string { - try { - return new URL(value).origin; - } catch { - return '(invalid endpoint)'; - } -} - -function boundedWebSocketCloseReason(reason: string): string { - const encoded = Buffer.from(reason, 'utf8'); - if (encoded.length <= MAX_WEBSOCKET_CLOSE_REASON_BYTES) return reason; - - const suffix = Buffer.from('…', 'utf8'); - let end = MAX_WEBSOCKET_CLOSE_REASON_BYTES - suffix.length; - // Do not cut through a UTF-8 continuation sequence. Excluding the leading - // byte at this boundary also excludes the incomplete code point. - while (end > 0 && (encoded[end] & 0xc0) === 0x80) end -= 1; - return Buffer.concat([encoded.subarray(0, end), suffix]).toString('utf8'); -} - -function diagnosticValue(value: string): string { - // JSON quoting keeps node names and upstream text from injecting terminal - // control characters into the operator-facing error. - return JSON.stringify(value); -} - -function resolvedNodeIdFromTerminalUrl(value: string): string | undefined { - try { - const match = /^\/v1\/nodes\/([^/]+)\/terminal\/connect$/.exec(new URL(value).pathname); - return match?.[1] ? decodeURIComponent(match[1]) : undefined; - } catch { - return undefined; - } -} - -function retryDelayBudgetMs(attempts: number, initialMs: number, maxMs: number): number { - let total = 0; - for (let attempt = 0; attempt < attempts; attempt += 1) { - total += Math.min(initialMs * 2 ** attempt, maxMs); - } - return total; -} - -function terminalSessionFailureSummary(error: TerminalSessionAttemptError): string { - if (error.code === 'node_not_found') return 'Control-plane node lookup found no matching record'; - if (error.code === 'node_unreachable') { - return /no terminal transport/i.test(error.message) - ? 'The node record was found, but its terminal transport was unavailable' - : 'The control plane classified the node as unreachable'; - } - if (error.code === 'terminal_session_unavailable') { - return 'The control plane could not allocate a terminal session'; - } - if (error.code === 'control_plane_timeout') return 'Control-plane terminal-session lookup timed out'; - if (error.code === 'control_plane_unavailable') return 'Control-plane terminal-session lookup failed'; - return 'The terminal-session request was rejected'; -} - -function isRetryableTerminalSessionFailure(code: string | undefined): boolean { - // These structured responses are emitted before a session is returned, so - // retrying cannot duplicate a successful allocation. A fetch timeout, - // network failure, or unclassified 5xx is different: this POST may already - // have completed server-side, and retrying it could create a second session. - return code === 'node_unreachable' || code === 'terminal_session_unavailable'; -} - -/** Start a broker-compatible loopback proxy for one remote terminal session. */ -export async function startFleetNodeAttachProxy( - options: FleetNodeAttachOptions -): Promise { - const env = options.env ?? process.env; - const fetchFn = options.fetch ?? globalThis.fetch; - const { workspaceKey, baseUrl: requestedBaseUrl } = resolveWorkspaceTransport({ - workspaceKey: options.workspaceKey, - baseUrl: options.baseUrl, - env, - }); - const baseUrl = validateFleetAttachBaseUrl(requestedBaseUrl ?? CANONICAL_RELAYCAST_ORIGIN); - const nodePath = safeNodePath(options.node); - const sessionEndpoint = `${baseUrl}/v1/nodes/${nodePath}/terminal/sessions`; - const sessionRequestTimeoutMs = options.sessionRequest?.timeoutMs ?? SESSION_REQUEST_TIMEOUT_MS; - const sessionRequestTotalTimeoutMs = - options.sessionRequest?.totalTimeoutMs ?? SESSION_REQUEST_TOTAL_TIMEOUT_MS; - const sessionRequestDeadline = Date.now() + sessionRequestTotalTimeoutMs; - const sessionRequestSleep = - options.sessionRequest?.sleep ?? - ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); - let sessionRequestAttempts = 0; - let sessionRequestBudgetExhaustedBetweenAttempts = false; - let lastSessionError: TerminalSessionAttemptError | undefined; - const sessionResult = await collectWithRetry( - 'terminal session request', - async () => { - const remainingRequestBudgetMs = sessionRequestDeadline - Date.now(); - if (remainingRequestBudgetMs <= 0) { - sessionRequestBudgetExhaustedBetweenAttempts = true; - const exhausted = - lastSessionError === undefined - ? new TerminalSessionAttemptError( - 'overall terminal-session request deadline exhausted', - 'control_plane_timeout', - undefined, - false, - false - ) - : new TerminalSessionAttemptError( - lastSessionError.message, - lastSessionError.code, - lastSessionError.status, - false, - lastSessionError.completionUnknown - ); - lastSessionError = exhausted; - throw exhausted; - } - sessionRequestAttempts += 1; - const controller = new AbortController(); - let timedOut = false; - const attemptTimeoutMs = Math.min(sessionRequestTimeoutMs, remainingRequestBudgetMs); - const timeout = setTimeout(() => { - timedOut = true; - controller.abort(); - }, attemptTimeoutMs); - let ticketResponse: Response; - let ticketPayload: FleetSessionResponse; - try { - ticketResponse = await fetchFn(sessionEndpoint, { - method: 'POST', - headers: { Authorization: `Bearer ${workspaceKey}`, 'Content-Type': 'application/json' }, - body: JSON.stringify({ agent: options.agent, mode: options.mode }), - signal: controller.signal, - }); - const parsedPayload = (await ticketResponse.json()) as unknown; - ticketPayload = - parsedPayload && typeof parsedPayload === 'object' && !Array.isArray(parsedPayload) - ? (parsedPayload as FleetSessionResponse) - : {}; - } catch (error) { - const detail = error instanceof Error ? error.message : String(error); - lastSessionError = new TerminalSessionAttemptError( - timedOut ? 'request exceeded its deadline' : detail, - timedOut ? 'control_plane_timeout' : 'control_plane_unavailable', - undefined, - false, - true - ); - throw lastSessionError; - } finally { - clearTimeout(timeout); - } - - const terminalUrl = ticketPayload.data?.terminal_url; - const sessionId = ticketPayload.data?.session_id; - const resumeToken = ticketPayload.data?.resume_token; - if (!ticketResponse.ok || !terminalUrl || !sessionId || !resumeToken) { - const code = ticketPayload.error?.code; - const message = - ticketPayload.error?.message ?? `terminal session request failed (HTTP ${ticketResponse.status})`; - lastSessionError = new TerminalSessionAttemptError( - message, - code, - ticketResponse.status, - isRetryableTerminalSessionFailure(code), - false - ); - throw lastSessionError; - } - return { terminalUrl, sessionId, resumeToken }; - }, - { - retries: SESSION_REQUEST_RETRIES, - baseDelayMs: SESSION_REQUEST_RETRY_DELAY_MS, - sleep: async (delayMs) => { - const remainingRequestBudgetMs = sessionRequestDeadline - Date.now(); - if (remainingRequestBudgetMs <= 0) return; - await sessionRequestSleep(Math.min(delayMs, remainingRequestBudgetMs)); - }, - shouldRetry: (error) => error instanceof TerminalSessionAttemptError && error.retryable, - } - ); - if (!sessionResult.ok) { - const failure = lastSessionError; - const status = failure?.status === undefined ? '' : ` HTTP ${failure.status};`; - const code = failure?.code === undefined ? '' : ` code ${failure.code};`; - const retryNote = sessionRequestBudgetExhaustedBetweenAttempts - ? sessionRequestAttempts > 1 - ? `retried ${sessionRequestAttempts - 1} time${sessionRequestAttempts === 2 ? '' : 's'};` + - ' overall budget exhausted before the next attempt' - : 'not retried because the overall budget was exhausted before the next attempt' - : sessionRequestAttempts > 1 - ? `retried ${sessionRequestAttempts - 1} time${sessionRequestAttempts === 2 ? '' : 's'}` + - (failure?.completionUnknown - ? '; final POST not retried because it may have completed server-side' - : '') - : failure?.completionUnknown - ? 'not retried because the POST may have completed server-side' - : 'not retried because the failure was terminal'; - const upstream = failure?.message ? ` Upstream message ${diagnosticValue(failure.message)}.` : ''; - throw new FleetNodeAttachError( - `Error: ${failure ? terminalSessionFailureSummary(failure) : 'Terminal-session request failed'}.` + - `${upstream} Node ref ${diagnosticValue(options.node.trim())}, resolved node id unavailable (session creation did not complete);` + - ` endpoint ${diagnosticValue(diagnosticEndpoint(sessionEndpoint))};${status}${code}` + - ` timeout ${sessionRequestTimeoutMs}ms per attempt; overall budget ${sessionRequestTotalTimeoutMs}ms;` + - ` attempts ${sessionRequestAttempts} (${retryNote}).`, - failure?.code - ); - } - const { terminalUrl, sessionId, resumeToken } = sessionResult.value; - const resolvedNodeId = resolvedNodeIdFromTerminalUrl(terminalUrl); - const remoteEndpoint = diagnosticEndpoint(terminalUrl); - const reconnectInitialDelayMs = options.reconnectDelay?.initialMs ?? INITIAL_RECONNECT_DELAY_MS; - const reconnectMaxDelayMs = options.reconnectDelay?.maxMs ?? MAX_RECONNECT_DELAY_MS; - const terminalHandshakeTimeoutMs = - options.reconnectDelay?.handshakeTimeoutMs ?? TERMINAL_CONNECT_TIMEOUT_MS; - const terminalReadyTimeoutMs = options.reconnectDelay?.readyTimeoutMs ?? TERMINAL_CONNECT_TIMEOUT_MS; - // A readiness-gated local request follows activeReadiness across reconnect - // generations. Its own deadline therefore has to cover the same complete, - // finite recovery path: every backoff plus every handshake/readiness pair. - const terminalWaitTimeoutMs = - retryDelayBudgetMs(MAX_RECONNECT_ATTEMPTS, reconnectInitialDelayMs, reconnectMaxDelayMs) + - MAX_RECONNECT_ATTEMPTS * (terminalHandshakeTimeoutMs + terminalReadyTimeoutMs); - - let connectionGeneration = 0; - const createReadiness = (): TerminalReadiness => { - let resolve!: () => void; - let reject!: (error: Error) => void; - const promise = new Promise((resolvePromise, rejectPromise) => { - resolve = resolvePromise; - reject = rejectPromise; - }); - // A failure can land before a snapshot request attaches its waiter. Keep - // the rejection observable while avoiding an unhandled-rejection warning. - void promise.catch(() => undefined); - return { generation: ++connectionGeneration, settled: false, promise, resolve, reject }; - }; - let activeReadiness = createReadiness(); - const resolveReadiness = (readiness: TerminalReadiness) => { - if (readiness.settled) return; - readiness.settled = true; - readiness.resolve(); - }; - const rejectReadiness = (readiness: TerminalReadiness, error: Error) => { - if (readiness.settled) return; - readiness.settled = true; - readiness.reject(error); - }; - const waitForCurrentReadiness = async (): Promise => { - for (;;) { - const readiness = activeReadiness; - await readiness.promise; - if (readiness === activeReadiness) return; - } - }; - /** - * Await the live readiness generation through both the WebSocket handshake - * and the post-open terminal.ready allowance. - * Every handler that must not run before `terminal.ready` — snapshot, - * delivery-mode PUT, resize — goes through this one helper so the - * timer/clearTimeout/settle logic cannot diverge between copies. - * - * Rejects with the underlying {@link FleetNodeAttachError} when the terminal - * failed (preserving its `code`), or a plain `Error` carrying - * `timeoutMessage` when the wait expired. - */ - const waitForTerminalReady = (timeoutMessage: string): Promise => - new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error(timeoutMessage)), terminalWaitTimeoutMs); - void waitForCurrentReadiness().then( - () => { - clearTimeout(timer); - resolve(); - }, - (error: Error) => { - clearTimeout(timer); - reject(error); - } - ); - }); - const snapshot: { screen: string; rows: number; cols: number; offset: number } = { - screen: '', - rows: 24, - cols: 80, - offset: 0, - }; - const eventSockets = new Set(); - const inputSockets = new Set(); - const outputHistory: Array<{ chunk: string; offset?: number }> = []; - let outputHistoryBytes = 0; - let remote: WebSocket | undefined; - let stopped = false; - let terminalEnded = false; - let terminalEverReady = false; - let reconnecting = false; - let reconnectAttempts = 0; - let reconnectTimer: ReturnType | undefined; - const terminalReadinessTimers = new Set>(); - /** Locally-tracked delivery mode, kept in sync with each broker reply. */ - let loopbackDeliveryMode: 'manual_flush' | 'auto_inject' = - options.mode === 'drive' ? 'manual_flush' : 'auto_inject'; - type DeliveryModeResult = { mode: string; flushed: number; matched: boolean; revision: string }; - type FlushResult = { - flushed: number; - dead_lettered: number; - held: number; - blocked_reason: string | null; - }; - let pendingFlush: { - requestId: string; - resolve: (result: FlushResult) => void; - reject: (error: Error) => void; - timer: ReturnType; - } | null = null; - /** At most one in-flight delivery-mode PUT at a time. */ - let pendingDeliveryMode: { - requestId: string; - resolve: (result: DeliveryModeResult) => void; - reject: (err: Error) => void; - timer: ReturnType; - } | null = null; - const loopbackApiKey = randomBytes(32).toString('base64url'); - const loopbackAuthorized = (headers: IncomingMessage['headers']) => - headers.authorization === `Bearer ${loopbackApiKey}` || headers['x-api-key'] === loopbackApiKey; - - const server = createServer(async (request, response) => { - if (!loopbackAuthorized(request.headers)) { - json(response, 401, { - error: { code: 'unauthorized', message: 'loopback terminal token is required' }, - }); - return; - } - const path = new URL(request.url ?? '/', 'http://127.0.0.1').pathname; - if (request.method === 'GET' && path === `/api/spawned/${encodeURIComponent(options.agent)}/snapshot`) { - try { - await waitForTerminalReady('terminal snapshot timed out'); - } catch (error) { - const terminalError = error instanceof FleetNodeAttachError ? error : undefined; - json(response, terminalErrorStatus(terminalError?.code), { - error: { - code: terminalError?.code ?? 'snapshot_unavailable', - message: - terminalError?.message ?? (error instanceof Error ? error.message : 'snapshot unavailable'), - }, - }); - return; - } - json(response, 200, { format: 'ansi', ...snapshot }); - return; - } - const name = encodeURIComponent(options.agent); - if (path === `/api/spawned/${name}/flush` && request.method === 'POST') { - // Forward the flush to the remote broker over the terminal websocket. - // Without this the `--node` form of `node agent message flush` reached - // only the LOCAL broker's worker registry and returned agent_not_found - // for a name that `node agent list` and `attach` both resolve. - try { - await waitForTerminalReady('terminal connection timed out'); - } catch (error) { - const terminalError = error instanceof FleetNodeAttachError ? error : undefined; - json(response, terminalErrorStatus(terminalError?.code), { - error: { - code: terminalError?.code ?? 'node_unreachable', - message: - terminalError?.message ?? (error instanceof Error ? error.message : 'terminal unavailable'), - }, - }); - return; - } - if (!remote || remote.readyState !== WebSocket.OPEN) { - json(response, 503, { - error: { code: 'node_unreachable', message: 'terminal transport is not connected' }, - }); - return; - } - if (pendingFlush) { - json(response, 503, { - error: { code: 'flush_conflict', message: 'a flush request is already in flight' }, - }); - return; - } - const requestId = randomBytes(8).toString('hex'); - const result = await new Promise((resolve) => { - const timer = setTimeout(() => { - pendingFlush = null; - resolve(new FleetNodeAttachError('flush request timed out', 'flush_timeout')); - }, DELIVERY_MODE_TIMEOUT_MS); - pendingFlush = { - requestId, - resolve: (r) => resolve(r), - reject: (e) => resolve(e), - timer, - }; - remote!.send( - JSON.stringify({ - type: 'terminal.flush_pending', - session_id: sessionId, - request_id: requestId, - }) - ); - }); - if (result instanceof Error) { - const errCode = - result instanceof FleetNodeAttachError ? (result.code ?? 'flush_failed') : 'flush_failed'; - json(response, terminalErrorStatus(errCode), { - error: { code: errCode, message: result.message }, - }); - return; - } - json(response, 200, result); - return; - } - if (path === `/api/spawned/${name}/delivery-mode`) { - if (request.method === 'GET') { - json(response, 200, { mode: loopbackDeliveryMode }); - return; - } - // PUT — forward the request to the remote broker via the terminal WS and - // await the broker's real reply. This is the path that was previously a - // static stub returning manual_flush, causing drive attach to fail with - // "broker remained in manual_flush mode". - const body = await readBody(request); - const requestedMode = - body.mode === 'auto_inject' ? 'auto_inject' : body.mode === 'manual_flush' ? 'manual_flush' : null; - if (requestedMode === null) { - json(response, 400, { - error: { code: 'invalid_mode', message: `unsupported delivery mode '${String(body.mode)}'` }, - }); - return; - } - // Drive attach changes delivery mode before it requests the initial - // snapshot. Gate the PUT on terminal.ready so a fast local caller does - // not lose a race with the remote websocket handshake and receive the - // misleading "terminal transport is not connected" failure. - try { - await waitForTerminalReady('terminal connection timed out'); - } catch (error) { - // Preserve the canonical status mapping here too: a readiness failure - // carrying `agent_not_found` has to reach the preflight as a 404 or - // the operator loses the cross-node placement hint that tells them - // which machine to run the attach on. - const terminalError = error instanceof FleetNodeAttachError ? error : undefined; - json(response, terminalErrorStatus(terminalError?.code), { - error: { - code: terminalError?.code ?? 'node_unreachable', - message: - terminalError?.message ?? (error instanceof Error ? error.message : 'terminal unavailable'), - }, - }); - return; - } - if (!remote || remote.readyState !== WebSocket.OPEN) { - json(response, 503, { - error: { code: 'node_unreachable', message: 'terminal transport is not connected' }, - }); - return; - } - if (pendingDeliveryMode) { - json(response, 503, { - error: { code: 'delivery_mode_conflict', message: 'a delivery mode request is already in flight' }, - }); - return; - } - const requestId = randomBytes(8).toString('hex'); - const result = await new Promise((resolve) => { - const timer = setTimeout(() => { - pendingDeliveryMode = null; - resolve(new FleetNodeAttachError('delivery mode request timed out', 'delivery_mode_timeout')); - }, DELIVERY_MODE_TIMEOUT_MS); - pendingDeliveryMode = { - requestId, - resolve: (r) => resolve(r), - reject: (e) => resolve(e), - timer, - }; - const frame: Record = { - type: 'terminal.set_delivery_mode', - session_id: sessionId, - mode: requestedMode, - request_id: requestId, - }; - if (typeof body.expected_mode === 'string') frame.expected_mode = body.expected_mode; - if (typeof body.expected_revision === 'string') frame.expected_revision = body.expected_revision; - remote!.send(JSON.stringify(frame)); - }); - if (result instanceof Error) { - const errCode = - result instanceof FleetNodeAttachError - ? (result.code ?? 'delivery_mode_failed') - : 'delivery_mode_failed'; - // Same canonical mapping as the readiness gate: 404 for - // agent_not_found so the attach preflight can produce the no-agent or - // cross-node placement error, 409 for unsupported_runtime, 503 for - // everything else. - json(response, terminalErrorStatus(errCode), { - error: { code: errCode, message: result.message }, - }); - return; - } - loopbackDeliveryMode = result.mode === 'manual_flush' ? 'manual_flush' : 'auto_inject'; - json(response, 200, { - mode: result.mode, - flushed: result.flushed, - matched: result.matched, - revision: result.revision, - }); - return; - } - if (request.method === 'GET' && path === `/api/spawned/${name}/pending`) { - json(response, 200, { pending: [] }); - return; - } - if (request.method === 'POST' && path === `/api/spawned/${name}/flush`) { - json(response, 200, { flushed: 0 }); - return; - } - if (request.method === 'GET' && path === '/api/spawned') { - json(response, 200, { agents: [{ name: options.agent, workerPid: 1 }] }); - return; - } - if (request.method === 'POST' && path === `/api/resize/${name}`) { - const body = await readBody(request); - if (body.release === true) { - json(response, 200, { name: options.agent, released: true }); - return; - } - const rows = typeof body.rows === 'number' ? body.rows : 0; - const cols = typeof body.cols === 'number' ? body.cols : 0; - if (!Number.isInteger(rows) || !Number.isInteger(cols) || rows < 1 || cols < 1) { - json(response, 400, { - error: { code: 'invalid_dimensions', message: 'rows and cols must be positive integers' }, - }); - return; - } - try { - await waitForTerminalReady('terminal resize timed out'); - } catch (error) { - json(response, 503, { - error: { - code: 'session_not_ready', - message: error instanceof Error ? error.message : 'terminal session is not ready', - }, - }); - return; - } - if (!remote || remote.readyState !== WebSocket.OPEN || remote.bufferedAmount > MAX_BUFFERED_BYTES) { - json(response, 503, { - error: { code: 'node_unreachable', message: 'terminal transport is unavailable' }, - }); - return; - } - remote.send(JSON.stringify({ type: 'terminal.resize', session_id: sessionId, rows, cols })); - json(response, 200, { name: options.agent, rows, cols, applied: true }); - return; - } - json(response, 404, { error: { code: 'not_found', message: 'loopback terminal endpoint not found' } }); - }); - const websocketServer = new WebSocketServer({ noServer: true }); - - const closeSocket = (socket: WebSocket, code: number, reason: string) => { - try { - socket.close(code, boundedWebSocketCloseReason(reason)); - } catch { - /* connection already gone */ - } - }; - const broadcast = (sockets: Set, payload: unknown): boolean => { - const encoded = JSON.stringify(payload); - let accepted = false; - for (const socket of sockets) { - if (socket.readyState !== WebSocket.OPEN) continue; - if (socket.bufferedAmount > MAX_BUFFERED_BYTES) { - closeSocket(socket, 1013, 'loopback client backpressure exceeded'); - sockets.delete(socket); - continue; - } - try { - socket.send(encoded); - accepted = true; - } catch { - sockets.delete(socket); - } - } - return accepted; - }; - const workerStreamEvent = (chunk: string, offset?: number) => ({ - kind: 'worker_stream', - name: options.agent, - stream: 'stdout', - chunk, - ...(offset === undefined ? {} : { offset }), - }); - const retainOutput = (chunk: string, offset: number | undefined): boolean => { - const bytes = Buffer.byteLength(chunk, 'utf8'); - if (outputHistoryBytes + bytes > MAX_BUFFERED_BYTES) return false; - outputHistory.push({ chunk, ...(offset === undefined ? {} : { offset }) }); - outputHistoryBytes += bytes; - return true; - }; - - websocketServer.on('connection', (socket, request) => { - const path = new URL(request.url ?? '/', 'http://127.0.0.1').pathname; - if (path === '/ws') { - eventSockets.add(socket); - socket.on('close', () => eventSockets.delete(socket)); - let replayed = 0; - for (const event of outputHistory) { - if (socket.readyState !== WebSocket.OPEN || socket.bufferedAmount > MAX_BUFFERED_BYTES) break; - try { - socket.send(JSON.stringify(workerStreamEvent(event.chunk, event.offset))); - } catch { - break; - } - replayed += 1; - } - if (replayed > 0) { - const sentBytes = outputHistory - .slice(0, replayed) - .reduce((total, event) => total + Buffer.byteLength(event.chunk, 'utf8'), 0); - outputHistory.splice(0, replayed); - outputHistoryBytes -= sentBytes; - } - return; - } - if (path === `/api/input/${encodeURIComponent(options.agent)}/stream`) { - inputSockets.add(socket); - socket.on('close', () => inputSockets.delete(socket)); - socket.send(JSON.stringify({ type: 'pty_input_ready', name: options.agent })); - socket.on('message', (data) => { - if (!remote || remote.readyState !== WebSocket.OPEN || remote.bufferedAmount > MAX_BUFFERED_BYTES) { - broadcast(inputSockets, { - type: 'pty_input_error', - code: 'node_unreachable', - message: 'terminal transport is unavailable', - }); - return; - } - const raw = rawDataToString(data); - remote.send( - JSON.stringify({ - type: 'terminal.input', - session_id: sessionId, - data_base64: Buffer.from(raw, 'utf8').toString('base64'), - }) - ); - }); - return; - } - closeSocket(socket, 1008, 'unknown loopback endpoint'); - }); - server.on('upgrade', (request, socket, head) => { - if (!loopbackAuthorized(request.headers)) { - socket.write('HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n'); - socket.destroy(); - return; - } - websocketServer.handleUpgrade(request, socket, head, (client) => - websocketServer.emit('connection', client, request) - ); - }); - server.listen(0, '127.0.0.1'); - await once(server, 'listening'); - const address = server.address(); - if (!address || typeof address === 'string') - throw new FleetNodeAttachError( - 'Error: could not allocate loopback terminal listener.', - 'loopback_unavailable' - ); - - const resumeUrl = new URL(terminalUrl); - resumeUrl.searchParams.delete('ticket'); - resumeUrl.searchParams.set('session_id', sessionId); - resumeUrl.searchParams.set('resume', resumeToken); - /** Reject and clear any in-flight delivery-mode PUT, if one is pending. */ - const rejectPendingDeliveryMode = (error: FleetNodeAttachError) => { - if (pendingFlush) { - const flush = pendingFlush; - pendingFlush = null; - clearTimeout(flush.timer); - // A flush never changes delivery mode, so it must not inherit the - // delivery-mode disconnect wording. An operator debugging a failed flush - // would otherwise be told the "delivery-mode change" was interrupted — - // pointing at an operation their command never performed. - const flushError = - error.code === 'delivery_mode_disconnected' - ? new FleetNodeAttachError( - 'terminal transport disconnected while the flush was in flight', - 'flush_disconnected' - ) - : error; - flush.reject(flushError); - } - if (!pendingDeliveryMode) return; - const pending = pendingDeliveryMode; - pendingDeliveryMode = null; - clearTimeout(pending.timer); - pending.reject(error); - }; - const endTerminal = (error: FleetNodeAttachError, eventCloseReason = error.message) => { - if (terminalEnded) return; - terminalEnded = true; - if (reconnectTimer) { - clearTimeout(reconnectTimer); - reconnectTimer = undefined; - } - for (const timer of terminalReadinessTimers) clearTimeout(timer); - terminalReadinessTimers.clear(); - rejectPendingDeliveryMode(error); - const activeRemote = remote; - remote = undefined; - rejectReadiness(activeReadiness, error); - broadcast(inputSockets, { type: 'pty_input_error', code: error.code, message: error.message }); - for (const socket of eventSockets) closeSocket(socket, 1011, eventCloseReason); - if (activeRemote && activeRemote.readyState !== WebSocket.CLOSED) { - try { - activeRemote.terminate(); - } catch { - // The socket may have closed between the state check and terminate. - } - } - }; - const failRemote = (message: string, eventCloseReason?: string) => { - endTerminal(new FleetNodeAttachError(message, 'node_unreachable'), eventCloseReason); - }; - const connect = (url: string, readiness: TerminalReadiness) => { - if (stopped || terminalEnded) return; - const socket = new WebSocket(asWsUrl(url), { handshakeTimeout: terminalHandshakeTimeoutMs }); - remote = socket; - let readinessTimer: ReturnType | undefined; - let readinessExpired = false; - const clearReadinessTimer = () => { - if (!readinessTimer) return; - clearTimeout(readinessTimer); - terminalReadinessTimers.delete(readinessTimer); - readinessTimer = undefined; - }; - socket.on('open', () => { - // `handshakeTimeout` independently bounds the HTTP upgrade. Start the - // terminal.ready allowance only after that upgrade succeeds so a slow - // but valid handshake cannot consume the readiness window. - readinessTimer = setTimeout(() => { - const expiredTimer = readinessTimer; - readinessTimer = undefined; - if (expiredTimer) terminalReadinessTimers.delete(expiredTimer); - if (remote !== socket || stopped || terminalEnded || readiness.settled) return; - // Mark this generation stale before terminating. The ws receiver may - // still deliver data already buffered on the socket while close is - // propagating; none of it may restore readiness or reset retry state. - readinessExpired = true; - if (!terminalEverReady) { - failRemote( - `terminal transport connected but did not become ready (node ref ${diagnosticValue(options.node.trim())},` + - ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + - ` readiness timeout ${terminalReadyTimeoutMs}ms, attempts 1; not retried because no terminal session became ready)` - ); - return; - } - // A successful WebSocket upgrade is not sufficient: Relaycast may - // accept a resume lane that never produces terminal.ready. Terminating - // it drives the same bounded close/retry path as a transport failure. - options.reconnectDelay?.beforeReadyTimeoutTerminate?.(socket); - socket.terminate(); - }, terminalReadyTimeoutMs); - terminalReadinessTimers.add(readinessTimer); - }); - socket.on('message', (data) => { - // A late frame from a transport superseded during reconnect must never - // overwrite the fresh snapshot or end the replacement session. - if (remote !== socket || stopped || terminalEnded || readinessExpired) return; - const frame = parseFrame(data); - if (!frame || frame.session_id !== sessionId) return; - if (frame.type === 'terminal.ready') { - clearReadinessTimer(); - snapshot.screen = typeof frame.screen === 'string' ? frame.screen : ''; - snapshot.rows = typeof frame.rows === 'number' ? frame.rows : 24; - snapshot.cols = typeof frame.cols === 'number' ? frame.cols : 80; - snapshot.offset = typeof frame.offset === 'number' ? frame.offset : 0; - // Seed loopbackDeliveryMode from the broker's actual state so that - // detach restores the correct mode even when the worker started in a - // different mode than our local inference at line 214. - if (frame.delivery_mode === 'manual_flush' || frame.delivery_mode === 'auto_inject') { - loopbackDeliveryMode = frame.delivery_mode; - } - terminalEverReady = true; - reconnectAttempts = 0; - if (readiness === activeReadiness) { - resolveReadiness(readiness); - // A reconnect gets a fresh ANSI grid but existing local `/ws` - // consumers have already performed their initial HTTP snapshot. - // Re-emit this screen without an offset so they repaint instead of - // retaining a stale pre-reconnect terminal image. - if (readiness.generation > 1 && snapshot.screen) { - broadcast(eventSockets, workerStreamEvent(snapshot.screen)); - } - } - } else if (frame.type === 'terminal.output' && typeof frame.chunk === 'string') { - const offset = typeof frame.offset === 'number' ? frame.offset : undefined; - if (!broadcast(eventSockets, workerStreamEvent(frame.chunk, offset))) { - if (!retainOutput(frame.chunk, offset)) { - endTerminal( - new FleetNodeAttachError( - 'terminal output exceeded the bounded loopback buffer', - 'output_backpressure' - ) - ); - } - } - } else if (frame.type === 'terminal.input_ack') { - broadcast(inputSockets, { - type: 'pty_input_ack', - name: options.agent, - bytes_written: typeof frame.bytes_written === 'number' ? frame.bytes_written : 0, - }); - } else if (frame.type === 'terminal.delivery_mode') { - const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; - if (pendingDeliveryMode && (frameRid === undefined || frameRid === pendingDeliveryMode.requestId)) { - const pending = pendingDeliveryMode; - pendingDeliveryMode = null; - clearTimeout(pending.timer); - pending.resolve({ - mode: typeof frame.mode === 'string' ? frame.mode : 'auto_inject', - flushed: typeof frame.flushed === 'number' ? frame.flushed : 0, - matched: typeof frame.matched === 'boolean' ? frame.matched : true, - revision: typeof frame.revision === 'string' ? frame.revision : '1', - }); - } - } else if (frame.type === 'terminal.flush_pending') { - const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; - // Exact match only. The proxy always sends a request_id, and unlike - // delivery-mode there is no older-broker reply shape to stay - // compatible with, so a reply without one is not ours — accepting it - // would resolve the caller's flush with an unrelated result. - if (pendingFlush && frameRid === pendingFlush.requestId) { - const pending = pendingFlush; - pendingFlush = null; - clearTimeout(pending.timer); - pending.resolve({ - flushed: typeof frame.flushed === 'number' ? frame.flushed : 0, - dead_lettered: typeof frame.dead_lettered === 'number' ? frame.dead_lettered : 0, - held: typeof frame.held === 'number' ? frame.held : 0, - blocked_reason: typeof frame.blocked_reason === 'string' ? frame.blocked_reason : null, - }); - } - } else if (frame.type === 'terminal.error') { - const message = typeof frame.message === 'string' ? frame.message : 'remote terminal failed'; - const code = typeof frame.code === 'string' ? frame.code : 'terminal_error'; - const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; - // Route the error to the pending delivery-mode request only when the - // request_id matches (or the broker sent no request_id at all — older - // broker compat). An unrelated session-level error must not cancel a - // live delivery-mode PUT and vice-versa. - if (pendingFlush && frameRid !== undefined && frameRid === pendingFlush.requestId) { - const pending = pendingFlush; - pendingFlush = null; - clearTimeout(pending.timer); - pending.reject(new FleetNodeAttachError(message, code)); - } else if ( - pendingDeliveryMode && - (frameRid === undefined || frameRid === pendingDeliveryMode.requestId) - ) { - const pending = pendingDeliveryMode; - pendingDeliveryMode = null; - clearTimeout(pending.timer); - pending.reject(new FleetNodeAttachError(message, code)); - } else if (!frameRid && readiness === activeReadiness && !readiness.settled) { - endTerminal(new FleetNodeAttachError(message, code)); - } else if (!frameRid) { - broadcast(inputSockets, { type: 'pty_input_error', code, message }); - } - } else if (frame.type === 'terminal.closed') { - endTerminal(new FleetNodeAttachError('remote terminal session closed', 'terminal_closed')); - } - }); - socket.on('error', () => { - readinessExpired = true; - clearReadinessTimer(); - // Initial connection failure has no terminal state worth preserving. - // Fail promptly with the canonical unavailable-node error instead of - // letting the HTTP snapshot timeout mask it. Once Ready has been seen, - // the close handler retains the bounded resume/backoff behaviour. - if (remote === socket && readiness === activeReadiness && !readiness.settled && !terminalEverReady) { - failRemote( - `terminal transport could not connect to the fleet node (node ref ${diagnosticValue(options.node.trim())},` + - ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + - ` handshake budget ${terminalHandshakeTimeoutMs}ms, attempts 1; not retried because no terminal session became ready)` - ); - } - }); - socket.on('close', () => { - readinessExpired = true; - clearReadinessTimer(); - if (remote !== socket || stopped || terminalEnded || reconnecting) return; - // Relaycast drops the old lane's terminal session state on disconnect, - // so a set_delivery_mode frame already sent on this dying socket is - // lost and will never get a reply on the replacement socket — even - // once reconnect succeeds. Fail the pending PUT fast with a retryable - // error instead of leaving it to hang out the full timeout. - rejectPendingDeliveryMode( - new FleetNodeAttachError( - 'terminal transport disconnected while the delivery-mode change was in flight', - 'delivery_mode_disconnected' - ) - ); - // Any waiter that observed the prior connection must retry against the - // fresh generation instead of receiving its stale resolved snapshot. - resolveReadiness(readiness); - const nextReadiness = createReadiness(); - activeReadiness = nextReadiness; - if (reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) { - const backoffBudgetMs = retryDelayBudgetMs( - MAX_RECONNECT_ATTEMPTS, - reconnectInitialDelayMs, - reconnectMaxDelayMs - ); - const message = - `terminal transport could not reconnect to the fleet node (node ref ${diagnosticValue(options.node.trim())},` + - ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + - ` handshake timeout ${terminalHandshakeTimeoutMs}ms, readiness timeout ${terminalReadyTimeoutMs}ms,` + - ` attempts ${reconnectAttempts},` + - ` backoff budget ${backoffBudgetMs}ms)`; - failRemote( - message, - `terminal reconnect failed; attempts=${reconnectAttempts}; budget=${backoffBudgetMs}ms;` + - ` endpoint=${diagnosticValue(diagnosticEndpointOrigin(terminalUrl))};` + - ` node=${diagnosticValue(resolvedNodeId ?? options.node.trim())}` - ); - return; - } - reconnecting = true; - const delay = Math.min(reconnectInitialDelayMs * 2 ** reconnectAttempts, reconnectMaxDelayMs); - reconnectAttempts += 1; - reconnectTimer = setTimeout(() => { - reconnectTimer = undefined; - reconnecting = false; - connect(resumeUrl.toString(), nextReadiness); - }, delay); - }); - }; - connect(terminalUrl, activeReadiness); - - return { - brokerUrl: `http://127.0.0.1:${address.port}`, - apiKey: loopbackApiKey, - requestTimeoutMs: terminalWaitTimeoutMs + DELIVERY_MODE_TIMEOUT_MS + LOOPBACK_REQUEST_TIMEOUT_MARGIN_MS, - async close() { - if (stopped) return; - stopped = true; - terminalEnded = true; - if (reconnectTimer) { - clearTimeout(reconnectTimer); - reconnectTimer = undefined; - } - for (const timer of terminalReadinessTimers) clearTimeout(timer); - terminalReadinessTimers.clear(); - rejectReadiness(activeReadiness, new FleetNodeAttachError('terminal attach closed', 'closed')); - rejectPendingDeliveryMode(new FleetNodeAttachError('terminal attach closed', 'closed')); - const activeRemote = remote; - remote = undefined; - if (activeRemote && activeRemote.readyState === WebSocket.OPEN) { - try { - activeRemote.send(JSON.stringify({ type: 'terminal.close', session_id: sessionId })); - } catch { - // Best effort; terminate below still prevents a late reconnect. - } - } - if (activeRemote && activeRemote.readyState !== WebSocket.CLOSED) { - try { - activeRemote.terminate(); - } catch { - // Socket is already gone. - } - } - for (const socket of [...eventSockets, ...inputSockets]) - closeSocket(socket, 1000, 'terminal attach closed'); - websocketServer.close(); - await new Promise((resolve) => server.close(() => resolve())); - }, - }; -} +export * from '@agent-relay/cloud/attach'; diff --git a/packages/cli/src/cli/lib/attach-mode.ts b/packages/cli/src/cli/lib/attach-mode.ts index b1107eece..b10726bba 100644 --- a/packages/cli/src/cli/lib/attach-mode.ts +++ b/packages/cli/src/cli/lib/attach-mode.ts @@ -1 +1 @@ -export type AttachMode = 'drive' | 'view' | 'passthrough'; +export type { AttachMode } from '@agent-relay/cloud/attach'; diff --git a/packages/cli/src/cli/lib/collect-with-retry.ts b/packages/cli/src/cli/lib/collect-with-retry.ts index fb726a231..71c417be1 100644 --- a/packages/cli/src/cli/lib/collect-with-retry.ts +++ b/packages/cli/src/cli/lib/collect-with-retry.ts @@ -1,42 +1 @@ -export interface CollectWithRetryOptions { - retries?: number; - baseDelayMs?: number; - sleep?: (ms: number) => Promise; - /** Return false for terminal failures that must be surfaced immediately. */ - shouldRetry?: (error: unknown, attempt: number) => boolean; -} - -/** - * Run a bounded operation with the fleet CLI's small retry delay. - * - * The result records whether another attempt actually ran. Callers that need - * richer failure metadata can retain the thrown value in their operation; - * this helper deliberately keeps the existing fleet-list rendering contract. - */ -export async function collectWithRetry( - label: string, - fn: () => Promise, - options?: CollectWithRetryOptions -): Promise<{ ok: true; value: T; retried: boolean } | { ok: false; error: string; retried: boolean }> { - const retries = options?.retries ?? 1; - const baseDelay = options?.baseDelayMs ?? 500; - const sleep = options?.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms))); - let lastError: unknown; - let attempts = 0; - for (let attempt = 0; attempt <= retries; attempt += 1) { - attempts += 1; - try { - const value = await fn(); - return { ok: true, value, retried: attempt > 0 }; - } catch (error) { - lastError = error; - if (attempt >= retries || options?.shouldRetry?.(error, attempt + 1) === false) break; - // Keep the established fleet-list delay shape while making later - // attempts wait slightly longer, without introducing test randomness. - const delay = baseDelay + Math.floor(baseDelay * (attempt / (retries + 1))); - await sleep(delay); - } - } - const message = lastError instanceof Error ? lastError.message : String(lastError); - return { ok: false, error: `${label}: ${message}`, retried: attempts > 1 }; -} +export * from '@agent-relay/cloud/attach'; diff --git a/packages/cli/src/cli/lib/sdk-client.ts b/packages/cli/src/cli/lib/sdk-client.ts index fe9bf1716..42ea056de 100644 --- a/packages/cli/src/cli/lib/sdk-client.ts +++ b/packages/cli/src/cli/lib/sdk-client.ts @@ -1,9 +1,18 @@ import path from 'node:path'; import { AgentRelay, type AgentRelayAgent } from '@agent-relay/sdk'; -import { AGENT37_RELAYCAST_ORIGIN, CANONICAL_RELAYCAST_ORIGIN } from '@agent-relay/cloud'; import { - resolveWorkspaceSelection as resolveCloudWorkspaceSelection, + resolveBaseUrl, + resolveWorkspaceSelection, + resolveWorkspaceTransport, +} from '@agent-relay/cloud/workspace-transport'; +export { + resolveBaseUrl, + resolveWorkspaceSelection, + resolveWorkspaceTransport, + type WorkspaceTransport, +} from '@agent-relay/cloud/workspace-transport'; +import { writeProjectWorkspaceTargetIfSelectionCurrent, type WorkspaceSelection, type WorkspaceKeySource, @@ -34,23 +43,6 @@ function trimOrUndefined(value: string | undefined): string | undefined { export type { WorkspaceKeySource }; export type { WorkspaceSelection }; -export type WorkspaceTransport = { - workspaceKey: string; - baseUrl?: string; - source: WorkspaceKeySource; -}; - -/** Resolve the selected key and any previously persisted Relay workspace identity. */ -export function resolveWorkspaceSelection(options: SdkClientOptions = {}): WorkspaceSelection | undefined { - const explicitProject = trimOrUndefined(env(options).AGENT_RELAY_PROJECT); - const projectRoot = explicitProject ? path.resolve(explicitProject) : options.projectRoot; - return resolveCloudWorkspaceSelection({ - workspaceKey: options.workspaceKey, - env: env(options), - ...(projectRoot ? { projectRoot } : {}), - }); -} - /** * Resolve the workspace key and report which source it came from. Precedence: * explicit flag → `RELAY_WORKSPACE_KEY`/`RELAY_API_KEY` env → the key the local @@ -70,120 +62,6 @@ export function resolveWorkspaceKey(options: SdkClientOptions = {}): string { return resolveWorkspaceKeyWithSource(options).key; } -export function resolveBaseUrl(options: SdkClientOptions = {}): string | undefined { - const selection = selectionForTransport(options); - return resolveBaseUrlForSelection(selection, options); -} - -function selectionForTransport(options: SdkClientOptions): WorkspaceSelection | undefined { - const selection = resolveWorkspaceSelection(options); - if (!selection || !options.ignorePersistedRelaycastTarget) return selection; - const { - relaycastRoute: _relaycastRoute, - relaycastBaseUrl: _relaycastBaseUrl, - relaycastApiKey: _relaycastApiKey, - relaycastApiKeyRef: _relaycastApiKeyRef, - ...canonicalSelection - } = selection; - return canonicalSelection; -} - -function resolveBaseUrlForSelection( - selection: WorkspaceSelection | undefined, - options: SdkClientOptions -): string | undefined { - const persisted = validatePersistedRelaycastBaseUrl(selection); - const requested = trimOrUndefined(options.baseUrl) ?? trimOrUndefined(env(options).RELAY_BASE_URL); - if (persisted && requested) { - let parsed: URL; - try { - parsed = new URL(requested); - } catch { - throw new Error('The requested Relaycast base URL is invalid.'); - } - const authority = /^https:\/\/([^/?#]+)/i.exec(requested)?.[1] ?? ''; - if ( - !/^https:\/\/[^/?#]+\/?$/i.test(requested) || - parsed.protocol !== 'https:' || - parsed.username || - parsed.password || - parsed.port || - /:\d+$/.test(authority) || - parsed.search || - parsed.hash || - (parsed.pathname !== '' && parsed.pathname !== '/') - ) { - throw new Error('The requested Relaycast base URL is not a trusted origin.'); - } - if (parsed.origin !== persisted) { - throw new Error('The requested Relaycast base URL does not match the persisted workspace route.'); - } - } - return persisted ?? requested; -} - -/** Resolve one credential/origin pair from one workspace selection. */ -export function resolveWorkspaceTransport(options: SdkClientOptions = {}): WorkspaceTransport { - const selection = selectionForTransport(options); - if (!selection) { - throw new Error( - 'No workspace key found. Pass --workspace-key, set RELAY_WORKSPACE_KEY, or run `relay workspace set_key `.' - ); - } - const baseUrl = resolveBaseUrlForSelection(selection, options); - // Project-session loading already validates the reference against the - // project/workspace/route/base tuple. Never re-read a ref here: doing so - // would let a tampered ref bypass that binding and pair an unrelated key - // with this route. - const routeCredential = trimOrUndefined(selection.relaycastApiKey); - if (selection.relaycastRoute === 'agent37-isolated' && !routeCredential) { - throw new Error( - 'The persisted isolated Relaycast credential is unavailable or mismatched; rerun the sandbox command to mint a fresh route.' - ); - } - return { - workspaceKey: routeCredential ?? selection.key, - ...(baseUrl ? { baseUrl } : {}), - source: selection.source, - }; -} - -function validatePersistedRelaycastBaseUrl(selection: WorkspaceSelection | undefined): string | undefined { - const baseUrl = trimOrUndefined(selection?.relaycastBaseUrl); - const route = selection?.relaycastRoute; - const relaycastApiKey = trimOrUndefined(selection?.relaycastApiKey); - if (!baseUrl && !route && !relaycastApiKey) return undefined; - if (!baseUrl || !route) { - throw new Error('The persisted Relaycast workspace route is incomplete.'); - } - let parsed: URL; - try { - parsed = new URL(baseUrl); - } catch { - throw new Error('The persisted Relaycast workspace route is invalid.'); - } - const expectedOrigin = - route === 'canonical' - ? CANONICAL_RELAYCAST_ORIGIN - : route === 'agent37-isolated' - ? AGENT37_RELAYCAST_ORIGIN - : undefined; - if ( - !expectedOrigin || - parsed.origin !== expectedOrigin || - parsed.protocol !== 'https:' || - parsed.username || - parsed.password || - parsed.port || - parsed.search || - parsed.hash || - (parsed.pathname !== '' && parsed.pathname !== '/') - ) { - throw new Error('The persisted Relaycast workspace route is not trusted.'); - } - return parsed.origin; -} - /** Persist a server-selected target only while the captured project selection is still current. */ export function persistWorkspaceRelaycastTarget( selection: WorkspaceSelection | undefined, diff --git a/packages/cloud/package.json b/packages/cloud/package.json index 54bafde6b..d6b63072e 100644 --- a/packages/cloud/package.json +++ b/packages/cloud/package.json @@ -49,7 +49,19 @@ "types": "./dist/workspace-key.d.ts", "import": "./dist/workspace-key.js" }, - "./package.json": "./package.json" + "./package.json": "./package.json", + "./attach": { + "types": "./dist/attach.d.ts", + "import": "./dist/attach.js" + }, + "./workspace-transport": { + "types": "./dist/workspace-transport.d.ts", + "import": "./dist/workspace-transport.js" + }, + "./fleet": { + "types": "./dist/fleet.d.ts", + "import": "./dist/fleet.js" + } }, "files": [ "dist", @@ -65,7 +77,8 @@ "@agent-relay/config": "12.2.4", "@aws-sdk/client-s3": "3.1020.0", "ignore": "^7.0.5", - "tar": "^7.5.21" + "tar": "^7.5.21", + "ws": "^8.18.3" }, "optionalDependencies": { "ssh2": "^1.17.0" @@ -73,7 +86,8 @@ "devDependencies": { "@types/node": "^22.19.3", "@types/ssh2": "^1.15.5", - "vitest": "^4.1.0" + "vitest": "^4.1.0", + "@types/ws": "^8.18.1" }, "publishConfig": { "access": "public" diff --git a/packages/cloud/src/attach-mode.ts b/packages/cloud/src/attach-mode.ts new file mode 100644 index 000000000..b1107eece --- /dev/null +++ b/packages/cloud/src/attach-mode.ts @@ -0,0 +1 @@ +export type AttachMode = 'drive' | 'view' | 'passthrough'; diff --git a/packages/cloud/src/attach.test.ts b/packages/cloud/src/attach.test.ts new file mode 100644 index 000000000..53c58bb4b --- /dev/null +++ b/packages/cloud/src/attach.test.ts @@ -0,0 +1,110 @@ +import { once } from 'node:events'; +import { connect, type AddressInfo, type Socket } from 'node:net'; +import { stat } from 'node:fs/promises'; +import { WebSocketServer, type WebSocket } from 'ws'; +import { afterEach, expect, it, vi } from 'vitest'; +import { startFleetNodeAttachProxy, type FleetNodeAttachProxy } from './attach.js'; + +let proxy: FleetNodeAttachProxy | undefined; +let server: WebSocketServer | undefined; +let local: Socket | undefined; +afterEach(async () => { + local?.destroy(); + await proxy?.close(); + proxy = undefined; + if (server) { + for (const client of server.clients) client.terminate(); + await new Promise((resolve) => server!.close(() => resolve())); + } + server = undefined; +}); + +async function setup(mode: 'drive' | 'view' = 'drive') { + server = new WebSocketServer({ host: '127.0.0.1', port: 0 }); + await once(server, 'listening'); + const remoteConnection = once(server, 'connection'); + const fetch = vi.fn(async (url: string | URL | Request, _init?: RequestInit) => + Response.json({ + ok: true, + data: String(url).endsWith('/agents') + ? [{ agentName: 'worker' }] + : { + session_id: 'session', + resume_token: 'resume', + terminal_url: `ws://127.0.0.1:${(server!.address() as AddressInfo).port}/terminal`, + }, + }) + ); + proxy = await startFleetNodeAttachProxy({ nodeId: 'node', mode, workspaceKey: 'rk_test', env: {}, fetch }); + const [remote] = (await remoteConnection) as [WebSocket]; + return { remote, fetch }; +} +function send(remote: WebSocket, type: string, data = {}) { + remote.send(JSON.stringify({ session_id: 'session', type, ...data })); +} + +it('discovers one agent and pipes raw bytes through a private socket, then removes it', async () => { + const { remote, fetch } = await setup(); + expect(JSON.parse(String(fetch.mock.calls[1]?.[1]?.body)).agent).toBe('worker'); + expect((await stat(proxy!.socketPath)).mode & 0o777).toBe(0o600); + local = connect(proxy!.socketPath); + await once(local, 'connect'); + send(remote, 'terminal.ready', { screen: 'hello' }); + expect((await once(local, 'data'))[0].toString()).toBe('hello'); + const frame = once(remote, 'message'); + const bytes = Buffer.from([0x00, 0x1b, 0xff, 0xc3, 0xa9]); + local.write(bytes); + expect(JSON.parse((await frame)[0].toString())).toMatchObject({ + type: 'terminal.input', + data_base64: bytes.toString('base64'), + }); + const output = once(local, 'data'); + send(remote, 'terminal.output', { chunk: 'world' }); + expect((await output)[0].toString()).toBe('world'); + const socketPath = proxy!.socketPath; + send(remote, 'terminal.closed'); + await expect(proxy!.finished).resolves.toBe(0); + await Promise.all([proxy!.close(), proxy!.close()]); + await expect(stat(socketPath)).rejects.toMatchObject({ code: 'ENOENT' }); +}); + +it('settles finished on explicit close before terminal.ready', async () => { + await setup(); + await proxy!.close(); + await expect(proxy!.finished).resolves.toBe(0); +}); + +it('resolves failure without an unhandled rejection', async () => { + const { remote } = await setup(); + send(remote, 'terminal.error', { code: 'node_unreachable', message: 'offline' }); + await expect(proxy!.finished).resolves.toBe(1); +}); + +it('view sockets cannot send terminal input and reject additional clients', async () => { + const { remote } = await setup('view'); + local = connect(proxy!.socketPath); + await once(local, 'connect'); + send(remote, 'terminal.ready', { screen: 'ready' }); + await once(local, 'data'); + const received = vi.fn(); + remote.on('message', received); + local.write('ignored'); + const extra = connect(proxy!.socketPath); + await once(extra, 'close'); + expect(received).not.toHaveBeenCalled(); +}); + +it.each([[], [{ agentName: 'a' }, { agentName: 'b' }]])( + 'rejects missing or ambiguous agents', + async (...agents) => { + await expect( + startFleetNodeAttachProxy({ + nodeId: 'node', + mode: 'drive', + workspaceKey: 'rk_test', + env: {}, + fetch: vi.fn(async () => Response.json({ data: agents.length === 0 ? [] : agents })), + }) + ).rejects.toMatchObject({ code: 'ambiguous_agent' }); + } +); diff --git a/packages/cloud/src/attach.ts b/packages/cloud/src/attach.ts new file mode 100644 index 000000000..d0d116199 --- /dev/null +++ b/packages/cloud/src/attach.ts @@ -0,0 +1,1362 @@ +/** + * Ticketed fleet-node attach adapter. + * + * The established attach clients intentionally continue to speak the local + * broker HTTP/WebSocket contract. This short-lived loopback adapter maps that + * contract onto Relaycast's authenticated terminal session, so view/drive and + * passthrough retain their behaviour without exposing a remote broker listener + * or copying a broker API key off a physical or Daytona node. + */ + +import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'; +import { once } from 'node:events'; +import { Buffer } from 'node:buffer'; +import { randomBytes } from 'node:crypto'; +import { createServer as createSocketServer, type Socket } from 'node:net'; +import { mkdtemp, chmod, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import WebSocket, { WebSocketServer } from 'ws'; + +import { AGENT37_RELAYCAST_ORIGIN, CANONICAL_RELAYCAST_ORIGIN } from './fleet-sandbox.js'; +import type { AttachMode } from './attach-mode.js'; +import { collectWithRetry } from './collect-with-retry.js'; +import { resolveWorkspaceTransport } from './workspace-transport.js'; + +const MAX_BUFFERED_BYTES = 1024 * 1024; +const MAX_WEBSOCKET_CLOSE_REASON_BYTES = 123; +// Use the same finite 30s request window as the broker's Relaycast HTTP calls. +// Terminal-session creation can exceed the shorter startup-handshake latency +// under load, and this POST cannot be replayed safely after an ambiguous client +// timeout because allocation may have completed server-side. +const SESSION_REQUEST_TIMEOUT_MS = 30_000; +// Immediate structured reachability failures still receive the complete +// five-attempt/31.2s retry schedule, while slow responses cannot multiply the +// per-attempt timeout into a roughly three-minute CLI hang. +const SESSION_REQUEST_TOTAL_TIMEOUT_MS = 90_000; +const SESSION_REQUEST_RETRIES = 4; +// The fleet node publishes liveness every 12s. With four retries, the helper's +// deterministic delays are 6s, 7.2s, 8.4s, and 9.6s: 31.2s total. That spans +// more than two heartbeat intervals and matches the established terminal +// transport's bounded recovery window instead of exhausting every retry inside +// the same stale control-plane read. +const SESSION_REQUEST_RETRY_DELAY_MS = 6_000; +const TERMINAL_CONNECT_TIMEOUT_MS = 10_000; +const INITIAL_RECONNECT_DELAY_MS = 500; +const MAX_RECONNECT_DELAY_MS = 30_000; +// Six delays (0.5s + 1s + 2s + 4s + 8s + 16s) cover the node terminal +// transport's independent 30s maximum reconnect backoff without leaving this +// client unbounded. Five attempts previously stopped after only 15.5s. +const MAX_RECONNECT_ATTEMPTS = 6; +// Bounds the acknowledgement for a delivery-mode command after readiness. +// The command is not replayable across reconnects, so this is intentionally a +// separate post-readiness phase rather than another full reconnect window. +const DELIVERY_MODE_TIMEOUT_MS = 10_000; +// The caller starts its HTTP deadline before the loopback handler starts its +// readiness timer. Leave a small response-delivery margin so the proxy's +// actionable readiness error wins that race. +const LOOPBACK_REQUEST_TIMEOUT_MARGIN_MS = 1_000; + +type FleetSessionResponse = { + ok?: boolean; + data?: { + session_id?: string; + terminal_url?: string; + resume_token?: string; + }; + error?: { code?: string; message?: string }; +}; + +type TerminalFrame = Record & { type?: string; session_id?: string }; + +type TerminalReadiness = { + generation: number; + settled: boolean; + promise: Promise; + resolve: () => void; + reject: (error: Error) => void; +}; + +export interface FleetNodeAttachOptions { + /** Defaults to the only agent on the node; ambiguous nodes require an explicit agent. */ + agent?: string; + node?: string; + nodeId?: string; + mode: AttachMode; + env?: NodeJS.ProcessEnv; + baseUrl?: string; + workspaceKey?: string; + fetch?: typeof globalThis.fetch; + /** Deterministic test seam for the bounded session-request retry delay. */ + sessionRequest?: { + timeoutMs?: number; + totalTimeoutMs?: number; + sleep?: (ms: number) => Promise; + }; + /** Deterministic test seam for established-session reconnect timing. */ + reconnectDelay?: { + initialMs?: number; + maxMs?: number; + handshakeTimeoutMs?: number; + readyTimeoutMs?: number; + beforeReadyTimeoutTerminate?: (socket: WebSocket) => void; + }; +} + +/** Internal compatibility transport for broker-based CLI attach clients. */ +export interface FleetNodeAttachTransport { + brokerUrl: string; + apiKey: string; + requestTimeoutMs: number; +} + +export interface FleetNodeAttachProxy extends FleetNodeAttachTransport { + /** Private UNIX byte-stream socket (named pipe on Windows). */ + socketPath: string; + /** Inferred status: 0 on terminal.closed or detach, 1 on transport failure. + * Relaycast does not currently transmit the remote process exit code. */ + finished: Promise; + close(): Promise; +} + +export class FleetNodeAttachError extends Error { + constructor( + message: string, + readonly code?: string + ) { + super(message); + this.name = 'FleetNodeAttachError'; + } +} + +const TRUSTED_RELAYCAST_ORIGINS = new Set([CANONICAL_RELAYCAST_ORIGIN, AGENT37_RELAYCAST_ORIGIN]); + +export function validateFleetAttachBaseUrl(value: string): string { + let parsed: URL; + try { + parsed = new URL(value); + } catch { + throw new FleetNodeAttachError('Fleet node attach requires a trusted Relaycast origin.'); + } + if ( + parsed.protocol !== 'https:' || + parsed.username || + parsed.password || + parsed.port || + parsed.search || + parsed.hash || + (parsed.pathname !== '' && parsed.pathname !== '/') || + !TRUSTED_RELAYCAST_ORIGINS.has(parsed.origin) + ) { + throw new FleetNodeAttachError('Fleet node attach requires a trusted Relaycast origin.'); + } + return parsed.origin; +} + +class TerminalSessionAttemptError extends FleetNodeAttachError { + constructor( + message: string, + code: string | undefined, + readonly status: number | undefined, + readonly retryable: boolean, + readonly completionUnknown: boolean + ) { + super(message, code); + } +} + +/** + * Canonical HTTP status for a terminal failure code. + * + * `agent_not_found` must stay 404: {@link switchInboundDeliveryModeOrAbort} + * only emits the "no agent named X" message and the cross-node placement hint + * on a 404, so collapsing it into 503 replaces actionable guidance with an + * opaque unreachable-node error. `unsupported_runtime` stays 409. Everything + * else is a transport-level failure and reports 503. + */ +function terminalErrorStatus(code: string | undefined): number { + if (code === 'agent_not_found') return 404; + if (code === 'unsupported_runtime') return 409; + return 503; +} + +function json(response: ServerResponse, status: number, payload: unknown): void { + response.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); + response.end(JSON.stringify(payload)); +} + +function readBody(request: IncomingMessage): Promise> { + return new Promise((resolve) => { + let settled = false; + const finish = (body: Record) => { + if (settled) return; + settled = true; + resolve(body); + }; + let body = ''; + request.setEncoding('utf8'); + request.on('data', (chunk: string) => { + body += chunk; + }); + request.on('end', () => { + try { + const parsed = JSON.parse(body) as unknown; + finish( + parsed && typeof parsed === 'object' && !Array.isArray(parsed) + ? (parsed as Record) + : {} + ); + } catch { + finish({}); + } + }); + request.on('error', () => finish({})); + request.on('aborted', () => finish({})); + }); +} + +function asWsUrl(value: string): string { + const lower = value.toLowerCase(); + if (lower.startsWith('https://')) return 'wss://' + value.slice(8); + if (lower.startsWith('http://')) return 'ws://' + value.slice(7); + return value; +} + +function safeNodePath(node: string): string { + const trimmed = node.trim().replace(/^#/, ''); + if (!trimmed) throw new FleetNodeAttachError('Error: --node requires a node name or id.', 'invalid_node'); + return encodeURIComponent(trimmed); +} + +function parseFrame(data: WebSocket.RawData): TerminalFrame | null { + try { + const parsed = JSON.parse(rawDataToString(data)) as unknown; + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as TerminalFrame) : null; + } catch { + return null; + } +} + +function rawDataToString(data: WebSocket.RawData): string { + if (Buffer.isBuffer(data)) return data.toString('utf8'); + if (data instanceof ArrayBuffer) return Buffer.from(data).toString('utf8'); + if (Array.isArray(data)) return Buffer.concat(data).toString('utf8'); + return String(data); +} + +function diagnosticEndpoint(value: string): string { + try { + const url = new URL(value); + url.username = ''; + url.password = ''; + url.search = ''; + url.hash = ''; + return url.toString(); + } catch { + return '(invalid endpoint)'; + } +} + +function diagnosticEndpointOrigin(value: string): string { + try { + return new URL(value).origin; + } catch { + return '(invalid endpoint)'; + } +} + +function boundedWebSocketCloseReason(reason: string): string { + const encoded = Buffer.from(reason, 'utf8'); + if (encoded.length <= MAX_WEBSOCKET_CLOSE_REASON_BYTES) return reason; + + const suffix = Buffer.from('…', 'utf8'); + let end = MAX_WEBSOCKET_CLOSE_REASON_BYTES - suffix.length; + // Do not cut through a UTF-8 continuation sequence. Excluding the leading + // byte at this boundary also excludes the incomplete code point. + while (end > 0 && (encoded[end] & 0xc0) === 0x80) end -= 1; + return Buffer.concat([encoded.subarray(0, end), suffix]).toString('utf8'); +} + +function diagnosticValue(value: string): string { + // JSON quoting keeps node names and upstream text from injecting terminal + // control characters into the operator-facing error. + return JSON.stringify(value); +} + +function resolvedNodeIdFromTerminalUrl(value: string): string | undefined { + try { + const match = /^\/v1\/nodes\/([^/]+)\/terminal\/connect$/.exec(new URL(value).pathname); + return match?.[1] ? decodeURIComponent(match[1]) : undefined; + } catch { + return undefined; + } +} + +function retryDelayBudgetMs(attempts: number, initialMs: number, maxMs: number): number { + let total = 0; + for (let attempt = 0; attempt < attempts; attempt += 1) { + total += Math.min(initialMs * 2 ** attempt, maxMs); + } + return total; +} + +function terminalSessionFailureSummary(error: TerminalSessionAttemptError): string { + if (error.code === 'node_not_found') return 'Control-plane node lookup found no matching record'; + if (error.code === 'node_unreachable') { + return /no terminal transport/i.test(error.message) + ? 'The node record was found, but its terminal transport was unavailable' + : 'The control plane classified the node as unreachable'; + } + if (error.code === 'terminal_session_unavailable') { + return 'The control plane could not allocate a terminal session'; + } + if (error.code === 'control_plane_timeout') return 'Control-plane terminal-session lookup timed out'; + if (error.code === 'control_plane_unavailable') return 'Control-plane terminal-session lookup failed'; + return 'The terminal-session request was rejected'; +} + +function isRetryableTerminalSessionFailure(code: string | undefined): boolean { + // These structured responses are emitted before a session is returned, so + // retrying cannot duplicate a successful allocation. A fetch timeout, + // network failure, or unclassified 5xx is different: this POST may already + // have completed server-side, and retrying it could create a second session. + return code === 'node_unreachable' || code === 'terminal_session_unavailable'; +} + +/** Start a broker-compatible loopback proxy for one remote terminal session. */ +export async function startFleetNodeAttachProxy( + input: FleetNodeAttachOptions +): Promise { + const options = { ...input, node: input.node ?? input.nodeId ?? '', agent: input.agent ?? '' }; + if (input.node && input.nodeId && input.node !== input.nodeId) { + throw new FleetNodeAttachError('node and nodeId must identify the same node.', 'invalid_node'); + } + const env = options.env ?? process.env; + const fetchFn = options.fetch ?? globalThis.fetch; + const { workspaceKey, baseUrl: requestedBaseUrl } = resolveWorkspaceTransport({ + workspaceKey: options.workspaceKey, + baseUrl: options.baseUrl, + env, + }); + const baseUrl = validateFleetAttachBaseUrl(requestedBaseUrl ?? CANONICAL_RELAYCAST_ORIGIN); + const nodePath = safeNodePath(options.node); + if (!options.agent) { + const response = await fetchFn(`${baseUrl}/v1/nodes/${nodePath}/agents`, { + headers: { Authorization: `Bearer ${workspaceKey}` }, + signal: AbortSignal.timeout(SESSION_REQUEST_TIMEOUT_MS), + }); + const payload = (await response.json()) as { data?: Array<{ agentName?: string }> }; + if (!response.ok || !Array.isArray(payload.data)) { + throw new FleetNodeAttachError('Could not discover the fleet node agent.', 'agent_discovery_failed'); + } + const names = [ + ...new Set( + payload.data + .map((agent) => agent.agentName) + .filter((name): name is string => typeof name === 'string' && !!name.trim()) + ), + ]; + if (names.length !== 1) { + throw new FleetNodeAttachError( + 'Specify agent explicitly unless the node has exactly one agent.', + 'ambiguous_agent' + ); + } + options.agent = names[0]; + } + const sessionEndpoint = `${baseUrl}/v1/nodes/${nodePath}/terminal/sessions`; + const sessionRequestTimeoutMs = options.sessionRequest?.timeoutMs ?? SESSION_REQUEST_TIMEOUT_MS; + const sessionRequestTotalTimeoutMs = + options.sessionRequest?.totalTimeoutMs ?? SESSION_REQUEST_TOTAL_TIMEOUT_MS; + const sessionRequestDeadline = Date.now() + sessionRequestTotalTimeoutMs; + const sessionRequestSleep = + options.sessionRequest?.sleep ?? + ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + let sessionRequestAttempts = 0; + let sessionRequestBudgetExhaustedBetweenAttempts = false; + let lastSessionError: TerminalSessionAttemptError | undefined; + const sessionResult = await collectWithRetry( + 'terminal session request', + async () => { + const remainingRequestBudgetMs = sessionRequestDeadline - Date.now(); + if (remainingRequestBudgetMs <= 0) { + sessionRequestBudgetExhaustedBetweenAttempts = true; + const exhausted = + lastSessionError === undefined + ? new TerminalSessionAttemptError( + 'overall terminal-session request deadline exhausted', + 'control_plane_timeout', + undefined, + false, + false + ) + : new TerminalSessionAttemptError( + lastSessionError.message, + lastSessionError.code, + lastSessionError.status, + false, + lastSessionError.completionUnknown + ); + lastSessionError = exhausted; + throw exhausted; + } + sessionRequestAttempts += 1; + const controller = new AbortController(); + let timedOut = false; + const attemptTimeoutMs = Math.min(sessionRequestTimeoutMs, remainingRequestBudgetMs); + const timeout = setTimeout(() => { + timedOut = true; + controller.abort(); + }, attemptTimeoutMs); + let ticketResponse: Response; + let ticketPayload: FleetSessionResponse; + try { + ticketResponse = await fetchFn(sessionEndpoint, { + method: 'POST', + headers: { Authorization: `Bearer ${workspaceKey}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ agent: options.agent, mode: options.mode }), + signal: controller.signal, + }); + const parsedPayload = (await ticketResponse.json()) as unknown; + ticketPayload = + parsedPayload && typeof parsedPayload === 'object' && !Array.isArray(parsedPayload) + ? (parsedPayload as FleetSessionResponse) + : {}; + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + lastSessionError = new TerminalSessionAttemptError( + timedOut ? 'request exceeded its deadline' : detail, + timedOut ? 'control_plane_timeout' : 'control_plane_unavailable', + undefined, + false, + true + ); + throw lastSessionError; + } finally { + clearTimeout(timeout); + } + + const terminalUrl = ticketPayload.data?.terminal_url; + const sessionId = ticketPayload.data?.session_id; + const resumeToken = ticketPayload.data?.resume_token; + if (!ticketResponse.ok || !terminalUrl || !sessionId || !resumeToken) { + const code = ticketPayload.error?.code; + const message = + ticketPayload.error?.message ?? `terminal session request failed (HTTP ${ticketResponse.status})`; + lastSessionError = new TerminalSessionAttemptError( + message, + code, + ticketResponse.status, + isRetryableTerminalSessionFailure(code), + false + ); + throw lastSessionError; + } + return { terminalUrl, sessionId, resumeToken }; + }, + { + retries: SESSION_REQUEST_RETRIES, + baseDelayMs: SESSION_REQUEST_RETRY_DELAY_MS, + sleep: async (delayMs) => { + const remainingRequestBudgetMs = sessionRequestDeadline - Date.now(); + if (remainingRequestBudgetMs <= 0) return; + await sessionRequestSleep(Math.min(delayMs, remainingRequestBudgetMs)); + }, + shouldRetry: (error) => error instanceof TerminalSessionAttemptError && error.retryable, + } + ); + if (!sessionResult.ok) { + const failure = lastSessionError; + const status = failure?.status === undefined ? '' : ` HTTP ${failure.status};`; + const code = failure?.code === undefined ? '' : ` code ${failure.code};`; + const retryNote = sessionRequestBudgetExhaustedBetweenAttempts + ? sessionRequestAttempts > 1 + ? `retried ${sessionRequestAttempts - 1} time${sessionRequestAttempts === 2 ? '' : 's'};` + + ' overall budget exhausted before the next attempt' + : 'not retried because the overall budget was exhausted before the next attempt' + : sessionRequestAttempts > 1 + ? `retried ${sessionRequestAttempts - 1} time${sessionRequestAttempts === 2 ? '' : 's'}` + + (failure?.completionUnknown + ? '; final POST not retried because it may have completed server-side' + : '') + : failure?.completionUnknown + ? 'not retried because the POST may have completed server-side' + : 'not retried because the failure was terminal'; + const upstream = failure?.message ? ` Upstream message ${diagnosticValue(failure.message)}.` : ''; + throw new FleetNodeAttachError( + `Error: ${failure ? terminalSessionFailureSummary(failure) : 'Terminal-session request failed'}.` + + `${upstream} Node ref ${diagnosticValue(options.node.trim())}, resolved node id unavailable (session creation did not complete);` + + ` endpoint ${diagnosticValue(diagnosticEndpoint(sessionEndpoint))};${status}${code}` + + ` timeout ${sessionRequestTimeoutMs}ms per attempt; overall budget ${sessionRequestTotalTimeoutMs}ms;` + + ` attempts ${sessionRequestAttempts} (${retryNote}).`, + failure?.code + ); + } + const { terminalUrl, sessionId, resumeToken } = sessionResult.value; + const resolvedNodeId = resolvedNodeIdFromTerminalUrl(terminalUrl); + const remoteEndpoint = diagnosticEndpoint(terminalUrl); + const reconnectInitialDelayMs = options.reconnectDelay?.initialMs ?? INITIAL_RECONNECT_DELAY_MS; + const reconnectMaxDelayMs = options.reconnectDelay?.maxMs ?? MAX_RECONNECT_DELAY_MS; + const terminalHandshakeTimeoutMs = + options.reconnectDelay?.handshakeTimeoutMs ?? TERMINAL_CONNECT_TIMEOUT_MS; + const terminalReadyTimeoutMs = options.reconnectDelay?.readyTimeoutMs ?? TERMINAL_CONNECT_TIMEOUT_MS; + // A readiness-gated local request follows activeReadiness across reconnect + // generations. Its own deadline therefore has to cover the same complete, + // finite recovery path: every backoff plus every handshake/readiness pair. + const terminalWaitTimeoutMs = + retryDelayBudgetMs(MAX_RECONNECT_ATTEMPTS, reconnectInitialDelayMs, reconnectMaxDelayMs) + + MAX_RECONNECT_ATTEMPTS * (terminalHandshakeTimeoutMs + terminalReadyTimeoutMs); + + let connectionGeneration = 0; + const createReadiness = (): TerminalReadiness => { + let resolve!: () => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + // A failure can land before a snapshot request attaches its waiter. Keep + // the rejection observable while avoiding an unhandled-rejection warning. + void promise.catch(() => undefined); + return { generation: ++connectionGeneration, settled: false, promise, resolve, reject }; + }; + let activeReadiness = createReadiness(); + const resolveReadiness = (readiness: TerminalReadiness) => { + if (readiness.settled) return; + readiness.settled = true; + readiness.resolve(); + }; + const rejectReadiness = (readiness: TerminalReadiness, error: Error) => { + if (readiness.settled) return; + readiness.settled = true; + readiness.reject(error); + }; + const waitForCurrentReadiness = async (): Promise => { + for (;;) { + const readiness = activeReadiness; + await readiness.promise; + if (readiness === activeReadiness) return; + } + }; + /** + * Await the live readiness generation through both the WebSocket handshake + * and the post-open terminal.ready allowance. + * Every handler that must not run before `terminal.ready` — snapshot, + * delivery-mode PUT, resize — goes through this one helper so the + * timer/clearTimeout/settle logic cannot diverge between copies. + * + * Rejects with the underlying {@link FleetNodeAttachError} when the terminal + * failed (preserving its `code`), or a plain `Error` carrying + * `timeoutMessage` when the wait expired. + */ + const waitForTerminalReady = (timeoutMessage: string): Promise => + new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(timeoutMessage)), terminalWaitTimeoutMs); + void waitForCurrentReadiness().then( + () => { + clearTimeout(timer); + resolve(); + }, + (error: Error) => { + clearTimeout(timer); + reject(error); + } + ); + }); + const snapshot: { screen: string; rows: number; cols: number; offset: number } = { + screen: '', + rows: 24, + cols: 80, + offset: 0, + }; + const eventSockets = new Set(); + const inputSockets = new Set(); + const outputHistory: Array<{ chunk: string; offset?: number }> = []; + let outputHistoryBytes = 0; + let remote: WebSocket | undefined; + let rawSocket: Socket | undefined; + let rawConnected = false; + let rawReady = false; + let finish!: (code: number) => void; + const finished = new Promise((resolve) => { + finish = resolve; + }); + let closePromise: Promise | undefined; + let stopped = false; + let terminalEnded = false; + let terminalEverReady = false; + let reconnecting = false; + let reconnectAttempts = 0; + let reconnectTimer: ReturnType | undefined; + const terminalReadinessTimers = new Set>(); + /** Locally-tracked delivery mode, kept in sync with each broker reply. */ + let loopbackDeliveryMode: 'manual_flush' | 'auto_inject' = + options.mode === 'drive' ? 'manual_flush' : 'auto_inject'; + type DeliveryModeResult = { mode: string; flushed: number; matched: boolean; revision: string }; + type FlushResult = { + flushed: number; + dead_lettered: number; + held: number; + blocked_reason: string | null; + }; + let pendingFlush: { + requestId: string; + resolve: (result: FlushResult) => void; + reject: (error: Error) => void; + timer: ReturnType; + } | null = null; + /** At most one in-flight delivery-mode PUT at a time. */ + let pendingDeliveryMode: { + requestId: string; + resolve: (result: DeliveryModeResult) => void; + reject: (err: Error) => void; + timer: ReturnType; + } | null = null; + const loopbackApiKey = randomBytes(32).toString('base64url'); + const loopbackAuthorized = (headers: IncomingMessage['headers']) => + headers.authorization === `Bearer ${loopbackApiKey}` || headers['x-api-key'] === loopbackApiKey; + + const server = createServer(async (request, response) => { + if (!loopbackAuthorized(request.headers)) { + json(response, 401, { + error: { code: 'unauthorized', message: 'loopback terminal token is required' }, + }); + return; + } + const path = new URL(request.url ?? '/', 'http://127.0.0.1').pathname; + if (request.method === 'GET' && path === `/api/spawned/${encodeURIComponent(options.agent)}/snapshot`) { + try { + await waitForTerminalReady('terminal snapshot timed out'); + } catch (error) { + const terminalError = error instanceof FleetNodeAttachError ? error : undefined; + json(response, terminalErrorStatus(terminalError?.code), { + error: { + code: terminalError?.code ?? 'snapshot_unavailable', + message: + terminalError?.message ?? (error instanceof Error ? error.message : 'snapshot unavailable'), + }, + }); + return; + } + json(response, 200, { format: 'ansi', ...snapshot }); + return; + } + const name = encodeURIComponent(options.agent); + if (path === `/api/spawned/${name}/flush` && request.method === 'POST') { + // Forward the flush to the remote broker over the terminal websocket. + // Without this the `--node` form of `node agent message flush` reached + // only the LOCAL broker's worker registry and returned agent_not_found + // for a name that `node agent list` and `attach` both resolve. + try { + await waitForTerminalReady('terminal connection timed out'); + } catch (error) { + const terminalError = error instanceof FleetNodeAttachError ? error : undefined; + json(response, terminalErrorStatus(terminalError?.code), { + error: { + code: terminalError?.code ?? 'node_unreachable', + message: + terminalError?.message ?? (error instanceof Error ? error.message : 'terminal unavailable'), + }, + }); + return; + } + if (!remote || remote.readyState !== WebSocket.OPEN) { + json(response, 503, { + error: { code: 'node_unreachable', message: 'terminal transport is not connected' }, + }); + return; + } + if (pendingFlush) { + json(response, 503, { + error: { code: 'flush_conflict', message: 'a flush request is already in flight' }, + }); + return; + } + const requestId = randomBytes(8).toString('hex'); + const result = await new Promise((resolve) => { + const timer = setTimeout(() => { + pendingFlush = null; + resolve(new FleetNodeAttachError('flush request timed out', 'flush_timeout')); + }, DELIVERY_MODE_TIMEOUT_MS); + pendingFlush = { + requestId, + resolve: (r) => resolve(r), + reject: (e) => resolve(e), + timer, + }; + remote!.send( + JSON.stringify({ + type: 'terminal.flush_pending', + session_id: sessionId, + request_id: requestId, + }) + ); + }); + if (result instanceof Error) { + const errCode = + result instanceof FleetNodeAttachError ? (result.code ?? 'flush_failed') : 'flush_failed'; + json(response, terminalErrorStatus(errCode), { + error: { code: errCode, message: result.message }, + }); + return; + } + json(response, 200, result); + return; + } + if (path === `/api/spawned/${name}/delivery-mode`) { + if (request.method === 'GET') { + json(response, 200, { mode: loopbackDeliveryMode }); + return; + } + // PUT — forward the request to the remote broker via the terminal WS and + // await the broker's real reply. This is the path that was previously a + // static stub returning manual_flush, causing drive attach to fail with + // "broker remained in manual_flush mode". + const body = await readBody(request); + const requestedMode = + body.mode === 'auto_inject' ? 'auto_inject' : body.mode === 'manual_flush' ? 'manual_flush' : null; + if (requestedMode === null) { + json(response, 400, { + error: { code: 'invalid_mode', message: `unsupported delivery mode '${String(body.mode)}'` }, + }); + return; + } + // Drive attach changes delivery mode before it requests the initial + // snapshot. Gate the PUT on terminal.ready so a fast local caller does + // not lose a race with the remote websocket handshake and receive the + // misleading "terminal transport is not connected" failure. + try { + await waitForTerminalReady('terminal connection timed out'); + } catch (error) { + // Preserve the canonical status mapping here too: a readiness failure + // carrying `agent_not_found` has to reach the preflight as a 404 or + // the operator loses the cross-node placement hint that tells them + // which machine to run the attach on. + const terminalError = error instanceof FleetNodeAttachError ? error : undefined; + json(response, terminalErrorStatus(terminalError?.code), { + error: { + code: terminalError?.code ?? 'node_unreachable', + message: + terminalError?.message ?? (error instanceof Error ? error.message : 'terminal unavailable'), + }, + }); + return; + } + if (!remote || remote.readyState !== WebSocket.OPEN) { + json(response, 503, { + error: { code: 'node_unreachable', message: 'terminal transport is not connected' }, + }); + return; + } + if (pendingDeliveryMode) { + json(response, 503, { + error: { code: 'delivery_mode_conflict', message: 'a delivery mode request is already in flight' }, + }); + return; + } + const requestId = randomBytes(8).toString('hex'); + const result = await new Promise((resolve) => { + const timer = setTimeout(() => { + pendingDeliveryMode = null; + resolve(new FleetNodeAttachError('delivery mode request timed out', 'delivery_mode_timeout')); + }, DELIVERY_MODE_TIMEOUT_MS); + pendingDeliveryMode = { + requestId, + resolve: (r) => resolve(r), + reject: (e) => resolve(e), + timer, + }; + const frame: Record = { + type: 'terminal.set_delivery_mode', + session_id: sessionId, + mode: requestedMode, + request_id: requestId, + }; + if (typeof body.expected_mode === 'string') frame.expected_mode = body.expected_mode; + if (typeof body.expected_revision === 'string') frame.expected_revision = body.expected_revision; + remote!.send(JSON.stringify(frame)); + }); + if (result instanceof Error) { + const errCode = + result instanceof FleetNodeAttachError + ? (result.code ?? 'delivery_mode_failed') + : 'delivery_mode_failed'; + // Same canonical mapping as the readiness gate: 404 for + // agent_not_found so the attach preflight can produce the no-agent or + // cross-node placement error, 409 for unsupported_runtime, 503 for + // everything else. + json(response, terminalErrorStatus(errCode), { + error: { code: errCode, message: result.message }, + }); + return; + } + loopbackDeliveryMode = result.mode === 'manual_flush' ? 'manual_flush' : 'auto_inject'; + json(response, 200, { + mode: result.mode, + flushed: result.flushed, + matched: result.matched, + revision: result.revision, + }); + return; + } + if (request.method === 'GET' && path === `/api/spawned/${name}/pending`) { + json(response, 200, { pending: [] }); + return; + } + if (request.method === 'POST' && path === `/api/spawned/${name}/flush`) { + json(response, 200, { flushed: 0 }); + return; + } + if (request.method === 'GET' && path === '/api/spawned') { + json(response, 200, { agents: [{ name: options.agent, workerPid: 1 }] }); + return; + } + if (request.method === 'POST' && path === `/api/resize/${name}`) { + const body = await readBody(request); + if (body.release === true) { + json(response, 200, { name: options.agent, released: true }); + return; + } + const rows = typeof body.rows === 'number' ? body.rows : 0; + const cols = typeof body.cols === 'number' ? body.cols : 0; + if (!Number.isInteger(rows) || !Number.isInteger(cols) || rows < 1 || cols < 1) { + json(response, 400, { + error: { code: 'invalid_dimensions', message: 'rows and cols must be positive integers' }, + }); + return; + } + try { + await waitForTerminalReady('terminal resize timed out'); + } catch (error) { + json(response, 503, { + error: { + code: 'session_not_ready', + message: error instanceof Error ? error.message : 'terminal session is not ready', + }, + }); + return; + } + if (!remote || remote.readyState !== WebSocket.OPEN || remote.bufferedAmount > MAX_BUFFERED_BYTES) { + json(response, 503, { + error: { code: 'node_unreachable', message: 'terminal transport is unavailable' }, + }); + return; + } + remote.send(JSON.stringify({ type: 'terminal.resize', session_id: sessionId, rows, cols })); + json(response, 200, { name: options.agent, rows, cols, applied: true }); + return; + } + json(response, 404, { error: { code: 'not_found', message: 'loopback terminal endpoint not found' } }); + }); + const websocketServer = new WebSocketServer({ noServer: true }); + + const closeSocket = (socket: WebSocket, code: number, reason: string) => { + try { + socket.close(code, boundedWebSocketCloseReason(reason)); + } catch { + /* connection already gone */ + } + }; + const broadcast = (sockets: Set, payload: unknown): boolean => { + const encoded = JSON.stringify(payload); + let accepted = false; + for (const socket of sockets) { + if (socket.readyState !== WebSocket.OPEN) continue; + if (socket.bufferedAmount > MAX_BUFFERED_BYTES) { + closeSocket(socket, 1013, 'loopback client backpressure exceeded'); + sockets.delete(socket); + continue; + } + try { + socket.send(encoded); + accepted = true; + } catch { + sockets.delete(socket); + } + } + return accepted; + }; + const workerStreamEvent = (chunk: string, offset?: number) => ({ + kind: 'worker_stream', + name: options.agent, + stream: 'stdout', + chunk, + ...(offset === undefined ? {} : { offset }), + }); + const retainOutput = (chunk: string, offset: number | undefined): boolean => { + const bytes = Buffer.byteLength(chunk, 'utf8'); + if (outputHistoryBytes + bytes > MAX_BUFFERED_BYTES) return false; + outputHistory.push({ chunk, ...(offset === undefined ? {} : { offset }) }); + outputHistoryBytes += bytes; + return true; + }; + + websocketServer.on('connection', (socket, request) => { + const path = new URL(request.url ?? '/', 'http://127.0.0.1').pathname; + if (path === '/ws') { + eventSockets.add(socket); + socket.on('close', () => eventSockets.delete(socket)); + let replayed = 0; + for (const event of outputHistory) { + if (socket.readyState !== WebSocket.OPEN || socket.bufferedAmount > MAX_BUFFERED_BYTES) break; + try { + socket.send(JSON.stringify(workerStreamEvent(event.chunk, event.offset))); + } catch { + break; + } + replayed += 1; + } + if (replayed > 0) { + const sentBytes = outputHistory + .slice(0, replayed) + .reduce((total, event) => total + Buffer.byteLength(event.chunk, 'utf8'), 0); + outputHistory.splice(0, replayed); + outputHistoryBytes -= sentBytes; + } + return; + } + if (path === `/api/input/${encodeURIComponent(options.agent)}/stream`) { + inputSockets.add(socket); + socket.on('close', () => inputSockets.delete(socket)); + socket.send(JSON.stringify({ type: 'pty_input_ready', name: options.agent })); + socket.on('message', (data) => { + if (!remote || remote.readyState !== WebSocket.OPEN || remote.bufferedAmount > MAX_BUFFERED_BYTES) { + broadcast(inputSockets, { + type: 'pty_input_error', + code: 'node_unreachable', + message: 'terminal transport is unavailable', + }); + return; + } + const raw = rawDataToString(data); + remote.send( + JSON.stringify({ + type: 'terminal.input', + session_id: sessionId, + data_base64: Buffer.from(raw, 'utf8').toString('base64'), + }) + ); + }); + return; + } + closeSocket(socket, 1008, 'unknown loopback endpoint'); + }); + server.on('upgrade', (request, socket, head) => { + if (!loopbackAuthorized(request.headers)) { + socket.write('HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n'); + socket.destroy(); + return; + } + websocketServer.handleUpgrade(request, socket, head, (client) => + websocketServer.emit('connection', client, request) + ); + }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') + throw new FleetNodeAttachError( + 'Error: could not allocate loopback terminal listener.', + 'loopback_unavailable' + ); + + let socketDirectory: string | undefined; + let socketPath: string; + const rawServer = createSocketServer((socket) => { + if (rawConnected || stopped || terminalEnded) { + socket.destroy(); + return; + } + rawConnected = true; + rawSocket = socket; + socket.pause(); + socket.on('error', () => socket.destroy()); + socket.on('close', () => { + if (rawSocket === socket) rawSocket = undefined; + }); + socket.on('data', (data: Buffer) => { + if (options.mode === 'view') return; + if ( + !remote || + remote.readyState !== WebSocket.OPEN || + remote.bufferedAmount + data.length > MAX_BUFFERED_BYTES + ) { + endTerminal(new FleetNodeAttachError('terminal input transport unavailable', 'node_unreachable')); + return; + } + remote.send( + JSON.stringify({ + type: 'terminal.input', + session_id: sessionId, + data_base64: data.toString('base64'), + }) + ); + }); + void waitForTerminalReady('terminal socket readiness timed out').then( + () => { + if (socket.destroyed || stopped || terminalEnded) return; + if (snapshot.screen) socket.write(snapshot.screen); + for (const entry of outputHistory) socket.write(entry.chunk); + outputHistory.length = 0; + outputHistoryBytes = 0; + rawReady = true; + socket.resume(); + }, + () => socket.destroy() + ); + }); + try { + socketPath = + process.platform === 'win32' + ? `\\\\.\\pipe\\agent-relay-attach-${randomBytes(16).toString('hex')}` + : join((socketDirectory = await mkdtemp(join(tmpdir(), 'relay-attach-'))), 'stdio.sock'); + rawServer.listen(socketPath); + await once(rawServer, 'listening'); + if (socketDirectory) await chmod(socketPath, 0o600); + } catch (error) { + rawServer.close(); + websocketServer.close(); + await new Promise((resolve) => server.close(() => resolve())); + if (socketDirectory) await rm(socketDirectory, { recursive: true, force: true }); + throw error; + } + + const resumeUrl = new URL(terminalUrl); + resumeUrl.searchParams.delete('ticket'); + resumeUrl.searchParams.set('session_id', sessionId); + resumeUrl.searchParams.set('resume', resumeToken); + /** Reject and clear any in-flight delivery-mode PUT, if one is pending. */ + const rejectPendingDeliveryMode = (error: FleetNodeAttachError) => { + if (pendingFlush) { + const flush = pendingFlush; + pendingFlush = null; + clearTimeout(flush.timer); + // A flush never changes delivery mode, so it must not inherit the + // delivery-mode disconnect wording. An operator debugging a failed flush + // would otherwise be told the "delivery-mode change" was interrupted — + // pointing at an operation their command never performed. + const flushError = + error.code === 'delivery_mode_disconnected' + ? new FleetNodeAttachError( + 'terminal transport disconnected while the flush was in flight', + 'flush_disconnected' + ) + : error; + flush.reject(flushError); + } + if (!pendingDeliveryMode) return; + const pending = pendingDeliveryMode; + pendingDeliveryMode = null; + clearTimeout(pending.timer); + pending.reject(error); + }; + const endTerminal = (error: FleetNodeAttachError, eventCloseReason = error.message) => { + if (terminalEnded) return; + terminalEnded = true; + finish(error.code === 'terminal_closed' ? 0 : 1); + rawSocket?.end(); + if (reconnectTimer) { + clearTimeout(reconnectTimer); + reconnectTimer = undefined; + } + for (const timer of terminalReadinessTimers) clearTimeout(timer); + terminalReadinessTimers.clear(); + rejectPendingDeliveryMode(error); + const activeRemote = remote; + remote = undefined; + rejectReadiness(activeReadiness, error); + broadcast(inputSockets, { type: 'pty_input_error', code: error.code, message: error.message }); + for (const socket of eventSockets) closeSocket(socket, 1011, eventCloseReason); + if (activeRemote && activeRemote.readyState !== WebSocket.CLOSED) { + try { + activeRemote.terminate(); + } catch { + // The socket may have closed between the state check and terminate. + } + } + }; + const failRemote = (message: string, eventCloseReason?: string) => { + endTerminal(new FleetNodeAttachError(message, 'node_unreachable'), eventCloseReason); + }; + const connect = (url: string, readiness: TerminalReadiness) => { + if (stopped || terminalEnded) return; + const socket = new WebSocket(asWsUrl(url), { handshakeTimeout: terminalHandshakeTimeoutMs }); + remote = socket; + let readinessTimer: ReturnType | undefined; + let readinessExpired = false; + const clearReadinessTimer = () => { + if (!readinessTimer) return; + clearTimeout(readinessTimer); + terminalReadinessTimers.delete(readinessTimer); + readinessTimer = undefined; + }; + socket.on('open', () => { + // `handshakeTimeout` independently bounds the HTTP upgrade. Start the + // terminal.ready allowance only after that upgrade succeeds so a slow + // but valid handshake cannot consume the readiness window. + readinessTimer = setTimeout(() => { + const expiredTimer = readinessTimer; + readinessTimer = undefined; + if (expiredTimer) terminalReadinessTimers.delete(expiredTimer); + if (remote !== socket || stopped || terminalEnded || readiness.settled) return; + // Mark this generation stale before terminating. The ws receiver may + // still deliver data already buffered on the socket while close is + // propagating; none of it may restore readiness or reset retry state. + readinessExpired = true; + if (!terminalEverReady) { + failRemote( + `terminal transport connected but did not become ready (node ref ${diagnosticValue(options.node.trim())},` + + ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + + ` readiness timeout ${terminalReadyTimeoutMs}ms, attempts 1; not retried because no terminal session became ready)` + ); + return; + } + // A successful WebSocket upgrade is not sufficient: Relaycast may + // accept a resume lane that never produces terminal.ready. Terminating + // it drives the same bounded close/retry path as a transport failure. + options.reconnectDelay?.beforeReadyTimeoutTerminate?.(socket); + socket.terminate(); + }, terminalReadyTimeoutMs); + terminalReadinessTimers.add(readinessTimer); + }); + socket.on('message', (data) => { + // A late frame from a transport superseded during reconnect must never + // overwrite the fresh snapshot or end the replacement session. + if (remote !== socket || stopped || terminalEnded || readinessExpired) return; + const frame = parseFrame(data); + if (!frame || frame.session_id !== sessionId) return; + if (frame.type === 'terminal.ready') { + clearReadinessTimer(); + snapshot.screen = typeof frame.screen === 'string' ? frame.screen : ''; + snapshot.rows = typeof frame.rows === 'number' ? frame.rows : 24; + snapshot.cols = typeof frame.cols === 'number' ? frame.cols : 80; + snapshot.offset = typeof frame.offset === 'number' ? frame.offset : 0; + // Seed loopbackDeliveryMode from the broker's actual state so that + // detach restores the correct mode even when the worker started in a + // different mode than our local inference at line 214. + if (frame.delivery_mode === 'manual_flush' || frame.delivery_mode === 'auto_inject') { + loopbackDeliveryMode = frame.delivery_mode; + } + terminalEverReady = true; + reconnectAttempts = 0; + if (readiness === activeReadiness) { + resolveReadiness(readiness); + // A reconnect gets a fresh ANSI grid but existing local `/ws` + // consumers have already performed their initial HTTP snapshot. + // Re-emit this screen without an offset so they repaint instead of + // retaining a stale pre-reconnect terminal image. + if (readiness.generation > 1 && snapshot.screen) { + broadcast(eventSockets, workerStreamEvent(snapshot.screen)); + rawSocket?.write(snapshot.screen); + } + } + } else if (frame.type === 'terminal.output' && typeof frame.chunk === 'string') { + const offset = typeof frame.offset === 'number' ? frame.offset : undefined; + if (rawReady && rawSocket && !rawSocket.destroyed) { + if (rawSocket.writableLength + Buffer.byteLength(frame.chunk) > MAX_BUFFERED_BYTES) { + endTerminal( + new FleetNodeAttachError( + 'terminal output exceeded the bounded socket buffer', + 'output_backpressure' + ) + ); + return; + } + rawSocket.write(frame.chunk); + } + if (!broadcast(eventSockets, workerStreamEvent(frame.chunk, offset)) && !rawReady) { + if (!retainOutput(frame.chunk, offset)) { + endTerminal( + new FleetNodeAttachError( + 'terminal output exceeded the bounded loopback buffer', + 'output_backpressure' + ) + ); + } + } + } else if (frame.type === 'terminal.input_ack') { + broadcast(inputSockets, { + type: 'pty_input_ack', + name: options.agent, + bytes_written: typeof frame.bytes_written === 'number' ? frame.bytes_written : 0, + }); + } else if (frame.type === 'terminal.delivery_mode') { + const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; + if (pendingDeliveryMode && (frameRid === undefined || frameRid === pendingDeliveryMode.requestId)) { + const pending = pendingDeliveryMode; + pendingDeliveryMode = null; + clearTimeout(pending.timer); + pending.resolve({ + mode: typeof frame.mode === 'string' ? frame.mode : 'auto_inject', + flushed: typeof frame.flushed === 'number' ? frame.flushed : 0, + matched: typeof frame.matched === 'boolean' ? frame.matched : true, + revision: typeof frame.revision === 'string' ? frame.revision : '1', + }); + } + } else if (frame.type === 'terminal.flush_pending') { + const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; + // Exact match only. The proxy always sends a request_id, and unlike + // delivery-mode there is no older-broker reply shape to stay + // compatible with, so a reply without one is not ours — accepting it + // would resolve the caller's flush with an unrelated result. + if (pendingFlush && frameRid === pendingFlush.requestId) { + const pending = pendingFlush; + pendingFlush = null; + clearTimeout(pending.timer); + pending.resolve({ + flushed: typeof frame.flushed === 'number' ? frame.flushed : 0, + dead_lettered: typeof frame.dead_lettered === 'number' ? frame.dead_lettered : 0, + held: typeof frame.held === 'number' ? frame.held : 0, + blocked_reason: typeof frame.blocked_reason === 'string' ? frame.blocked_reason : null, + }); + } + } else if (frame.type === 'terminal.error') { + const message = typeof frame.message === 'string' ? frame.message : 'remote terminal failed'; + const code = typeof frame.code === 'string' ? frame.code : 'terminal_error'; + const frameRid = typeof frame.request_id === 'string' ? frame.request_id : undefined; + // Route the error to the pending delivery-mode request only when the + // request_id matches (or the broker sent no request_id at all — older + // broker compat). An unrelated session-level error must not cancel a + // live delivery-mode PUT and vice-versa. + if (pendingFlush && frameRid !== undefined && frameRid === pendingFlush.requestId) { + const pending = pendingFlush; + pendingFlush = null; + clearTimeout(pending.timer); + pending.reject(new FleetNodeAttachError(message, code)); + } else if ( + pendingDeliveryMode && + (frameRid === undefined || frameRid === pendingDeliveryMode.requestId) + ) { + const pending = pendingDeliveryMode; + pendingDeliveryMode = null; + clearTimeout(pending.timer); + pending.reject(new FleetNodeAttachError(message, code)); + } else if (!frameRid && readiness === activeReadiness && !readiness.settled) { + endTerminal(new FleetNodeAttachError(message, code)); + } else if (!frameRid) { + broadcast(inputSockets, { type: 'pty_input_error', code, message }); + } + } else if (frame.type === 'terminal.closed') { + endTerminal(new FleetNodeAttachError('remote terminal session closed', 'terminal_closed')); + } + }); + socket.on('error', () => { + readinessExpired = true; + clearReadinessTimer(); + // Initial connection failure has no terminal state worth preserving. + // Fail promptly with the canonical unavailable-node error instead of + // letting the HTTP snapshot timeout mask it. Once Ready has been seen, + // the close handler retains the bounded resume/backoff behaviour. + if (remote === socket && readiness === activeReadiness && !readiness.settled && !terminalEverReady) { + failRemote( + `terminal transport could not connect to the fleet node (node ref ${diagnosticValue(options.node.trim())},` + + ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + + ` handshake budget ${terminalHandshakeTimeoutMs}ms, attempts 1; not retried because no terminal session became ready)` + ); + } + }); + socket.on('close', () => { + readinessExpired = true; + clearReadinessTimer(); + if (remote !== socket || stopped || terminalEnded || reconnecting) return; + // Relaycast drops the old lane's terminal session state on disconnect, + // so a set_delivery_mode frame already sent on this dying socket is + // lost and will never get a reply on the replacement socket — even + // once reconnect succeeds. Fail the pending PUT fast with a retryable + // error instead of leaving it to hang out the full timeout. + rejectPendingDeliveryMode( + new FleetNodeAttachError( + 'terminal transport disconnected while the delivery-mode change was in flight', + 'delivery_mode_disconnected' + ) + ); + // Any waiter that observed the prior connection must retry against the + // fresh generation instead of receiving its stale resolved snapshot. + resolveReadiness(readiness); + const nextReadiness = createReadiness(); + activeReadiness = nextReadiness; + if (reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) { + const backoffBudgetMs = retryDelayBudgetMs( + MAX_RECONNECT_ATTEMPTS, + reconnectInitialDelayMs, + reconnectMaxDelayMs + ); + const message = + `terminal transport could not reconnect to the fleet node (node ref ${diagnosticValue(options.node.trim())},` + + ` resolved node id ${diagnosticValue(resolvedNodeId ?? 'unavailable')}, endpoint ${diagnosticValue(remoteEndpoint)},` + + ` handshake timeout ${terminalHandshakeTimeoutMs}ms, readiness timeout ${terminalReadyTimeoutMs}ms,` + + ` attempts ${reconnectAttempts},` + + ` backoff budget ${backoffBudgetMs}ms)`; + failRemote( + message, + `terminal reconnect failed; attempts=${reconnectAttempts}; budget=${backoffBudgetMs}ms;` + + ` endpoint=${diagnosticValue(diagnosticEndpointOrigin(terminalUrl))};` + + ` node=${diagnosticValue(resolvedNodeId ?? options.node.trim())}` + ); + return; + } + reconnecting = true; + const delay = Math.min(reconnectInitialDelayMs * 2 ** reconnectAttempts, reconnectMaxDelayMs); + reconnectAttempts += 1; + reconnectTimer = setTimeout(() => { + reconnectTimer = undefined; + reconnecting = false; + connect(resumeUrl.toString(), nextReadiness); + }, delay); + }); + }; + connect(terminalUrl, activeReadiness); + + return { + socketPath, + finished, + brokerUrl: `http://127.0.0.1:${address.port}`, + apiKey: loopbackApiKey, + requestTimeoutMs: terminalWaitTimeoutMs + DELIVERY_MODE_TIMEOUT_MS + LOOPBACK_REQUEST_TIMEOUT_MARGIN_MS, + close() { + return (closePromise ??= (async () => { + if (stopped) return; + stopped = true; + finish(0); + rawSocket?.destroy(); + terminalEnded = true; + if (reconnectTimer) { + clearTimeout(reconnectTimer); + reconnectTimer = undefined; + } + for (const timer of terminalReadinessTimers) clearTimeout(timer); + terminalReadinessTimers.clear(); + rejectReadiness(activeReadiness, new FleetNodeAttachError('terminal attach closed', 'closed')); + rejectPendingDeliveryMode(new FleetNodeAttachError('terminal attach closed', 'closed')); + const activeRemote = remote; + remote = undefined; + if (activeRemote && activeRemote.readyState === WebSocket.OPEN) { + try { + activeRemote.send(JSON.stringify({ type: 'terminal.close', session_id: sessionId })); + } catch { + // Best effort; terminate below still prevents a late reconnect. + } + } + if (activeRemote && activeRemote.readyState !== WebSocket.CLOSED) { + try { + activeRemote.terminate(); + } catch { + // Socket is already gone. + } + } + for (const socket of [...eventSockets, ...inputSockets]) + closeSocket(socket, 1000, 'terminal attach closed'); + websocketServer.close(); + await Promise.all([ + new Promise((resolve) => server.close(() => resolve())), + new Promise((resolve) => rawServer.close(() => resolve())), + ]); + if (socketDirectory) await rm(socketDirectory, { recursive: true, force: true }); + })()); + }, + }; +} + +export type { AttachMode } from './attach-mode.js'; +export { collectWithRetry } from './collect-with-retry.js'; diff --git a/packages/cloud/src/collect-with-retry.ts b/packages/cloud/src/collect-with-retry.ts new file mode 100644 index 000000000..fb726a231 --- /dev/null +++ b/packages/cloud/src/collect-with-retry.ts @@ -0,0 +1,42 @@ +export interface CollectWithRetryOptions { + retries?: number; + baseDelayMs?: number; + sleep?: (ms: number) => Promise; + /** Return false for terminal failures that must be surfaced immediately. */ + shouldRetry?: (error: unknown, attempt: number) => boolean; +} + +/** + * Run a bounded operation with the fleet CLI's small retry delay. + * + * The result records whether another attempt actually ran. Callers that need + * richer failure metadata can retain the thrown value in their operation; + * this helper deliberately keeps the existing fleet-list rendering contract. + */ +export async function collectWithRetry( + label: string, + fn: () => Promise, + options?: CollectWithRetryOptions +): Promise<{ ok: true; value: T; retried: boolean } | { ok: false; error: string; retried: boolean }> { + const retries = options?.retries ?? 1; + const baseDelay = options?.baseDelayMs ?? 500; + const sleep = options?.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms))); + let lastError: unknown; + let attempts = 0; + for (let attempt = 0; attempt <= retries; attempt += 1) { + attempts += 1; + try { + const value = await fn(); + return { ok: true, value, retried: attempt > 0 }; + } catch (error) { + lastError = error; + if (attempt >= retries || options?.shouldRetry?.(error, attempt + 1) === false) break; + // Keep the established fleet-list delay shape while making later + // attempts wait slightly longer, without introducing test randomness. + const delay = baseDelay + Math.floor(baseDelay * (attempt / (retries + 1))); + await sleep(delay); + } + } + const message = lastError instanceof Error ? lastError.message : String(lastError); + return { ok: false, error: `${label}: ${message}`, retried: attempts > 1 }; +} diff --git a/packages/cloud/src/fleet-sandbox.test.ts b/packages/cloud/src/fleet-sandbox.test.ts index d90e1800f..242c1240f 100644 --- a/packages/cloud/src/fleet-sandbox.test.ts +++ b/packages/cloud/src/fleet-sandbox.test.ts @@ -1114,6 +1114,7 @@ describe('Cloud fleet sandbox client', () => { requiredCapability: 'spawn:claude', forceProvision: true, relayfilePaths: ['/live-review/run-123/**'], + readonlyPaths: ['/live-review/run-123/reference/**'], }); const ensureCall = mocks.authorizedApiFetch.mock.calls[1]; @@ -1122,9 +1123,36 @@ describe('Cloud fleet sandbox client', () => { requiredCapability: 'spawn:claude', forceProvision: true, relayfilePaths: ['/live-review/run-123/**'], + readonlyPaths: ['/live-review/run-123/reference/**'], }); }); + it.each([[], ['/'], ['relative/**'], ['/foo'], ['/foo/../bar/**'], ['/foo/*/**']])( + 'rejects invalid readonlyPaths before authentication: %j', + async (...paths) => { + await expect( + ensureCloudFleetSandbox({ + workspaceId: 'rw_abc', + requiredCapability: 'spawn:claude', + readonlyPaths: paths as string[], + }) + ).rejects.toThrow('readonlyPaths'); + expect(mocks.ensureCloudSession).not.toHaveBeenCalled(); + } + ); + + it('rejects readonly paths without a mount', async () => { + await expect( + ensureCloudFleetSandbox({ + workspaceId: 'rw_abc', + requiredCapability: 'spawn:claude', + mountRelayfile: false, + readonlyPaths: ['/foo/**'], + }) + ).rejects.toThrow('requires mounting'); + expect(mocks.ensureCloudSession).not.toHaveBeenCalled(); + }); + it('rejects incomplete revision maps before Cloud authentication', async () => { await expect( ensureCloudFleetSandbox({ diff --git a/packages/cloud/src/fleet-sandbox.ts b/packages/cloud/src/fleet-sandbox.ts index a698a92a5..0f66de9ac 100644 --- a/packages/cloud/src/fleet-sandbox.ts +++ b/packages/cloud/src/fleet-sandbox.ts @@ -119,6 +119,8 @@ export type EnsureCloudFleetSandboxInput = { * must use the explicit `/path/**` subtree form accepted by Cloud. */ relayfilePaths?: readonly string[]; + /** Read-only Relayfile subtrees in explicit `/path/**` form. Cloud owns chmod enforcement. */ + readonlyPaths?: readonly string[]; forceProvision?: boolean; /** Constrain provisioning to a provider that Cloud has enabled for routing. */ providerId?: CloudFleetSandboxProviderId; @@ -906,6 +908,24 @@ export async function ensureCloudFleetSandbox( if (input.relayfilePaths !== undefined && input.relayfilePaths.length === 0) { throw new Error('At least one Relayfile subtree path is required when relayfilePaths is provided.'); } + if (input.readonlyPaths !== undefined) { + if ( + input.readonlyPaths.length === 0 || + input.readonlyPaths.some( + (path) => + !/^\/(?:[^/*?\\\x00-\x1f]+\/)+\*\*$/.test(path) || + path + .slice(1, -3) + .split('/') + .some((segment) => segment === '.' || segment === '..') + ) + ) { + throw new Error( + 'readonlyPaths must contain at least one explicit /path/** subtree without traversal or wildcards.' + ); + } + if (input.mountRelayfile === false) throw new Error('readonlyPaths requires mounting Relayfile.'); + } validateRequestedRepos(input.repos); const repoRevisions = validateRepoRevisions(input.repos, input.repoRevisions); @@ -932,6 +952,7 @@ export async function ensureCloudFleetSandbox( ...(input.maxAgents !== undefined ? { maxAgents: input.maxAgents } : {}), ...(input.mountRelayfile !== undefined ? { mountRelayfile: input.mountRelayfile } : {}), ...(input.relayfilePaths === undefined ? {} : { relayfilePaths: [...input.relayfilePaths] }), + ...(input.readonlyPaths === undefined ? {} : { readonlyPaths: [...input.readonlyPaths] }), ...(input.forceProvision !== undefined ? { forceProvision: input.forceProvision } : {}), ...(input.providerId !== undefined ? { providerId: input.providerId } : {}), ...(input.workloadProfile !== undefined ? { workloadProfile: input.workloadProfile } : {}), diff --git a/packages/cloud/src/fleet.ts b/packages/cloud/src/fleet.ts index 36c67c570..f34cc4836 100644 --- a/packages/cloud/src/fleet.ts +++ b/packages/cloud/src/fleet.ts @@ -382,3 +382,13 @@ export function resolveActiveFleetNodeEnrollment( `Multiple fleet node enrollments match; pass baseUrl and workspaceId to disambiguate. Candidates: ${candidates}.` ); } + +/** Fleet sandbox provisioning primitives. */ +export { + ensureCloudFleetSandbox, + deleteCloudFleetSandbox, + CloudFleetSandboxProvisionError, + type EnsureCloudFleetSandboxInput, + type EnsureCloudFleetSandboxResult, + type CloudFleetSandboxRequestOptions, +} from './fleet-sandbox.js'; diff --git a/packages/cloud/src/public-entries.test.ts b/packages/cloud/src/public-entries.test.ts new file mode 100644 index 000000000..17063bf70 --- /dev/null +++ b/packages/cloud/src/public-entries.test.ts @@ -0,0 +1,28 @@ +import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import { execFileSync } from 'node:child_process'; +import { expect, it } from 'vitest'; + +it('resolves the published Cloud and SDK fleet/attach ESM entries after build', () => { + for (const name of ['cloud', 'sdk']) { + const manifest = JSON.parse(readFileSync(new URL(`../../${name}/package.json`, import.meta.url), 'utf8')); + for (const subpath of ['fleet', 'attach']) { + const entry = manifest.exports[`./${subpath}`]; + expect(entry.types).toBe(`./dist/${subpath}.d.ts`); + expect(entry.import).toBe(`./dist/${subpath}.js`); + expect(readFileSync(new URL(`../../${name}/${entry.types}`, import.meta.url), 'utf8')).toBeTruthy(); + const exportedFunction = subpath === 'attach' ? 'startFleetNodeAttachProxy' : 'ensureCloudFleetSandbox'; + expect( + execFileSync( + process.execPath, + [ + '--input-type=module', + '-e', + `import { ${exportedFunction} } from '@agent-relay/${name}/${subpath}'; console.log(typeof ${exportedFunction});`, + ], + { cwd: fileURLToPath(new URL('../../..', import.meta.url)), encoding: 'utf8' } + ).trim() + ).toBe('function'); + } + } +}); diff --git a/packages/cloud/src/workspace-transport.ts b/packages/cloud/src/workspace-transport.ts new file mode 100644 index 000000000..663091a53 --- /dev/null +++ b/packages/cloud/src/workspace-transport.ts @@ -0,0 +1,160 @@ +import path from 'node:path'; +import { AGENT37_RELAYCAST_ORIGIN, CANONICAL_RELAYCAST_ORIGIN } from './fleet-sandbox.js'; +import { + resolveWorkspaceSelection as resolveCloudWorkspaceSelection, + type WorkspaceSelection, + type WorkspaceKeySource, +} from './workspace-key.js'; +/** Options shared by the SDK-backed (Relaycast) CLI command groups. */ +export interface WorkspaceTransportOptions { + workspaceKey?: string; + token?: string; + baseUrl?: string; + env?: NodeJS.ProcessEnv; + /** Explicit project root for nested invocations such as packages/web. */ + projectRoot?: string; + /** Use the canonical gateway instead of a persisted server-selected route. */ + ignorePersistedRelaycastTarget?: boolean; +} + +function env(options: WorkspaceTransportOptions): NodeJS.ProcessEnv { + return options.env ?? process.env; +} + +function trimOrUndefined(value: string | undefined): string | undefined { + const trimmed = value?.trim(); + return trimmed ? trimmed : undefined; +} + +export type WorkspaceTransport = { + workspaceKey: string; + baseUrl?: string; + source: WorkspaceKeySource; +}; + +/** Resolve the selected key and any previously persisted Relay workspace identity. */ +export function resolveWorkspaceSelection( + options: WorkspaceTransportOptions = {} +): WorkspaceSelection | undefined { + const explicitProject = trimOrUndefined(env(options).AGENT_RELAY_PROJECT); + const projectRoot = explicitProject ? path.resolve(explicitProject) : options.projectRoot; + return resolveCloudWorkspaceSelection({ + workspaceKey: options.workspaceKey, + env: env(options), + ...(projectRoot ? { projectRoot } : {}), + }); +} + +export function resolveBaseUrl(options: WorkspaceTransportOptions = {}): string | undefined { + const selection = selectionForTransport(options); + return resolveBaseUrlForSelection(selection, options); +} + +function selectionForTransport(options: WorkspaceTransportOptions): WorkspaceSelection | undefined { + const selection = resolveWorkspaceSelection(options); + if (!selection || !options.ignorePersistedRelaycastTarget) return selection; + const { + relaycastRoute: _relaycastRoute, + relaycastBaseUrl: _relaycastBaseUrl, + relaycastApiKey: _relaycastApiKey, + relaycastApiKeyRef: _relaycastApiKeyRef, + ...canonicalSelection + } = selection; + return canonicalSelection; +} + +function resolveBaseUrlForSelection( + selection: WorkspaceSelection | undefined, + options: WorkspaceTransportOptions +): string | undefined { + const persisted = validatePersistedRelaycastBaseUrl(selection); + const requested = trimOrUndefined(options.baseUrl) ?? trimOrUndefined(env(options).RELAY_BASE_URL); + if (persisted && requested) { + let parsed: URL; + try { + parsed = new URL(requested); + } catch { + throw new Error('The requested Relaycast base URL is invalid.'); + } + const authority = /^https:\/\/([^/?#]+)/i.exec(requested)?.[1] ?? ''; + if ( + !/^https:\/\/[^/?#]+\/?$/i.test(requested) || + parsed.protocol !== 'https:' || + parsed.username || + parsed.password || + parsed.port || + /:\d+$/.test(authority) || + parsed.search || + parsed.hash || + (parsed.pathname !== '' && parsed.pathname !== '/') + ) { + throw new Error('The requested Relaycast base URL is not a trusted origin.'); + } + if (parsed.origin !== persisted) { + throw new Error('The requested Relaycast base URL does not match the persisted workspace route.'); + } + } + return persisted ?? requested; +} + +/** Resolve one credential/origin pair from one workspace selection. */ +export function resolveWorkspaceTransport(options: WorkspaceTransportOptions = {}): WorkspaceTransport { + const selection = selectionForTransport(options); + if (!selection) { + throw new Error( + 'No workspace key found. Pass --workspace-key, set RELAY_WORKSPACE_KEY, or run `relay workspace set_key `.' + ); + } + const baseUrl = resolveBaseUrlForSelection(selection, options); + // Project-session loading already validates the reference against the + // project/workspace/route/base tuple. Never re-read a ref here: doing so + // would let a tampered ref bypass that binding and pair an unrelated key + // with this route. + const routeCredential = trimOrUndefined(selection.relaycastApiKey); + if (selection.relaycastRoute === 'agent37-isolated' && !routeCredential) { + throw new Error( + 'The persisted isolated Relaycast credential is unavailable or mismatched; rerun the sandbox command to mint a fresh route.' + ); + } + return { + workspaceKey: routeCredential ?? selection.key, + ...(baseUrl ? { baseUrl } : {}), + source: selection.source, + }; +} + +function validatePersistedRelaycastBaseUrl(selection: WorkspaceSelection | undefined): string | undefined { + const baseUrl = trimOrUndefined(selection?.relaycastBaseUrl); + const route = selection?.relaycastRoute; + const relaycastApiKey = trimOrUndefined(selection?.relaycastApiKey); + if (!baseUrl && !route && !relaycastApiKey) return undefined; + if (!baseUrl || !route) { + throw new Error('The persisted Relaycast workspace route is incomplete.'); + } + let parsed: URL; + try { + parsed = new URL(baseUrl); + } catch { + throw new Error('The persisted Relaycast workspace route is invalid.'); + } + const expectedOrigin = + route === 'canonical' + ? CANONICAL_RELAYCAST_ORIGIN + : route === 'agent37-isolated' + ? AGENT37_RELAYCAST_ORIGIN + : undefined; + if ( + !expectedOrigin || + parsed.origin !== expectedOrigin || + parsed.protocol !== 'https:' || + parsed.username || + parsed.password || + parsed.port || + parsed.search || + parsed.hash || + (parsed.pathname !== '' && parsed.pathname !== '/') + ) { + throw new Error('The persisted Relaycast workspace route is not trusted.'); + } + return parsed.origin; +} diff --git a/packages/sdk/package.json b/packages/sdk/package.json index 01015b946..e26d7757e 100644 --- a/packages/sdk/package.json +++ b/packages/sdk/package.json @@ -37,6 +37,16 @@ "types": "./dist/capabilities.d.ts", "import": "./dist/capabilities.js", "default": "./dist/capabilities.js" + }, + "./attach": { + "types": "./dist/attach.d.ts", + "import": "./dist/attach.js", + "default": "./dist/attach.js" + }, + "./fleet": { + "types": "./dist/fleet.d.ts", + "import": "./dist/fleet.js", + "default": "./dist/fleet.js" } }, "files": [ @@ -64,6 +74,7 @@ "dependencies": { "@relaycast/sdk": "^8.0.7", "@relaycast/types": "^8.0.7", - "zod": "^4.4.3" + "zod": "^4.4.3", + "@agent-relay/cloud": "12.2.4" } } diff --git a/packages/sdk/src/attach.ts b/packages/sdk/src/attach.ts new file mode 100644 index 000000000..71c417be1 --- /dev/null +++ b/packages/sdk/src/attach.ts @@ -0,0 +1 @@ +export * from '@agent-relay/cloud/attach'; diff --git a/packages/sdk/src/fleet.ts b/packages/sdk/src/fleet.ts new file mode 100644 index 000000000..caf0e4fd8 --- /dev/null +++ b/packages/sdk/src/fleet.ts @@ -0,0 +1 @@ +export * from '@agent-relay/cloud/fleet'; diff --git a/packages/sdk/tsconfig.build.json b/packages/sdk/tsconfig.build.json index 0c970cfdd..975f69885 100644 --- a/packages/sdk/tsconfig.build.json +++ b/packages/sdk/tsconfig.build.json @@ -28,7 +28,9 @@ "src/messaging/**/*.ts", "src/delivery/**/*.ts", "src/actions/**/*.ts", - "src/session/**/*.ts" + "src/session/**/*.ts", + "src/attach.ts", + "src/fleet.ts" ], "exclude": ["src/__tests__/**"] } diff --git a/summary.md b/summary.md new file mode 100644 index 000000000..e43f0d704 --- /dev/null +++ b/summary.md @@ -0,0 +1,35 @@ +## Summary + +Draft implementation of the attach and read-only-request portions of the workforce#338 upstream work. **This does not yet unblock workforce#338.** The reviewed plan's D4 decision is pending: `spawnFleetSandbox` must either orchestrate a running agent or expose a narrower ensure-only contract. No placeholder `spawnFleetSandbox` is exported. + +- Move the attach adapter and workspace transport resolution into Cloud, retaining CLI re-exports and existing broker transport fields. Add Cloud and SDK `/attach` entries, nodeId/optional-agent discovery, a private single-client raw stdio socket, idempotent cleanup, and a completion promise. +- Expose existing provisioning primitives through Cloud and SDK `/fleet`; route the CLI's ensure import through the SDK entry. Preserve existing Cloud enrollment APIs. The running-agent spawn wrapper remains pending D4. +- Validate and forward `readonlyPaths` as explicit `/path/**` subtrees; expose `fleet spawn --sandbox-readonly-path` with mount guards. +- Add socket, request-validation, CLI, and built-entry regression coverage. Add an independent read-only-request RelayFlow proof; leave historical case 1630 unchanged. + +The proxy retains broker transport fields alongside socketPath/finished/close, as the reviewed plan recommends: workforce duck-types the three consumer fields and ignores extras. Completion is an **inferred** status (0 for normal terminal closure/detach, 1 for transport failure), not a remote harness exit code. SDK re-exports add a dependency on Cloud and its transitive dependencies. No package versions changed. + +## Remaining requirements + +- Resolve reviewed-plan.md D4 and implement/test `spawnFleetSandbox` registration, placement confirmation, input mapping, and teardown. +- Cloud's server-side ensure handler/mount builder is absent from this repository. Forwarding does not enforce chmod-444; server implementation and a live Daytona write-denial proof are still required. +- Exports follow the reviewed plan's existing ESM conventions. No CommonJS build or `require` condition was added. +- No live CLI sandbox-spawn or node-attach parity run was performed. + +## Validation + +- `npm run typecheck`: passed. +- `npm run build`: passed (Rust build skipped automatically because Cargo is unavailable). +- SDK suite: 182 passed; fleet suite: 43 passed. +- Cloud suite: 450 passed, 4 skipped after removing inherited Cloud authentication variables. +- Existing and new attach tests: 42 passed, using actual local WebSocket and UNIX socket servers. +- CLI suite via `npx vitest run packages/cli/ --maxWorkers=2`: 1,652 passed, 29 skipped. `npm --prefix packages/cli test` is unavailable because that package has no test script. Tests ran without inherited Cloud/Relay credentials and with local Git file transport enabled for the temporary-repository fixture. +- Built Cloud/SDK entry-resolution regression: passed. +- New proof runner executed locally against base `c978302` and implementation `b6c3bb1`: base `absent`, head `fixed`. Both generated probes and compiled CLI-help checks passed. This was not a hosted Cloud or live Daytona run. + +## RelayFlow Proof + +- Change type: `feature` +- RelayFlow case: `sandbox-readonly-paths` + +This generated base/head probe verifies exact request forwarding and compiled CLI help. It does not provision a sandbox or prove chmod enforcement. diff --git a/tests/relayflows/cases/sandbox-readonly-paths/case.json b/tests/relayflows/cases/sandbox-readonly-paths/case.json new file mode 100644 index 000000000..fd0d6c299 --- /dev/null +++ b/tests/relayflows/cases/sandbox-readonly-paths/case.json @@ -0,0 +1,20 @@ +{ + "version": 1, + "id": "sandbox-readonly-paths", + "kind": "feature", + "title": "Forward read-only Relayfile subtree requests to Cloud", + "runner": { + "command": ["node", "tests/relayflows/cases/sandbox-readonly-paths/run.mjs"] + }, + "timeoutSeconds": 900, + "expected": { + "base": { + "outcome": "absent", + "signature": "sandbox_readonly_paths_not_forwarded" + }, + "head": { + "outcome": "fixed", + "signature": "sandbox_readonly_paths_forwarded" + } + } +} diff --git a/tests/relayflows/cases/sandbox-readonly-paths/run.mjs b/tests/relayflows/cases/sandbox-readonly-paths/run.mjs new file mode 100644 index 000000000..693fdc8ee --- /dev/null +++ b/tests/relayflows/cases/sandbox-readonly-paths/run.mjs @@ -0,0 +1,264 @@ +import { execFileSync, spawnSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { lstat, mkdir, open, readFile, rename, rm, writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath } from 'node:url'; + +const CASE_ID = 'sandbox-readonly-paths'; +const COMMAND_TIMEOUT_MS = 5 * 60 * 1000; +const targetDir = requiredDirectory('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = requiredDirectory('RELAY_PR_PROOF_HARNESS_DIR'); +const resultPath = requiredValue('RELAY_PR_PROOF_RESULT_PATH'); +const arm = requiredValue('RELAY_PR_PROOF_ARM'); + +if (arm !== 'base' && arm !== 'head') { + throw new Error(`RELAY_PR_PROOF_ARM must be base or head, received ${JSON.stringify(arm)}.`); +} + +const expectedSha = + arm === 'base' ? process.env.RELAY_PR_PROOF_BASE_SHA : process.env.RELAY_PR_PROOF_HEAD_SHA; +if (!expectedSha) throw new Error(`Missing expected ${arm} SHA.`); +const targetSha = execFileSync('git', ['-C', targetDir, 'rev-parse', 'HEAD'], { + encoding: 'utf8', +}).trim(); +if (targetSha !== expectedSha) { + throw new Error(`Target checkout ${targetSha} does not match exact ${arm} SHA ${expectedSha}.`); +} + +const runnerPath = fileURLToPath(import.meta.url); +if (!isWithin(harnessDir, runnerPath)) { + throw new Error('The RelayFlow runner must execute from the exact-head harness checkout.'); +} + +const probePath = path.join(targetDir, 'packages/cloud/src/.relayflow-sandbox-readonly-paths.test.ts'); +const probeObservationPath = path.join(targetDir, '.relayflow-sandbox-readonly-paths-observation.json'); +const probeConfigPath = path.join(targetDir, '.relayflow-sandbox-readonly-paths.vitest.config.mjs'); + +const probeSource = String.raw`import { afterEach, expect, test, vi } from 'vitest'; +import { writeFile } from 'node:fs/promises'; + +const mocks = vi.hoisted(() => ({ + ensureCloudSession: vi.fn(), + authorizedApiFetch: vi.fn(), +})); + +vi.mock('./auth.js', () => ({ + ensureCloudSession: mocks.ensureCloudSession, + authorizedApiFetch: mocks.authorizedApiFetch, +})); + +import { ensureCloudFleetSandbox } from './fleet-sandbox.js'; + +const auth = { + accessToken: 'relayflow-probe-access', + refreshToken: 'relayflow-probe-refresh', + accessTokenExpiresAt: '2099-01-01T00:00:00Z', + apiUrl: 'https://relayflow.invalid', +}; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +test('observes exact Relayfile subtree forwarding', async () => { + const observationPath = process.env.RELAY_READONLY_OBSERVATION_PATH; + if (!observationPath) throw new Error('Missing RELAY_READONLY_OBSERVATION_PATH.'); + + mocks.ensureCloudSession.mockResolvedValue({ auth, client: {} }); + mocks.authorizedApiFetch + .mockResolvedValueOnce({ + response: Response.json({ cloudWorkspaceId: '50587328-441d-4acb-b8f3-dbe1b3c5de99' }), + auth, + }) + .mockResolvedValueOnce({ + response: Response.json( + { + outcome: 'provisioned', + nodeId: 'node-relayflow', + nodeName: 'scoped-relayflow', + sandboxId: 'sandbox-relayflow', + relayWorkspaceId: 'rw_relayflow', + relayfileMounted: true, + providerId: 'daytona', + providerSandboxId: '223e4567-e89b-42d3-a456-426614174000', + }, + { status: 201 } + ), + auth, + }); + + await ensureCloudFleetSandbox({ + workspaceId: 'rw_relayflow', + requiredCapability: 'spawn:claude', + mountRelayfile: true, + readonlyPaths: ['/live-review/run-123/**', '/live-review/run-123/reviews/**'], + }); + + const ensureRequest = mocks.authorizedApiFetch.mock.calls[1]?.[2]; + expect(ensureRequest?.body).toEqual(expect.any(String)); + const ensureBody = JSON.parse(ensureRequest.body); + + await writeFile( + observationPath, + JSON.stringify({ readonlyPaths: ensureBody.readonlyPaths ?? null }), + 'utf8' + ); +}); +`; + +const probeConfigSource = `export default { + test: { + environment: 'node', + include: ['packages/cloud/src/.relayflow-sandbox-readonly-paths.test.ts'], + setupFiles: [], + }, +};\n`; + +try { + run('npm', ['ci', '--ignore-scripts'], targetDir, 'workspace dependency installation'); + run('npm', ['run', 'build:session'], targetDir, 'session package build'); + run('npm', ['run', 'build:config'], targetDir, 'configuration package build'); + run('npm', ['run', 'build:cloud'], targetDir, 'Cloud package build'); + run('npm', ['run', 'build:utils'], targetDir, 'utilities package build'); + run('npm', ['run', 'build:policy'], targetDir, 'policy package build'); + run('npm', ['run', 'build:sdk'], targetDir, 'SDK package build'); + run('npm', ['run', 'build:harness-driver'], targetDir, 'harness driver package build'); + run('npm', ['run', 'build:harnesses'], targetDir, 'harnesses package build'); + run('npm', ['run', 'build:fleet'], targetDir, 'fleet package build'); + run('npm', ['run', 'build:cli-surface'], targetDir, 'CLI surface package build'); + run('npm', ['run', 'build:cli'], targetDir, 'CLI package build'); + + await writeGeneratedFile(probePath, probeSource); + await writeGeneratedFile(probeConfigPath, probeConfigSource); + run( + 'npm', + ['exec', '--', 'vitest', 'run', '--config', path.relative(targetDir, probeConfigPath)], + targetDir, + 'Relayfile subtree forwarding probe', + { RELAY_READONLY_OBSERVATION_PATH: probeObservationPath } + ); + + const observation = JSON.parse(await readFile(probeObservationPath, 'utf8')); + const cliHelp = runCapture( + process.execPath, + [path.join(targetDir, 'packages/cli/dist/cli/index.js'), 'fleet', 'spawn', '--help'], + targetDir, + 'compiled fleet spawn help' + ); + const cliHasScopedMountFlag = cliHelp.includes('--sandbox-readonly-path '); + const requestedPaths = ['/live-review/run-123/**', '/live-review/run-123/reviews/**']; + const baseObserved = observation.readonlyPaths === null && !cliHasScopedMountFlag; + const headObserved = + JSON.stringify(observation.readonlyPaths) === JSON.stringify(requestedPaths) && cliHasScopedMountFlag; + + let outcome; + let signature; + let details; + if (baseObserved) { + outcome = 'absent'; + signature = 'sandbox_readonly_paths_not_forwarded'; + details = + 'The base Cloud client dropped requested Relayfile subtrees and the fleet spawn CLI exposed no read-only-mount flag.'; + } else if (headObserved) { + outcome = 'fixed'; + signature = 'sandbox_readonly_paths_forwarded'; + details = + 'The head Cloud client forwarded the exact requested Relayfile subtrees and the fleet spawn CLI exposed the read-only-mount flag.'; + } else { + throw new Error( + `Unexpected read-only Relayfile mount observation: ${JSON.stringify({ + ...observation, + cliHasScopedMountFlag, + })}.` + ); + } + + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details })}\n`, + 'utf8' + ); +} finally { + await rm(probePath, { force: true }); + await rm(probeConfigPath, { force: true }); + await rm(probeObservationPath, { force: true }); +} + +function requiredValue(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing required environment variable ${name}.`); + return value; +} + +function requiredDirectory(name) { + return path.resolve(requiredValue(name)); +} + +function isWithin(directory, candidate) { + const relative = path.relative(directory, candidate); + return ( + relative === '' || + (!relative.startsWith(`..${path.sep}`) && relative !== '..' && !path.isAbsolute(relative)) + ); +} + +async function writeGeneratedFile(targetPath, source) { + try { + const existing = await lstat(targetPath); + if (!existing.isFile()) { + throw new Error(`Refusing to replace non-regular generated file ${targetPath}.`); + } + } catch (error) { + if (error?.code !== 'ENOENT') throw error; + } + + const temporaryPath = `${targetPath}.tmp-${process.pid}-${randomUUID()}`; + try { + const handle = await open(temporaryPath, 'wx', 0o600); + try { + await handle.writeFile(source, 'utf8'); + } finally { + await handle.close(); + } + await rename(temporaryPath, targetPath); + } finally { + await rm(temporaryPath, { force: true }); + } +} + +function run(command, args, cwd, label, extraEnv = {}) { + const completed = spawnSync(command, args, { + cwd, + env: { ...process.env, ...extraEnv }, + stdio: ['ignore', 'inherit', 'inherit'], + timeout: COMMAND_TIMEOUT_MS, + }); + if (completed.error) throw new Error(`${label} could not start: ${completed.error.message}`); + if (completed.status !== 0) { + throw new Error( + `${label} failed with ${ + completed.signal ? `signal ${completed.signal}` : `exit code ${completed.status ?? 'unknown'}` + }.` + ); + } +} + +function runCapture(command, args, cwd, label, extraEnv = {}) { + const completed = spawnSync(command, args, { + cwd, + env: { ...process.env, ...extraEnv }, + encoding: 'utf8', + timeout: COMMAND_TIMEOUT_MS, + }); + if (completed.error) throw new Error(`${label} could not start: ${completed.error.message}`); + if (completed.status !== 0) { + throw new Error( + `${label} failed with ${ + completed.signal ? `signal ${completed.signal}` : `exit code ${completed.status ?? 'unknown'}` + }: ${completed.stderr?.slice(-2000) ?? ''}` + ); + } + return completed.stdout; +}