`server.js`

248 lines
// The AI Factory — orchestrator (:9890)
// Stub server. Pipeline stages are no-ops; see PLAN.md before wiring.

require('dotenv').config();
const express = require('express');
const helmet = require('helmet');
const { Pool } = require('pg');

const PG_DATABASE = process.env.PG_DATABASE || 'ai_factory';

// Hard safety guard: this project must NEVER touch dw_unified.
// Steve's standing rule (2026-04-30) — The AI Factory has its own standalone DB.
if (PG_DATABASE === 'dw_unified') {
  console.error('[ai-factory] FATAL: PG_DATABASE=dw_unified is forbidden. Use the standalone `ai_factory` DB.');
  process.exit(1);
}

const fs = require('node:fs/promises');
const path = require('node:path');
const os = require('node:os');
const { runPipeline } = require('./src/pipeline');
const { runMemoryWrite } = require('./src/stages/memory_write');

const CLAUDE_AGENTS_DIR = path.join(os.homedir(), '.claude', 'agents');
const CLAUDE_SKILLS_DIR = path.join(os.homedir(), '.claude', 'skills');

const app = express();
// Security headers via helmet (added 2026-05-04 overnight YOLO loop)
app.use(helmet({ contentSecurityPolicy: false }));
app.use(express.json({ limit: '1mb' }));

// LAN/tailnet auth gate (skip /health for watchdog probes)
const BASIC_AUTH = process.env.BASIC_AUTH || 'admin:DWSecure2024!';
const AUTH_HEADER = 'Basic ' + Buffer.from(BASIC_AUTH).toString('base64');
app.use((req, res, next) => {
  if (req.path === '/health' || req.path === '/healthz') return next();
  if (req.get('authorization') === AUTH_HEADER) return next();
  res.set('WWW-Authenticate', 'Basic realm="mac2-pm2"');
  res.status(401).send('auth required');
});

