From d907961a1940a125585cb155622c9bebcec64867 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 17 Oct 2026 21:19:52 -0700 Subject: [PATCH 1/2] feat(sdk): accept run-scoped Relaycast token file for agent communication Agent communication now accepts RELAY_RUN_TOKEN_FILE (access-only {"token":"rt_live_...","expiresAt":...}) alongside the legacy RELAY_API_KEY (rk_live_), preferring the file when both are set. - preflight validates the file (absolute path, readable, rt_live_ token, future expiresAt, no refresh token) without ever echoing the value - the runtime registers and deletes its journal agent through a thin POST /v1/agents + DELETE /v1/agents/:name adapter that re-reads the file on every request; GET /v1/agents/:name is never used in this mode - HarnessDriverClient.spawn receives runTokenFile (the path), never the value - RELAY_RUN_TOKEN_FILE is blanked for communication agents; redaction covers rt_ Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/communication/environment.ts | 2 +- packages/sdk/src/communication/preflight.ts | 9 +- packages/sdk/src/communication/relay.ts | 30 +++-- packages/sdk/src/communication/run-token.ts | 121 +++++++++++++++++ packages/sdk/src/redact.ts | 2 +- ...ommunication-environment-preflight.test.ts | 51 ++++++- .../sdk/tests/communication-run-token.test.ts | 125 ++++++++++++++++++ packages/sdk/tests/redact.test.ts | 1 + 8 files changed, 324 insertions(+), 17 deletions(-) create mode 100644 packages/sdk/src/communication/run-token.ts create mode 100644 packages/sdk/tests/communication-run-token.test.ts diff --git a/packages/sdk/src/communication/environment.ts b/packages/sdk/src/communication/environment.ts index 53aff3403..e716cd6f8 100644 --- a/packages/sdk/src/communication/environment.ts +++ b/packages/sdk/src/communication/environment.ts @@ -7,7 +7,7 @@ const terminalNames = ['TERM', 'COLORTERM', 'XDG_CONFIG_HOME', 'XDG_DATA_HOME', const networkNames = ['HTTP_PROXY', 'HTTPS_PROXY', 'ALL_PROXY', 'NO_PROXY', 'http_proxy', 'https_proxy', 'all_proxy', 'no_proxy']; const relayNames = ['RELAY_API_KEY', 'RELAY_WORKSPACE_KEY', 'AGENT_RELAY_WORKSPACE_KEY', - 'RELAY_BROKER_API_KEY', 'RELAY_AGENT_TOKEN', 'RELAY_NODE_TOKEN', 'RELAY_WORKSPACES_JSON']; + 'RELAY_BROKER_API_KEY', 'RELAY_AGENT_TOKEN', 'RELAY_NODE_TOKEN', 'RELAY_WORKSPACES_JSON', 'RELAY_RUN_TOKEN_FILE']; // Multi-provider CLIs select their provider in their own configuration. const multiProvider = ['OPENAI_API_KEY', 'OPENAI_BASE_URL', 'ANTHROPIC_API_KEY', 'ANTHROPIC_BASE_URL', 'GEMINI_API_KEY', 'GOOGLE_API_KEY', 'OPENROUTER_API_KEY'] as const; diff --git a/packages/sdk/src/communication/preflight.ts b/packages/sdk/src/communication/preflight.ts index c3abfa5f6..d59940d01 100644 --- a/packages/sdk/src/communication/preflight.ts +++ b/packages/sdk/src/communication/preflight.ts @@ -4,6 +4,7 @@ import { spawnSync } from 'node:child_process'; import { pathToFileURL } from 'node:url'; import type { FlowSpec, KernelRunSpec } from '../spec.js'; import { communicationInstruction } from './spec.js'; +import { checkRunTokenFile, relayCredential } from './run-token.js'; const require = createRequire(import.meta.url); export class CommunicationEnvironmentError extends Error {} @@ -11,8 +12,12 @@ export class CommunicationEnvironmentError extends Error {} /** Local-only probes: no broker, workspace creation, or optional runtime import in this process. */ export function checkCommunicationEnvironment(spec: FlowSpec | KernelRunSpec): void { if (!spec.steps.some(step => step.type === 'agent' && communicationInstruction(step.instruction))) return; - if (!process.env.RELAY_API_KEY?.trim().startsWith('rk_live_')) { - throw new CommunicationEnvironmentError('Agent communication requires RELAY_API_KEY for an existing workspace.'); + // A run token file wins over RELAY_API_KEY; its value is validated, never logged. + try { + const credential = relayCredential(); + if (credential.kind === 'run-token') checkRunTokenFile(credential.file); + } catch (error) { + throw new CommunicationEnvironmentError((error as Error).message); } let driverPath: string; try { diff --git a/packages/sdk/src/communication/relay.ts b/packages/sdk/src/communication/relay.ts index 70f5cd9ee..264528eed 100644 --- a/packages/sdk/src/communication/relay.ts +++ b/packages/sdk/src/communication/relay.ts @@ -3,6 +3,7 @@ import { mkdir } from 'node:fs/promises'; import { join } from 'node:path'; import { createProjection } from './projection.js'; import { brokerEnvironment } from './environment.js'; +import { relayCredential, runTokenAgents, runTokenProvider } from './run-token.js'; export interface RelayEvent { kind: string; name?: string; event_id?: string; delivery_id?: string; verification?: string; [key: string]: unknown } export interface RelayHandle { @@ -40,9 +41,9 @@ export async function loadRelayModules(): Promise { } } export function relayWorkspaceKey(env = process.env): string { - const key = env.RELAY_API_KEY?.trim(); - if (!key?.startsWith('rk_live_')) throw new Error('Agent communication requires RELAY_API_KEY for an existing workspace.'); - return key; + const credential = relayCredential({ RELAY_API_KEY: env.RELAY_API_KEY }); + if (credential.kind !== 'workspace-key') throw new Error('Agent communication requires RELAY_API_KEY for an existing workspace.'); + return credential.workspaceKey; } export interface RelayRuntime { broker: RelayBroker; @@ -75,28 +76,35 @@ export async function acquireRelayRuntime(dataDir: string, runId: string): Promi } } async function createRuntime(dataDir: string, runId: string): Promise { - const workspaceKey = relayWorkspaceKey(); + const credential = relayCredential(); const { HarnessDriverClient, RelaycastMessagingClient } = await loadRelayModules(); const hash = createHash('sha256').update(runId).digest('hex').slice(0, 12); const prefix = `flow-${hash}-${randomUUID().slice(0, 8)}`; const channel = `wf-${runId.toLowerCase()}`; - const options = { workspaceKey, ...(process.env.RELAY_BASE_URL ? { baseUrl: process.env.RELAY_BASE_URL } : {}) }; - const workspace = new RelaycastMessagingClient(options); + const baseUrl = process.env.RELAY_BASE_URL ? { baseUrl: process.env.RELAY_BASE_URL } : {}; + // Run-token mode: agent admin re-reads the file per request, the broker gets + // only the file path, and no workspace key exists anywhere in this process. + const workspace = credential.kind === 'run-token' + ? runTokenAgents(runTokenProvider(credential.file), baseUrl) + : new RelaycastMessagingClient({ workspaceKey: credential.workspaceKey, ...baseUrl }).agents; + const brokerCredential = credential.kind === 'run-token' + ? { runTokenFile: credential.file } : { workspaceKey: credential.workspaceKey }; const publisher = `${prefix}-journal`; - const registration = await workspace.agents.register({ name: publisher, type: 'agent' }); + const registration = await workspace.register({ name: publisher, type: 'agent' }); let broker: RelayBroker | undefined; try { - const messaging = new RelaycastMessagingClient({ ...options, agentToken: registration.token }); + const messaging = new RelaycastMessagingClient({ ...(credential.kind === 'workspace-key' + ? { workspaceKey: credential.workspaceKey } : {}), ...baseUrl, agentToken: registration.token }); const cwd = join(dataDir, 'communication', prefix); await mkdir(cwd, { recursive: true, mode: 0o700 }); - broker = await HarnessDriverClient.spawn({ workspaceKey, brokerName: prefix, cwd, channels: [], + broker = await HarnessDriverClient.spawn({ ...brokerCredential, brokerName: prefix, cwd, channels: [], ...(process.env.RELAYFLOW_RELAY_BROKER_BIN ? { binaryPath: process.env.RELAYFLOW_RELAY_BROKER_BIN } : {}), env: brokerEnvironment(process.env) }); return { broker, messaging, prefix, channel, project: createProjection(messaging, channel, runId), close: async () => { - try { await broker!.shutdown(); } finally { await workspace.agents.delete(publisher); } + try { await broker!.shutdown(); } finally { await workspace.delete(publisher); } } }; } catch (error) { - try { await broker?.shutdown(); } finally { await workspace.agents.delete(publisher); } + try { await broker?.shutdown(); } finally { await workspace.delete(publisher); } throw error; } } diff --git a/packages/sdk/src/communication/run-token.ts b/packages/sdk/src/communication/run-token.ts new file mode 100644 index 000000000..9a27d5ee8 --- /dev/null +++ b/packages/sdk/src/communication/run-token.ts @@ -0,0 +1,121 @@ +// Run-scoped Relaycast credential (`rt_live_`), delivered as an access-only +// file named by RELAY_RUN_TOKEN_FILE. Cloud refreshes the file outside the +// sandbox, so the value is re-read on every request and never cached, copied +// into env or argv, or written anywhere else. Errors never carry the value or +// the file contents. +import { readFileSync } from 'node:fs'; +import { readFile } from 'node:fs/promises'; +import { isAbsolute } from 'node:path'; + +const RUN_TOKEN_PREFIX = 'rt_live_'; +const WORKSPACE_KEY_PREFIX = 'rk_live_'; +const DEFAULT_BASE_URL = 'https://cast.agentrelay.com'; + +export type RelayCredential = + | { kind: 'run-token'; file: string } + | { kind: 'workspace-key'; workspaceKey: string }; + +export const MISSING_RELAY_CREDENTIAL = + 'Agent communication requires RELAY_API_KEY for an existing workspace or RELAY_RUN_TOKEN_FILE for a run-scoped token.'; + +/** Selects the Relay credential; a run token file wins over a workspace key. */ +export function relayCredential(env: NodeJS.ProcessEnv = process.env): RelayCredential { + const file = env.RELAY_RUN_TOKEN_FILE?.trim(); + if (file) return { kind: 'run-token', file }; + const key = env.RELAY_API_KEY?.trim(); + if (key?.startsWith(WORKSPACE_KEY_PREFIX)) return { kind: 'workspace-key', workspaceKey: key }; + throw new Error(MISSING_RELAY_CREDENTIAL); +} + +/** Validates the access file's JSON and returns the token. Messages never include the value. */ +export function parseRunTokenFile(text: string, now = Date.now()): string { + let parsed: unknown; + try { parsed = JSON.parse(text); } catch { + // JSON.parse messages quote the input; replace them wholesale. + throw new Error('RELAY_RUN_TOKEN_FILE does not contain valid JSON.'); + } + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + throw new Error('RELAY_RUN_TOKEN_FILE must contain a JSON object.'); + } + const record = parsed as Record; + if ('refreshToken' in record || 'refresh_token' in record) { + throw new Error('RELAY_RUN_TOKEN_FILE must be access-only and must not contain a refresh token.'); + } + const token = typeof record.token === 'string' ? record.token.trim() : ''; + if (!token.startsWith(RUN_TOKEN_PREFIX) || token.length === RUN_TOKEN_PREFIX.length) { + throw new Error('RELAY_RUN_TOKEN_FILE token must be a non-empty rt_live_ run token.'); + } + const expiresAt = typeof record.expiresAt === 'string' ? Date.parse(record.expiresAt) : Number.NaN; + if (!Number.isFinite(expiresAt)) throw new Error('RELAY_RUN_TOKEN_FILE expiresAt must be an ISO 8601 time.'); + if (expiresAt <= now) throw new Error('RELAY_RUN_TOKEN_FILE run token has expired.'); + return token; +} + +function checkPath(file: string): void { + if (!isAbsolute(file)) throw new Error('RELAY_RUN_TOKEN_FILE must be an absolute path.'); +} + +/** Synchronous check for preflight. Returns nothing so the value cannot escape. */ +export function checkRunTokenFile(file: string, now = Date.now()): void { + checkPath(file); + let text: string; + try { text = readFileSync(file, 'utf8'); } catch { + throw new Error('RELAY_RUN_TOKEN_FILE does not name a readable file.'); + } + parseRunTokenFile(text, now); +} + +/** A credential provider that reads the file on every call. */ +export function runTokenProvider(file: string): () => Promise { + checkPath(file); + return async () => { + let text: string; + try { text = await readFile(file, 'utf8'); } catch { + throw new Error('RELAY_RUN_TOKEN_FILE does not name a readable file.'); + } + return parseRunTokenFile(text); + }; +} + +export interface RunTokenAgents { + register(input: { name: string; type: string }): Promise<{ token: string }>; + delete(name: string): Promise; +} + +/** + * Minimal Relaycast agent admin for run tokens: `POST /v1/agents` and + * `DELETE /v1/agents/:name` only, the two agent routes that accept + * `rt_live_` (relaycast#559). The Agent Relay messaging client takes one + * static credential at construction, so it cannot re-read a rotating file. + */ +export function runTokenAgents(credential: () => Promise, + options: { baseUrl?: string; fetch?: typeof fetch } = {}): RunTokenAgents { + const base = (options.baseUrl?.trim() || DEFAULT_BASE_URL).replace(/\/+$/, ''); + const doFetch = options.fetch ?? fetch; + async function call(method: string, path: string, body?: unknown): Promise { + const token = await credential(); + const response = await doFetch(`${base}${path}`, { + method, + headers: { authorization: `Bearer ${token}`, ...(body === undefined ? {} : { 'content-type': 'application/json' }) }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); + if (response.status === 204) return undefined; + let envelope: { ok?: unknown; data?: unknown; error?: { code?: unknown; message?: unknown } } | undefined; + try { envelope = await response.json() as typeof envelope; } catch { envelope = undefined; } + if (response.ok && envelope?.ok === true) return envelope.data; + const code = typeof envelope?.error?.code === 'string' ? envelope.error.code : 'unexpected_response'; + throw new Error(`Relaycast ${method} ${path} failed (${response.status} ${code}).`); + } + return { + async register(input) { + const data = await call('POST', '/v1/agents', input) as { token?: unknown } | undefined; + if (typeof data?.token !== 'string' || !data.token) { + throw new Error('Relaycast POST /v1/agents returned no agent token.'); + } + return { token: data.token }; + }, + async delete(name) { + await call('DELETE', `/v1/agents/${encodeURIComponent(name)}`); + }, + }; +} diff --git a/packages/sdk/src/redact.ts b/packages/sdk/src/redact.ts index 7f5e2d923..23fa7589d 100644 --- a/packages/sdk/src/redact.ts +++ b/packages/sdk/src/redact.ts @@ -21,7 +21,7 @@ const MIN_SECRET_LENGTH = 8; * bearer, header and vendor shapes a gate render can carry. */ const TOKEN_PATTERNS: readonly RegExp[] = [ - /\b(?:at|rk|nt|ot|br|arr)_(?:live_)?[A-Za-z0-9_-]+/g, + /\b(?:at|rk|rt|nt|ot|br|arr)_(?:live_)?[A-Za-z0-9_-]+/g, /\bsk-[A-Za-z0-9_-]{16,}/g, /\bghp_[A-Za-z0-9]{20,}/g, /\bxox[abp]-[A-Za-z0-9-]+/g, diff --git a/packages/sdk/tests/communication-environment-preflight.test.ts b/packages/sdk/tests/communication-environment-preflight.test.ts index d59a956e8..3151942fa 100644 --- a/packages/sdk/tests/communication-environment-preflight.test.ts +++ b/packages/sdk/tests/communication-environment-preflight.test.ts @@ -1,14 +1,29 @@ -import { afterEach, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, expect, it, vi } from 'vitest'; import type { FlowSpec } from '../src/spec.js'; const mocks = vi.hoisted(() => ({ resolve: vi.fn(), spawn: vi.fn(), access: vi.fn() })); vi.mock('node:module', () => ({ createRequire: () => ({ resolve: mocks.resolve }) })); vi.mock('node:child_process', () => ({ spawnSync: mocks.spawn })); -vi.mock('node:fs', () => ({ accessSync: mocks.access, constants: { X_OK: 1 } })); +vi.mock('node:fs', async importOriginal => ({ ...await importOriginal(), + accessSync: mocks.access, constants: { X_OK: 1 } })); +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { checkCommunicationEnvironment } from '../src/communication/preflight.js'; const spec: FlowSpec = { version: '0.1.0', steps: [{ id: 'a', type: 'agent', cli: 'codex', instruction: JSON.stringify({ type: 'relayflows.communication.v1', instruction: '', incoming: ['b'], outgoing: [], timeoutMs: 1000, }) }] }; +const dir = mkdtempSync(join(tmpdir(), 'flows-run-token-')); +const runFile = join(dir, 'relaycast-run.json'); +const future = new Date(Date.now() + 600_000).toISOString(); +// Synthetic value; never a real credential. +const SYNTHETIC = 'rt_live_synthetic0preflight'; +function ready() { + mocks.resolve.mockReturnValue('/installed/broker-path.js'); + mocks.spawn.mockReturnValue({ status: 0, stdout: '/installed/broker' }); +} +beforeEach(() => { vi.stubEnv('RELAY_RUN_TOKEN_FILE', undefined); }); afterEach(() => { vi.unstubAllEnvs(); vi.resetAllMocks(); }); +process.on('exit', () => rmSync(dir, { recursive: true, force: true })); it('ordinary flows never inspect Relay credentials, packages, or broker', () => { vi.stubEnv('RELAY_API_KEY', undefined); checkCommunicationEnvironment({ version: '0.1.0', steps: [{ id: 'plain', type: 'agent', instruction: 'hello' }] }); @@ -30,3 +45,35 @@ it.each(['missing', 'not-executable', 'ready'])('checks the broker executable: % if (state === 'ready') expect(() => checkCommunicationEnvironment(spec)).not.toThrow(); else expect(() => checkCommunicationEnvironment(spec)).toThrow('executable Relay broker'); }); +it('accepts a valid run token file without RELAY_API_KEY', () => { + writeFileSync(runFile, JSON.stringify({ token: SYNTHETIC, expiresAt: future }), { mode: 0o600 }); + vi.stubEnv('RELAY_API_KEY', undefined); vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); ready(); + expect(() => checkCommunicationEnvironment(spec)).not.toThrow(); +}); +it('prefers the run token file over RELAY_API_KEY', () => { + writeFileSync(runFile, JSON.stringify({ token: SYNTHETIC, expiresAt: new Date(Date.now() - 1000).toISOString() })); + vi.stubEnv('RELAY_API_KEY', 'rk_live_test'); vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); ready(); + expect(() => checkCommunicationEnvironment(spec)).toThrow('expired'); +}); +it.each([ + ['missing file', null, 'readable file'], + ['invalid JSON', `{"token":"${SYNTHETIC}"`, 'valid JSON'], + ['workspace key', JSON.stringify({ token: 'rk_live_synthetic', expiresAt: future }), 'rt_live_'], + ['empty token', JSON.stringify({ token: 'rt_live_', expiresAt: future }), 'rt_live_'], + ['no expiry', JSON.stringify({ token: SYNTHETIC }), 'expiresAt'], + ['expired', JSON.stringify({ token: SYNTHETIC, expiresAt: '2020-01-01T00:00:00Z' }), 'expired'], + ['refresh token', JSON.stringify({ token: SYNTHETIC, expiresAt: future, refreshToken: 'rt_refresh_x' }), 'refresh token'], +])('rejects a bad run token file: %s', (_name, contents, message) => { + rmSync(runFile, { force: true }); + if (contents !== null) writeFileSync(runFile, contents); + vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); ready(); + let thrown: Error | undefined; + try { checkCommunicationEnvironment(spec); } catch (error) { thrown = error as Error; } + expect(thrown?.message).toContain(message); + expect(thrown?.message).not.toContain('synthetic'); + expect(mocks.spawn).not.toHaveBeenCalled(); +}); +it('rejects a relative run token file path', () => { + vi.stubEnv('RELAY_RUN_TOKEN_FILE', 'relative/run.json'); + expect(() => checkCommunicationEnvironment(spec)).toThrow('absolute path'); +}); diff --git a/packages/sdk/tests/communication-run-token.test.ts b/packages/sdk/tests/communication-run-token.test.ts new file mode 100644 index 000000000..bfde72b5b --- /dev/null +++ b/packages/sdk/tests/communication-run-token.test.ts @@ -0,0 +1,125 @@ +import { mkdtempSync, renameSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterAll, afterEach, beforeEach, expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + spawn: vi.fn(), + clients: [] as Array>, + register: vi.fn(async () => ({ token: 'at_live_legacy' })), + remove: vi.fn(async () => {}), +})); +vi.mock('@agent-relay/harness-driver', () => ({ HarnessDriverClient: { spawn: mocks.spawn } })); +vi.mock('@agent-relay/sdk', () => ({ + RelaycastMessagingClient: class { + agents = { register: mocks.register, delete: mocks.remove }; + channels = {}; messages = {}; + constructor(options: Record) { mocks.clients.push(options); } + }, +})); +import { acquireRelayRuntime } from '../src/communication/relay.js'; +import { runTokenAgents } from '../src/communication/run-token.js'; + +// Synthetic values; never real credentials. +const FIRST = 'rt_live_synthetic0first'; +const SECOND = 'rt_live_synthetic0second'; +const dir = mkdtempSync(join(tmpdir(), 'flows-run-token-runtime-')); +const runFile = join(dir, 'relaycast-run.json'); +const future = () => new Date(Date.now() + 600_000).toISOString(); +function writeRunFile(token: string) { + // Cloud's worker contract: write a 0600 temp file, then rename over the target. + const temp = `${runFile}.tmp`; + writeFileSync(temp, JSON.stringify({ token, expiresAt: future() }), { mode: 0o600 }); + renameSync(temp, runFile); +} +interface Call { method: string; url: string; authorization: string; body?: unknown } +let calls: Call[]; +function stubFetch() { + calls = []; + vi.stubGlobal('fetch', vi.fn(async (url: string, init: RequestInit) => { + const headers = init.headers as Record; + calls.push({ method: init.method!, url, authorization: headers.authorization!, + ...(init.body ? { body: JSON.parse(String(init.body)) } : {}) }); + if (init.method === 'DELETE') return new Response(null, { status: 204 }); + return Response.json({ ok: true, data: { name: 'x', token: 'at_live_synthetic0agent', run_id: 'run-1' } }, { status: 201 }); + })); +} +const broker = { spawnPty: vi.fn(), onEvent: vi.fn(), shutdown: vi.fn(async () => {}) }; + +beforeEach(() => { + mocks.clients.length = 0; + mocks.spawn.mockResolvedValue(broker); + vi.stubEnv('RELAY_BASE_URL', 'https://relaycast.test'); + vi.stubEnv('RELAYFLOW_RELAY_BROKER_BIN', undefined); + stubFetch(); +}); +afterEach(() => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); vi.clearAllMocks(); }); +afterAll(() => rmSync(dir, { recursive: true, force: true })); + +it('registers and deletes with the current file token and gives the broker only the path', async () => { + writeRunFile(FIRST); + vi.stubEnv('RELAY_API_KEY', 'rk_live_legacy0should0lose'); + vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); + const runtime = await acquireRelayRuntime(dir, 'run-1'); + + expect(calls[0]).toMatchObject({ method: 'POST', url: 'https://relaycast.test/v1/agents', + authorization: `Bearer ${FIRST}`, body: { name: `${runtime.prefix}-journal`, type: 'agent' } }); + expect(mocks.register).not.toHaveBeenCalled(); + // The messaging client gets only the per-agent token, never a workspace key or run token. + expect(mocks.clients).toEqual([{ baseUrl: 'https://relaycast.test', agentToken: 'at_live_synthetic0agent' }]); + + const spawnOptions = mocks.spawn.mock.calls[0]![0] as Record; + expect(spawnOptions.runTokenFile).toBe(runFile); + expect(spawnOptions).not.toHaveProperty('workspaceKey'); + const serialized = JSON.stringify(spawnOptions); + expect(serialized).not.toContain('rt_live_'); + expect(serialized).not.toContain('rk_live_'); + const env = spawnOptions.env as Record; + expect(Object.entries(env).filter(([, value]) => value !== undefined).map(([name]) => name)) + .not.toEqual(expect.arrayContaining(['RELAY_API_KEY', 'RELAY_RUN_TOKEN_FILE'])); + + // Cloud rotates the file; the next call reads the new token. + writeRunFile(SECOND); + await runtime.close(); + expect(calls[1]).toMatchObject({ method: 'DELETE', + url: `https://relaycast.test/v1/agents/${runtime.prefix}-journal`, authorization: `Bearer ${SECOND}` }); + // GET /v1/agents/:name does not accept run tokens and is never called. + expect(calls.every(call => call.method !== 'GET')).toBe(true); +}); + +it('keeps the legacy workspace key path when no run token file is set', async () => { + vi.stubEnv('RELAY_API_KEY', 'rk_live_legacy0key'); + vi.stubEnv('RELAY_RUN_TOKEN_FILE', undefined); + const runtime = await acquireRelayRuntime(dir, 'run-2'); + expect(mocks.register).toHaveBeenCalledWith({ name: `${runtime.prefix}-journal`, type: 'agent' }); + expect(mocks.clients[0]).toEqual({ workspaceKey: 'rk_live_legacy0key', baseUrl: 'https://relaycast.test' }); + expect(mocks.spawn.mock.calls[0]![0]).toMatchObject({ workspaceKey: 'rk_live_legacy0key' }); + expect(mocks.spawn.mock.calls[0]![0]).not.toHaveProperty('runTokenFile'); + await runtime.close(); + expect(mocks.remove).toHaveBeenCalledWith(`${runtime.prefix}-journal`); + expect(calls).toEqual([]); +}); + +it('surfaces envelope errors without the token value', async () => { + writeRunFile(FIRST); + const fetchImpl = vi.fn(async () => Response.json( + { ok: false, error: { code: 'run_token_forbidden', message: `bad ${FIRST}` } }, { status: 401 })); + const agents = runTokenAgents(async () => FIRST, { baseUrl: 'https://relaycast.test', fetch: fetchImpl as typeof fetch }); + const error = await agents.register({ name: 'a', type: 'agent' }).catch(e => e as Error); + expect(error.message).toBe('Relaycast POST /v1/agents failed (401 run_token_forbidden).'); + expect(error.message).not.toContain('rt_live_'); +}); + +it('refuses to register with an expired file and never sends a request', async () => { + writeFileSync(runFile, JSON.stringify({ token: FIRST, expiresAt: '2020-01-01T00:00:00Z' })); + vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); + await expect(acquireRelayRuntime(dir, 'run-3')).rejects.toThrow('expired'); + expect(calls).toEqual([]); + expect(mocks.spawn).not.toHaveBeenCalled(); +}); + +it('never hands the run token file to communication agents', async () => { + const { agentEnvironment } = await import('../src/communication/environment.js'); + expect(agentEnvironment('codex', { RELAY_RUN_TOKEN_FILE: runFile, OPENAI_API_KEY: 'codex-key' })) + .toMatchObject({ RELAY_RUN_TOKEN_FILE: '', OPENAI_API_KEY: 'codex-key' }); +}); diff --git a/packages/sdk/tests/redact.test.ts b/packages/sdk/tests/redact.test.ts index f5417d0c2..8526ad3f6 100644 --- a/packages/sdk/tests/redact.test.ts +++ b/packages/sdk/tests/redact.test.ts @@ -36,6 +36,7 @@ describe('redact', () => { it.each([ ['relay token', 'key rk_live_abc123DEF', 'key [redacted]'], + ['relaycast run token', 'run rt_live_abc123-DEF_x', 'run [redacted]'], ['observer token', 'see ot_live_xyz-9', 'see [redacted]'], ['agent token', 'at_0123456789', '[redacted]'], ['bearer', 'Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.x.y', 'Authorization: Bearer [redacted]'], From 87b5bfa15af2af06272da2e1c4db2261f21ec074 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 17 Oct 2026 21:34:14 -0700 Subject: [PATCH 2/2] feat(sdk): rotate run-mode agent tokens before expiry and retry 401s relaycast#550 expires at_live_ tokens after 15 minutes while runs can last hours. In run-token mode the SDK now rotates the journal agent's token via POST /v1/agents/:name/rotate-token (authenticated by its own at_live_) at two thirds of the lifetime or five minutes before expiry, whichever is earlier, matching the relay broker's schedule and fail-closed policy (401/403/404 stop; transient failures back off 2/5/10/20/30s). The token stays in memory; the messaging facade rebuilds its client on rotation and forces one rotation plus a single retry on 401. Run-token register/delete retry once on 401 after re-reading the file. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/communication/agent-token.ts | 162 ++++++++++++++++ packages/sdk/src/communication/relay.ts | 15 +- packages/sdk/src/communication/run-token.ts | 57 ++++-- .../tests/communication-agent-token.test.ts | 173 ++++++++++++++++++ 4 files changed, 387 insertions(+), 20 deletions(-) create mode 100644 packages/sdk/src/communication/agent-token.ts create mode 100644 packages/sdk/tests/communication-agent-token.test.ts diff --git a/packages/sdk/src/communication/agent-token.ts b/packages/sdk/src/communication/agent-token.ts new file mode 100644 index 000000000..414b7e0cc --- /dev/null +++ b/packages/sdk/src/communication/agent-token.ts @@ -0,0 +1,162 @@ +// Proactive rotation of short-lived Relaycast agent tokens (`at_live_`) in +// run-token mode. Relaycast #550 gives each agent token a 15-minute expiry, +// while a run can last hours, and run-token mode holds no workspace key that +// could take an identity over. So each agent token this process registered is +// rotated through `POST /v1/agents/:name/rotate-token`, authenticated by that +// agent's own current token, well before expiry. Schedule and failure policy +// mirror the relay broker (crates/broker/src/relaycast/agent_token_rotation.rs). +// +// Tokens live in process memory only and never appear in errors. +import type { RelayMessaging } from './relay.js'; +import { RelaycastHttpError, relaycastBaseUrl, relaycastRequest, type AgentRegistration } from './run-token.js'; + +/** Rotate no later than this long before expiry. */ +export const ROTATE_BEFORE_EXPIRY_MS = 5 * 60_000; +/** A token with no known expiry is rotated once after this delay to learn it. */ +export const UNKNOWN_EXPIRY_PROBE_MS = 60_000; +/** Backoff after a transient rotation failure; Relaycast keeps the old token valid for a 60s grace window. */ +export const ROTATION_RETRY_BACKOFFS_MS = [2_000, 5_000, 10_000, 20_000, 30_000] as const; + +/** + * Delay until rotation: two thirds of the remaining lifetime or + * ROTATE_BEFORE_EXPIRY_MS before expiry, whichever is earlier. A token already + * inside that window rotates immediately. + */ +export function rotationDelay(now: number, expiresAt: number): number { + const lifetime = expiresAt - now; + if (!(lifetime > 0)) return 0; + return Math.min(lifetime * 2 / 3, Math.max(0, lifetime - ROTATE_BEFORE_EXPIRY_MS)); +} + +function parseExpiry(raw: unknown): number | undefined { + if (typeof raw !== 'string') return undefined; + const at = Date.parse(raw); + return Number.isFinite(at) ? at : undefined; +} + +export function isUnauthorized(error: unknown): boolean { + const record = error as { status?: unknown; statusCode?: unknown } | null; + return typeof record === 'object' && record !== null && (record.status === 401 || record.statusCode === 401); +} + +export interface AgentTokenSessionOptions { + baseUrl?: string; + fetch?: typeof fetch; + now?: () => number; +} + +/** One agent's current token, rotated on a timer and on demand. */ +export class AgentTokenSession { + private token: string; + private expiresAt: number | undefined; + private timer: ReturnType | undefined; + private inFlight: Promise | undefined; + private stopped = false; + private readonly base: string; + private readonly doFetch: typeof fetch; + private readonly now: () => number; + + constructor(readonly name: string, initial: AgentRegistration, options: AgentTokenSessionOptions = {}) { + this.token = initial.token; + this.expiresAt = parseExpiry(initial.tokenExpiresAt); + this.base = relaycastBaseUrl(options.baseUrl); + this.doFetch = options.fetch ?? ((...args) => fetch(...args)); + this.now = options.now ?? Date.now; + } + + current(): string { return this.token; } + + /** Begin proactive rotation. */ + start(): void { this.schedule(); } + + stop(): void { + this.stopped = true; + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + } + + /** Rotate now (one attempt; concurrent callers share it) and return the new token. */ + rotate(): Promise { + this.inFlight ??= this.rotateOnce().finally(() => { this.inFlight = undefined; }); + return this.inFlight; + } + + private async rotateOnce(): Promise { + const path = `/v1/agents/${encodeURIComponent(this.name)}/rotate-token`; + const data = await relaycastRequest(this.doFetch, this.base, this.token, 'POST', path, {}) as + { token?: unknown; token_expires_at?: unknown } | undefined; + if (typeof data?.token !== 'string' || !data.token.trim()) { + throw new Error(`Relaycast agent token rotation for '${this.name}' returned no token.`); + } + // The previous token stays valid for Relaycast's grace window, so requests + // already in flight with it still succeed. + this.token = data.token; + this.expiresAt = parseExpiry(data.token_expires_at); + // A rotated token with no expiry comes from an engine that predates #550 + // and needs no further rotation. + if (this.expiresAt === undefined) this.stop(); else this.schedule(); + return this.token; + } + + private schedule(): void { + if (this.stopped) return; + if (this.timer) clearTimeout(this.timer); + const delay = this.expiresAt === undefined ? UNKNOWN_EXPIRY_PROBE_MS : rotationDelay(this.now(), this.expiresAt); + this.timer = setTimeout(() => { void this.scheduledRotation(0); }, delay); + this.timer.unref?.(); + } + + private async scheduledRotation(attempt: number): Promise { + this.timer = undefined; + if (this.stopped) return; + try { + await this.rotate(); + } catch (error) { + // 401/403/404 fail closed: run-token mode never takes over or re-registers. + if (error instanceof RelaycastHttpError && [401, 403, 404].includes(error.status)) { this.stop(); return; } + const expired = this.expiresAt !== undefined && this.expiresAt <= this.now(); + if (attempt >= ROTATION_RETRY_BACKOFFS_MS.length || expired || this.stopped) { this.stop(); return; } + this.timer = setTimeout(() => { void this.scheduledRotation(attempt + 1); }, ROTATION_RETRY_BACKOFFS_MS[attempt]); + this.timer.unref?.(); + } + } +} + +/** + * A messaging facade that always uses the session's current token. The + * underlying client is rebuilt whenever the token rotates; a 401 forces one + * rotation and a single retry. Sends carry idempotency keys, so a retry + * cannot duplicate a message. + */ +export function rotatingMessaging(session: AgentTokenSession, + build: (agentToken: string) => RelayMessaging): RelayMessaging { + let token = session.current(); + let client = build(token); + const current = () => { + const latest = session.current(); + if (latest !== token) { token = latest; client = build(latest); } + return client; + }; + async function call(operation: (messaging: RelayMessaging) => Promise): Promise { + try { return await operation(current()); } catch (error) { + if (!isUnauthorized(error)) throw error; + await session.rotate(); + return operation(current()); + } + } + return { + agents: { + register: input => call(m => m.agents.register(input)), + delete: name => call(m => m.agents.delete(name)), + }, + channels: { + create: input => call(m => m.channels.create(input)), + get: name => call(m => m.channels.get(name)), + join: name => call(m => m.channels.join(name)), + }, + messages: { + direct: input => call(m => m.messages.direct(input)), + send: input => call(m => m.messages.send(input)), + }, + }; +} diff --git a/packages/sdk/src/communication/relay.ts b/packages/sdk/src/communication/relay.ts index 264528eed..eeefe0a06 100644 --- a/packages/sdk/src/communication/relay.ts +++ b/packages/sdk/src/communication/relay.ts @@ -4,6 +4,7 @@ import { join } from 'node:path'; import { createProjection } from './projection.js'; import { brokerEnvironment } from './environment.js'; import { relayCredential, runTokenAgents, runTokenProvider } from './run-token.js'; +import { AgentTokenSession, rotatingMessaging } from './agent-token.js'; export interface RelayEvent { kind: string; name?: string; event_id?: string; delivery_id?: string; verification?: string; [key: string]: unknown } export interface RelayHandle { @@ -92,19 +93,25 @@ async function createRuntime(dataDir: string, runId: string): Promise { session?.stop(); await workspace.delete(publisher); }; try { - const messaging = new RelaycastMessagingClient({ ...(credential.kind === 'workspace-key' - ? { workspaceKey: credential.workspaceKey } : {}), ...baseUrl, agentToken: registration.token }); + session?.start(); + const messaging = credential.kind === 'workspace-key' + ? new RelaycastMessagingClient({ workspaceKey: credential.workspaceKey, ...baseUrl, agentToken: registration.token }) + : rotatingMessaging(session!, agentToken => new RelaycastMessagingClient({ ...baseUrl, agentToken })); const cwd = join(dataDir, 'communication', prefix); await mkdir(cwd, { recursive: true, mode: 0o700 }); broker = await HarnessDriverClient.spawn({ ...brokerCredential, brokerName: prefix, cwd, channels: [], ...(process.env.RELAYFLOW_RELAY_BROKER_BIN ? { binaryPath: process.env.RELAYFLOW_RELAY_BROKER_BIN } : {}), env: brokerEnvironment(process.env) }); return { broker, messaging, prefix, channel, project: createProjection(messaging, channel, runId), close: async () => { - try { await broker!.shutdown(); } finally { await workspace.delete(publisher); } + try { await broker!.shutdown(); } finally { await release(); } } }; } catch (error) { - try { await broker?.shutdown(); } finally { await workspace.delete(publisher); } + try { await broker?.shutdown(); } finally { await release(); } throw error; } } diff --git a/packages/sdk/src/communication/run-token.ts b/packages/sdk/src/communication/run-token.ts index 9a27d5ee8..33661db87 100644 --- a/packages/sdk/src/communication/run-token.ts +++ b/packages/sdk/src/communication/run-token.ts @@ -77,8 +77,9 @@ export function runTokenProvider(file: string): () => Promise { }; } +export interface AgentRegistration { token: string; tokenExpiresAt?: string } export interface RunTokenAgents { - register(input: { name: string; type: string }): Promise<{ token: string }>; + register(input: { name: string; type: string }): Promise; delete(name: string): Promise; } @@ -88,31 +89,55 @@ export interface RunTokenAgents { * `rt_live_` (relaycast#559). The Agent Relay messaging client takes one * static credential at construction, so it cannot re-read a rotating file. */ +export function relaycastBaseUrl(baseUrl?: string): string { + return (baseUrl?.trim() || DEFAULT_BASE_URL).replace(/\/+$/, ''); +} + +/** A Relaycast HTTP failure. Carries status and code only, never a credential. */ +export class RelaycastHttpError extends Error { + constructor(readonly method: string, readonly path: string, readonly status: number, readonly code: string) { + super(`Relaycast ${method} ${path} failed (${status} ${code}).`); + } +} + +/** One envelope-aware Relaycast request authenticated by `token`. */ +export async function relaycastRequest(doFetch: typeof fetch, base: string, token: string, + method: string, path: string, body?: unknown): Promise { + const response = await doFetch(`${base}${path}`, { + method, + headers: { authorization: `Bearer ${token}`, ...(body === undefined ? {} : { 'content-type': 'application/json' }) }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); + if (response.status === 204) return undefined; + let envelope: { ok?: unknown; data?: unknown; error?: { code?: unknown; message?: unknown } } | undefined; + try { envelope = await response.json() as typeof envelope; } catch { envelope = undefined; } + if (response.ok && envelope?.ok === true) return envelope.data; + const code = typeof envelope?.error?.code === 'string' ? envelope.error.code : 'unexpected_response'; + throw new RelaycastHttpError(method, path, response.status, code); +} + export function runTokenAgents(credential: () => Promise, options: { baseUrl?: string; fetch?: typeof fetch } = {}): RunTokenAgents { - const base = (options.baseUrl?.trim() || DEFAULT_BASE_URL).replace(/\/+$/, ''); + const base = relaycastBaseUrl(options.baseUrl); const doFetch = options.fetch ?? fetch; + // A 401 can mean Cloud rotated the file between our read and the request; + // retry once, and the provider re-reads the file for the retry. async function call(method: string, path: string, body?: unknown): Promise { - const token = await credential(); - const response = await doFetch(`${base}${path}`, { - method, - headers: { authorization: `Bearer ${token}`, ...(body === undefined ? {} : { 'content-type': 'application/json' }) }, - ...(body === undefined ? {} : { body: JSON.stringify(body) }), - }); - if (response.status === 204) return undefined; - let envelope: { ok?: unknown; data?: unknown; error?: { code?: unknown; message?: unknown } } | undefined; - try { envelope = await response.json() as typeof envelope; } catch { envelope = undefined; } - if (response.ok && envelope?.ok === true) return envelope.data; - const code = typeof envelope?.error?.code === 'string' ? envelope.error.code : 'unexpected_response'; - throw new Error(`Relaycast ${method} ${path} failed (${response.status} ${code}).`); + try { + return await relaycastRequest(doFetch, base, await credential(), method, path, body); + } catch (error) { + if (!(error instanceof RelaycastHttpError) || error.status !== 401) throw error; + return relaycastRequest(doFetch, base, await credential(), method, path, body); + } } return { async register(input) { - const data = await call('POST', '/v1/agents', input) as { token?: unknown } | undefined; + const data = await call('POST', '/v1/agents', input) as { token?: unknown; token_expires_at?: unknown } | undefined; if (typeof data?.token !== 'string' || !data.token) { throw new Error('Relaycast POST /v1/agents returned no agent token.'); } - return { token: data.token }; + return { token: data.token, + ...(typeof data.token_expires_at === 'string' ? { tokenExpiresAt: data.token_expires_at } : {}) }; }, async delete(name) { await call('DELETE', `/v1/agents/${encodeURIComponent(name)}`); diff --git a/packages/sdk/tests/communication-agent-token.test.ts b/packages/sdk/tests/communication-agent-token.test.ts new file mode 100644 index 000000000..228ae43f1 --- /dev/null +++ b/packages/sdk/tests/communication-agent-token.test.ts @@ -0,0 +1,173 @@ +import { mkdtempSync, renameSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + spawn: vi.fn(), + clients: [] as Array>, + send: vi.fn(async (_agentToken: unknown, _input: unknown): Promise => ({ id: 'm1' })), +})); +vi.mock('@agent-relay/harness-driver', () => ({ HarnessDriverClient: { spawn: mocks.spawn } })); +vi.mock('@agent-relay/sdk', () => ({ + RelaycastMessagingClient: class { + agents = {}; channels = {}; + messages: Record; + constructor(options: Record) { + mocks.clients.push(options); + this.messages = { send: (input: unknown) => mocks.send(options.agentToken, input) }; + } + }, +})); +import { acquireRelayRuntime } from '../src/communication/relay.js'; +import { runTokenAgents, runTokenProvider } from '../src/communication/run-token.js'; +import { AgentTokenSession, rotationDelay } from '../src/communication/agent-token.js'; + +// Synthetic values; never real credentials. +const RUN_FIRST = 'rt_live_synthetic0first'; +const RUN_SECOND = 'rt_live_synthetic0second'; +const AGENT_FIRST = 'at_live_synthetic0agent1'; +const AGENT_SECOND = 'at_live_synthetic0agent2'; +const T0 = Date.parse('2026-10-17T12:00:00Z'); +const MIN = 60_000; +const BASE = 'https://relaycast.test'; + +const dir = mkdtempSync(join(tmpdir(), 'flows-agent-token-')); +const runFile = join(dir, 'relaycast-run.json'); +function writeRunFile(token: string) { + const temp = `${runFile}.tmp`; + writeFileSync(temp, JSON.stringify({ token, expiresAt: new Date(Date.now() + 15 * MIN).toISOString() }), { mode: 0o600 }); + renameSync(temp, runFile); +} +interface Call { method: string; path: string; authorization: string } +let calls: Call[]; +type Responder = (call: Call) => Response | Promise; +function fakeFetch(responder: Responder) { + calls = []; + return vi.fn(async (url: string, init: RequestInit) => { + const call = { method: init.method!, path: url.slice(BASE.length), + authorization: (init.headers as Record).authorization! }; + calls.push(call); + return responder(call); + }) as unknown as typeof fetch; +} +const ok = (data: unknown, status = 200) => Response.json({ ok: true, data }, { status }); +const expiresIn = (ms: number) => new Date(Date.now() + ms).toISOString(); + +beforeEach(() => { mocks.clients.length = 0; }); +afterEach(() => { vi.useRealTimers(); vi.unstubAllEnvs(); vi.unstubAllGlobals(); vi.clearAllMocks(); }); +afterAll(() => rmSync(dir, { recursive: true, force: true })); + +describe('rotation schedule', () => { + it('rotates at two thirds of the lifetime or five minutes before expiry, whichever is earlier', () => { + expect(rotationDelay(T0, T0 + 60 * MIN)).toBe(40 * MIN); + expect(rotationDelay(T0, T0 + 15 * MIN)).toBe(10 * MIN); + expect(rotationDelay(T0, T0 + 6 * MIN)).toBe(1 * MIN); + expect(rotationDelay(T0, T0 + 3 * MIN)).toBe(0); + expect(rotationDelay(T0, T0 - 1)).toBe(0); + }); + + it('rotates the agent token before it expires, authenticated by the agent itself', async () => { + vi.useFakeTimers({ now: T0 }); + const doFetch = fakeFetch(() => ok({ token: AGENT_SECOND, token_expires_at: expiresIn(15 * MIN) })); + const session = new AgentTokenSession('flow-journal', + { token: AGENT_FIRST, tokenExpiresAt: new Date(T0 + 15 * MIN).toISOString() }, { baseUrl: BASE, fetch: doFetch }); + session.start(); + await vi.advanceTimersByTimeAsync(10 * MIN - 1); + expect(calls).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + expect(calls).toEqual([{ method: 'POST', path: '/v1/agents/flow-journal/rotate-token', authorization: `Bearer ${AGENT_FIRST}` }]); + expect(session.current()).toBe(AGENT_SECOND); + // The next rotation follows the new token's expiry, still before it lapses. + await vi.advanceTimersByTimeAsync(10 * MIN); + expect(calls).toHaveLength(2); + expect(calls[1]!.authorization).toBe(`Bearer ${AGENT_SECOND}`); + session.stop(); + await vi.advanceTimersByTimeAsync(60 * MIN); + expect(calls).toHaveLength(2); + }); + + it('fails closed on a rejected rotation and retries transient failures', async () => { + vi.useFakeTimers({ now: T0 }); + let status = 503; + const doFetch = fakeFetch(() => Response.json({ ok: false, error: { code: 'x', message: 'x' } }, { status })); + const session = new AgentTokenSession('a', { token: AGENT_FIRST, tokenExpiresAt: new Date(T0 + 15 * MIN).toISOString() }, + { baseUrl: BASE, fetch: doFetch }); + session.start(); + await vi.advanceTimersByTimeAsync(10 * MIN); + expect(calls).toHaveLength(1); + status = 401; + await vi.advanceTimersByTimeAsync(2_000); + expect(calls).toHaveLength(2); + await vi.advanceTimersByTimeAsync(30 * MIN); + expect(calls).toHaveLength(2); + expect(session.current()).toBe(AGENT_FIRST); + }); +}); + +describe('401 retry', () => { + it('register retries once with a run token file rotated between attempts', async () => { + writeRunFile(RUN_FIRST); + const doFetch = fakeFetch(call => { + if (call.authorization === `Bearer ${RUN_FIRST}`) { + writeRunFile(RUN_SECOND); // Cloud rotated the file while the request was in flight. + return Response.json({ ok: false, error: { code: 'unauthorized', message: 'expired' } }, { status: 401 }); + } + return ok({ token: AGENT_FIRST, token_expires_at: expiresIn(15 * MIN) }, 201); + }); + const agents = runTokenAgents(runTokenProvider(runFile), { baseUrl: BASE, fetch: doFetch }); + await expect(agents.register({ name: 'a', type: 'agent' })).resolves.toMatchObject({ token: AGENT_FIRST }); + expect(calls.map(call => call.authorization)).toEqual([`Bearer ${RUN_FIRST}`, `Bearer ${RUN_SECOND}`]); + }); + + it('retries only once', async () => { + writeRunFile(RUN_FIRST); + const doFetch = fakeFetch(() => Response.json({ ok: false, error: { code: 'unauthorized', message: 'x' } }, { status: 401 })); + const agents = runTokenAgents(runTokenProvider(runFile), { baseUrl: BASE, fetch: doFetch }); + await expect(agents.delete('a')).rejects.toThrow('(401 unauthorized)'); + expect(calls).toHaveLength(2); + }); +}); + +describe('runtime messaging', () => { + function stubRelaycast() { + vi.stubGlobal('fetch', fakeFetch(call => { + if (call.path.endsWith('/rotate-token')) return ok({ token: AGENT_SECOND, token_expires_at: expiresIn(15 * MIN) }); + if (call.method === 'DELETE') return new Response(null, { status: 204 }); + return ok({ name: 'x', token: AGENT_FIRST, token_expires_at: expiresIn(15 * MIN), run_id: 'run-1' }, 201); + })); + } + beforeEach(() => { + mocks.spawn.mockResolvedValue({ spawnPty: vi.fn(), onEvent: vi.fn(), shutdown: vi.fn(async () => {}) }); + vi.stubEnv('RELAY_BASE_URL', BASE); + vi.stubEnv('RELAYFLOW_RELAY_BROKER_BIN', undefined); + writeRunFile(RUN_FIRST); + vi.stubEnv('RELAY_RUN_TOKEN_FILE', runFile); + }); + const message = { channel: 'wf-run', text: 'hi', idempotencyKey: 'k1', metadata: {} }; + + it('uses the rotated agent token after a scheduled rotation', async () => { + vi.useFakeTimers({ now: T0, toFake: ['setTimeout', 'clearTimeout', 'Date'] }); + stubRelaycast(); + const runtime = await acquireRelayRuntime(dir, 'run-rotate'); + await runtime.messaging.messages.send(message); + expect(mocks.send).toHaveBeenLastCalledWith(AGENT_FIRST, message); + await vi.advanceTimersByTimeAsync(10 * MIN); + expect(calls.at(-1)).toMatchObject({ path: `/v1/agents/${runtime.prefix}-journal/rotate-token`, authorization: `Bearer ${AGENT_FIRST}` }); + await runtime.messaging.messages.send(message); + expect(mocks.send).toHaveBeenLastCalledWith(AGENT_SECOND, message); + // Neither agent token ever reaches the broker. + expect(JSON.stringify(mocks.spawn.mock.calls)).not.toMatch(/at_live_|rt_live_/); + await runtime.close(); + }); + + it('forces one rotation and retries when messaging gets a 401', async () => { + stubRelaycast(); + const runtime = await acquireRelayRuntime(dir, 'run-401'); + mocks.send.mockImplementationOnce(async () => { throw Object.assign(new Error('unauthorized'), { status: 401 }); }); + await expect(runtime.messaging.messages.send(message)).resolves.toEqual({ id: 'm1' }); + expect(mocks.send.mock.calls.map(([token]) => token)).toEqual([AGENT_FIRST, AGENT_SECOND]); + expect(calls.filter(call => call.path.endsWith('/rotate-token'))).toHaveLength(1); + await runtime.close(); + }); +});