ci: gate pull requests on tests and typecheck

.github/workflows/ci.yml

@@ -0,0 +1,41 @@
+name: CI
+
+on:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    name: Test and typecheck
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          # This job builds and runs code from the pull request head. Leaving
+          # the checkout token in the runner's git config would expose it to
+          # every npm lifecycle script and test subprocess that follows.
+          persist-credentials: false
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: '22'
+          cache: npm
+
+      - name: Install dependencies
+        run: npm ci
+
+      # `npm run typecheck` already runs `build:primitives` and builds core, and
+      # `npm test` needs the primitives built. Running typecheck first means one
+      # build serves both gates instead of building the primitives twice.
+      - name: Typecheck
+        run: npm run typecheck
+
+      # Always report the test result, so a typecheck failure does not hide it.
+      - name: Test
+        if: ${{ !cancelled() }}
+        run: npm test

package.json

@@ -14,7 +14,7 @@
   "scripts": {
     "build:primitives": "npm run build --workspace=packages/github-primitive --workspace=packages/slack-primitive --workspace=packages/browser-primitive",
     "build": "npm run build:primitives && npm run build --workspace=packages/core && npm run build --workspace=packages/cli",
-    "typecheck": "npm run build:primitives && npm run typecheck --workspace=packages/core && npm run typecheck --workspace=packages/cli",
+    "typecheck": "npm run build:primitives && npm run build --workspace=packages/core && npm run typecheck --workspace=packages/core && npm run typecheck --workspace=packages/cli",
     "test": "npm run test --workspace=packages/core"
   },
   "devDependencies": {

packages/core/src/__tests__/pty-rekey.test.ts

@@ -0,0 +1,120 @@
+/**
+ * The broker can assign an agent a different name than the one we asked for.
+ * Re-keying the PTY maps to that name is one critical section with an
+ * unavoidable `await` in it — the old log stream must be closed before its file
+ * can be renamed — and a `worker_stream` for the old name can arrive inside
+ * that window. These tests hold that window open deliberately and assert that
+ * nothing addressed to the old name is lost.
+ */
+import { createWriteStream, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { WorkflowRunner, type WorkflowDb } from '../runner.js';
+import type { WorkflowRunRow, WorkflowStep, WorkflowStepRow } from '../types.js';
+
+function makeDb(): WorkflowDb {
+  const runs = new Map<string, WorkflowRunRow>();
+  const steps = new Map<string, WorkflowStepRow>();
+  return {
+    insertRun: vi.fn(async (run) => runs.set(run.id, { ...run })),
+    updateRun: vi.fn(async (id, patch) => {
+      const run = runs.get(id);
+      if (run) runs.set(id, { ...run, ...patch });
+    }),
+    getRun: vi.fn(async (id) => {
+      const run = runs.get(id);
+      return run ? { ...run } : null;
+    }),
+    insertStep: vi.fn(async (step) => steps.set(step.id, { ...step })),
+    updateStep: vi.fn(async (id, patch) => {
+      const step = steps.get(id);
+      if (step) steps.set(id, { ...step, ...patch });
+    }),
+    getStepsByRunId: vi.fn(async (runId) =>
+      [...steps.values()].filter((s) => s.runId === runId).map((s) => ({ ...s }))
+    ),
+  };
+}
+
+const step: WorkflowStep = { name: 'worker', type: 'agent', agent: 'a', task: 't' };
+
+describe('PTY re-key when the broker renames an agent', () => {
+  const tempDirs: string[] = [];
+  afterEach(() => {
+    for (const dir of tempDirs.splice(0)) rmSync(dir, { recursive: true, force: true });
+  });
+
+  function setup() {
+    const logsDir = mkdtempSync(path.join(os.tmpdir(), 'relayflows-pty-rekey-'));
+    tempDirs.push(logsDir);
+    const runner = new WorkflowRunner({ db: makeDb(), cwd: logsDir, workspaceId: 'ws-test' }) as any;
+
+    const oldLogPath = path.join(logsDir, 'old-name.log');
+    writeFileSync(oldLogPath, 'before\n');
+    runner.ptyOutputBuffers.set('old-name', ['before\n']);
+    runner.ptyLogStreams.set('old-name', createWriteStream(oldLogPath, { flags: 'a' }));
+    runner.ptyListeners.set('old-name', () => {
+      throw new Error('the pre-rekey listener must be replaced, not invoked');
+    });
+
+    return { runner, logsDir };
+  }
+
+  async function flush(runner: any, name: string) {
+    await new Promise<void>((resolve) => runner.ptyLogStreams.get(name)?.end(resolve));
+  }
+
+  it('captures a chunk that arrives for the old name during the await window', async () => {
+    const { runner, logsDir } = setup();
+    const seen: string[] = [];
+
+    // Do NOT await: this leaves the critical section suspended on the log
+    // stream close, which is exactly the window the race lives in.
+    const rekey = runner.rekeyPtyStreams({
+      oldName: 'old-name',
+      newName: 'new-name',
+      logsDir,
+      step,
+      humanAssistanceConfig: undefined,
+      onChunk: (info: { agentName: string; chunk: string }) => seen.push(`${info.agentName}:${info.chunk}`),
+    });
+
+    const inFlight = runner.ptyListeners.get('old-name');
+    expect(inFlight, 'the old name must still route somewhere mid-swap').toBeTypeOf('function');
+    inFlight('during\n');
+
+    await rekey;
+    await flush(runner, 'new-name');
+
+    // The chunk reached the buffer, under the new name.
+    expect(runner.ptyOutputBuffers.get('new-name')).toContain('during\n');
+    expect(runner.ptyOutputBuffers.has('old-name')).toBe(false);
+    // It was parked while no stream existed, then written to the renamed file.
+    expect(readFileSync(path.join(logsDir, 'new-name.log'), 'utf8')).toBe('before\nduring\n');
+    // And it was reported to the caller as the new agent.
+    expect(seen).toEqual(['new-name:during\n']);
+    // The old key is retired only once the swap is complete.
+    expect(runner.ptyListeners.has('old-name')).toBe(false);
+    expect(runner.ptyListeners.has('new-name')).toBe(true);
+  });
+
+  it('keeps the earlier buffer contents and routes chunks after the swap', async () => {
+    const { runner, logsDir } = setup();
+
+    await runner.rekeyPtyStreams({
+      oldName: 'old-name',
+      newName: 'new-name',
+      logsDir,
+      step,
+      humanAssistanceConfig: undefined,
+    });
+
+    runner.ptyListeners.get('new-name')('after\n');
+    await flush(runner, 'new-name');
+
+    expect(runner.ptyOutputBuffers.get('new-name')).toEqual(['before\n', 'after\n']);
+    expect(readFileSync(path.join(logsDir, 'new-name.log'), 'utf8')).toBe('before\nafter\n');
+  });
+});

packages/core/src/runner.ts

@@ -295,6 +295,16 @@ function sleepMs(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+function closeWriteStream(stream: WriteStream): Promise<void> {
+  if (stream.closed) return Promise.resolve();
+
+  return new Promise((resolve) => {
+    const settle = () => resolve();
+    stream.once('error', settle);
+    stream.end(settle);
+  });
+}
+
 // ── DB adapter interface ────────────────────────────────────────────────────
 
 /** Minimal DB adapter so the runner is not coupled to a specific driver. */
@@ -7646,7 +7656,7 @@ export class WorkflowRunner {
         combined: combinedOutput,
       });
       stopHeartbeat?.();
-      logStream.end();
+      await closeWriteStream(logStream);
       this.unregisterWorker(agentName);
     }
   }