// SECURITY (P1 fix 2026-05-04): wildcard CORS was "safe since loopback" but
// CORS only restricts BROWSERS — any malicious page the user opens can fetch
// 127.0.0.1:9890 with credentials and trigger /runs or /runs/:id/activate.
// Allowlist the viewer origin only.
const VIEWER_ORIGIN = process.env.VIEWER_ORIGIN || 'http://127.0.0.1:9891';
app.use((req, res, next) => {
  const origin = req.headers.origin;
  if (origin === VIEWER_ORIGIN) {
    res.set('Access-Control-Allow-Origin', origin);
    res.set('Vary', 'Origin');
    res.set('Access-Control-Allow-Credentials', 'true');
  }
  res.set('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
  res.set('Access-Control-Allow-Headers', 'Content-Type, Authorization');
  if (req.method === 'OPTIONS') return res.sendStatus(204);
  next();
});

// SECURITY (P0 fix 2026-05-04): ADMIN_TOKEN was declared in .env but never
// enforced — any local process could trigger /runs or activate artifacts into
// ~/.claude/. Mutating routes now require Authorization: Bearer $ADMIN_TOKEN.
const ADMIN_TOKEN = process.env.ADMIN_TOKEN || '';
function requireToken(req, res, next) {
  if (!ADMIN_TOKEN) return res.status(503).json({ error: 'ADMIN_TOKEN not configured on server' });
  const auth = String(req.headers['authorization'] || '');
  if (auth !== `Bearer ${ADMIN_TOKEN}`) return res.status(401).json({ error: 'unauthorized' });
  next();
}

const PORT = Number(process.env.ORCHESTRATOR_PORT || 9890);

// Build pool config; only include `password` if one is actually set, so that
// Unix-socket peer auth (the default on this Mac) works without forcing SCRAM.
const poolCfg = {
  host: process.env.PG_HOST || '/tmp',
  port: Number(process.env.PG_PORT || 5432),
  database: PG_DATABASE,
  user: process.env.PG_USER || 'stevestudio2'
};
if (process.env.PG_PASSWORD) poolCfg.password = process.env.PG_PASSWORD;
const pg = new Pool(poolCfg);

app.get('/health', (_req, res) => {
  res.json({ ok: true, service: 'ai-factory-orchestrator', port: PORT, db: PG_DATABASE });
});

app.get('/runs', async (_req, res) => {
  try {
    const { rows } = await pg.query(
      'SELECT id, prompt, artifact_type, artifact_name, status, current_stage, created_at, finished_at FROM runs ORDER BY id DESC LIMIT 100'
    );
    res.json({ runs: rows });
  } catch (err) {
    res.status(500).json({ error: err.message, hint: 'Did you `createdb ai_factory` and apply sql/001_init.sql?' });
  }
});

app.post('/runs', requireToken, async (req, res) => {
  const { prompt, artifact_type = null, artifact_name = null } = req.body || {};
  if (!prompt || typeof prompt !== 'string') {
    return res.status(400).json({ error: 'prompt (string) required' });
  }
  try {
    const { rows } = await pg.query(
      'INSERT INTO runs (prompt, artifact_type, artifact_name) VALUES ($1,$2,$3) RETURNING *',
      [prompt, artifact_type, artifact_name]
    );
    const run = rows[0];
    // Fire-and-forget pipeline; errors are persisted to events/runs.
    runPipeline(pg, run).catch(err => console.error('[ai-factory] pipeline crash:', err));
    res.json({ run });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

/**
 * Manual activation: copy a sandboxed artifact into ~/.claude/agents or ~/.claude/skills.
 * Hard-gated:
 *   - Run status must be `awaiting_activation*` (the critic approved the final version),
 *     unless `?force=true` is passed.
 *   - Will not overwrite an existing target file unless `?overwrite=true` is passed.
 *   - Reads the artifact location from the artifacts table — only files in our sandbox
 *     under ~/Projects/the-ai-factory/output/ are eligible.
 */
app.post('/runs/:id/activate', requireToken, async (req, res) => {
  const runId = Number(req.params.id);
  const force = req.query.force === 'true';
  const overwrite = req.query.overwrite === 'true';
  try {
    const { rows: runRows } = await pg.query('SELECT * FROM runs WHERE id=$1', [runId]);
    if (!runRows.length) return res.status(404).json({ error: 'run not found' });
    const run = runRows[0];

    if (!force && !String(run.status).startsWith('awaiting_activation')) {
      return res.status(409).json({
        error: `run status is ${run.status}; activation requires awaiting_activation* (or ?force=true)`,
        hint: 'critic did not approve. Inspect the artifact, fix manually, and pass ?force=true to override.'
      });
    }

    const { rows: artRows } = await pg.query(
      "SELECT * FROM artifacts WHERE run_id=$1 ORDER BY id DESC LIMIT 1",
      [runId]
    );
    if (!artRows.length) return res.status(409).json({ error: 'no artifact for this run' });
    const art = artRows[0];

    const sandboxRoot = path.resolve(path.join(__dirname, 'output'));
    const src = path.resolve(art.location);
    if (!src.startsWith(sandboxRoot + path.sep)) {
      return res.status(400).json({ error: `refusing to activate from outside sandbox: ${src}` });
    }

    // SECURITY (P0 fix 2026-05-04): re-validate artifact_name from DB before
    // building the dest path. Intake regex ran earlier but a tampered DB row
    // (or a future bug in intake) could let `../../.ssh/authorized_keys`
    // resolve outside CLAUDE_AGENTS_DIR / CLAUDE_SKILLS_DIR.
    if (!/^[a-z0-9-]{2,60}$/.test(String(art.artifact_name || ''))) {
      return res.status(400).json({ error: 'artifact_name failed validation' });
    }
    let dest, allowedRoot;
    if (art.artifact_type === 'subagent') {
      allowedRoot = path.resolve(CLAUDE_AGENTS_DIR);
      dest = path.join(CLAUDE_AGENTS_DIR, `${art.artifact_name}.md`);
    } else if (art.artifact_type === 'skill') {
      allowedRoot = path.resolve(CLAUDE_SKILLS_DIR);
      dest = path.join(CLAUDE_SKILLS_DIR, art.artifact_name, 'SKILL.md');
    } else {
      return res.status(400).json({ error: `unsupported artifact_type: ${art.artifact_type}` });
    }
    // Defense-in-depth: confirm resolved dest stays under allowedRoot.
    const resolvedDest = path.resolve(dest);
    if (!resolvedDest.startsWith(allowedRoot + path.sep)) {
      return res.status(400).json({ error: 'dest path traversal blocked' });
    }

    let existed = false;
    try { await fs.access(dest); existed = true; } catch { /* ok */ }
    if (existed && !overwrite) {
      return res.status(409).json({
        error: `target already exists: ${dest}`,
        hint: 'pass ?overwrite=true to replace'
      });
    }

    await fs.mkdir(path.dirname(dest), { recursive: true });
    await fs.copyFile(src, dest);

    await pg.query("UPDATE artifacts SET status='activated', location=$1 WHERE id=$2", [dest, art.id]);
    await pg.query("UPDATE runs SET status='activated', updated_at=now() WHERE id=$1", [runId]);
    await pg.query(
      "INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
      [runId, 10, 'activate', 'info', existed ? 'overwrote target' : 'wrote target', JSON.stringify({ src, dest, force, overwrite })]
    );

    // Stage 9 — memory_write. Best-effort: log but don't fail activation if the
    // index append blows up. Spec is recovered from the intake event payload.
    let memory = null;
    try {
      const { rows: specRows } = await pg.query(
        "SELECT payload FROM events WHERE run_id=$1 AND stage_name='intake' AND message='spec extracted' ORDER BY id DESC LIMIT 1",
        [runId]
      );
      const spec = specRows[0]?.payload?.spec || null;
      memory = await runMemoryWrite({
        runId,
        artifactType: art.artifact_type,
        artifactName: art.artifact_name,
        spec,
        dest
      });
      await pg.query(
        "INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
        [runId, 9, 'memory_write', 'info',
          memory.skipped ? `skipped: ${memory.reason}` : 'appended to MEMORY.md',
          JSON.stringify({ line: memory.line })]
      );
    } catch (err) {
      await pg.query(
        "INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
        [runId, 9, 'memory_write', 'error', 'memory_write failed', JSON.stringify({ error: err.message })]
      );
    }

    res.json({ ok: true, src, dest, existed_before: existed, memory });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

app.get('/runs/:id/events', async (req, res) => {
  try {
    const { rows } = await pg.query(
      'SELECT id, stage, stage_name, level, message, payload, created_at FROM events WHERE run_id=$1 ORDER BY id ASC',
      [Number(req.params.id)]
    );
    res.json({ events: rows });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

app.listen(PORT, '0.0.0.0', () => {
  console.log(`[ai-factory] orchestrator listening on http://127.0.0.1:${PORT} (db=${PG_DATABASE})`);
});