From 9ca922933c904b61279ad98f3aad03f0b8709e6e Mon Sep 17 00:00:00 2001 From: MagMueller Date: Sat, 5 Sep 2026 15:45:26 -0700 Subject: [PATCH] fix: keep newer card revisions independent of old outcomes --- README.md | 8 ++ app/api/agent-jobs/route.ts | 22 +++- app/api/tasks/route.ts | 2 +- db/index.ts | 33 +++--- lib/job-lifecycle.ts | 17 +++ tests/job-reconciliation.test.mjs | 182 ++++++++++++++++++++++++++++++ 6 files changed, 239 insertions(+), 25 deletions(-) create mode 100644 tests/job-reconciliation.test.mjs diff --git a/README.md b/README.md index 8962694..8a182df 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,14 @@ only after that action and its validation succeed. Terminal updates cannot resurrect a job; a same-terminal-status retry does not rewrite the result. Results are limited to 20,000 characters. Check newer jobs before finalizing or acting on old approval. +A terminal outcome updates only the job's captured card version by default. If you published your own +replacement while doing that job, pass its exact returned `version` as `expectedIdeaVersion` when +finishing. A newer job, an intervening card revision or Skip prevents that outcome changing the card; +the completed job and its result remain recorded. `ideaStatus: null` means no card state was changed. +A competing terminal update returns 409 if its job-status comparison lost. Do not blindly retry it. +For legacy jobs without a captured version, automatic reconciliation requires an unrevised card or +a job created strictly after that card revision. Same-second ordering after a replacement is ambiguous. + If work is complete, record the result without inventing a new decision. If the premise changed or work is blocked, replace the misleading card with the current facts and any meaningful next option. Never add an Acknowledge/Keep blocked no-op. Keep actual in-flight checks Working until they finish. diff --git a/app/api/agent-jobs/route.ts b/app/api/agent-jobs/route.ts index ac1b63a..8e71f9d 100644 --- a/app/api/agent-jobs/route.ts +++ b/app/api/agent-jobs/route.ts @@ -1,6 +1,6 @@ import { env } from "cloudflare:workers"; import { ensureDatabase } from "../../../db"; -import { canUpdateJob, ideaStatusForOutcome, jobLeaseWindow, MAX_CONCURRENT_JOBS, resolveTicketOutcome, type StoredJobStatus, type TicketOutcome } from "../../../lib/job-lifecycle"; +import { canUpdateJob, ideaStatusForOutcome, JOB_MATCHES_IDEA_SQL, jobLeaseWindow, MAX_CONCURRENT_JOBS, resolveTicketOutcome, type StoredJobStatus, type TicketOutcome } from "../../../lib/job-lifecycle"; function canUseQueue(request: Request) { const origin = request.headers.get("origin"); @@ -63,11 +63,12 @@ export async function GET(request: Request) { export async function POST(request: Request) { if (!canUseQueue(request)) return Response.json({ error: "Missing agent key" }, { status: 401 }); - const payload = (await request.json()) as { id?: number; status?: "running" | "done" | "failed"; result?: string; ticketOutcome?: TicketOutcome }; + const payload = (await request.json()) as { id?: number; status?: "running" | "done" | "failed"; result?: string; ticketOutcome?: TicketOutcome; expectedIdeaVersion?: number }; if (!payload.id || !["running", "done", "failed"].includes(payload.status ?? "")) return Response.json({ error: "Invalid job update" }, { status: 400 }); if (payload.ticketOutcome && !["completed", "review", "blocked"].includes(payload.ticketOutcome)) return Response.json({ error: "Invalid ticket outcome" }, { status: 400 }); if (payload.status === "done" && payload.ticketOutcome === "blocked") return Response.json({ error: "A done job cannot be blocked" }, { status: 400 }); if (payload.status === "failed" && payload.ticketOutcome && payload.ticketOutcome !== "blocked") return Response.json({ error: "A failed job must be blocked" }, { status: 400 }); + if (payload.expectedIdeaVersion !== undefined && (!Number.isInteger(payload.expectedIdeaVersion) || payload.expectedIdeaVersion < 1)) return Response.json({ error: "Invalid expected idea version" }, { status: 400 }); const db = await ensureDatabase(); const job = await db.prepare("SELECT idea_id AS ideaId, action, status FROM agent_jobs WHERE id = ?").bind(payload.id).first<{ ideaId: number; action: string; status: StoredJobStatus }>(); if (!job) return Response.json({ error: "Job not found" }, { status: 404 }); @@ -84,15 +85,26 @@ export async function POST(request: Request) { ]; if ((payload.status === "done" || payload.status === "failed") && job.action !== "no") { const ideaStatus = ideaStatusForOutcome(ticketOutcome); + // In this atomic batch, changes() is the preceding job CAS, not another request. updates.push(db.prepare(` UPDATE ideas SET status = ? WHERE id = ? AND status IN ('new', 'working') + AND changes() = 1 AND NOT EXISTS ( SELECT 1 FROM agent_jobs newer WHERE newer.idea_id = ? AND newer.id > ? ) - `).bind(ideaStatus, job.ideaId, job.ideaId, payload.id)); + AND EXISTS ( + SELECT 1 FROM agent_jobs job + WHERE job.id = ? AND job.status = ? AND job.ticket_outcome = ? + AND CASE WHEN ? IS NOT NULL THEN ideas.version = ? + ELSE (${JOB_MATCHES_IDEA_SQL}) END + ) + `).bind(ideaStatus, job.ideaId, job.ideaId, payload.id, payload.id, payload.status, ticketOutcome, payload.expectedIdeaVersion ?? null, payload.expectedIdeaVersion ?? null)); + } + const [updatedJob, updatedIdea] = await db.batch(updates); + if (updatedJob.meta.changes !== 1) { + return Response.json({ error: "Job changed while completing it" }, { status: 409 }); } - await db.batch(updates); - return Response.json({ ok: true, ticketOutcome, ideaStatus: ideaStatusForOutcome(ticketOutcome) }); + return Response.json({ ok: true, ticketOutcome, ideaStatus: updatedIdea?.meta.changes ? ideaStatusForOutcome(ticketOutcome) : null }); } diff --git a/app/api/tasks/route.ts b/app/api/tasks/route.ts index b7a037c..4440070 100644 --- a/app/api/tasks/route.ts +++ b/app/api/tasks/route.ts @@ -53,7 +53,7 @@ export async function POST(request: Request) { source: "Agency New Task", boundary: "Complete safe private work first. Stop before an external, destructive, paid, merge, or deploy action unless the task explicitly and exactly approves it.", }); - const idea = await db.prepare("INSERT INTO ideas (project, category, headline, why_matters, impact, finished_work, primary_action, secondary_action, external_action, card_html, agent_context, score, rise_reach, rise_impact, rise_strategic_fit, rise_ease, source_label, source_url, agent_name, preview_kind, preview_title, preview_body, preview_asset, dedupe_key, status) VALUES ('Agency', 'Quick task', ?, '', '', '', '', '', '', ?, ?, 100, 25, 25, 25, 25, 'User-created task', '', 'Agency · Task Runner', 'html', '', '', '', ?, 'working') RETURNING id, project, category, headline, card_html AS cardHtml, agent_context AS agentContext, source_label AS sourceLabel, source_url AS sourceUrl, dedupe_key AS dedupeKey") + const idea = await db.prepare("INSERT INTO ideas (project, category, headline, why_matters, impact, finished_work, primary_action, secondary_action, external_action, card_html, agent_context, score, rise_reach, rise_impact, rise_strategic_fit, rise_ease, source_label, source_url, agent_name, preview_kind, preview_title, preview_body, preview_asset, dedupe_key, status) VALUES ('Agency', 'Quick task', ?, '', '', '', '', '', '', ?, ?, 100, 25, 25, 25, 25, 'User-created task', '', 'Agency · Task Runner', 'html', '', '', '', ?, 'working') RETURNING id, version, project, category, headline, card_html AS cardHtml, agent_context AS agentContext, source_label AS sourceLabel, source_url AS sourceUrl, dedupe_key AS dedupeKey") .bind(headline, taskCard(task), agentContext, taskKey).first(); if (!idea?.id) return Response.json({ error: "Task card could not be created." }, { status: 500 }); diff --git a/db/index.ts b/db/index.ts index 07b59f6..2bb95c1 100644 --- a/db/index.ts +++ b/db/index.ts @@ -1,4 +1,5 @@ import { env } from "cloudflare:workers"; +import { JOB_MATCHES_IDEA_SQL } from "../lib/job-lifecycle"; export function getD1() { if (!env.DB) throw new Error("The local Agency database is unavailable."); @@ -103,32 +104,26 @@ async function runMaintenance(db: ReturnType) { if (!jobNames.has("ticket_outcome")) { await db.prepare("ALTER TABLE agent_jobs ADD COLUMN ticket_outcome TEXT").run(); } - // A replacement card can set an idea back to New before the agent posts its - // terminal outcome. Reconcile from the latest explicit marker so completed, - // review, and blocked remain card states instead of agent-run states. Never - // revive a card the user already dismissed. + // Only the latest terminal job for this card version may reconcile its state. + // A fresh reply must not inherit completion from an older card version. await db.prepare(` UPDATE ideas - SET status = CASE ( - SELECT latest.ticket_outcome - FROM agent_jobs latest - WHERE latest.idea_id = ideas.id - ORDER BY latest.id DESC - LIMIT 1 - ) + SET status = CASE job.ticket_outcome WHEN 'completed' THEN 'done' WHEN 'review' THEN 'new' WHEN 'blocked' THEN 'new' - ELSE status + ELSE ideas.status END - WHERE status IN ('new', 'working', 'done') + FROM agent_jobs job + WHERE job.idea_id = ideas.id + AND job.id = (SELECT MAX(latest.id) FROM agent_jobs latest WHERE latest.idea_id = ideas.id) + AND ideas.status IN ('new', 'working', 'done') + AND job.action <> 'no' AND ( - SELECT latest.ticket_outcome - FROM agent_jobs latest - WHERE latest.idea_id = ideas.id - ORDER BY latest.id DESC - LIMIT 1 - ) IS NOT NULL + (job.status = 'done' AND job.ticket_outcome IN ('completed', 'review')) + OR (job.status = 'failed' AND job.ticket_outcome = 'blocked') + ) + AND (${JOB_MATCHES_IDEA_SQL}) `).run(); // Early Agency audits used the same decision fields as real user clicks. Mark // those known maintenance labels once so decision timing measures the user. diff --git a/lib/job-lifecycle.ts b/lib/job-lifecycle.ts index 8eae5e9..3bdcbe1 100644 --- a/lib/job-lifecycle.ts +++ b/lib/job-lifecycle.ts @@ -43,3 +43,20 @@ export function ideaStatusForOutcome(outcome: TicketOutcome | null) { if (outcome === "review" || outcome === "blocked") return "new"; return "working"; } + +const legacyJobMatchesIdea = "(ideas.version = 1 OR job.created_at > ideas.created_at)"; + +// A captured version wins over second-resolution timestamps. Legacy same-second +// ordering is ambiguous after a replacement, so only an unrevised card is safe. +export const JOB_MATCHES_IDEA_SQL = ` + CASE WHEN json_valid(job.card_context) THEN + CASE + WHEN json_type(job.card_context, '$.idea.version') = 'integer' + THEN json_extract(job.card_context, '$.idea.version') = ideas.version + WHEN json_type(job.card_context, '$.idea.version') IS NULL + THEN ${legacyJobMatchesIdea} + ELSE 0 + END + ELSE ${legacyJobMatchesIdea} + END +`; diff --git a/tests/job-reconciliation.test.mjs b/tests/job-reconciliation.test.mjs new file mode 100644 index 0000000..adefea0 --- /dev/null +++ b/tests/job-reconciliation.test.mjs @@ -0,0 +1,182 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { DatabaseSync } from "node:sqlite"; +import test from "node:test"; +import { pathToFileURL } from "node:url"; + +const source = (variable, relative) => process.env[variable] + ? pathToFileURL(process.env[variable]) : new URL(relative, import.meta.url); +const { JOB_MATCHES_IDEA_SQL } = await import(source("AGENCY_LIFECYCLE_SOURCE", "../lib/job-lifecycle.ts")); +const databaseSource = readFileSync(source("AGENCY_DB_SOURCE", "../db/index.ts"), "utf8"); +const jobsSource = readFileSync(source("AGENCY_JOBS_SOURCE", "../app/api/agent-jobs/route.ts"), "utf8"); +const tasksSource = readFileSync(source("AGENCY_TASKS_SOURCE", "../app/api/tasks/route.ts"), "utf8"); +const expand = (sql) => sql.replaceAll("${JOB_MATCHES_IDEA_SQL}", JOB_MATCHES_IDEA_SQL); +const reconcileSql = expand(databaseSource.match(/await db\.prepare\(`\s*(UPDATE ideas\n[\s\S]*?)`\)\.run\(\);/)[1]); +const finishJobSql = jobsSource.match(/db\.prepare\("(UPDATE agent_jobs SET status = [^"]+)"\)/)[1]; +const finishIdeaSql = expand(jobsSource.match(/updates\.push\(db\.prepare\(`([\s\S]*?)`\)/)[1]); +const T0 = "2026-09-05 10:00:00"; +const T1 = "2026-09-05 10:00:01"; +const T2 = "2026-09-05 10:00:02"; + +function fixture(idea = {}, jobs = [{}]) { + const db = new DatabaseSync(":memory:"); + db.function("current_timestamp", () => T2); + db.exec(` + CREATE TABLE ideas(id INTEGER PRIMARY KEY, version INTEGER, status TEXT, created_at TEXT, score INTEGER, rise_impact INTEGER); + CREATE TABLE agent_jobs(id INTEGER PRIMARY KEY, idea_id INTEGER, action TEXT, status TEXT, ticket_outcome TEXT, card_context TEXT, created_at TEXT, updated_at TEXT, result TEXT); + CREATE TABLE feedback(id INTEGER PRIMARY KEY, note TEXT); + CREATE TABLE card_attention(idea_id INTEGER, idea_version INTEGER, active_ms INTEGER); + CREATE TABLE card_interactions(id INTEGER PRIMARY KEY, action TEXT); + INSERT INTO feedback VALUES(1, 'Synthetic earlier feedback'); + INSERT INTO card_attention VALUES(1, 2, 12345); + INSERT INTO card_interactions VALUES(1, 'do'); + `); + const card = { id: 1, version: 2, status: "new", createdAt: T0, score: 72, riseImpact: 17, ...idea }; + db.prepare("INSERT INTO ideas VALUES (?, ?, ?, ?, ?, ?)").run(card.id, card.version, card.status, card.createdAt, card.score, card.riseImpact); + for (const override of jobs) { + const job = { id: 10, ideaId: card.id, action: "do", status: "done", outcome: "completed", context: JSON.stringify({ idea: { id: card.id, version: 2 } }), createdAt: T1, updatedAt: T2, result: "Earlier verified result", ...override }; + db.prepare("INSERT INTO agent_jobs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)").run(job.id, job.ideaId, job.action, job.status, job.outcome, job.context, job.createdAt, job.updatedAt, job.result); + } + return db; +} + +const rows = (db, table) => db.prepare(`SELECT * FROM ${table} ORDER BY 1`).all().map((row) => ({ ...row })); +const cardStatus = (db) => db.prepare("SELECT status FROM ideas").get().status; +const history = (db) => ["agent_jobs", "feedback", "card_attention", "card_interactions"].map((table) => rows(db, table)); +const cardData = (db) => rows(db, "ideas").map((row) => { + delete row.status; + return row; +}); + +function reconcile(db, expected) { + const beforeHistory = history(db); + const beforeCards = cardData(db); + db.exec(reconcileSql); + assert.equal(cardStatus(db), expected); + assert.deepEqual(history(db), beforeHistory); + assert.deepEqual(cardData(db), beforeCards); + db.exec(reconcileSql); + assert.equal(cardStatus(db), expected, "maintenance must be idempotent"); +} + +for (const [name, idea, jobs, expected] of [ + ["current completion", {}, [{}], "done"], + ["current working completion", { status: "working" }, [{}], "done"], + ["newer replacement after completion", { version: 3, createdAt: T2 }, [{}], "new"], + ["same-second newer replacement", { version: 3, createdAt: T1 }, [{ updatedAt: T1 }], "new"], + ["old job completes after replacement", { version: 3, createdAt: T1 }, [{ updatedAt: T2 }], "new"], + ["captured version is authoritative over clocks", { createdAt: T2 }, [{ createdAt: T0, updatedAt: T0 }], "done"], + ["current review", { status: "done" }, [{ outcome: "review" }], "new"], + ["current blocked", { status: "working" }, [{ status: "failed", outcome: "blocked" }], "new"], + ["stale review cannot reopen newer completed version", { version: 3, status: "done" }, [{ outcome: "review" }], "done"], + ["stale blocked cannot reopen newer completed version", { version: 3, status: "done" }, [{ status: "failed", outcome: "blocked" }], "done"], + ["Skip stays rejected", { status: "rejected" }, [{}], "rejected"], + ["old Skip job is not an execution outcome", {}, [{ action: "no" }], "new"], + ["newer queued job wins", { status: "working" }, [{}, { id: 11, status: "queued", outcome: null }], "working"], + ["newer running job wins", { status: "working" }, [{}, { id: 11, status: "running", outcome: null }], "working"], + ["in-flight marker cannot complete a card", { status: "working" }, [{ status: "running" }], "working"], + ["invalid terminal combination ignored", { status: "working" }, [{ status: "failed", outcome: "completed" }], "working"], + ["legacy first revision same-second completion", { version: 1 }, [{ context: "{}", createdAt: T0, updatedAt: T0 }], "done"], + ["legacy job unambiguously created for current revision", {}, [{ context: "{}", createdAt: T1 }], "done"], + ["legacy job predates replacement even if it finishes later", { version: 3, createdAt: T1 }, [{ context: "{}", createdAt: T0, updatedAt: T2 }], "new"], + ["legacy same-second replacement is ambiguous", { version: 3, createdAt: T1 }, [{ context: "{}", createdAt: T1, updatedAt: T2 }], "new"], + ["malformed historical context cannot abort maintenance", { version: 3, createdAt: T1 }, [{ context: "not JSON", createdAt: T0 }], "new"], + ["explicit string version is not a legacy match", {}, [{ context: '{"idea":{"version":"2"}}' }], "new"], + ["explicit null version is not a legacy match", {}, [{ context: '{"idea":{"version":null}}' }], "new"], + ["legacy completed state without outcome preserved", { status: "done" }, [{ outcome: null }], "done"], +]) { + test(name, () => { + const db = fixture(idea, jobs); + try { reconcile(db, expected); } finally { db.close(); } + }); +} + +// Reproduce both requests having read status=running before either batch commits. +function finish(db, { id = 10, status = "done", outcome = "completed", expectedIdeaVersion = null } = {}) { + db.exec("BEGIN"); + try { + const job = db.prepare(finishJobSql).run(status, "Synthetic result", outcome, id, "running"); + const args = [outcome === "completed" ? "done" : "new", 1, 1, id, id, status, outcome, expectedIdeaVersion, expectedIdeaVersion]; + const idea = db.prepare(finishIdeaSql).run(...args.slice(0, (finishIdeaSql.match(/\?/g) ?? []).length)); + db.exec("COMMIT"); + return { jobChanges: job.changes, ideaChanges: idea.changes }; + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } +} + +test("default completion never applies to a replacement, even in the same second", () => { + const db = fixture({ version: 3, status: "new", createdAt: T2 }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db), { jobChanges: 1, ideaChanges: 0 }); + assert.equal(rows(db, "agent_jobs")[0].ticket_outcome, "completed"); + reconcile(db, "new"); + } finally { db.close(); } +}); + +test("explicit current replacement version can complete and later replacements stay New", () => { + const db = fixture({ version: 3, createdAt: T2 }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db, { expectedIdeaVersion: 3 }), { jobChanges: 1, ideaChanges: 1 }); + reconcile(db, "done"); + db.prepare("UPDATE ideas SET version=4, status='new', created_at=?").run(T2); + reconcile(db, "new"); + } finally { db.close(); } +}); + +test("explicit version cannot complete an intervening newer replacement", () => { + const db = fixture({ version: 4, createdAt: T2 }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db, { expectedIdeaVersion: 3 }), { jobChanges: 1, ideaChanges: 0 }); + reconcile(db, "new"); + } finally { db.close(); } +}); + +for (const [first, second, expected] of [ + [{ status: "done", outcome: "completed" }, { status: "failed", outcome: "blocked" }, "done"], + [{ status: "failed", outcome: "blocked" }, { status: "done", outcome: "completed" }, "new"], + [{ status: "done", outcome: "review" }, { status: "done", outcome: "completed" }, "new"], +]) { + test(`competing ${first.outcome}/${second.outcome} completions preserve the CAS winner`, () => { + const db = fixture({ status: "working" }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db, first), { jobChanges: 1, ideaChanges: 1 }); + assert.deepEqual(finish(db, second), { jobChanges: 0, ideaChanges: 0 }); + assert.equal(rows(db, "agent_jobs")[0].ticket_outcome, first.outcome); + reconcile(db, expected); + } finally { db.close(); } + }); +} + +test("a newer queued job prevents old completion applying even with an explicit version", () => { + const db = fixture({ status: "working" }, [{ status: "running", outcome: null }, { id: 11, status: "queued", outcome: null }]); + try { + assert.deepEqual(finish(db, { expectedIdeaVersion: 2 }), { jobChanges: 1, ideaChanges: 0 }); + reconcile(db, "working"); + } finally { db.close(); } +}); + +test("same-outcome CAS loser cannot apply a different explicit card version", () => { + const db = fixture({ version: 3, createdAt: T2 }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db, { expectedIdeaVersion: 2 }), { jobChanges: 1, ideaChanges: 0 }); + assert.deepEqual(finish(db, { expectedIdeaVersion: 3 }), { jobChanges: 0, ideaChanges: 0 }); + reconcile(db, "new"); + } finally { db.close(); } +}); + +test("Skip during an in-flight job survives completion", () => { + const db = fixture({ status: "rejected" }, [{ status: "running", outcome: null }]); + try { + assert.deepEqual(finish(db), { jobChanges: 1, ideaChanges: 0 }); + reconcile(db, "rejected"); + } finally { db.close(); } +}); + +test("API validates explicit versions, reports lost CAS, and new task snapshots include version", () => { + assert.match(jobsSource, /Number\.isInteger\(payload\.expectedIdeaVersion\)/); + assert.match(jobsSource, /updatedJob\.meta\.changes !== 1[\s\S]*status: 409/); + assert.match(jobsSource, /ideaStatus: updatedIdea\?\.meta\.changes \? ideaStatusForOutcome\(ticketOutcome\) : null/); + assert.match(tasksSource, /RETURNING id, version, project/); +});