From fffd383ee01be0cd0b075fcdca71e465f610654b Mon Sep 17 00:00:00 2001 From: Miya Date: Tue, 25 Aug 2026 09:15:27 +0200 Subject: [PATCH 1/4] ci: gate pull requests on tests and typecheck Session-Id: 01a037bb-4e8c-7c20-8962-621515f5e335 --- .github/workflows/ci.yml | 34 ++++++++++++++++++++++++++++++++++ package.json | 2 +- 2 files changed, 35 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..b1dff91 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,34 @@ +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 + + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: '22' + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Build primitives + run: npm run build:primitives + + - name: Test + run: npm test + + - name: Typecheck + run: npm run typecheck diff --git a/package.json b/package.json index 3c5fbfb..9cd8da4 100644 --- a/package.json +++ b/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": { From 475e25e012a245721fef85127f2bfdf29a4bfd25 Mon Sep 17 00:00:00 2001 From: Miya Date: Tue, 25 Aug 2026 09:20:37 +0200 Subject: [PATCH 2/4] fix: await worker log cleanup Session-Id: 01a037bb-4e8c-7c20-8962-621515f5e335 --- packages/core/src/runner.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index e9310d1..1f77d6f 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -295,6 +295,16 @@ function sleepMs(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +function closeWriteStream(stream: WriteStream): Promise { + 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); } } @@ -7848,7 +7858,7 @@ export class WorkflowRunner { const newLogPath = path.join(logsDir, `${agent.name}.log`); const oldLogStream = this.ptyLogStreams.get(oldName); if (oldLogStream) { - oldLogStream.end(); + await closeWriteStream(oldLogStream); this.ptyLogStreams.delete(oldName); try { renameSync(oldLogPath, newLogPath); @@ -8036,7 +8046,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); From 68ebb3d8ebe8524d462491c102a749afc9bb63f0 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 12:16:13 +0200 Subject: [PATCH 3/4] ci: do not persist the checkout token in the runner This job builds and runs pull-request code: npm ci executes lifecycle scripts and the suite spawns subprocesses. A token left in the runner's git config is reachable by all of them. Raised by coderabbitai on #39, fixed here because this workflow is #42's file. Session-Id: 08f4e75c-099d-49ad-b98e-26a37e4cf2c9 --- .github/workflows/ci.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b1dff91..031a218 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,6 +14,11 @@ jobs: 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 From 4050ece16df5cd410ff4945345d90d4d148d8915 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 12:20:55 +0200 Subject: [PATCH 4/4] fix(core): do not drop agent output while re-keying a renamed agent The broker can assign a different name than the one requested. Re-keying the PTY maps to that name is one critical section containing an unavoidable await -- 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. As it stood, the buffer was re-keyed before the await while the old listener stayed registered through it, so a chunk arriving mid-swap was looked up under a key that no longer existed and written to a stream that was closing: dropped twice over. Extracted to rekeyPtyStreams so the window is testable, and closed it: the buffer is captured by reference, chunks are parked while no stream exists and flushed in order, and the old listener key is retired only once the swap has finished. Raised as P2 by chatgpt-codex-connector on #42. The regression test holds the window open deliberately; it fails against the previous ordering. ci: build the primitives once (P3, cubic-dev-ai) and keep the checkout token out of the runner (coderabbitai). Session-Id: 08f4e75c-099d-49ad-b98e-26a37e4cf2c9 --- .github/workflows/ci.yml | 12 +- packages/core/src/__tests__/pty-rekey.test.ts | 120 ++++++++++++++++ packages/core/src/runner.ts | 134 ++++++++++++------ 3 files changed, 216 insertions(+), 50 deletions(-) create mode 100644 packages/core/src/__tests__/pty-rekey.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 031a218..273893c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,11 +29,13 @@ jobs: - name: Install dependencies run: npm ci - - name: Build primitives - run: npm run build:primitives + # `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 - - - name: Typecheck - run: npm run typecheck diff --git a/packages/core/src/__tests__/pty-rekey.test.ts b/packages/core/src/__tests__/pty-rekey.test.ts new file mode 100644 index 0000000..837260a --- /dev/null +++ b/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(); + const steps = new Map(); + 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((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'); + }); +}); diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index 1f77d6f..c077d01 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -7847,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) { - await closeWriteStream(oldLogStream); - 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; } @@ -8640,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 { + 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 | 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;