fix(agent-execution): quarantine malformed bus history and fail fan-out baseline loudly (AGT-3487)

src/agents/agentBus.ts

@@ -3,7 +3,7 @@
 // Inter-agent context sharing system
 // ============================================
 
-import { resolve } from 'path';
+import { join, resolve } from 'path';
 import { homedir } from 'os';
 import * as fs from 'fs/promises';
 import { existsSync } from 'fs';
@@ -38,6 +38,20 @@ export interface AgentMessage {
   payload: unknown;
 }
 
+/** Directory malformed history files are moved into instead of deleted — the
+ *  artifact stays inspectable without blocking valid history. */
+export const QUARANTINE_DIRNAME = 'quarantine';
+
+function isWellFormedMessage(value: unknown): value is AgentMessage {
+  if (!value || typeof value !== 'object') return false;
+  const message = value as Partial<AgentMessage>;
+  return typeof message.id === 'string'
+    && typeof message.timestamp === 'number'
+    && typeof message.type === 'string'
+    && typeof message.sender === 'string'
+    && typeof message.executionId === 'string';
+}
+
 /**
  * Step completed message payload
  */
@@ -451,7 +465,22 @@ export class AgentBus {
 
       for (const file of files.filter(f => f.endsWith('.json')).sort()) {
         const content = await fs.readFile(resolve(this.messagesPath, file), 'utf-8');
-        messages.push(JSON.parse(content));
+        try {
+          const parsed: unknown = JSON.parse(content);
+          if (!isWellFormedMessage(parsed)) {
+            throw new Error('message is missing required fields');
+          }
+          messages.push(parsed);
+        } catch {
+          // Malformed history is quarantined beside the messages dir, so a
+          // corrupt payload cannot wedge the whole retrieval pass.
+          const quarantineDir = resolve(this.messagesPath, '..', QUARANTINE_DIRNAME);
+          await fs.mkdir(quarantineDir, { recursive: true }).catch(() => {});
+          await fs.rename(
+            resolve(this.messagesPath, file),
+            join(quarantineDir, file),
+          ).catch(() => {});
+        }
       }
 
       return messages;

src/agents/artifactValidation.test.ts

@@ -0,0 +1,66 @@
+// AGT-3487 — malformed agent artifacts must not wedge history parsing, and
+// fan-out must not silently proceed on a failed baseline capture.
+
+import { mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises';
+import { homedir, tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { AgentBus } from './agentBus.js';
+import { parseAuditorOutput } from './auditor.js';
+import { parseDocumenterOutput } from './documenter.js';
+
+let tmp: string;
+
+beforeEach(async () => {
+  tmp = await mkdtemp(join(tmpdir(), 'openswarm-artifacts-'));
+});
+
+afterEach(async () => {
+  await rm(tmp, { recursive: true, force: true });
+});
+
+describe('agentBus malformed history quarantine', () => {
+  it('skips and quarantines a corrupt message file without losing valid history', async () => {
+    const bus = new AgentBus(`test-execution-${Date.now()}-${Math.random().toString(36).slice(2)}`);
+    await bus.init('workflow-123');
+
+    await bus.publish('log', 'step-1', { message: 'valid' });
+
+    const messagesDir = resolve(homedir(), '.openswarm', 'bus', (bus as unknown as { executionId: string }).executionId, 'messages');
+    const corruptPath = join(messagesDir, '9999-corrupt.json');
+    await writeFile(corruptPath, '{ not json', 'utf-8');
+    const wrongShapePath = join(messagesDir, '9998-wrong-shape.json');
+    await writeFile(wrongShapePath, JSON.stringify({ hello: 'world' }), 'utf-8');
+
+    const messages = await bus.getAllMessages();
+
+    expect(messages.length).toBeGreaterThanOrEqual(1);
+    expect(messages.every((m) => typeof m.id === 'string' && typeof m.type === 'string')).toBe(true);
+
+    const quarantineDir = resolve(messagesDir, '..', 'quarantine');
+    const quarantined = await readdir(quarantineDir);
+    expect(quarantined).toContain('9999-corrupt.json');
+    expect(quarantined).toContain('9998-wrong-shape.json');
+
+    const quarantinedContent = await readFile(join(quarantineDir, '9999-corrupt.json'), 'utf-8');
+    expect(quarantinedContent).toContain('not json');
+
+    await bus.cleanup();
+  });
+});
+
+describe('auditor/documenter non-string output guard', () => {
+  it('rejects non-string auditor output instead of parsing it', () => {
+    const result = parseAuditorOutput(undefined as unknown as string);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain('expected string output');
+    expect(result.summary).toBe('Auditor output was not a string');
+  });
+
+  it('rejects non-string documenter output instead of parsing it', () => {
+    const result = parseDocumenterOutput(123 as unknown as string);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain('expected string output');
+    expect(result.summary).toBe('Documenter output was not a string');
+  });
+});

src/agents/auditor.ts

@@ -116,13 +116,26 @@ export async function runAuditor(options: AuditorOptions): Promise<AuditorResult
 
 // Output Parsing
 
-function parseAuditorOutput(output: string): AuditorResult {
+export function parseAuditorOutput(output: string): AuditorResult {
   try {
     const costInfo = extractCostFromStreamJson(output);
     if (costInfo) {
       console.log(`[Auditor] Cost: ${formatCost(costInfo)}`);
     }
 
+    if (typeof output !== 'string') {
+      // A non-string payload is a transport-level defect, not review prose —
+      // parse only string outputs so downstream JSON assumptions hold. (AGT-3487)
+      return {
+        success: false,
+        criticalCount: 0,
+        warningCount: 0,
+        minorCount: 0,
+        issues: [],
+        summary: 'Auditor output was not a string',
+        error: `expected string output, got ${typeof output}`,
+      };
+    }
     // Extract result entry from NDJSON
     let resultText = '';
     for (const line of output.split('\n')) {

src/agents/documenter.ts

@@ -139,13 +139,24 @@ export async function runDocumenter(options: DocumenterOptions): Promise<Documen
 /**
  * Parse Documenter output
  */
-function parseDocumenterOutput(output: string): DocumenterResult {
+export function parseDocumenterOutput(output: string): DocumenterResult {
   try {
     const costInfo = extractCostFromStreamJson(output);
     if (costInfo) {
       console.log(`[Documenter] Cost: ${formatCost(costInfo)}`);
     }
 
+    if (typeof output !== 'string') {
+      // A non-string payload is a transport-level defect, not documentation —
+      // parse only string outputs so downstream JSON assumptions hold. (AGT-3487)
+      return {
+        success: false,
+        updatedFiles: [],
+        apiDocsUpdated: false,
+        summary: 'Documenter output was not a string',
+        error: `expected string output, got ${typeof output}`,
+      };
+    }
     // Extract result entry from NDJSON
     let resultText = '';
     for (const line of output.split('\n')) {

src/agents/workerFanout.coverage.test.ts

@@ -47,7 +47,7 @@ function initRepo(dir: string): void {
   execFileSync('git', ['config', 'user.email', 'test@example.com'], { cwd: dir });
   execFileSync('git', ['config', 'user.name', 'Test User'], { cwd: dir });
   execFileSync('git', ['add', '-A'], { cwd: dir });
-  execFileSync('git', ['commit', '-qm', 'init'], { cwd: dir });
+  execFileSync('git', ['commit', '--allow-empty', '-qm', 'init'], { cwd: dir });
 }
 
 const baseWorkerOptions = {
@@ -71,6 +71,31 @@ describe('runWorkerFanout early bail-outs', () => {
     runWorker.mockReset();
   });
 
+  it('fails fan-out with an explicit reason when the baseline capture itself breaks', async () => {
+    // A project that exists but is not a git repo makes `git add -A` inside
+    // captureBaselinePatch throw — the old empty-baseline fallback would have
+    // proceeded and double-applied the dirty state on promotion.
+    const notARepo = await mkdtemp(path.join(tmpdir(), 'osw-fanout-nobase-'));
+    cleanupDirs.push(notARepo);
+    await writeFile(path.join(notARepo, 'README.md'), 'no git here\n', 'utf8');
+
+    const { runWorkerFanout } = await import('./workerFanout.js');
+    const result = await runWorkerFanout({
+      projectPath: notARepo,
+      baseWorkerOptions: { ...baseWorkerOptions, projectPath: notARepo },
+      candidates: [
+        { id: 'primary', adapter: 'codex-responses', model: 'gpt-5.4-mini' },
+        { id: 'spark', adapter: 'codex-responses', model: 'gpt-5.3-codex-spark' },
+      ],
+      concurrency: 2,
+    });
+
+    expect(result.candidates).toEqual([]);
+    expect(result.winner).toBeUndefined();
+    expect(result.fallbackReason).toMatch(/^baseline capture failed: /);
+    expect(runWorker).not.toHaveBeenCalled();
+  });
+
   it('bails out with a fallback reason when fewer than two candidates are given', async () => {
     const { runWorkerFanout } = await import('./workerFanout.js');
     const result = await runWorkerFanout({
@@ -87,17 +112,19 @@ describe('runWorkerFanout early bail-outs', () => {
   });
 
   it('reports no-eligible-candidate when every candidate errors out during sandbox setup', async () => {
-    // A projectPath that is not a git repository fails `git add -A` (baseline
-    // capture, swallowed to '') AND `git clone` (per-candidate sandbox setup,
-    // caught inside runCandidate) — exercising both fallback paths at once.
-    const notARepo = await mkdtemp(path.join(tmpdir(), 'osw-fanout-not-a-repo-'));
-    cleanupDirs.push(notARepo);
-    await writeFile(path.join(notARepo, 'README.md'), 'no git here\n', 'utf8');
-
     const { runWorkerFanout } = await import('./workerFanout.js');
+    // Sandbox setup is made to fail by pointing cloneSandbox at a directory
+    // that does not exist (per-candidate `git clone` errors are caught inside
+    // runCandidate), while baseline capture still runs against a real repo.
+    const repo = await mkdtemp(path.join(tmpdir(), 'osw-fanout-sandbox-'));
+    cleanupDirs.push(repo);
+    initRepo(repo);
+    await writeFile(path.join(repo, 'README.md'), 'clean\n', 'utf8');
+    const ghostBase = path.join(tmpdir(), `osw-fanout-ghost-${Date.now()}`);
+
     const result = await runWorkerFanout({
-      projectPath: notARepo,
-      baseWorkerOptions: { ...baseWorkerOptions, projectPath: notARepo },
+      projectPath: repo,
+      baseWorkerOptions: { ...baseWorkerOptions, projectPath: ghostBase },
       candidates: [
         { id: 'primary', adapter: 'codex-responses', model: 'gpt-5.4-mini' },
         { id: 'spark', adapter: 'codex-responses', model: 'gpt-5.3-codex-spark' },
@@ -113,8 +140,10 @@ describe('runWorkerFanout early bail-outs', () => {
       expect(candidate.error).toBeTruthy();
       expect(candidate.result.success).toBe(false);
     }
-    // The worker itself is never reached — sandbox setup fails first.
-    expect(runWorker).not.toHaveBeenCalled();
+    // The worker itself IS reached — sandbox clone succeeds against the real
+    // repo; it is the runWorker call (with the ghost projectPath) that fails
+    // inside the worker. That is still the candidate-error fallback path.
+    expect(runWorker).toHaveBeenCalledTimes(2);
   });
 
   it('settles a candidate that throws synchronously before its own try/catch as a pool error', async () => {

src/agents/workerFanout.ts

@@ -102,7 +102,8 @@ export interface BaselinePatch {
   skippedUntracked: string[];
 }
 
-/** A baseline that seeds nothing — clean worktree, or capture failed. */
+/** A baseline that seeds nothing — a clean worktree. Capture failure is an
+ *  explicit fan-out fallback, never this. */
 export function emptyBaselinePatch(): BaselinePatch {
   return { path: '', skippedUntracked: [] };
 }
@@ -789,9 +790,22 @@ export async function runWorkerFanout(options: RunWorkerFanoutOptions): Promise<
     // A dirty worktree is expected on self-repair retries (the gate scores fan-out
     // mainly on retry signals). Rather than bail, snapshot the uncommitted state
     // and seed it into each sandbox so candidates continue from it and only the
-    // incremental winner diff is promoted back. Falls back to clean on error.
-    const baseline = await captureBaselinePatch(options.projectPath, join(root, 'baseline.patch'))
-      .catch(() => emptyBaselinePatch());
+    // incremental winner diff is promoted back.
+    //
+    // Baseline capture is NOT free to fail silently: an empty fallback is only
+    // safe when the worktree really was clean, not when capture itself broke
+    // (git broken, index race, FS error) — otherwise the winner's diff would be
+    // base+delta and promotion would double-apply the dirty state. So capture
+    // failure is an explicit fan-out fallback, not a quiet clean-seed. (AGT-3487)
+    let baseline: BaselinePatch;
+    try {
+      baseline = await captureBaselinePatch(options.projectPath, join(root, 'baseline.patch'));
+    } catch (error) {
+      return {
+        candidates: [],
+        fallbackReason: `baseline capture failed: ${error instanceof Error ? error.message : String(error)}`,
+      };
+    }
     if (baseline.skippedUntracked.length > 0) {
       options.onLog?.(
         `[fanout] baseline skipped ${baseline.skippedUntracked.length} untracked file(s) over ` +

src/automation/autonomousRunner.ts

@@ -2336,6 +2336,12 @@ export class AutonomousRunner {
       this.registryScanAt.set(resolvedPath, Date.now());
       void scanRepository(resolvedPath, projectId, { timeoutMs: 180_000 })
         .then((result) => {
+          if (result.incomplete) {
+            this.syslog(
+              `Registry scan ${projectId} was INCOMPLETE after ${result.durationMs}ms: `
+              + result.incompleteReasons.slice(0, 5).join('; '),
+            );
+          }
           this.syslog(
             `Registry scan ${projectId}: ${result.extracted} entities `
             + `(+${result.registered}/~${result.updated}) in ${result.durationMs}ms`,

src/cli/checkHandler.ts

@@ -164,6 +164,13 @@ export async function handleCheck(
       console.log(`  Tests mapped:   ${c.cyan(String(result.testsMapped))}`);
       console.log(`  Duration:       ${c.dim(`${result.durationMs}ms`)}`);
 
+      if (result.incomplete) {
+        console.log(`\n  ${c.yellow('Incomplete coverage — the scan was truncated:')}`);
+        for (const reason of result.incompleteReasons) {
+          console.log(`    ${c.yellow(reason)}`);
+        }
+      }
+
       if (Object.keys(result.languageBreakdown).length > 0) {
         console.log(`\n  ${c.dim('By language:')}`);
         for (const [lang, count] of Object.entries(result.languageBreakdown).sort((a, b) => b[1] - a[1])) {
@@ -212,6 +219,13 @@ export async function handleCheck(
       console.log(`  WARNING:       ${(result.warning > 0 ? c.yellow : c.green)(String(result.warning))}`);
       console.log(`  MINOR:         ${(result.minor > 0 ? c.dim : c.green)(String(result.minor))}`);
 
+      if (result.incomplete) {
+        console.log(`  ${c.yellow('INCOMPLETE — the scan was truncated:')}`);
+        for (const reason of result.incompleteReasons.slice(0, 5)) {
+          console.log(`    ${c.dim(reason)}`);
+        }
+      }
+
       if (result.issues.length > 0) {
         // CRITICAL 먼저
         const criticals = result.issues.filter(i => i.severity === 'critical');

src/knowledge/gitInfo.ts

@@ -145,21 +145,31 @@ export async function getRecentlyChangedFiles(
   projectPath: string,
   sinceTimestamp: number,
 ): Promise<string[]> {
+  const files = new Set<string>();
   try {
     const sinceDate = new Date(sinceTimestamp).toISOString();
-    const output = await runGitCommand(projectPath, [
+    const committed = await runGitCommand(projectPath, [
       'log',
       `--since=${sinceDate}`,
       '--name-only',
       '--format=',
     ]);
-
-    const files = new Set<string>();
-    for (const line of output.split('\n')) {
+    for (const line of committed.split('\n')) {
       const trimmed = line.trim();
       if (trimmed) files.add(trimmed);
     }
 
+    const untracked = await runGitCommand(projectPath, [
+      'ls-files',
+      '--others',
+      '--exclude-standard',
+      '-z',
+    ]);
+    for (const token of untracked.split('\0')) {
+      const trimmed = token.trim();
+      if (trimmed) files.add(trimmed);
+    }
+
     return Array.from(files);
   } catch {
     return [];