← back to Visual Factory
server.js
1168 lines
// Visual Factory orchestrator (:9892)
require('dotenv').config();
const fs = require('node:fs/promises');
const { constants: fsConstants } = require('node:fs');
const path = require('node:path');
const os = require('node:os');
const { spawn } = require('node:child_process');
const crypto = require('node:crypto');
const express = require('express');
const helmet = require('helmet');
const { Pool } = require('pg');
const PG_DATABASE = process.env.PG_DATABASE || 'visual_factory';
if (PG_DATABASE === 'dw_unified') {
console.error('[visual-factory] FATAL: PG_DATABASE=dw_unified is forbidden. Use the standalone `visual_factory` DB.');
process.exit(1);
}
const { runPipeline } = require('./src/pipeline');
// In-process FIFO queue. Caps how many pipelines run concurrently so a flurry
// of submissions doesn't crush Ollama / Playwright on this single Mac. Pending
// runs sit at status='pending' (set by the INSERT) until promoted by pump().
//
// Codex P2 #13 fix: clamp to integer ≥1; 0/negative/NaN would silently deadlock
// the queue (pump never moves anything) — fail fast on bad config.
const PIPELINE_CONCURRENCY = (() => {
const raw = process.env.PIPELINE_CONCURRENCY;
const n = Number.parseInt(raw, 10);
if (raw !== undefined && (!Number.isFinite(n) || n < 1)) {
console.error(`[visual-factory] FATAL: PIPELINE_CONCURRENCY=${JSON.stringify(raw)} invalid; must be integer ≥1`);
process.exit(1);
}
return Number.isFinite(n) && n >= 1 ? n : 2;
})();
const _inFlight = new Set();
const _queue = [];
function enqueueRun(_pg, run) {
_queue.push(run);
_pump(_pg);
}
function _pump(_pg) {
while (_inFlight.size < PIPELINE_CONCURRENCY && _queue.length) {
const r = _queue.shift();
_inFlight.add(r.id);
runPipeline(_pg, r)
.catch(err => console.error('[visual-factory] pipeline crash:', err))
.finally(() => { _inFlight.delete(r.id); _pump(_pg); });
}
}
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');
});
// Same-origin: the UI is served from this same Express, so CORS is unnecessary.
// Kept commented as a record of the prior split-port architecture.
// app.use((req, res, next) => { res.set('Access-Control-Allow-Origin','*'); ... });
// Block snapshot/backup artifacts from ever serving off static root (standing
// fleet-refactor rule). Catches `.bak`, `.bak.<anything>`, `.pre-<anything>`,
// `.orig`, and `~`-suffixed editor backups anywhere in the URL path.
app.use((req, res, next) => {
if (/(\.bak(\..*)?|\.pre-[^/]*|\.orig|~)(\?|$)/i.test(req.path)) {
return res.status(404).type('text/plain').send('not found');
}
next();
});
// Serve the static UI (public/index.html etc.) at the orchestrator root. This
// collapses the previous 9892 (API) + 9893 (UI) split into a single port.
app.use(express.static(path.join(__dirname, 'public')));
// SECURITY (P0 fix 2026-05-04): mutating routes (POST /runs, retry, activate,
// delete, post-ig) had no auth — any local browser tab or process could
// trigger pipeline runs / activate artifacts / delete data. Token check via
// X-Admin-Token header or ?token= query. If ADMIN_TOKEN not set, opt-out for
// pure-loopback dev (mirror ai-factory pattern but soft-default for now).
const ADMIN_TOKEN = process.env.ADMIN_TOKEN || '';
function requireToken(req, res, next) {
if (!ADMIN_TOKEN) return next(); // dev opt-out; set ADMIN_TOKEN in pm2 env to enforce
const t = req.get('x-admin-token') || req.query.token;
if (t !== ADMIN_TOKEN) return res.status(401).json({ error: 'unauthorized' });
next();
}
const PORT = Number(process.env.ORCHESTRATOR_PORT || 9892);
const PUBLISH_ROOT = process.env.PUBLISH_ROOT || path.join(os.homedir(), 'Pictures', 'visual-factory-published');
const MANIFEST_PATH = path.join(PUBLISH_ROOT, 'manifest.json');
const SANDBOX_OUTPUT = path.resolve(path.join(__dirname, 'output'));
const PUBLISH_ROOT_RESOLVED = path.resolve(PUBLISH_ROOT);
// Codex P1 #5/#6 fix: assert a path lives inside an allowlisted root before we
// fs.copyFile / sendFile it. DB-resident paths can be poisoned (a buggy stage
// or future bug could write `/etc/passwd` into artifacts.png_path); resolving
// against a fixed root + path.sep prefix-check defeats relative-path traversal.
function assertWithin(root, p) {
const resolvedRoot = path.resolve(root);
const resolved = path.resolve(p);
if (resolved !== resolvedRoot && !resolved.startsWith(resolvedRoot + path.sep)) {
const err = new Error(`path "${p}" escapes sandbox "${root}"`);
err.code = 'EPATHESCAPE';
throw err;
}
return resolved;
}
// Codex P1 #5 fix: artifact_name must be safe for use in a filename. Mirror the
// kebab-case rule the intake stage already applies, but enforce server-side
// before constructing destination paths.
const ARTIFACT_NAME_RE = /^[a-z0-9][a-z0-9-]{0,79}$/;
function isSafeArtifactName(name) {
return typeof name === 'string' && ARTIFACT_NAME_RE.test(name);
}
// Codex P1 #9 fix: manifest.json is shared mutable state across activations.
// A simple promise-chained mutex serializes read-modify-write so concurrent
// activations don't lose entries. Same-process only — fine since the
// orchestrator runs as a single Node process.
let _manifestChain = Promise.resolve();
function withManifestLock(fn) {
const next = _manifestChain.then(() => fn(), () => fn());
_manifestChain = next.catch(() => {});
return next;
}
// Append (upsert by artifact_name) an entry to the publish-folder manifest.
// Best-effort: never throws — activation must succeed even if manifest write fails.
// Serialized via withManifestLock so concurrent activations don't clobber each other.
// Atomic write via .tmp + rename so a crash mid-write can't leave a corrupt file.
async function appendManifest(entry) {
return withManifestLock(async () => {
try {
let arr = [];
try {
const raw = await fs.readFile(MANIFEST_PATH, 'utf8');
const parsed = JSON.parse(raw);
if (Array.isArray(parsed)) arr = parsed;
} catch (e) { if (e.code !== 'ENOENT') throw e; }
const idx = arr.findIndex(e => e.artifact_name === entry.artifact_name);
if (idx >= 0) arr[idx] = entry; else arr.push(entry);
const tmp = MANIFEST_PATH + '.tmp.' + process.pid;
await fs.writeFile(tmp, JSON.stringify(arr, null, 2), 'utf8');
await fs.rename(tmp, MANIFEST_PATH);
return { ok: true, count: arr.length };
} catch (err) {
return { ok: false, error: err.message };
}
});
}
// Codex P1 #7 fix (round 2): validate any URL we hand off to Norma to fetch.
// Reject non-HTTP(S), userinfo, private/loopback IPs (incl. bracketed IPv6,
// IPv4-mapped IPv6, and `localhost.` with trailing dot), and resolve DNS so
// names that resolve into private space are also blocked (DNS rebinding).
const dns = require('node:dns').promises;
const http = require('node:http');
const https = require('node:https');
function _normalizeHost(rawHost) {
let h = (rawHost || '').toLowerCase();
if (h.startsWith('[') && h.endsWith(']')) h = h.slice(1, -1); // strip IPv6 brackets
if (h.endsWith('.')) h = h.slice(0, -1); // strip trailing dot
return h;
}
function _isBlockedIPv4(addr) {
const v4 = addr.match(/^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/);
if (!v4) return false;
const [, a, b, c] = v4.map(Number);
if (a === 127 || a === 10 || a === 0) return true;
if (a === 169 && b === 254) return true; // link-local + AWS metadata
if (a === 172 && b >= 16 && b <= 31) return true;
if (a === 192 && b === 168) return true;
if (a === 100 && b >= 64 && b <= 127) return true; // CGNAT
// Audit P2: reserved/special-purpose ranges that shouldn't be reachable
// from a public-image fetcher. Cheap to block; prevents probing of internal
// protocol assignments and benchmarking infra.
if (a === 192 && b === 0 && c === 0) return true; // 192.0.0/24 IETF protocol assignments (incl. 192.0.0.170 NAT64 disco)
if (a === 192 && b === 0 && c === 2) return true; // 192.0.2/24 TEST-NET-1
if (a === 198 && b === 51 && c === 100) return true; // 198.51.100/24 TEST-NET-2
if (a === 203 && b === 0 && c === 113) return true; // 203.0.113/24 TEST-NET-3
if (a === 198 && (b === 18 || b === 19)) return true; // 198.18/15 RFC2544 benchmark
if (a >= 224 && a <= 239) return true; // 224/4 multicast
if (a >= 240) return true; // 240/4 reserved (incl. 255.255.255.255 broadcast)
return false;
}
// Expand fully-zero-compressed IPv4-mapped to dotted-quad. Handles forms
// like `0:0:0:0:0:ffff:7f00:1` that `dns.resolve6` may emit instead of the
// `::ffff:127.0.0.1` shorthand the original regex assumed.
function _expandV4Mapped(a) {
// Already in dotted form: ::ffff:1.2.3.4
let m = a.match(/^(?:0:){5}ffff:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$/);
if (m) return m[1];
m = a.match(/^::ffff:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$/);
if (m) return m[1];
// Hex pair form: ::ffff:7f00:1 or 0:0:0:0:0:ffff:7f00:1
m = a.match(/^(?:0:){5}ffff:([0-9a-f]{1,4}):([0-9a-f]{1,4})$/) ||
a.match(/^::ffff:([0-9a-f]{1,4}):([0-9a-f]{1,4})$/);
if (m) {
const hi = parseInt(m[1], 16), lo = parseInt(m[2], 16);
return `${(hi >> 8) & 0xff}.${hi & 0xff}.${(lo >> 8) & 0xff}.${lo & 0xff}`;
}
return null;
}
// Expand any IPv6 string (including compressed `::` and `100::1` shorthand)
// to a full 8-group form so prefix matching is reliable. Returns null on
// invalid input. Codex 2026-05-01 P2: textual prefix checks miss valid
// in-range forms like `100::1` (in 100::/64) and `fed0::1` (in fec0::/10).
function _expandIPv6(addr) {
if (!addr || typeof addr !== 'string') return null;
const a = addr.toLowerCase().split('%')[0]; // strip zone id if any
if (!a.includes(':')) return null;
const [head, tail = ''] = a.split('::');
const headParts = head ? head.split(':') : [];
const tailParts = tail ? tail.split(':') : [];
const fill = 8 - headParts.length - tailParts.length;
if (fill < 0) return null;
const groups = [...headParts, ...Array(fill).fill('0'), ...tailParts];
if (groups.length !== 8) return null;
// Pad each group to 4 hex chars so leading-zero forms compare correctly.
return groups.map(g => g.padStart(4, '0')).join(':');
}
// Numeric-prefix check: does `addr` fall within `prefix/bits`? Both args are
// expanded IPv6 strings (16 colon-separated 4-hex-char groups). Bits is the
// CIDR mask length. Returns false on parse error (caller should treat as
// untrusted and reject elsewhere).
function _ipv6InRange(addr, prefix, bits) {
const a = addr.replace(/:/g, '');
const p = prefix.replace(/:/g, '');
if (a.length !== 32 || p.length !== 32) return false;
// Compare bit-by-bit up to `bits` bits.
const fullNibbles = Math.floor(bits / 4);
if (a.slice(0, fullNibbles) !== p.slice(0, fullNibbles)) return false;
const remainder = bits % 4;
if (remainder === 0) return true;
const aN = parseInt(a[fullNibbles], 16);
const pN = parseInt(p[fullNibbles], 16);
const mask = 0xf << (4 - remainder);
return (aN & mask) === (pN & mask);
}
function _isBlockedIPv6(addr) {
const a = addr.toLowerCase();
if (a === '::1' || a === '::') return true;
// Expand once for all numeric range checks below — handles `100::1`,
// `fed0::1`, `fee0::1`, etc. that the old textual prefix regex missed.
const expanded = _expandIPv6(a);
if (expanded) {
// Reserved blocks specified by CIDR. Each entry: [prefix-expanded, bits, label]
const RANGES = [
['fe80:0000:0000:0000:0000:0000:0000:0000', 10, 'fe80::/10 link-local'],
['fc00:0000:0000:0000:0000:0000:0000:0000', 7, 'fc00::/7 ULA'],
['0100:0000:0000:0000:0000:0000:0000:0000', 64, '100::/64 RFC6666 discard'],
['0064:ff9b:0000:0000:0000:0000:0000:0000', 96, '64:ff9b::/96 NAT64'],
['2001:0db8:0000:0000:0000:0000:0000:0000', 32, '2001:db8::/32 documentation'],
['fec0:0000:0000:0000:0000:0000:0000:0000', 10, 'fec0::/10 deprecated site-local'],
['ff00:0000:0000:0000:0000:0000:0000:0000', 8, 'ff00::/8 multicast'],
];
for (const [prefix, bits] of RANGES) {
if (_ipv6InRange(expanded, prefix, bits)) return true;
}
}
// IPv4-mapped — handle every canonical and expanded form.
const v4 = _expandV4Mapped(a);
if (v4 && _isBlockedIPv4(v4)) return true;
return false;
}
function isAllowedExternalUrl(input) {
let u;
try { u = new URL(input); } catch { return false; }
if (!/^https?:$/.test(u.protocol)) return false;
// userinfo is a confusion vector — `http://user@127.0.0.1/` etc. — disallow.
if (u.username || u.password) return false;
// Audit P3: reject IPv6 zone IDs (`http://[fe80::1%eth0]/` style). The fe80
// prefix would already trip _isBlockedIPv6, but other v6 with zones could
// slip through and confuse http.request. Cheap and total.
if (u.hostname.includes('%')) return false;
const host = _normalizeHost(u.hostname);
if (!host) return false;
if (host === 'localhost') return false;
if (_isBlockedIPv4(host)) return false;
if (_isBlockedIPv6(host)) return false;
return true;
}
// Async second-line check: resolve hostname and reject if any A/AAAA points
// into private space. Defends against DNS rebinding + private-DNS names that
// the literal-only check above can't see.
async function isAllowedAfterDns(input) {
if (!isAllowedExternalUrl(input)) return false;
const u = new URL(input);
const host = _normalizeHost(u.hostname);
if (/^\d{1,3}(\.\d{1,3}){3}$/.test(host) || host.includes(':')) return true; // literal — done
try {
const [v4s, v6s] = await Promise.all([
dns.resolve4(host).catch(() => []),
dns.resolve6(host).catch(() => []),
]);
const all = [...v4s, ...v6s];
if (!all.length) return false;
return all.every(a => !_isBlockedIPv4(a) && !_isBlockedIPv6(a));
} catch { return false; }
}
// Reveal the published file in Finder. macOS-only; non-fatal if `open` is missing.
function revealInFinder(filePath) {
try {
const child = spawn('open', ['-R', filePath], { detached: true, stdio: 'ignore' });
child.unref();
return true;
} catch { return false; }
}
// Norma instagram-agent (post.skill) host — single Instagram Business account.
// Stays in simulation mode until IG_USER_ID + IG_ACCESS_TOKEN are set on Norma's
// side. We just hand it a URL + caption and let it decide simulate vs real.
//
// Reconciled 2026-05-01: actual port is 9810 (not 9870 as skill docs originally
// claimed). Norma uses Basic auth — default user `admin`, password from
// AUTH_PASSWORD env (empty if unset). PID 26855 is the long-running healthy
// instance; another pm2 entry with the same name crashes on require('express')
// due to pm2's require-in-the-middle hook + root-shim conflict, but that
// crashloop doesn't affect the running one.
const NORMA_HOST = process.env.NORMA_HOST || 'http://127.0.0.1:9810';
const NORMA_USER = process.env.NORMA_AUTH_USER || 'admin';
const NORMA_PASS = process.env.NORMA_AUTH_PASS || '';
const NORMA_AUTH_HEADER = 'Basic ' + Buffer.from(`${NORMA_USER}:${NORMA_PASS}`).toString('base64');
const CLAUDE_CLI = process.env.CLAUDE_CLI || 'claude';
async function assertExecutable(command) {
if (path.isAbsolute(command) || command.includes(path.sep)) {
const stat = await fs.stat(command);
if (stat.isDirectory()) throw new Error(`${command} is a directory`);
await fs.access(command, fsConstants.X_OK);
return;
}
for (const dir of (process.env.PATH || '').split(path.delimiter).filter(Boolean)) {
try {
const candidate = path.join(dir, command);
const stat = await fs.stat(candidate);
if (stat.isDirectory()) continue;
await fs.access(candidate, fsConstants.X_OK);
return;
} catch {}
}
throw new Error(`${command} not found on PATH`);
}
function buildIgCaption(run, art) {
const lines = [];
if (run.purpose) lines.push(run.purpose);
if (run.domain && run.domain.includes('designerwallcoverings')) {
lines.push('', 'Designer Wallcoverings · One Stop Luxury Resource', '#DesignerWallcoverings #wallcovering #interiordesign');
} else if (run.domain) {
lines.push('', `via ${run.domain}`);
}
return lines.join('\n').trim() || (run.brief || '').slice(0, 280);
}
// Pre-fetch a user-supplied image_url and re-host it under our own loopback
// route so Norma never resolves the user's hostname. This closes the
// DNS-rebinding TOCTOU window that `isAllowedAfterDns` alone leaves open
// (the validator and Norma's later fetch each do their own DNS resolution).
//
// Safety guards layered here:
// 1. DNS pinning — resolve hostname once, validate the IP, then force
// http(s).request to connect to *that* IP via custom `lookup`. Defeats
// DNS rebinding even within the validate→fetch window.
// 2. Reject 3xx — never follow redirects (each would re-open the rebinding
// window on a different host).
// 3. Content-Type allowlist — only image/{png,jpeg,jpg,webp,gif}.
// 4. Streaming size cap — reject mid-stream once total > _MAX_PREFETCH_BYTES,
// so a server lying about Content-Length (or omitting it) can't OOM us.
// 5. Magic-byte verification — refuse to save bytes whose signature doesn't
// match an allowed image format, even if Content-Type claims it does.
// 6. 15s overall timeout.
// 7. Caller must already have validated the URL via `isAllowedAfterDns`.
const _ALLOWED_IMG_CT = /^image\/(png|jpe?g|webp|gif)\b/i;
const _MAX_PREFETCH_BYTES = 25 * 1024 * 1024;
// Magic-byte signatures for the formats we accept. Don't trust Content-Type.
function _detectImageMagic(buf) {
if (buf.length < 12) return null;
// PNG: 89 50 4E 47 0D 0A 1A 0A
if (buf[0] === 0x89 && buf[1] === 0x50 && buf[2] === 0x4E && buf[3] === 0x47 &&
buf[4] === 0x0D && buf[5] === 0x0A && buf[6] === 0x1A && buf[7] === 0x0A) return 'png';
// JPEG: FF D8 FF
if (buf[0] === 0xFF && buf[1] === 0xD8 && buf[2] === 0xFF) return 'jpg';
// GIF87a / GIF89a: 47 49 46 38 [37|39] 61
if (buf[0] === 0x47 && buf[1] === 0x49 && buf[2] === 0x46 && buf[3] === 0x38 &&
(buf[4] === 0x37 || buf[4] === 0x39) && buf[5] === 0x61) return 'gif';
// WebP: "RIFF" .... "WEBP"
if (buf[0] === 0x52 && buf[1] === 0x49 && buf[2] === 0x46 && buf[3] === 0x46 &&
buf[8] === 0x57 && buf[9] === 0x45 && buf[10] === 0x42 && buf[11] === 0x50) return 'webp';
return null;
}
// Resolve hostname → pick a public IP, then force the GET to that exact IP
// via http(s).request's `lookup` callback. Streams the body so Content-Length
// lies can't blow past maxBytes. Throws on any failure; caller catches.
async function safeFetchToBuffer(input, { maxBytes, timeoutMs }) {
const u = new URL(input);
const hostNorm = _normalizeHost(u.hostname);
// 1. Resolve hostname (skip if already an IP literal).
let pinnedIp;
let pinnedFamily;
const isV4Literal = /^\d{1,3}(\.\d{1,3}){3}$/.test(hostNorm);
const isV6Literal = hostNorm.includes(':');
if (isV4Literal) { pinnedIp = hostNorm; pinnedFamily = 4; }
else if (isV6Literal) { pinnedIp = hostNorm; pinnedFamily = 6; }
else {
const [v4s, v6s] = await Promise.all([
dns.resolve4(hostNorm).catch(() => []),
dns.resolve6(hostNorm).catch(() => []),
]);
const v4Pick = v4s.find(a => !_isBlockedIPv4(a));
const v6Pick = v6s.find(a => !_isBlockedIPv6(a));
if (v4Pick) { pinnedIp = v4Pick; pinnedFamily = 4; }
else if (v6Pick) { pinnedIp = v6Pick; pinnedFamily = 6; }
else throw new Error('no public IP for hostname');
}
// Defense in depth: even if isAllowedAfterDns just passed, recheck the IP
// we're about to pin in case DNS rebound between validation and now.
if (pinnedFamily === 4 && _isBlockedIPv4(pinnedIp)) throw new Error('pinned IP is private');
if (pinnedFamily === 6 && _isBlockedIPv6(pinnedIp)) throw new Error('pinned IP is private');
const lib = u.protocol === 'https:' ? https : http;
return await new Promise((resolve, reject) => {
let settled = false;
const finish = (fn, val) => { if (!settled) { settled = true; fn(val); } };
const opts = {
method: 'GET',
protocol: u.protocol,
hostname: u.hostname, // SNI + Host header
port: u.port || (u.protocol === 'https:' ? 443 : 80),
path: (u.pathname || '/') + (u.search || ''),
headers: { Host: u.host, 'User-Agent': 'visual-factory-prefetch/1' },
timeout: timeoutMs,
// agent:false forces a fresh socket per request so the global keep-alive
// pool can't hand back a connection that bypasses our `lookup` callback.
// Marginally slower; eliminates ambiguity on whether the pin was honored.
agent: false,
// Custom DNS lookup → always returns the validated IP, regardless of
// what real DNS says now. This is the DNS pin.
lookup: (_h, _o, cb) => cb(null, pinnedIp, pinnedFamily),
};
const req = lib.request(opts, (res) => {
// Refuse 3xx — following would re-open the rebinding window on a new host.
if (res.statusCode >= 300 && res.statusCode < 400) {
res.resume();
req.destroy();
return finish(reject, new Error(`redirect to ${res.headers.location || '?'} not allowed (would reopen DNS-rebinding window)`));
}
if (res.statusCode < 200 || res.statusCode >= 300) {
res.resume();
req.destroy();
return finish(reject, new Error(`http ${res.statusCode}`));
}
const ct = res.headers['content-type'] || '';
const m = ct.match(_ALLOWED_IMG_CT);
if (!m) {
res.resume();
req.destroy();
return finish(reject, new Error(`content-type "${ct}" not allowed; image/{png,jpeg,webp,gif} only`));
}
// Quick reject if server promised a too-big body.
const cl = Number(res.headers['content-length']);
if (Number.isFinite(cl) && cl > maxBytes) {
res.resume();
req.destroy();
return finish(reject, new Error(`Content-Length ${cl} exceeds ${maxBytes} cap`));
}
const chunks = [];
let total = 0;
res.on('data', (chunk) => {
total += chunk.length;
if (total > maxBytes) {
req.destroy();
return finish(reject, new Error(`body exceeded ${maxBytes} bytes mid-stream`));
}
chunks.push(chunk);
});
res.on('end', () => {
if (settled) return;
finish(resolve, {
buf: Buffer.concat(chunks),
contentType: ct,
ext: (m[1] || 'png').toLowerCase().replace('jpeg', 'jpg'),
});
});
res.on('error', (err) => finish(reject, err));
});
req.on('timeout', () => {
req.destroy();
finish(reject, new Error(`prefetch timed out after ${timeoutMs}ms`));
});
req.on('error', (err) => finish(reject, err));
req.end();
});
}
async function prefetchAndRehost(url, runId) {
try {
const fetched = await safeFetchToBuffer(url, { maxBytes: _MAX_PREFETCH_BYTES, timeoutMs: 15_000 });
// Trust bytes, not headers. Magic must match AND agree with Content-Type.
const detected = _detectImageMagic(fetched.buf);
if (!detected) {
return { ok: false, error: 'body bytes do not match any allowed image format (png/jpg/gif/webp)' };
}
if (detected !== fetched.ext) {
return { ok: false, error: `content-type ${fetched.contentType} disagrees with magic bytes (${detected})` };
}
const fname = `external-${Date.now()}.${detected}`;
const fpath = path.join(__dirname, 'output', `run-${runId}`, fname);
await fs.mkdir(path.dirname(fpath), { recursive: true });
await fs.writeFile(fpath, fetched.buf);
return { ok: true, path: fpath, fname, size: fetched.buf.length, contentType: fetched.contentType };
} catch (err) {
return { ok: false, error: err.message };
}
}
// Post the activated artifact to Norma's instagram-agent. Best-effort.
// `imageUrl` should be a URL Norma can reach — for simulation mode any string
// works; for real Meta-Graph posting it must be publicly fetchable.
async function postToInstagramViaNorma({ run, art, imageUrl }) {
const caption = buildIgCaption(run, art);
const ctrl = new AbortController();
const t = setTimeout(() => ctrl.abort(), 30_000);
try {
const r = await fetch(`${NORMA_HOST}/api/skill/post`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': NORMA_AUTH_HEADER,
},
body: JSON.stringify({ image_url: imageUrl, caption, pipeline_id: `vf-run-${run.id}` }),
signal: ctrl.signal
});
clearTimeout(t);
const body = await r.json().catch(() => ({}));
return { ok: r.ok && body.success !== false, status: r.status, caption, response: body };
} catch (err) {
clearTimeout(t);
return { ok: false, error: err.message, caption };
}
}
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: 'visual-factory-orchestrator', port: PORT, db: PG_DATABASE });
});
// Deep health: probes every dependency the pipeline needs. The viewer hits this
// on load so the user gets a clear "Ollama is down" message before submitting
// a run that's doomed to fail at compose. Returns 503 when any required dep is
// unhealthy so external monitors can alert.
app.get('/health/deep', async (_req, res) => {
const checks = {
db: 'unknown',
ollama: 'unknown',
claude_cli: 'unknown',
playwright: 'unknown',
};
const ollamaModels = { compose: process.env.COMPOSE_MODEL || 'qwen3:14b', vision: process.env.VISION_MODEL || 'llava:latest' };
await Promise.allSettled([
pg.query('SELECT 1').then(() => { checks.db = 'ok'; }, e => { checks.db = `error: ${e.message}`; }),
(async () => {
try {
const ctrl = new AbortController();
const t = setTimeout(() => ctrl.abort(), 3000);
const r = await fetch(`${process.env.OLLAMA_HOST || 'http://127.0.0.1:11434'}/api/tags`, { signal: ctrl.signal });
clearTimeout(t);
if (!r.ok) { checks.ollama = `http ${r.status}`; return; }
const j = await r.json();
const have = new Set((j.models || []).map(m => m.name));
const missing = Object.values(ollamaModels).filter(m => !have.has(m));
checks.ollama = missing.length ? `ok but missing: ${missing.join(', ')}` : 'ok';
} catch (e) { checks.ollama = `error: ${e.message}`; }
})(),
(async () => {
try {
await assertExecutable(CLAUDE_CLI);
checks.claude_cli = 'ok';
} catch { checks.claude_cli = 'missing — critic stage will fail'; }
})(),
(async () => {
try {
require.resolve('playwright');
checks.playwright = 'ok';
} catch { checks.playwright = 'missing'; }
})(),
]);
// Codex P1 #12 fix: every required dep gates `ok`. The route docstring says
// "probes every dependency the pipeline needs," so missing claude_cli or
// playwright must surface as 503, not silently green. Ollama "ok but missing"
// (model not pulled) also blocks ok — pipeline will fail at compose/vision.
const ok =
checks.db === 'ok' &&
checks.ollama === 'ok' &&
checks.claude_cli === 'ok' &&
checks.playwright === 'ok';
res.status(ok ? 200 : 503).json({
ok,
checks,
queue: { in_flight: _inFlight.size, queued: _queue.length, capacity: PIPELINE_CONCURRENCY },
});
});
app.get('/runs', async (_req, res) => {
try {
// Enrich with critic score (latest artifact) and last_error message (latest
// error event) so the UI can show both on the card without per-row fetches.
const { rows } = await pg.query(`
SELECT r.id, r.brief, r.domain, r.purpose, r.visual_type, r.artifact_name,
r.status, r.current_stage, r.width, r.height, r.created_at, r.finished_at,
a.score, e.message AS last_error
FROM runs r
LEFT JOIN LATERAL (
SELECT score FROM artifacts WHERE run_id = r.id ORDER BY id DESC LIMIT 1
) a ON TRUE
LEFT JOIN LATERAL (
SELECT message FROM events
WHERE run_id = r.id AND level = 'error'
ORDER BY id DESC LIMIT 1
) e ON TRUE
ORDER BY r.id DESC
LIMIT 100
`);
res.json({ runs: rows });
} catch (err) {
res.status(500).json({ error: err.message, hint: 'Did you createdb visual_factory + apply sql/001_init.sql?' });
}
});
app.post('/runs', requireToken, async (req, res) => {
let { brief, domain, purpose, details } = req.body || {};
domain = (domain || '').trim();
purpose = (purpose || '').trim();
details = (details || '').trim();
brief = (brief || '').trim();
// P1 fix 2026-05-04: cap inputs. Without this an 8000-char brief drives a
// full qwen3:14b inference (locks GPU for minutes) — trivial DoS.
if (brief.length > 8000) return res.status(400).json({ error: 'brief too long (max 8000 chars)' });
if (domain.length > 256 || purpose.length > 256 || details.length > 4000) {
return res.status(400).json({ error: 'domain/purpose/details too long' });
}
// If domain+purpose given, construct a brief. Free-text brief still wins if provided.
if (!brief) {
if (!domain || !purpose) {
return res.status(400).json({ error: 'either `brief` (string) OR `domain` + `purpose` required' });
}
brief = `Create a visual for **${domain}**.\n\nPurpose: ${purpose}.`;
if (details) brief += `\n\nDetails: ${details}`;
}
try {
const { rows } = await pg.query(
'INSERT INTO runs (brief, domain, purpose) VALUES ($1,$2,$3) RETURNING *',
[brief, domain || null, purpose || null]
);
const run = rows[0];
enqueueRun(pg, run);
res.json({ run });
} catch (err) {
res.status(500).json({ error: err.message });
}
});
// Retry a failed / unapproved run. Resets status + current_stage, wipes its
// events and artifact rows, deletes its output dir, then re-fires the pipeline
// against the original brief/domain/purpose.
//
// Codex P0 #2 fix: refuse on `running`/`pending` (would race with an active
// pipeline writing the same row); use an atomic UPDATE-with-WHERE-status guard
// so two concurrent retry calls can't both proceed and double-fire the queue.
app.post('/runs/:id/retry', requireToken, async (req, res) => {
const runId = Number(req.params.id);
const force = req.query.force === 'true';
try {
const { rows: lookupRows } = await pg.query('SELECT * FROM runs WHERE id=$1', [runId]);
if (!lookupRows.length) return res.status(404).json({ error: 'run not found' });
const run = lookupRows[0];
// `running` / `pending` mean a live pipeline may still be writing this row
// (or it's queued behind one). Refuse unless force=true. `awaiting_activation*`
// has value worth preserving — delete-and-resubmit is the right move.
const retryable = /^(failed|completed_unapproved|unapproved)/.test(String(run.status));
if (!retryable && !force) {
return res.status(409).json({
error: `retry refused — status=${run.status}; pass ?force=true to override`,
hint: run.status === 'running' || run.status === 'pending'
? 'live pipeline; force only after confirming the orchestrator no longer holds it'
: undefined,
});
}
// Atomic guard: only the first concurrent retry that flips this row from
// its eligible status into 'pending' wins. The losers see rowCount=0.
const eligibleStatusList = force
? ['failed','failed_intake','failed_orphan','completed_unapproved','unapproved','running','pending']
: ['failed','failed_intake','failed_orphan','completed_unapproved','unapproved'];
const claim = await pg.query(
`UPDATE runs
SET status='pending', current_stage=0, finished_at=NULL, updated_at=now()
WHERE id=$1 AND status = ANY($2::text[])
RETURNING id`,
[runId, eligibleStatusList]
);
if (claim.rowCount === 0) {
return res.status(409).json({ error: 'retry race lost — status changed between read and claim; refresh and try again' });
}
await pg.query('DELETE FROM artifacts WHERE run_id=$1', [runId]);
await pg.query('DELETE FROM events WHERE run_id=$1', [runId]);
try { await fs.rm(path.join(__dirname, 'output', `run-${runId}`), { recursive: true, force: true }); } catch {}
enqueueRun(pg, { ...run, status: 'pending', current_stage: 0 });
res.json({ ok: true, runId });
} catch (err) { res.status(500).json({ error: err.message }); }
});
// Hard-delete a run: cascades to events, removes artifact rows, wipes output dir.
// Refuses on activated rows by default (since publishing dir lives outside output/
// the activated PNG survives; but the audit trail is gone — that's the user's call).
//
// Codex P1 #4 fix: also refuse on `running`/`pending` unless force=true. Deleting
// an in-flight run leaves the pipeline writing to a row that no longer exists,
// which fails the next logEvent insert and races against the fs.rm of output/run-N.
app.delete('/runs/:id', requireToken, async (req, res) => {
const runId = Number(req.params.id);
const force = req.query.force === 'true';
try {
const { rows } = await pg.query('SELECT status FROM runs WHERE id=$1', [runId]);
if (!rows.length) return res.status(404).json({ error: 'run not found' });
const status = rows[0].status;
const inFlight = status === 'running' || status === 'pending';
if (status === 'activated' && !force) {
return res.status(409).json({ error: 'run is activated; pass ?force=true to delete anyway' });
}
if (inFlight && !force) {
return res.status(409).json({
error: `run is ${status}; deleting an in-flight run will race the pipeline. pass ?force=true if you've confirmed it's not actually running.`,
});
}
await pg.query('DELETE FROM artifacts WHERE run_id=$1', [runId]);
await pg.query('DELETE FROM runs WHERE id=$1', [runId]); // events CASCADE
try { await fs.rm(path.join(__dirname, 'output', `run-${runId}`), { recursive: true, force: true }); } catch {}
res.json({ ok: true, runId });
} 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 }); }
});
// Static-serve sandbox PNGs so the viewer can show thumbnails.
//
// Serve a pre-fetched external image for a run. Used by the post-ig flow
// when a user supplies image_url — we download it once into the sandbox,
// then hand Norma a loopback URL pointing here. Path-traversal defense:
// fname must match a strict pattern, and the resolved path is asserted
// to live under the per-run output dir.
app.get('/runs/:id/external/:fname', async (req, res) => {
const runId = Number(req.params.id);
const fname = String(req.params.fname || '');
if (!Number.isFinite(runId) || !/^external-\d+\.(png|jpg|webp|gif)$/.test(fname)) {
return res.status(400).json({ error: 'invalid run id or fname' });
}
const fpath = path.join(__dirname, 'output', `run-${runId}`, fname);
let safe;
try { safe = assertWithin(SANDBOX_OUTPUT, fpath); }
catch { return res.status(400).json({ error: 'path outside sandbox' }); }
// Codex P2 fix: realpath defeats symlinks pointing outside the sandbox.
// sendFile follows symlinks, so a future bug (or attacker who can drop a
// symlink into output/) could otherwise serve arbitrary files. Resolve
// the link target and re-assert containment.
let real;
try { real = await fs.realpath(safe); }
catch (e) { return res.status(404).json({ error: 'not found' }); }
try { assertWithin(SANDBOX_OUTPUT, real); }
catch { return res.status(400).json({ error: 'symlink target outside sandbox' }); }
res.sendFile(real, err => {
if (err && !res.headersSent) res.status(404).json({ error: 'not found' });
});
});
// Codex P1 #6 fix: assert png_path lives under our output sandbox (or the
// publish root for activated artifacts) before sendFile, so a poisoned DB row
// can't make us serve arbitrary files (e.g., /etc/passwd).
app.get('/runs/:id/png', async (req, res) => {
try {
const { rows } = await pg.query("SELECT png_path, published_path FROM artifacts WHERE run_id=$1 ORDER BY id DESC LIMIT 1", [Number(req.params.id)]);
if (!rows.length || !rows[0].png_path) return res.status(404).json({ error: 'no png yet' });
const candidate = rows[0].published_path || rows[0].png_path;
let safe;
try {
safe = (() => {
try { return assertWithin(SANDBOX_OUTPUT, candidate); }
catch { return assertWithin(PUBLISH_ROOT_RESOLVED, candidate); }
})();
} catch (e) {
console.error('[visual-factory] png path escapes sandbox:', candidate);
return res.status(403).json({ error: 'png path outside sandbox' });
}
// P0 fix 2026-05-04: realpath the resolved path to defeat symlinks planted
// in output/ — assertWithin's lexical check passes a symlink, but sendFile
// follows it. Must re-validate after realpath. Mirrors the /external/:fname
// hardening already in place.
let real;
try { real = await fs.realpath(safe); }
catch { return res.status(404).json({ error: 'not found' }); }
try {
try { assertWithin(SANDBOX_OUTPUT, real); }
catch { assertWithin(PUBLISH_ROOT_RESOLVED, real); }
} catch {
return res.status(403).json({ error: 'png realpath outside sandbox' });
}
res.sendFile(real);
} catch (err) { res.status(500).json({ error: err.message }); }
});
/**
* Post an activated run to Instagram via Norma's instagram-agent.
* Pre-conditions:
* - run must be 'activated' (we want a published_path that exists on disk)
* - Norma on :9870 reachable; if not, returns 502
* Caveats:
* - Norma is in simulation mode unless IG_USER_ID + IG_ACCESS_TOKEN set on Norma's side
* - Norma posts to a single Instagram Business account regardless of run.domain
* - For real posting, image_url must be publicly fetchable; we pass our own
* /runs/:id/png URL which is loopback-only, so real posting will fail at
* Meta until this PNG is hosted somewhere public. Simulation mode doesn't
* care — it just logs the URL.
*/
app.post('/runs/:id/post-ig', requireToken, async (req, res) => {
const runId = Number(req.params.id);
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 (run.status !== 'activated' && req.query.force !== 'true') {
return res.status(409).json({ error: `run status is ${run.status}; post requires activated (or ?force=true)` });
}
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];
// Loopback URL — Norma can fetch this if it lives on this Mac (same host),
// but Meta cannot reach it. For real posting, host the file publicly and
// override via req.body.image_url.
//
// Codex P1 #7 fix (SSRF): the override URL is forwarded to Norma, which
// fetches it. Reject anything that would let a poisoned body reach private
// ranges, file://, or AWS metadata. Only public http(s) hosts allowed.
let imageUrl = `http://127.0.0.1:${PORT}/runs/${runId}/png`;
if (req.body && typeof req.body.image_url === 'string' && req.body.image_url.length) {
// 1. Validator (synchronous + DNS-resolved) gates the URL we're about to fetch.
if (!(await isAllowedAfterDns(req.body.image_url))) {
return res.status(400).json({
error: 'image_url rejected — must be public http(s); private/loopback/metadata addresses blocked (DNS-resolved)',
});
}
// 2. Pre-fetch the image into our sandbox and re-host on a loopback URL
// so Norma never resolves the user's hostname (closes the DNS-rebinding
// TOCTOU window: validate-then-fetch could resolve to different IPs).
const fetched = await prefetchAndRehost(req.body.image_url, runId);
if (!fetched.ok) {
return res.status(400).json({ error: `image_url prefetch failed: ${fetched.error}` });
}
imageUrl = `http://127.0.0.1:${PORT}/runs/${runId}/external/${fetched.fname}`;
}
const result = await postToInstagramViaNorma({ run, art, imageUrl });
await pg.query(
"INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
[runId, 9, 'social_post', result.ok ? 'info' : 'warn',
result.ok
? `posted to IG (simulated=${result.response?.result?.simulated ?? 'unknown'}, media=${result.response?.result?.media_id || 'n/a'})`
: `IG post failed: ${result.error || 'http ' + result.status}`,
JSON.stringify(result)]
);
res.status(result.ok ? 200 : 502).json(result);
} catch (err) { res.status(500).json({ error: err.message }); }
});
/**
* Manual activation: copy approved PNG (and HTML) to PUBLISH_ROOT.
* Gates: run must be `awaiting_activation*` unless ?force=true. Won't overwrite without ?overwrite=true.
*/
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)` });
}
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];
// Codex P1 #5 fix: validate every input before constructing destination
// paths. artifact_name from the DB could be poisoned with `../etc/passwd`
// — re-validate against the kebab-case rule. html_path needs the same
// sandbox check we already had on png_path.
if (!isSafeArtifactName(art.artifact_name)) {
return res.status(400).json({ error: `unsafe artifact_name: ${JSON.stringify(art.artifact_name)}` });
}
let srcPng;
try { srcPng = assertWithin(SANDBOX_OUTPUT, art.png_path); }
catch { return res.status(400).json({ error: 'png_path escapes sandbox' }); }
let srcHtml = null;
if (art.html_path) {
try { srcHtml = assertWithin(SANDBOX_OUTPUT, art.html_path); }
catch { return res.status(400).json({ error: 'html_path escapes sandbox' }); }
}
await fs.mkdir(PUBLISH_ROOT, { recursive: true });
// Resolve destinations and confirm they live under PUBLISH_ROOT_RESOLVED
// (defense-in-depth: artifact_name is already validated, but path.join
// could surprise us on weird edge cases).
let destPng, destHtml;
try {
destPng = assertWithin(PUBLISH_ROOT_RESOLVED, path.join(PUBLISH_ROOT, `${art.artifact_name}.png`));
destHtml = assertWithin(PUBLISH_ROOT_RESOLVED, path.join(PUBLISH_ROOT, `${art.artifact_name}.html`));
} catch { return res.status(400).json({ error: 'destination escapes publish root' }); }
// Codex round-2 P1 fix: activate overwrite race. fs.access() then rename
// is non-atomic — two concurrent activations can both pass the access
// check. When overwrite=false, claim destination via fs.link() (atomic;
// fails with EEXIST if the target already exists). When overwrite=true,
// accept the inherent rename-clobber race (atomic at the FS level — one
// of two concurrent renames wins, neither corrupts).
// Cycle-2 Claude review P1: stat the destination unconditionally so the
// audit event correctly reports `existed_before=true` on overwrite. The
// 409 short-circuit only applies when overwrite=false.
let existed = false;
try { await fs.access(destPng); existed = true; } catch {}
if (existed && !overwrite) {
return res.status(409).json({ error: `target exists: ${destPng}`, hint: 'pass ?overwrite=true' });
}
// Codex P1 #8 fix: copy to .tmp first, claim destination atomically, then
// commit DB. Failure path cleans up tmp files so we don't leave halves.
const tmpPng = destPng + '.tmp.' + process.pid + '.' + Date.now();
const tmpHtml = srcHtml ? destHtml + '.tmp.' + process.pid + '.' + Date.now() : null;
let pngCommitted = false, htmlCommitted = false;
try {
await fs.copyFile(srcPng, tmpPng);
if (tmpHtml) await fs.copyFile(srcHtml, tmpHtml);
if (overwrite) {
// Cycle-3 Claude review P2: surface EXDEV/ENOTSUP on the overwrite
// path too (not just the link path). Without this, cross-device
// PUBLISH_ROOT fails with an opaque 500 instead of the helpful hint.
try { await fs.rename(tmpPng, destPng); pngCommitted = true; }
catch (e) {
if (e.code === 'EXDEV' || e.code === 'ENOTSUP') {
await fs.rm(tmpPng, { force: true }).catch(() => {});
if (tmpHtml) await fs.rm(tmpHtml, { force: true }).catch(() => {});
return res.status(500).json({
error: `cross-device rename not supported (${e.code})`,
hint: 'PUBLISH_ROOT and output/ are on different filesystems. Set PUBLISH_ROOT to a path on the same device.',
});
}
throw e;
}
if (tmpHtml) {
try { await fs.rename(tmpHtml, destHtml); htmlCommitted = true; }
catch (e) {
if (e.code === 'EXDEV' || e.code === 'ENOTSUP') {
// PNG already committed; tolerate HTML failure with partial
// status (matches existing partial-success pattern below).
await fs.rm(tmpHtml, { force: true }).catch(() => {});
throw new Error(`HTML rename cross-device (${e.code}); PNG committed`);
}
throw e;
}
}
} else {
// Atomic claim. If two activations race on the same artifact_name,
// exactly one wins; the loser sees EEXIST and 409s.
try { await fs.link(tmpPng, destPng); }
catch (e) {
if (e.code === 'EEXIST') {
await fs.rm(tmpPng, { force: true }).catch(() => {});
if (tmpHtml) await fs.rm(tmpHtml, { force: true }).catch(() => {});
return res.status(409).json({ error: `target exists: ${destPng} (lost activation race)`, hint: 'pass ?overwrite=true' });
}
// Cycle-2 Claude review P2: PUBLISH_ROOT on a different filesystem
// from output/ produces EXDEV (or ENOTSUP on some FS). Surface a
// hint instead of an opaque 500.
if (e.code === 'EXDEV' || e.code === 'ENOTSUP') {
await fs.rm(tmpPng, { force: true }).catch(() => {});
if (tmpHtml) await fs.rm(tmpHtml, { force: true }).catch(() => {});
return res.status(500).json({
error: `cross-device link not supported (${e.code})`,
hint: 'PUBLISH_ROOT and output/ are on different filesystems. Set PUBLISH_ROOT to a path on the same device, or accept the rename-clobber semantics with ?overwrite=true (which will hit the same EXDEV).',
});
}
throw e;
}
// Cycle-4 review polish: set pngCommitted immediately after the link
// succeeds (mirrors the overwrite=true branch), so the outer catch can
// never see a state where the link landed but the flag wasn't set.
pngCommitted = true;
await fs.rm(tmpPng, { force: true }).catch(() => {});
if (tmpHtml) {
try { await fs.link(tmpHtml, destHtml); htmlCommitted = true; }
catch (e) { if (e.code !== 'EEXIST') throw e; /* PNG won, HTML clash unlikely; tolerate */ }
await fs.rm(tmpHtml, { force: true }).catch(() => {});
}
}
await pg.query("UPDATE artifacts SET status='activated', published_path=$1 WHERE id=$2", [destPng, art.id]);
await pg.query("UPDATE runs SET status='activated', updated_at=now() WHERE id=$1", [runId]);
} catch (err) {
// Best-effort rollback: drop the partials. If we already renamed into
// dest but DB failed, leave the file (manifest will be inconsistent for
// one entry, which is recoverable; deleting a published file is worse).
// Note: when pngCommitted=true, tmpPng was either renamed away (overwrite
// path) or already removed via fs.rm (link path); the rm below is a
// no-op in those cases — kept for the partial-write paths only.
try { await fs.rm(tmpPng, { force: true }); } catch {}
if (tmpHtml) { try { await fs.rm(tmpHtml, { force: true }); } catch {} }
if (!pngCommitted) {
return res.status(500).json({ error: `activate failed before publish: ${err.message}` });
}
// We did publish PNG but failed later; still respond with a partial-failure status.
return res.status(500).json({
error: `activate partially succeeded: ${err.message}`,
png_committed: pngCommitted,
html_committed: htmlCommitted,
});
}
// Codex round-2 P2: was firing this INSERT twice (dup audit row).
await pg.query(
"INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
[runId, 8, 'activate', 'info', existed ? 'overwrote target' : 'wrote target', JSON.stringify({ srcPng, destPng })]
);
// Manifest write + Finder reveal happen AFTER the source-of-truth DB writes succeed.
// Both are best-effort — failures here don't roll back the activation.
const manifestEntry = {
run_id: runId,
artifact_name: art.artifact_name,
visual_type: art.visual_type,
domain: run.domain || null,
purpose: run.purpose || null,
brief: run.brief,
width: art.width,
height: art.height,
score: art.score,
dest_png: destPng,
dest_html: art.html_path ? destHtml : null,
activated_at: new Date().toISOString()
};
const manifestResult = await appendManifest(manifestEntry);
const finderOpened = req.query.no_finder === 'true' ? false : revealInFinder(destPng);
await pg.query(
"INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
[runId, 8, 'activate', manifestResult.ok ? 'info' : 'warn',
`manifest=${manifestResult.ok ? `count ${manifestResult.count}` : 'failed'}, finder=${finderOpened}`,
JSON.stringify({ manifest: manifestResult, finderOpened })]
);
// Optional one-shot IG post on activation (?post_ig=true). Best-effort.
// Norma is in simulation mode by default — see instagram-post-scheduler skill.
let igResult = null;
if (req.query.post_ig === 'true') {
const imageUrl = `http://127.0.0.1:${PORT}/runs/${runId}/png`;
igResult = await postToInstagramViaNorma({ run, art, imageUrl });
await pg.query(
"INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)",
[runId, 9, 'social_post', igResult.ok ? 'info' : 'warn',
igResult.ok
? `auto-posted to IG (simulated=${igResult.response?.result?.simulated ?? 'unknown'})`
: `auto IG post failed: ${igResult.error || 'http ' + igResult.status}`,
JSON.stringify(igResult)]
);
}
res.json({
ok: true,
srcPng,
destPng,
destHtml,
existed_before: existed,
manifest_path: MANIFEST_PATH,
manifest_ok: manifestResult.ok,
finder_opened: finderOpened,
ig_posted: igResult ? igResult.ok : null,
ig_result: igResult
});
} catch (err) { res.status(500).json({ error: err.message }); }
});
// Boot-time orphan cleanup: any run still at 'running' or 'pending' is by
// definition orphaned, because at boot the in-memory queue and in-flight set
// are both empty — no live process is working on them. Drop the 10-min age
// filter (which would strand recent crashes); mark every non-terminal row.
// Codex P0 #1 fix.
async function reapOrphans() {
try {
const { rowCount, rows } = await pg.query(
`UPDATE runs
SET status='failed_orphan', finished_at=now(), updated_at=now()
WHERE status IN ('running','pending')
RETURNING id, current_stage`
);
if (rowCount > 0) {
console.log(`[visual-factory] reaped ${rowCount} orphan run(s):`, rows.map(r => `#${r.id}@stage${r.current_stage}`).join(', '));
for (const r of rows) {
await pg.query(
'INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)',
[r.id, r.current_stage, 'reap', 'error', 'orchestrator restarted while this run was in flight; marked orphaned', null]
);
}
}
} catch (err) { console.error('[visual-factory] reapOrphans failed:', err.message); }
}
// Codex round-2 P1: reap before opening the listener, so a request that lands
// in the brief boot window can't get insta-orphaned by a reaper that hasn't
// run yet. Top-level await isn't available in a CommonJS module, so we wrap
// in an async IIFE.
(async () => {
try { await reapOrphans(); } catch (e) { console.error('[visual-factory] boot reaper threw:', e.message); }
app.listen(PORT, '0.0.0.0', () => {
console.log(`[visual-factory] orchestrator listening on http://127.0.0.1:${PORT} (db=${PG_DATABASE})`);
});
})();