@@ -7837,52 +7847,16 @@ export class WorkflowRunner {
         }
       }
 
-      // Re-key PTY maps if broker assigned a different name than requested
+      // Re-key PTY maps if broker assigned a different name than requested.
       if (agent.name !== agentName) {
-        const oldName = agentName;
-        this.ptyOutputBuffers.set(agent.name, this.ptyOutputBuffers.get(oldName) ?? []);
-        this.ptyOutputBuffers.delete(oldName);
-
-        // Close old log stream and rename the file to match the new agent name
-        const oldLogPath = path.join(logsDir, `${oldName}.log`);
-        const newLogPath = path.join(logsDir, `${agent.name}.log`);
-        const oldLogStream = this.ptyLogStreams.get(oldName);
-        if (oldLogStream) {
-          oldLogStream.end();
-          this.ptyLogStreams.delete(oldName);
-          try {
-            renameSync(oldLogPath, newLogPath);
-          } catch {
-            // File may not exist yet if no output was written
-          }
-        }
-
-        // Open new log stream with the correct name
-        const newLogStream = createWriteStream(newLogPath, { flags: 'a' });
-        this.ptyLogStreams.set(agent.name, newLogStream);
-
-        // Update listener to use the new log stream
-        const oldListener = this.ptyListeners.get(oldName);
-        if (oldListener) {
-          this.ptyListeners.delete(oldName);
-          const resolvedAgentName = agent.name;
-          this.ptyListeners.set(resolvedAgentName, (chunk: string) => {
-            const stripped = WorkflowRunner.stripAnsi(chunk);
-            const buffer = this.ptyOutputBuffers.get(resolvedAgentName);
-            buffer?.push(stripped);
-            newLogStream.write(chunk);
-            if (this.isSlackHumanAssistanceEnabled(humanAssistanceConfig)) {
-              this.observeHumanAssistanceOutput({
-                agentName: resolvedAgentName,
-                step,
-                config: humanAssistanceConfig,
-                output: buffer?.join('') ?? stripped,
-              });
-            }
-            options.onChunk?.({ agentName: resolvedAgentName, chunk });
-          });
-        }
-
+        await this.rekeyPtyStreams({
+          oldName: agentName,
+          newName: agent.name,
+          logsDir,
+          step,
+          humanAssistanceConfig,
+          onChunk: options.onChunk,
+        });
         agentName = agent.name;
       }
 
