← back to Visual Factory
src/pipeline.js
244 lines
// Visual Factory pipeline. 9 stages.
// 1 intake → 2 design_system (folded into intake) → 3 compose → 4 render
// → 5 vision_check ‖ 6 critic-prep → 6 critic → 7 iterate (re-render+re-vision+re-critic)
// → 8 catalog → 9 activate (manual gate, separate endpoint)
const fs = require('node:fs/promises');
const { runIntake } = require('./stages/intake');
const { runCompose } = require('./stages/compose');
const { runRender } = require('./stages/render');
const { runVisionCheck } = require('./stages/vision_check');
const { runCritic } = require('./stages/critic');
const { runIterate } = require('./stages/iterate');
const STAGES = ['intake','compose','render','vision_check','critic','iterate','catalog','activate','cleanup'];
// Per-stage timeouts (ms). Tuned for local Ollama / Playwright / Claude CLI on
// the Mac Studio. If a stage hangs longer than this, abort the whole run so the
// UI doesn't show "running" forever. Critic + compose get longer budgets
// because qwen3:14b on a complex HTML brief or claude CLI on a tricky review
// can legitimately take 60–90s.
const STAGE_TIMEOUT_MS = {
intake: 45_000,
compose: 180_000,
render: 90_000,
vision_check: 90_000,
critic: 120_000,
iterate: 300_000,
};
// Codex P0 #3 fix: real cancellation. Each stage receives an AbortSignal that
// the underlying ollama fetch / Playwright browser / Claude CLI child honors,
// so when the timeout fires we actually kill the work instead of leaking a
// background browser/child holding GPU/CPU while the queue moves on.
//
// Usage: `await withTimeout('compose', signal => runCompose({runId, spec, signal}))`
function withTimeout(name, factory) {
const ms = STAGE_TIMEOUT_MS[name] || 60_000;
const controller = new AbortController();
let timer;
const timeoutP = new Promise((_, rej) => {
timer = setTimeout(() => {
controller.abort();
rej(new Error(`stage "${name}" timed out after ${ms}ms`));
}, ms);
});
// Run factory inside try to convert sync throws to rejections.
let workP;
try { workP = Promise.resolve(factory(controller.signal)); }
catch (e) { workP = Promise.reject(e); }
return Promise.race([workP, timeoutP]).finally(() => clearTimeout(timer));
}
async function logEvent(pg, runId, stage, name, level, message, payload) {
await pg.query(
'INSERT INTO events (run_id, stage, stage_name, level, message, payload) VALUES ($1,$2,$3,$4,$5,$6)',
[runId, stage, name, level, message, payload ? JSON.stringify(payload) : null]
);
}
async function setStage(pg, runId, stage, status) {
await pg.query('UPDATE runs SET current_stage=$1, status=$2, updated_at=now() WHERE id=$3', [stage, status, runId]);
}
async function finishRun(pg, runId, status) {
await pg.query('UPDATE runs SET status=$1, finished_at=now(), updated_at=now() WHERE id=$2', [status, runId]);
}
async function runPipeline(pg, run) {
const runId = run.id;
// Codex P1 #11 fix: track the actual current stage locally so the crash
// handler logs the right one (run.current_stage is frozen at runPipeline
// call time and won't reflect the stage we were in when we threw).
let currentStage = run.current_stage || 0;
let currentStageName = STAGES[currentStage - 1] || 'unknown';
const trackStage = (n, name) => { currentStage = n; currentStageName = name; };
try {
// 1: intake
trackStage(1, 'intake');
await setStage(pg, runId, 1, 'running');
await logEvent(pg, runId, 1, 'intake', 'info', 'starting', null);
const { spec, errors: ie } = await withTimeout('intake', signal => runIntake({ brief: run.brief, signal }));
if (ie.length) {
await logEvent(pg, runId, 1, 'intake', 'error', 'spec validation failed', { spec, errors: ie });
await finishRun(pg, runId, 'failed_intake');
return;
}
await pg.query('UPDATE runs SET visual_type=$1, artifact_name=$2, width=$3, height=$4, updated_at=now() WHERE id=$5',
[spec.visual_type, spec.artifact_name, spec.width, spec.height, runId]);
await logEvent(pg, runId, 1, 'intake', 'info', 'spec extracted', { spec });
// 2 (folded into intake — palette/typography/layout already in spec)
// 3: compose
trackStage(2, 'compose');
await setStage(pg, runId, 2, 'running');
await logEvent(pg, runId, 2, 'compose', 'info', 'starting', null);
let composed = await withTimeout('compose', signal => runCompose({ runId, spec, signal }));
if (composed.errors.length) {
await logEvent(pg, runId, 2, 'compose', 'warn', 'validation issues', { errors: composed.errors });
} else {
await logEvent(pg, runId, 2, 'compose', 'info', 'wrote html', { htmlPath: composed.htmlPath, bytes: composed.html.length });
}
// 4: render
trackStage(3, 'render');
await setStage(pg, runId, 3, 'running');
await logEvent(pg, runId, 3, 'render', 'info', 'starting (playwright headless)', null);
let rendered = await withTimeout('render', signal => runRender({ htmlPath: composed.htmlPath, width: spec.width, height: spec.height, signal }));
await logEvent(pg, runId, 3, 'render', 'info', 'wrote png', { pngPath: rendered.pngPath, bytes: rendered.bytes });
await pg.query(
'INSERT INTO artifacts (run_id, visual_type, artifact_name, html_path, png_path, width, height, status) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)',
[runId, spec.visual_type, spec.artifact_name, composed.htmlPath, rendered.pngPath, spec.width, spec.height, 'draft']
);
// 5: vision_check
trackStage(4, 'vision_check');
await setStage(pg, runId, 4, 'running');
await logEvent(pg, runId, 4, 'vision_check', 'info', 'starting (llava)', null);
let vision = await withTimeout('vision_check', signal => runVisionCheck({ pngPath: rendered.pngPath, signal }));
await logEvent(pg, runId, 4, 'vision_check', vision.issues.length ? 'warn' : 'info',
`quality=${vision.overall_quality} text_seen=${vision.text_content.length}`, vision);
// 6: critic
trackStage(5, 'critic');
await setStage(pg, runId, 5, 'running');
await logEvent(pg, runId, 5, 'critic', 'info', 'starting (claude CLI)', null);
let review;
try {
review = await withTimeout('critic', signal => runCritic({ brief: run.brief, spec, vision, signal }));
await logEvent(pg, runId, 5, 'critic', review.approved ? 'info' : 'warn',
`score=${review.score} approved=${review.approved}`, { review });
} catch (err) {
// Abort/timeout MUST propagate — the outer crash handler marks the run
// failed instead of cataloging it as `completed_unapproved` (which looks
// like a quality reject, not a cancelled stage).
if (err?.name === 'AbortError' || /aborted|timed out/i.test(err?.message || '')) {
await logEvent(pg, runId, 5, 'critic', 'error', 'critic stage aborted/timed out', { error: err.message });
throw err;
}
review = { approved: false, score: 0, issues: [`critic call failed: ${err.message}`], fix_instructions: [] };
await logEvent(pg, runId, 5, 'critic', 'error', 'critic call failed', { error: err.message });
}
// 7: iterate — loop up to MAX_ITERATIONS times, breaking on approval.
// Each round: apply fixes → re-render → re-vision → re-critic.
// Track iteration count so the UI can label v2/v3/etc and we can cap budget.
trackStage(6, 'iterate');
await setStage(pg, runId, 6, 'running');
let finalReview = review;
let iterationCount = 0;
const MAX_ITERATIONS = Number(process.env.MAX_ITERATIONS || 3);
while (
!finalReview.approved &&
Array.isArray(finalReview.fix_instructions) &&
finalReview.fix_instructions.length &&
iterationCount < MAX_ITERATIONS
) {
const iterNum = iterationCount + 1;
await logEvent(pg, runId, 6, 'iterate', 'info',
`iteration ${iterNum}/${MAX_ITERATIONS}: applying ${finalReview.fix_instructions.length} fixes (prev score=${finalReview.score})`,
{ iteration: iterNum, fixes: finalReview.fix_instructions });
try {
const iter = await withTimeout('iterate', signal => runIterate({ htmlPath: composed.htmlPath, html: composed.html, spec, review: finalReview, signal }));
composed.html = iter.html;
await logEvent(pg, runId, 6, 'iterate', 'info', `iter ${iterNum}: rewrote html`, { iteration: iterNum, bytes: iter.html.length });
rendered = await withTimeout('render', signal => runRender({ htmlPath: composed.htmlPath, width: spec.width, height: spec.height, signal }));
await logEvent(pg, runId, 6, 'iterate', 'info', `iter ${iterNum}: re-rendered png`, { iteration: iterNum, bytes: rendered.bytes });
vision = await withTimeout('vision_check', signal => runVisionCheck({ pngPath: rendered.pngPath, signal }));
await logEvent(pg, runId, 6, 'iterate', vision.issues.length ? 'warn' : 'info',
`iter ${iterNum}: re-vision quality=${vision.overall_quality}`, { iteration: iterNum, ...vision });
// Codex P1 #10 fix: if re-critic fails, the rewritten HTML/PNG is already
// on disk but we can't fairly score it. Mark the run failed_iteration
// and bail — better than cataloging a stale review against a brand-new
// artifact (the user would see a v2 with score from v1).
try {
const reviewN = await withTimeout('critic', signal => runCritic({ brief: run.brief, spec, vision, signal }));
await logEvent(pg, runId, 6, 'iterate', reviewN.approved ? 'info' : 'warn',
`iter ${iterNum}: re-critic score=${reviewN.score} approved=${reviewN.approved} (was ${finalReview.score})`,
{ iteration: iterNum, review: reviewN });
finalReview = reviewN;
} catch (err) {
await logEvent(pg, runId, 6, 'iterate', 'error', `iter ${iterNum}: re-critic failed — abandoning run`, { iteration: iterNum, error: err.message });
await pg.query("UPDATE artifacts SET status='failed_iteration' WHERE run_id=$1", [runId]);
await finishRun(pg, runId, 'failed_iteration');
return; // skip catalog/activate — the artifact never got a fair review
}
iterationCount = iterNum;
} catch (err) {
await logEvent(pg, runId, 6, 'iterate', 'error', `iter ${iterNum}: failed mid-loop`, { iteration: iterNum, error: err.message });
// Same reasoning: a partial iteration left the artifact in an unknown
// state vs the spec. Don't catalog it as awaiting_activation.
await pg.query("UPDATE artifacts SET status='failed_iteration' WHERE run_id=$1", [runId]);
await finishRun(pg, runId, 'failed_iteration');
return;
}
}
if (iterationCount === 0) {
await logEvent(pg, runId, 6, 'iterate', 'info',
review.approved ? 'critic approved — no iteration needed' : 'no fix_instructions to act on', null);
} else if (!finalReview.approved && iterationCount >= MAX_ITERATIONS) {
await logEvent(pg, runId, 6, 'iterate', 'warn',
`hit MAX_ITERATIONS=${MAX_ITERATIONS} without approval (final score=${finalReview.score})`, null);
}
// 8: catalog (update artifact row with final score + status)
trackStage(7, 'catalog');
await setStage(pg, runId, 7, 'running');
await pg.query("UPDATE artifacts SET score=$1, status=$2 WHERE run_id=$3",
[finalReview.score, finalReview.approved ? 'awaiting_activation' : 'unapproved', runId]);
await logEvent(pg, runId, 7, 'catalog', 'info', `final score=${finalReview.score}`, null);
// 9: activate (manual gate — see POST /runs/:id/activate)
trackStage(8, 'activate');
await setStage(pg, runId, 8, 'running');
await logEvent(pg, runId, 8, 'activate', 'info', 'awaiting manual activation (POST /runs/:id/activate)', null);
const finalStatus = finalReview.approved
? (iterationCount > 0 ? `awaiting_activation_v${iterationCount + 1}` : 'awaiting_activation')
: 'completed_unapproved';
await finishRun(pg, runId, finalStatus);
} catch (err) {
// Codex P1 #11 fix: log against the stage we were ACTUALLY in (tracked
// locally), not run.current_stage which is frozen at function-entry. And
// wrap the logEvent in its own try so a logging failure can't skip the
// finishRun — leaving the run stuck "running" is the worst outcome.
try {
await logEvent(pg, runId, currentStage, currentStageName, 'error', err.message, { stack: err.stack, stage: currentStage });
} catch (logErr) {
console.error('[visual-factory] crash handler logEvent failed:', logErr.message);
}
try {
await finishRun(pg, runId, 'failed');
} catch (finishErr) {
console.error('[visual-factory] crash handler finishRun failed:', finishErr.message);
}
}
}
module.exports = { STAGES, runPipeline };