feat(sdk): accept run-scoped Relaycast token file (RELAY_RUN_TOKEN_FILE)

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<typeof setTimeout> | undefined;
+  private inFlight: Promise<string> | 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<string> {
+    this.inFlight ??= this.rotateOnce().finally(() => { this.inFlight = undefined; });
+    return this.inFlight;
+  }
+
+  private async rotateOnce(): Promise<string> {
+    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<void> {
+    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<T>(operation: (messaging: RelayMessaging) => Promise<T>): Promise<T> {
+    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)),
+    },
+  };
+}

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;

packages/sdk/src/communication/preflight.ts

@@ -4,15 +4,20 @@ 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 {}
 
 /** 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 {

packages/sdk/src/communication/relay.ts

@@ -3,6 +3,8 @@ 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';
+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 {
@@ -40,9 +42,9 @@ export async function loadRelayModules(): Promise<RelayModules> {
   }
 }
 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 +77,41 @@ export async function acquireRelayRuntime(dataDir: string, runId: string): Promi
   }
 }
 async function createRuntime(dataDir: string, runId: string): Promise<RelayRuntime> {
-  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;
+  // Run-token mode: the journal agent's at_live_ expires (relaycast#550), so it
+  // is rotated in memory and the messaging client follows the current token.
+  const session = credential.kind === 'run-token' ? new AgentTokenSession(publisher, registration, baseUrl) : undefined;
+  const release = async () => { session?.stop(); await workspace.delete(publisher); };
   try {
-    const messaging = new RelaycastMessagingClient({ ...options, 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({ 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 release(); }
     } };
   } catch (error) {
-    try { await broker?.shutdown(); } finally { await workspace.agents.delete(publisher); }
+    try { await broker?.shutdown(); } finally { await release(); }
     throw error;
   }
 }