diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..273893c --- /dev/null +++ b/.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 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": { 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 e9310d1..c077d01 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); } } @@ -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 { + 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;