@@ -8036,7 +8010,7 @@ export class WorkflowRunner {
       this.ptyListeners.delete(agentName);
       const stream = this.ptyLogStreams.get(agentName);
       if (stream) {
-        stream.end();
+        await closeWriteStream(stream);
         this.ptyLogStreams.delete(agentName);
       }
       this.unregisterWorker(agentName);
@@ -8630,6 +8604,86 @@ export class WorkflowRunner {
     return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
   }
 
+  /**
+   * Move the PTY buffer, log stream and output listener from the name we asked
+   * the broker for to the name it actually assigned.
+   *
+   * This is one critical section containing an unavoidable `await`: the old log
+   * stream has to be closed before its file can be renamed. A `worker_stream`
+   * for the old name can arrive inside that window, so nothing the old listener
+   * depends on may be torn down before it. The buffer is therefore captured by
+   * reference rather than looked up by a key that is about to change, chunks
+   * are parked while no log stream exists, and the old listener key is dropped
+   * only once the swap has finished.
+   */
+  private async rekeyPtyStreams(params: {
+    oldName: string;
+    newName: string;
+    logsDir: string;
+    step: WorkflowStep;
+    humanAssistanceConfig: HumanAssistanceConfig | undefined;
+    onChunk?: (info: { agentName: string; chunk: string }) => void;
+  }): Promise<void> {
+    const { oldName, newName, logsDir, step, humanAssistanceConfig, onChunk } = params;
+
+    const buffer = this.ptyOutputBuffers.get(oldName) ?? [];
+    this.ptyOutputBuffers.set(newName, buffer);
+    this.ptyOutputBuffers.delete(oldName);
+
+    const oldLogPath = path.join(logsDir, `${oldName}.log`);
+    const newLogPath = path.join(logsDir, `${newName}.log`);
+    const oldLogStream = this.ptyLogStreams.get(oldName);
+
+    let newLogStream: ReturnType<typeof createWriteStream> | undefined;
+    const parkedChunks: string[] = [];
+    const writeToLog = (chunk: string) => {
+      if (newLogStream) {
+        newLogStream.write(chunk);
+      } else {
+        parkedChunks.push(chunk);
+      }
+    };
+
+    if (this.ptyListeners.has(oldName)) {
+      const rekeyedListener = (chunk: string) => {
+        const stripped = WorkflowRunner.stripAnsi(chunk);
+        buffer.push(stripped);
+        writeToLog(chunk);
+        if (this.isSlackHumanAssistanceEnabled(humanAssistanceConfig)) {
+          this.observeHumanAssistanceOutput({
+            agentName: newName,
+            step,
+            config: humanAssistanceConfig,
+            output: buffer.join('') || stripped,
+          });
+        }
+        onChunk?.({ agentName: newName, chunk });
+      };
+      // Registered under both names across the await window so a chunk still
+      // addressed to the old name is captured rather than dropped.
+      this.ptyListeners.set(newName, rekeyedListener);
+      this.ptyListeners.set(oldName, rekeyedListener);
+    }
+
+    if (oldLogStream) {
+      await closeWriteStream(oldLogStream);
+      this.ptyLogStreams.delete(oldName);
+      try {
+        renameSync(oldLogPath, newLogPath);
+      } catch {
+        // File may not exist yet if no output was written
+      }
+    }
+
+    newLogStream = createWriteStream(newLogPath, { flags: 'a' });
+    this.ptyLogStreams.set(newName, newLogStream);
+    for (const parked of parkedChunks.splice(0)) {
+      newLogStream.write(parked);
+    }
+
+    this.ptyListeners.delete(oldName);
+  }
+
   private resolveHumanAssistanceConfig(step: WorkflowStep): HumanAssistanceConfig | undefined {
     if (step.humanAssistance === false) return undefined;
     return step.humanAssistance ?? this.currentConfig?.swarm.humanAssistance;