// Mazzy Command Center
// Copyright (c) 2025 Mazurov N.N. (https://github.com/mazurovn)
// PolyForm Noncommercial 1.0.0 — free for noncommercial use (personal, research,
// education). Commercial use requires a separate license. See LICENSE.

import { createHash, randomUUID } from "node:crypto";
import { mkdirSync } from "node:fs";
import { dirname } from "node:path";
import { DatabaseSync } from "node:sqlite";
import { ALLOWED_TRANSITIONS, CONTROL_COMMANDS, CONTROL_REQUEST_STATES, isAllowedTransition, MAX_CONTROL_INSTRUCTIONS_LENGTH, MAX_REPORT_FIELD_LENGTH, MAX_TASK_COMMENT_LENGTH, RUN_LIFECYCLES, TASK_STATES, TASK_TYPES, UI_TRANSITIONS, type AddTaskCommentInput, type AssignRunInput, type ClaimControlRequestInput, type CompleteControlRequestInput, type CompletionAttestation, type CompletionAttestationInput, type CreateControlRequestInput, type FailControlRequestInput, type MazzyControlRequest, type MazzyEvent, type MazzyEvidence, type MazzyReviewReport, type MazzyRunBinding, type ControlPlanePort, type MazzySnapshot, type MazzyTask, type MazzyTaskComment, type MazzyTaskDetail, type QualityGate, type QualityGateCategory, type QualityGateStatus, type QualityGateSummary, type RecordEvidenceInput, type ReviewReportInput, type ReviewerEvidenceInput, type RunBindingState, type RunLifecycle, type RunRole, type TaskState, type TransferRunInput, type UpdateRunActivityInput, type UpdateTaskInput, type CreateTaskInput, type CommentResponseInput, type TaskCommentRole } from "./types.ts";

type Row = Record<string, unknown>;
const now = () => new Date().toISOString();
const clean = (value: string, label: string) => { const result = value.trim(); if (!result) throw new Error(`${label} is required`); return result; };
/** Priority is a bounded integer. Reject NaN/Infinity/float so an unvalidated JSON body cannot poison ORDER BY priority. */
const clampPriority = (value: number | undefined): number => { const p = value ?? 0; if (!Number.isFinite(p)) throw new Error("priority must be a finite integer"); if (!Number.isInteger(p)) throw new Error("priority must be an integer"); return Math.max(-100, Math.min(100, p)); };
/** The store is the durable trust boundary: validate verdict here, not only at the HTTP layer, so any caller writing evidence is held to the enum. */
const EVIDENCE_VERDICTS = ["PASS", "FAIL", "UNCERTAIN"] as const;
const cleanVerdict = (value: unknown): string => { if (typeof value !== "string" || !(EVIDENCE_VERDICTS as readonly string[]).includes(value)) throw new Error("verdict must be one of PASS, FAIL, UNCERTAIN"); return value; };
/** Work-item type is a bounded enum; default "task" when unset. */
const cleanType = (value: string | undefined, fallback = "task"): string => { if (value === undefined) return fallback; if (!(TASK_TYPES as readonly string[]).includes(value)) throw new Error(`type must be one of ${TASK_TYPES.join(", ")}`); return value; };
const digest = (value: unknown) => createHash("sha256").update(JSON.stringify(value)).digest("hex");
const contentDigest = (title: string, description: string) => digest({ title, description });
function fingerprint(action: string, taskId: string, revision: number, role: string, runId: string | undefined, agent: string | undefined, payload: unknown): string { return digest({ action, taskId, revision, role, runId: runId ?? null, agent: agent ?? null, payload: payload ?? null }); }
function task(row: Row): MazzyTask { return { id: String(row.id), title: String(row.title), description: String(row.description), type: (TASK_TYPES as readonly string[]).includes(String(row.type)) ? row.type as MazzyTask["type"] : "task", state: row.state as TaskState, priority: Number(row.priority), risk: row.risk as MazzyTask["risk"], executorActor: row.executor_actor ? String(row.executor_actor) : undefined, revision: Number(row.revision), acceptanceRevision: Number(row.acceptance_revision), acceptanceDigest: String(row.acceptance_digest), createdAt: String(row.created_at), updatedAt: String(row.updated_at) }; }
function binding(row: Row): MazzyRunBinding { return { id: String(row.id), taskId: String(row.task_id), taskRevision: Number(row.task_revision), acceptanceRevision: Number(row.acceptance_revision), acceptanceDigest: row.acceptance_digest ? String(row.acceptance_digest) : undefined, runId: String(row.run_id), agent: String(row.agent), role: row.role as RunRole, state: row.state as RunBindingState, idempotencyKey: String(row.idempotency_key), operationFingerprint: String(row.operation_fingerprint), parentSessionId: row.parent_session_id ? String(row.parent_session_id) : undefined, childSessionId: row.child_session_id ? String(row.child_session_id) : undefined, lifecycle: row.lifecycle as RunLifecycle, model: row.model ? String(row.model) : undefined, cycle: row.cycle === null || row.cycle === undefined ? undefined : Number(row.cycle), lastActivityAt: row.last_activity_at ? String(row.last_activity_at) : undefined, currentActivity: row.current_activity ? String(row.current_activity) : undefined, currentTool: row.current_tool ? String(row.current_tool) : undefined, createdAt: String(row.created_at), updatedAt: String(row.updated_at) }; }
function request(row: Row): MazzyControlRequest { return { id: String(row.id), idempotencyKey: String(row.idempotency_key), operationFingerprint: String(row.operation_fingerprint), taskId: String(row.task_id), expectedTaskRevision: Number(row.expected_task_revision), command: row.command as MazzyControlRequest["command"], state: row.state as MazzyControlRequest["state"], approvedAgent: row.approved_agent ? String(row.approved_agent) : undefined, instructions: row.instructions ? String(row.instructions) : undefined, maxCycles: Number(row.max_cycles), targetRunId: row.target_run_id ? String(row.target_run_id) : undefined, parentSessionId: row.parent_session_id ? String(row.parent_session_id) : undefined, childSessionId: row.child_session_id ? String(row.child_session_id) : undefined, childRunId: row.child_run_id ? String(row.child_run_id) : undefined, requestedAt: String(row.requested_at), deliveredAt: row.delivered_at ? String(row.delivered_at) : undefined, claimedAt: row.claimed_at ? String(row.claimed_at) : undefined, completedAt: row.completed_at ? String(row.completed_at) : undefined, failedAt: row.failed_at ? String(row.failed_at) : undefined, cancelledAt: row.cancelled_at ? String(row.cancelled_at) : undefined, recoveredAt: row.recovered_at ? String(row.recovered_at) : undefined, recoveryReason: row.recovery_reason ? String(row.recovery_reason) : undefined, error: row.error ? String(row.error) : undefined }; }

export class MazzyStore implements ControlPlanePort {
  readonly db: DatabaseSync; readonly path: string;
  private readonly eventListeners = new Set<(event: MazzyEvent) => void>();
  private committedEvents: MazzyEvent[] | undefined;
  constructor(path: string) { this.path = path; mkdirSync(dirname(path), { recursive: true }); this.db = new DatabaseSync(path); this.db.exec("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000;"); this.migrate(); }
  close(): void { this.db.close(); }

  private migrate(): void {
    this.db.exec(`CREATE TABLE IF NOT EXISTS schema_migrations(version INTEGER PRIMARY KEY,applied_at TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY,title TEXT NOT NULL,description TEXT NOT NULL DEFAULT '',state TEXT NOT NULL CHECK(state IN (${TASK_STATES.map((s) => `'${s}'`).join(",")})),priority INTEGER NOT NULL DEFAULT 0,risk TEXT NOT NULL DEFAULT 'medium',revision INTEGER NOT NULL DEFAULT 1,created_at TEXT NOT NULL,updated_at TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY AUTOINCREMENT,task_id TEXT NOT NULL REFERENCES tasks(id),type TEXT NOT NULL,payload_json TEXT NOT NULL,actor TEXT NOT NULL,created_at TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS evidence(id TEXT PRIMARY KEY,task_id TEXT NOT NULL REFERENCES tasks(id),task_revision INTEGER NOT NULL,kind TEXT NOT NULL,verdict TEXT NOT NULL,actor TEXT NOT NULL,payload_json TEXT NOT NULL,created_at TEXT NOT NULL);
      INSERT OR IGNORE INTO schema_migrations VALUES(1,datetime('now'));`);
    const cols = (name: string) => (this.db.prepare(`PRAGMA table_info(${name})`).all() as Array<{ name: string }>).map((x) => x.name);
    const add = (table: string, name: string, definition: string) => { if (!cols(table).includes(name)) this.db.exec(`ALTER TABLE ${table} ADD COLUMN ${definition}`); };
    add("tasks", "executor_actor", "executor_actor TEXT");
    add("tasks", "type", "type TEXT"); // Work-item classification (epic/feature/task/bug); legacy rows backfill to 'task' below.
    // Creation keys are a distinct durable operation namespace from run/control keys.
    add("tasks", "create_idempotency_key", "create_idempotency_key TEXT"); add("tasks", "create_operation_fingerprint", "create_operation_fingerprint TEXT");
    add("tasks", "acceptance_revision", "acceptance_revision INTEGER"); add("tasks", "acceptance_digest", "acceptance_digest TEXT");
    add("evidence", "run_id", "run_id TEXT"); add("evidence", "binding_id", "binding_id TEXT"); add("evidence", "acceptance_revision", "acceptance_revision INTEGER");
    this.db.exec(`CREATE TABLE IF NOT EXISTS run_bindings(id TEXT PRIMARY KEY,task_id TEXT NOT NULL REFERENCES tasks(id),task_revision INTEGER NOT NULL,run_id TEXT NOT NULL UNIQUE,agent TEXT NOT NULL,role TEXT NOT NULL CHECK(role IN('worker','reviewer')),state TEXT NOT NULL CHECK(state IN('active','superseded','completed','failed')),idempotency_key TEXT NOT NULL UNIQUE,created_at TEXT NOT NULL,updated_at TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS task_comments(id TEXT PRIMARY KEY,task_id TEXT NOT NULL REFERENCES tasks(id),body TEXT NOT NULL,actor TEXT NOT NULL,created_at TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS orchestration_requests(id TEXT PRIMARY KEY,idempotency_key TEXT NOT NULL UNIQUE,task_id TEXT NOT NULL REFERENCES tasks(id),expected_task_revision INTEGER NOT NULL,command TEXT NOT NULL CHECK(command IN('GO','PAUSE','STOP')),state TEXT NOT NULL CHECK(state IN('REQUESTED','DELIVERED','CLAIMED','COMPLETED','FAILED','CANCELLED')),approved_agent TEXT,instructions TEXT,max_cycles INTEGER NOT NULL CHECK(max_cycles BETWEEN 1 AND 10),target_run_id TEXT,parent_session_id TEXT,child_session_id TEXT,child_run_id TEXT,requested_at TEXT NOT NULL,delivered_at TEXT,claimed_at TEXT,completed_at TEXT,failed_at TEXT,cancelled_at TEXT,error TEXT);`);
    add("run_bindings", "parent_session_id", "parent_session_id TEXT"); add("run_bindings", "child_session_id", "child_session_id TEXT"); add("run_bindings", "acceptance_revision", "acceptance_revision INTEGER"); add("run_bindings", "acceptance_digest", "acceptance_digest TEXT"); add("run_bindings", "operation_fingerprint", "operation_fingerprint TEXT"); add("run_bindings", "lifecycle", "lifecycle TEXT"); add("run_bindings", "model", "model TEXT"); add("run_bindings", "cycle", "cycle INTEGER"); add("run_bindings", "last_activity_at", "last_activity_at TEXT"); add("run_bindings", "current_activity", "current_activity TEXT"); add("run_bindings", "current_tool", "current_tool TEXT");
    add("task_comments", "role", "role TEXT"); add("task_comments", "reply_to", "reply_to TEXT"); add("task_comments", "client_message_id", "client_message_id TEXT"); add("task_comments", "run_id", "run_id TEXT"); add("task_comments", "session_id", "session_id TEXT"); add("task_comments", "delivery_state", "delivery_state TEXT"); add("task_comments", "acknowledged_at", "acknowledged_at TEXT"); add("task_comments", "error", "error TEXT"); add("task_comments", "operation_fingerprint", "operation_fingerprint TEXT");
    add("orchestration_requests", "operation_fingerprint", "operation_fingerprint TEXT"); add("orchestration_requests", "recovered_at", "recovered_at TEXT"); add("orchestration_requests", "recovery_reason", "recovery_reason TEXT");
    // Legacy schemas did not consistently enforce one active binding per role. Keep
    // the newest attestation before creating partial unique indexes.
    this.db.exec(`UPDATE run_bindings AS old SET state='superseded',updated_at=datetime('now') WHERE old.state='active' AND EXISTS (SELECT 1 FROM run_bindings AS newer WHERE newer.task_id=old.task_id AND newer.role=old.role AND newer.state='active' AND (newer.updated_at>old.updated_at OR (newer.updated_at=old.updated_at AND newer.id>old.id)));
      CREATE TABLE IF NOT EXISTS review_reports(id TEXT PRIMARY KEY,task_id TEXT NOT NULL REFERENCES tasks(id),acceptance_revision INTEGER NOT NULL,worker_run_id TEXT NOT NULL,agent TEXT NOT NULL,parent_session_id TEXT,child_session_id TEXT,summary TEXT NOT NULL,what_changed TEXT NOT NULL,checks TEXT NOT NULL,how_to_use TEXT NOT NULL,acceptance_criteria_json TEXT NOT NULL,results_json TEXT NOT NULL,limitations TEXT NOT NULL,model TEXT,session_id TEXT,run_id TEXT,cycle INTEGER,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,UNIQUE(task_id,acceptance_revision,worker_run_id));
      CREATE TABLE IF NOT EXISTS comment_notifications(comment_id TEXT NOT NULL REFERENCES task_comments(id),session_id TEXT NOT NULL,notified_at TEXT NOT NULL,PRIMARY KEY(comment_id,session_id));
      CREATE INDEX IF NOT EXISTS idx_evidence_acceptance ON evidence(task_id,acceptance_revision,created_at); CREATE INDEX IF NOT EXISTS idx_bindings_task ON run_bindings(task_id,task_revision,role,state); CREATE INDEX IF NOT EXISTS idx_requests_pending ON orchestration_requests(state,requested_at); CREATE INDEX IF NOT EXISTS idx_comments_task ON task_comments(task_id,created_at,id);
      CREATE UNIQUE INDEX IF NOT EXISTS idx_task_create_idempotency ON tasks(create_idempotency_key) WHERE create_idempotency_key IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS idx_one_active_worker ON run_bindings(task_id) WHERE role='worker' AND state='active'; CREATE UNIQUE INDEX IF NOT EXISTS idx_one_active_reviewer ON run_bindings(task_id) WHERE role='reviewer' AND state='active'; CREATE UNIQUE INDEX IF NOT EXISTS idx_comment_client_message ON task_comments(task_id,client_message_id) WHERE client_message_id IS NOT NULL; DROP INDEX IF EXISTS idx_go_outstanding; CREATE INDEX IF NOT EXISTS idx_go_outstanding ON orchestration_requests(task_id,expected_task_revision) WHERE command='GO' AND state IN ('REQUESTED','DELIVERED','CLAIMED');`);
    const oldTasks = this.db.prepare("SELECT id,title,description,revision,acceptance_revision,acceptance_digest FROM tasks").all() as Row[];
    for (const row of oldTasks) { const ar = Number(row.acceptance_revision) || Number(row.revision); const ad = row.acceptance_digest ? String(row.acceptance_digest) : contentDigest(String(row.title), String(row.description)); this.db.prepare("UPDATE tasks SET acceptance_revision=?,acceptance_digest=? WHERE id=?").run(ar, ad, String(row.id)); }
    this.db.exec("UPDATE tasks SET type='task' WHERE type IS NULL OR type='' OR type NOT IN ('epic','feature','task','bug');");
    this.db.exec("UPDATE evidence SET acceptance_revision=task_revision WHERE acceptance_revision IS NULL; UPDATE run_bindings SET acceptance_revision=task_revision WHERE acceptance_revision IS NULL; UPDATE run_bindings SET acceptance_digest=(SELECT acceptance_digest FROM tasks WHERE tasks.id=run_bindings.task_id) WHERE acceptance_digest IS NULL; UPDATE run_bindings SET lifecycle=CASE WHEN state='completed' THEN 'completed' WHEN state='failed' THEN 'failed' ELSE 'queued' END WHERE lifecycle IS NULL; UPDATE task_comments SET role='user' WHERE role IS NULL OR role='' OR role='web'; UPDATE task_comments SET delivery_state='sent' WHERE delivery_state IS NULL OR delivery_state='' OR delivery_state='sending';");
    this.db.exec("UPDATE task_comments SET role=COALESCE((SELECT b.role FROM run_bindings b WHERE b.task_id=task_comments.task_id AND b.agent=task_comments.actor ORDER BY b.updated_at DESC,b.id DESC LIMIT 1),'orchestrator') WHERE role='agent';");
    for (const row of this.db.prepare("SELECT id,task_id,body,actor,role,reply_to,client_message_id,run_id,session_id FROM task_comments WHERE operation_fingerprint IS NULL OR operation_fingerprint='' ").all() as Row[]) this.db.prepare("UPDATE task_comments SET operation_fingerprint=? WHERE id=?").run(`legacy:${digest(row)}`, String(row.id));
    for (const row of this.db.prepare("SELECT id,task_id,task_revision,role,run_id,agent,idempotency_key,operation_fingerprint FROM run_bindings WHERE operation_fingerprint IS NULL OR operation_fingerprint='' ").all() as Row[]) this.db.prepare("UPDATE run_bindings SET operation_fingerprint=? WHERE id=?").run(`legacy:${digest({ key: row.idempotency_key, id: row.id })}`, String(row.id));
    for (const row of this.db.prepare("SELECT id,idempotency_key,operation_fingerprint FROM orchestration_requests WHERE operation_fingerprint IS NULL OR operation_fingerprint='' ").all() as Row[]) this.db.prepare("UPDATE orchestration_requests SET operation_fingerprint=? WHERE id=?").run(`legacy:${digest({ key: row.idempotency_key, id: row.id })}`, String(row.id));
    this.db.prepare("INSERT OR IGNORE INTO schema_migrations(version,applied_at) VALUES(8,datetime('now'))").run();
  }

  createTask(input: CreateTaskInput): MazzyTask { const title = clean(input.title, "Task title"), description = input.description?.trim() ?? "", id = randomUUID(), at = now(), type = cleanType(input.type), state = input.state ?? "BACKLOG", priority = clampPriority(input.priority), risk = input.risk ?? "medium", key = input.idempotencyKey?.trim(); if (state !== "DRAFT" && state !== "BACKLOG") throw new Error("Tasks must be created in DRAFT or BACKLOG"); if (key && key.length > 200) throw new Error("Idempotency-Key is too long"); const fp = digest({ title, description, type, state, priority, risk }); return this.tx(() => { if (key) { const prior = this.db.prepare("SELECT * FROM tasks WHERE create_idempotency_key=?").get(key) as Row | undefined; if (prior) { if (String(prior.create_operation_fingerprint) !== fp) throw new Error("Idempotency key conflict: task payload differs"); return task(prior); } } this.db.prepare("INSERT INTO tasks(id,title,description,type,state,priority,risk,revision,acceptance_revision,acceptance_digest,create_idempotency_key,create_operation_fingerprint,created_at,updated_at) VALUES(?,?,?,?,?,?,?,1,1,?,?,?,?,?)").run(id, title, description, type, state, priority, risk, contentDigest(title, description), key ?? null, key ? fp : null, at, at); this.event(id, "task.created", { title, state, type }, input.actor ?? "user"); return this.getTask(id)!; }); }
  getTask(id: string): MazzyTask | undefined { const row = this.db.prepare("SELECT * FROM tasks WHERE id=?").get(id) as Row | undefined; return row ? task(row) : undefined; }
  listTasks(state?: TaskState): MazzyTask[] { return ((state ? this.db.prepare("SELECT * FROM tasks WHERE state=? ORDER BY priority DESC,created_at").all(state) : this.db.prepare("SELECT * FROM tasks ORDER BY priority DESC,created_at").all()) as Row[]).map(task); }

  updateTask(id: string, input: UpdateTaskInput): MazzyTask { return this.tx(() => { const current = this.requireRevision(id, input.expectedRevision); const title = input.title === undefined ? current.title : clean(input.title, "Task title"), description = input.description === undefined ? current.description : input.description.trim(), contentChanged = title !== current.title || description !== current.description, acceptanceRestart = contentChanged && ["RUNNING", "REVIEW"].includes(current.state); if (contentChanged && current.state === "DONE") throw new Error("Cannot edit content of a DONE task: reopen it before changing acceptance, otherwise it would remain DONE under an unreviewed acceptance"); let state = input.state ?? current.state; if (acceptanceRestart) state = "READY"; if (state !== current.state && !(acceptanceRestart && state === "READY") && !isAllowedTransition(current.state, state)) throw new Error(`Invalid transition: ${current.state} -> ${state}`); if (state === "RUNNING" && current.state !== "RUNNING") throw new Error("RUNNING requires a current active worker binding; use parent assignment"); if (state === "REVIEW" && current.state !== "REVIEW") throw new Error("REVIEW requires a completion attestation with a structured report"); const at = now();
      // BEGIN IMMEDIATE makes this read authoritative. The DONE gate is deliberately
      // the final read before the CAS, so a reviewer verdict cannot slip in between.
      if (state === "DONE") this.requireCurrentReviewerPass(current);
      const changed = this.db.prepare("UPDATE tasks SET title=?,description=?,type=?,state=?,priority=?,risk=?,executor_actor=?,revision=revision+1,acceptance_revision=acceptance_revision+?,acceptance_digest=?,updated_at=? WHERE id=? AND revision=? AND state=?").run(title, description, cleanType(input.type, current.type), state, clampPriority(input.priority ?? current.priority), input.risk ?? current.risk, acceptanceRestart ? null : current.executorActor ?? null, contentChanged ? 1 : 0, contentChanged ? contentDigest(title, description) : current.acceptanceDigest, at, id, current.revision, current.state);
      if (Number(changed.changes) !== 1) throw new Error("Concurrent task update conflict");
      const released = acceptanceRestart ? this.releaseActiveBindings(id, "superseded", at) : 0;
      const terminalReleased = ["DONE", "FAILED", "CANCELLED"].includes(state) ? this.releaseActiveBindings(id, state === "FAILED" ? "failed" : "superseded", at) : 0;
      if (released) this.event(id, "run.superseded-for-acceptance-edit", { count: released, toState: "READY" }, input.actor ?? "user"); if (terminalReleased) this.event(id, "run.released", { count: terminalReleased }, input.actor ?? "user"); this.event(id, "task.updated", { fromState: current.state, toState: state, previousRevision: current.revision, acceptanceChanged: contentChanged }, input.actor ?? "user");
      return task(this.db.prepare("SELECT * FROM tasks WHERE id=? AND revision=?").get(id, current.revision + 1) as Row);
    }); }

  assignRun(input: AssignRunInput): MazzyRunBinding { const role = input.role, key = clean(input.idempotencyKey, "Idempotency key"), runId = clean(input.runId, "Run id"), agent = clean(input.agent, "Agent"), fp = fingerprint("assign", input.taskId, input.expectedTaskRevision, role, runId, agent, { payload: input.payload, parentSessionId: input.parentSessionId, childSessionId: input.childSessionId, model: input.model, cycle: input.cycle }); return this.tx(() => { const old = this.byBindingKey(key); if (old) return this.replayBinding(old, fp); const current = this.requireRevision(input.taskId, input.expectedTaskRevision); if (this.byRun(runId)) throw new Error("Run id is already bound"); let revision = current.revision; if (role === "worker") { if (!isAllowedTransition(current.state, "RUNNING") || this.activeWorker(current.id)) throw new Error(`Cannot assign worker from ${current.state}`); this.releaseActiveReviewers(current.id, now()); revision++; this.assignmentTask(current, "RUNNING", agent, revision); } else { if (current.state !== "REVIEW" || !this.latestCompletedWorker(current)) throw new Error("Reviewer assignment requires REVIEW and a completed worker for the current acceptance"); this.releaseActiveReviewers(current.id, now()); } const result = this.insertBinding({ task: current, revision, runId, agent, role, key, fp, parentSessionId: input.parentSessionId, childSessionId: input.childSessionId, model: input.model, cycle: input.cycle }); this.event(current.id, "run.assigned", { runId, role, taskRevision: revision, acceptanceRevision: current.acceptanceRevision }, input.actor); return result; }); }

  transferRun(input: TransferRunInput): MazzyRunBinding { const key = clean(input.idempotencyKey, "Idempotency key"), runId = clean(input.runId, "Run id"), agent = clean(input.agent, "Agent"), fp = fingerprint("transfer", input.taskId, input.expectedTaskRevision, "worker", runId, agent, { payload: input.payload, parentSessionId: input.parentSessionId, childSessionId: input.childSessionId, model: input.model, cycle: input.cycle }); return this.tx(() => { const old = this.byBindingKey(key); if (old) return this.replayBinding(old, fp); const current = this.requireRevision(input.taskId, input.expectedTaskRevision); if (current.state !== "RUNNING") throw new Error(`Cannot transfer worker from ${current.state}`); if (this.byRun(runId)) throw new Error("Run id is already bound"); const active = this.activeWorker(current.id); if (active) this.release(active.id, "superseded", now()); this.releaseActiveReviewers(current.id, now()); const revision = current.revision + 1; this.assignmentTask(current, "RUNNING", agent, revision); const result = this.insertBinding({ task: current, revision, runId, agent, role: "worker", key, fp, parentSessionId: input.parentSessionId, childSessionId: input.childSessionId, model: input.model, cycle: input.cycle }); this.event(current.id, active ? "run.transferred" : "run.recovered", { fromRunId: active?.runId, toRunId: runId, taskRevision: revision }, input.actor); return result; }); }

  /* All authority reads, the idempotent-replay decision, and the writes happen inside one BEGIN IMMEDIATE so a concurrent completer cannot make this path upsert a report over a binding it did not actually complete. The active-binding UPDATE is a CAS (changes===1); a lost race re-reads terminal state and returns the idempotent replay/conflict instead of silently overwriting. */
  attestCompletion(input: CompletionAttestationInput): CompletionAttestation { const report = input.report; return this.tx(() => { const b = this.byRun(input.runId); if (!b || b.taskId !== input.taskId || b.role !== "worker") return { accepted: false, reason: "unknown-run" }; const current = this.getTask(input.taskId); if (!current || b.acceptanceRevision !== current.acceptanceRevision || b.acceptanceDigest !== current.acceptanceDigest) return { accepted: false, reason: "revision-conflict", binding: b }; if (!report) return { accepted: false, reason: "report-required", binding: b }; const existingReport = this.getReviewReport(input.taskId, current.acceptanceRevision), completed = this.latestCompletedWorker(current); if (b.state === "completed" && current.state === "REVIEW" && completed?.runId === b.runId && existingReport?.workerRunId === b.runId) return this.sameReport(existingReport, b, report) ? { accepted: true, task: current, binding: b } : { accepted: false, reason: "report-conflict", binding: b }; if (b.state !== "active") return { accepted: false, reason: "stale-or-superseded", binding: b }; if (!["RUNNING", "REVIEW"].includes(current.state) || (current.state === "RUNNING" && current.executorActor !== b.agent)) return { accepted: false, reason: "invalid-task-state", binding: b }; const at = now(); if (current.state === "RUNNING") { const changed = this.db.prepare("UPDATE tasks SET state='REVIEW',revision=revision+1,updated_at=? WHERE id=? AND revision=? AND state='RUNNING'").run(at, current.id, current.revision); if (Number(changed.changes) !== 1) throw new Error("Concurrent task update conflict"); } const completedRow = this.db.prepare("UPDATE run_bindings SET state='completed',lifecycle='completed',last_activity_at=?,updated_at=? WHERE id=? AND state='active'").run(at, at, b.id); if (Number(completedRow.changes) !== 1) return { accepted: false, reason: "stale-or-superseded", binding: b }; this.upsertReport(current, b, report, at); this.event(current.id, "run.completion-attested", { runId: b.runId, acceptanceRevision: current.acceptanceRevision, lifecycleRevisionDrift: current.revision !== input.expectedTaskRevision || b.taskRevision !== current.revision, payload: input.payload ?? {} }, input.actor); return { accepted: true, task: this.getTask(input.taskId), binding: this.byRun(input.runId) }; }); }

  importReviewReport(taskId: string, runId: string, expectedRevision: number, report: ReviewReportInput, actor: string): MazzyReviewReport { return this.tx(() => { const current = this.requireRevision(taskId, expectedRevision), b = this.latestCompletedWorker(current); if (!b || b.runId !== runId) throw new Error("Completed matching worker binding is required for the current acceptance"); const result = this.upsertReport(current, b, report, now()); this.event(taskId, "report.imported", { runId, acceptanceRevision: current.acceptanceRevision }, actor); return result; }); }

  recordReviewerEvidence(taskId: string, input: ReviewerEvidenceInput): MazzyEvidence { const id = randomUUID(), at = now(), kind = clean(input.kind, "Evidence kind"); return this.tx(() => { const current = this.requireRevision(taskId, input.expectedTaskRevision); const reviewer = this.byRun(input.runId); if (current.state !== "REVIEW" || !reviewer || reviewer.taskId !== taskId || reviewer.role !== "reviewer" || reviewer.state !== "active" || reviewer.acceptanceRevision !== current.acceptanceRevision || reviewer.acceptanceDigest !== current.acceptanceDigest) throw new Error("Active reviewer binding is required for current acceptance"); const verdict = cleanVerdict(input.verdict); const worker = this.latestCompletedWorker(current); if (!worker || worker.runId === reviewer.runId || worker.agent === reviewer.agent) throw new Error("Reviewer must use a different run and agent than worker");
      // The guarded completion makes this reviewer binding a one-shot attestation.
      // A racing retry re-reads the completed binding after BEGIN IMMEDIATE and emits nothing.
      const completed = this.db.prepare("UPDATE run_bindings SET state='completed',lifecycle='completed',last_activity_at=?,updated_at=? WHERE id=? AND task_id=? AND role='reviewer' AND state='active' AND acceptance_revision=? AND acceptance_digest=?").run(at, at, reviewer.id, taskId, current.acceptanceRevision, current.acceptanceDigest);
      if (Number(completed.changes) !== 1) throw new Error("Concurrent reviewer evidence conflict");
      this.db.prepare("INSERT INTO evidence(id,task_id,task_revision,acceptance_revision,kind,verdict,actor,payload_json,created_at,run_id,binding_id) VALUES(?,?,?,?,?,?,?,?,?,?,?)").run(id, taskId, current.revision, current.acceptanceRevision, kind, verdict, reviewer.agent, JSON.stringify(input.payload ?? {}), at, reviewer.runId, reviewer.id); this.event(taskId, "evidence.attested", { evidenceId: id, verdict, acceptanceRevision: current.acceptanceRevision }, input.actor);
      return this.evidence(this.db.prepare("SELECT * FROM evidence WHERE id=?").get(id) as Row);
    }); }
  /** Verifier evidence has no caller idempotency key: each accepted invocation is one durable attestation, atomically bound to the revision and acceptance current at BEGIN IMMEDIATE. */
  recordEvidence(taskId: string, input: RecordEvidenceInput): MazzyEvidence { const id = randomUUID(), at = now(), actor = clean(input.actor, "Evidence actor"), kind = clean(input.kind, "Evidence kind"), verdict = cleanVerdict(input.verdict); return this.tx(() => { const current = this.requireRevision(taskId, input.expectedTaskRevision); if (current.state !== "REVIEW") throw new Error("Verifier evidence can only be recorded in REVIEW"); this.db.prepare("INSERT INTO evidence(id,task_id,task_revision,acceptance_revision,kind,verdict,actor,payload_json,created_at) VALUES(?,?,?,?,?,?,?,?,?)").run(id, taskId, current.revision, current.acceptanceRevision, kind, verdict, actor, JSON.stringify(input.payload ?? {}), at); this.event(taskId, "evidence.recorded", { evidenceId: id }, actor); return this.evidence(this.db.prepare("SELECT * FROM evidence WHERE id=?").get(id) as Row); }); }

  addComment(taskId: string, input: AddTaskCommentInput): MazzyTaskComment { const body = clean(input.body, "Comment body"); if (body.length > MAX_TASK_COMMENT_LENGTH) throw new Error(`Comment body must be at most ${MAX_TASK_COMMENT_LENGTH} characters`); const actor = clean(input.actor, "Comment actor"), role = input.role ?? "user", clientMessageId = input.clientMessageId?.trim() || undefined; if (!['user','orchestrator','worker','reviewer','system'].includes(role)) throw new Error("Unknown comment role"); if (clientMessageId && clientMessageId.length > 200) throw new Error("clientMessageId is too long"); const fp = digest({ body, replyTo: input.replyTo ?? null, actor, role, runId: input.runId ?? null, sessionId: input.sessionId ?? null }); return this.tx(() => { if (!this.getTask(taskId)) throw new Error(`Task not found: ${taskId}`); if (clientMessageId) { const prior = this.db.prepare("SELECT * FROM task_comments WHERE task_id=? AND client_message_id=?").get(taskId, clientMessageId) as Row | undefined; if (prior) { if (String(prior.operation_fingerprint) !== fp) throw new Error("Idempotency key conflict: comment content differs"); return this.comment(prior); } } if (input.replyTo && !this.db.prepare("SELECT 1 FROM task_comments WHERE id=? AND task_id=?").get(input.replyTo, taskId)) throw new Error("replyTo must reference a comment on this task"); const out: MazzyTaskComment = { id: randomUUID(), taskId, body, actor, role, deliveryState: "sent", clientMessageId, replyTo: input.replyTo, runId: input.runId, sessionId: input.sessionId, createdAt: now() }; this.db.prepare("INSERT INTO task_comments(id,task_id,body,actor,role,client_message_id,reply_to,run_id,session_id,delivery_state,operation_fingerprint,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)").run(out.id, taskId, out.body, out.actor, out.role, out.clientMessageId ?? null, out.replyTo ?? null, out.runId ?? null, out.sessionId ?? null, out.deliveryState, fp, out.createdAt); this.event(taskId, "comment.added", { commentId: out.id, replyTo: out.replyTo, role: out.role }, out.actor); return out; }); }
  respondToDiscussion(taskId: string, input: CommentResponseInput, parentActor: string, parentSessionId?: string): MazzyTaskComment { const current = this.getTask(taskId); if (!current) throw new Error(`Task not found: ${taskId}`); let actor = clean(parentActor, "Parent actor"), role: TaskCommentRole = "orchestrator", runId: string | undefined; if (input.runId) { const binding = this.byRun(input.runId); if (!binding || binding.taskId !== taskId || !["worker", "reviewer"].includes(binding.role) || !["active", "completed"].includes(binding.state) || binding.acceptanceRevision !== current.acceptanceRevision || binding.acceptanceDigest !== current.acceptanceDigest) throw new Error("Current matching run binding is required for agent discussion response"); actor = binding.agent; role = binding.role; runId = binding.runId; } const comment = this.addComment(taskId, { body: input.body, replyTo: input.replyTo, actor, role, runId, sessionId: parentSessionId }); if (input.replyTo) this.acknowledgeUserComment(taskId, input.replyTo); return comment; }
  importAgentComment(taskId: string, runId: string, body: string, replyTo: string | undefined, actor: string): MazzyTaskComment { return this.respondToDiscussion(taskId, { body, replyTo, runId }, actor); }
  acknowledgeUserComment(taskId: string, commentId: string): MazzyTaskComment | undefined { const existing = this.db.prepare("SELECT * FROM task_comments WHERE id=? AND task_id=?").get(commentId, taskId) as Row | undefined; if (!existing || existing.role !== "user") return existing ? this.comment(existing) : undefined; this.tx(() => { const changed = this.db.prepare("UPDATE task_comments SET delivery_state='acknowledged',acknowledged_at=? WHERE id=? AND task_id=? AND role='user' AND delivery_state='sent'").run(now(), commentId, taskId); /* Emit only on a real transition (changes===1), so a no-op re-acknowledge does not append phantom events or duplicate SSE fan-out (same CAS-gated emit rule as markDelivered). */ if (Number(changed.changes) === 1) this.event(taskId, "comment.acknowledged", { commentId }, "system"); }); return this.comment(this.db.prepare("SELECT * FROM task_comments WHERE id=?").get(commentId) as Row); }
  listComments(taskId: string): MazzyTaskComment[] { return (this.db.prepare("SELECT * FROM task_comments WHERE task_id=? ORDER BY created_at,id").all(taskId) as Row[]).map((r) => this.comment(r)); }
  /** A durable per-parent-session receipt makes retries and session-start redelivery one-shot. */
  claimCommentNotification(taskId: string, commentId: string, sessionId: string): boolean { return this.tx(() => { if (!this.db.prepare("SELECT 1 FROM task_comments WHERE id=? AND task_id=?").get(commentId, taskId)) return false; const result = this.db.prepare("INSERT OR IGNORE INTO comment_notifications(comment_id,session_id,notified_at) VALUES(?,?,?)").run(commentId, clean(sessionId, "Session id"), now()); return Number(result.changes) === 1; }); }
  unnotifiedComments(sessionId: string): MazzyTaskComment[] { return (this.db.prepare("SELECT c.* FROM task_comments c WHERE c.role='user' AND NOT EXISTS (SELECT 1 FROM comment_notifications n WHERE n.comment_id=c.id AND n.session_id=?) ORDER BY c.created_at,c.id").all(clean(sessionId, "Session id")) as Row[]).map((row) => this.comment(row)); }

  updateRunActivity(input: UpdateRunActivityInput): MazzyRunBinding { const b = this.byRun(input.runId); if (!b || b.taskId !== input.taskId) throw new Error("Matching run binding is required"); if (!RUN_LIFECYCLES.includes(input.lifecycle)) throw new Error("Unknown run lifecycle"); if (input.cycle !== undefined && (!Number.isInteger(input.cycle) || input.cycle < 0 || input.cycle > 10_000)) throw new Error("cycle must be a bounded integer"); for (const [name, value] of [["currentActivity", input.currentActivity], ["currentTool", input.currentTool], ["model", input.model]]) if (value !== undefined && value !== null && value.length > 500) throw new Error(`${name} is too long`); const at = now(); this.tx(() => { this.db.prepare("UPDATE run_bindings SET lifecycle=?,model=COALESCE(?,model),cycle=COALESCE(?,cycle),last_activity_at=?,current_activity=?,current_tool=?,updated_at=? WHERE id=?").run(input.lifecycle, input.model?.trim() || null, input.cycle ?? null, at, input.currentActivity?.trim() || null, input.currentTool?.trim() || null, at, b.id); this.event(input.taskId, "run.activity-attested", { runId: input.runId, lifecycle: input.lifecycle }, input.actor); }); return this.byRun(input.runId)!; }

  createControlRequest(input: CreateControlRequestInput): MazzyControlRequest { const key = clean(input.idempotencyKey, "Idempotency key"), max = input.maxCycles ?? 1, instructions = input.instructions?.trim(); if (!CONTROL_COMMANDS.includes(input.command) || !Number.isInteger(max) || max < 1 || max > 10) throw new Error("Invalid control request"); if (instructions && instructions.length > MAX_CONTROL_INSTRUCTIONS_LENGTH) throw new Error(`Operator instructions must be at most ${MAX_CONTROL_INSTRUCTIONS_LENGTH} characters`); const fp = fingerprint("control", input.taskId, input.expectedTaskRevision, input.command, input.targetRunId, input.approvedAgent, { instructions, maxCycles: max, parentSessionId: input.parentSessionId }); return this.tx(() => { const old = this.byRequestKey(key); if (old) return this.replayRequest(old, fp); const current = this.requireRevision(input.taskId, input.expectedTaskRevision); this.controlApplicable(current, input.command, input.targetRunId); if (input.command === "GO") { const outstanding = this.db.prepare("SELECT * FROM orchestration_requests WHERE task_id=? AND expected_task_revision=? AND command='GO' AND state IN ('REQUESTED','DELIVERED','CLAIMED') ORDER BY requested_at,id LIMIT 1").get(input.taskId, input.expectedTaskRevision) as Row | undefined; if (outstanding) return { ...request(outstanding), coalesced: true }; } const id = randomUUID(), at = now(); this.db.prepare("INSERT INTO orchestration_requests(id,idempotency_key,operation_fingerprint,task_id,expected_task_revision,command,state,approved_agent,instructions,max_cycles,target_run_id,parent_session_id,requested_at) VALUES(?,?,?,?,?,?, 'REQUESTED',?,?,?,?,?,?)").run(id, key, fp, current.id, current.revision, input.command, input.approvedAgent?.trim() || null, instructions || null, max, input.command === "GO" ? null : input.targetRunId ?? null, input.parentSessionId?.trim() || null, at); this.event(current.id, "orchestration.requested", { requestId: id, command: input.command }, "web"); return this.getControlRequest(id)!; }); }
  getControlRequest(id: string): MazzyControlRequest | undefined { const r = this.db.prepare("SELECT * FROM orchestration_requests WHERE id=?").get(id) as Row | undefined; return r ? request(r) : undefined; }
  listControlRequests(taskId: string): MazzyControlRequest[] { return (this.db.prepare("SELECT * FROM orchestration_requests WHERE task_id=? ORDER BY requested_at DESC,id DESC").all(taskId) as Row[]).map(request); }
  nextUndeliveredControlRequest(): MazzyControlRequest | undefined { const r = this.db.prepare("SELECT * FROM orchestration_requests WHERE state='REQUESTED' ORDER BY requested_at,id LIMIT 1").get() as Row | undefined; return r ? request(r) : undefined; }
  markDelivered(id: string): MazzyControlRequest { const existing = this.getControlRequest(id); if (!existing) throw new Error(`Control request not found: ${id}`); this.tx(() => { const r = this.getControlRequest(id); if (!r || r.state !== "REQUESTED") return; const changed = this.db.prepare("UPDATE orchestration_requests SET state='DELIVERED',delivered_at=? WHERE id=? AND state='REQUESTED'").run(now(), id); if (Number(changed.changes) === 1) this.event(r.taskId, "orchestration.delivered", { requestId: id }, "bridge"); }); return this.getControlRequest(id)!; }
  claimControlRequest(input: ClaimControlRequestInput): MazzyControlRequest { const parentSessionId = clean(input.parentSessionId, "Parent session id"); return this.tx(() => { const r = this.getControlRequest(input.id); if (!r) throw new Error(`Control request not found: ${input.id}`); const current = r.command === "GO" ? this.requireRevision(r.taskId, r.expectedTaskRevision) : this.getTask(r.taskId); if (!current) throw new Error(`Task not found: ${r.taskId}`); this.controlApplicable(current, r.command, r.targetRunId); if (r.state === "CLAIMED") return r; if (!["REQUESTED", "DELIVERED"].includes(r.state)) throw new Error(`Control request is ${r.state}`); this.db.prepare("UPDATE orchestration_requests SET state='CLAIMED',claimed_at=?,parent_session_id=? WHERE id=?").run(now(), parentSessionId, r.id); this.event(r.taskId, "orchestration.claimed", { requestId: r.id }, "pi-parent"); return this.getControlRequest(r.id)!; }); }
  completeControlRequest(input: CompleteControlRequestInput): MazzyControlRequest { return this.tx(() => { const r = this.getControlRequest(input.id); if (!r) throw new Error(`Control request not found: ${input.id}`); if (r.state === "COMPLETED") return r; if (r.state !== "CLAIMED") throw new Error(`Control request is ${r.state}`); if (r.command === "GO") { const b = input.childRunId ? this.byRun(input.childRunId) : undefined; /* GO completion must bind an actual current-acceptance worker for THIS request's task, spawned at/after the request's expected revision, not merely any historical run sharing the session ids. */ if (!input.childSessionId || !b || b.taskId !== r.taskId || b.role !== "worker" || b.childSessionId !== input.childSessionId || b.parentSessionId !== r.parentSessionId || b.taskRevision < r.expectedTaskRevision) throw new Error("GO completion requires a matching current parent-attested worker binding"); } else if (!clean(input.outcome ?? "", "Observed interrupt/stop outcome")) throw new Error("PAUSE/STOP completion requires a real observed outcome"); const changed = this.db.prepare("UPDATE orchestration_requests SET state='COMPLETED',completed_at=?,child_session_id=?,child_run_id=? WHERE id=? AND state='CLAIMED'").run(now(), input.childSessionId ?? null, input.childRunId ?? null, r.id); if (Number(changed.changes) !== 1) throw new Error("Concurrent control request transition conflict"); this.event(r.taskId, "orchestration.completed", { requestId: r.id, outcome: input.outcome }, "pi-parent"); return this.getControlRequest(r.id)!; }); }
  failControlRequest(input: FailControlRequestInput): MazzyControlRequest { return this.tx(() => { const r = this.getControlRequest(input.id); if (!r) throw new Error(`Control request not found: ${input.id}`); /* Read terminal state before validating the error string, so an idempotent retry of an already-terminal request returns it instead of throwing on a malformed error payload. */ if (["COMPLETED", "FAILED", "CANCELLED"].includes(r.state)) return r; const error = clean(input.error, "Control request error").slice(0, 1000); const changed = this.db.prepare("UPDATE orchestration_requests SET state='FAILED',failed_at=?,error=? WHERE id=? AND state IN ('REQUESTED','DELIVERED','CLAIMED')").run(now(), error, r.id); if (Number(changed.changes) !== 1) throw new Error("Concurrent control request transition conflict"); this.event(r.taskId, "orchestration.failed", { requestId: r.id }, "pi-parent"); return this.getControlRequest(r.id)!; }); }
  /** One bounded recovery candidate. Owner availability is injected by the parent/server, never guessed. */
  reconcileOneClaimedRequest(ownerAvailable: (sessionId: string) => boolean): MazzyControlRequest | undefined { const r = this.db.prepare("SELECT * FROM orchestration_requests WHERE state='CLAIMED' ORDER BY claimed_at,id LIMIT 1").get() as Row | undefined; if (!r) return undefined; const value = request(r); if (value.parentSessionId && ownerAvailable(value.parentSessionId)) return value; if (this.requestHasMatchingRun(value)) return value; const reason = value.command === "GO" ? "owner unavailable and no matching binding/run" : "owner unavailable and control target is no longer current"; this.tx(() => { if (value.command === "GO") { this.db.prepare("UPDATE orchestration_requests SET state='REQUESTED',recovered_at=?,recovery_reason=? WHERE id=? AND state='CLAIMED'").run(now(), reason, value.id); this.event(value.taskId, "orchestration.requeued", { requestId: value.id, reason }, "recovery"); } else { this.db.prepare("UPDATE orchestration_requests SET state='FAILED',failed_at=?,recovered_at=?,recovery_reason=?,error=? WHERE id=? AND state='CLAIMED'").run(now(), now(), reason, reason, value.id); this.event(value.taskId, "orchestration.failed", { requestId: value.id, reason }, "recovery"); } }); return this.getControlRequest(value.id); }

  listBindings(taskId: string): MazzyRunBinding[] { return (this.db.prepare("SELECT * FROM run_bindings WHERE task_id=? ORDER BY created_at,id").all(taskId) as Row[]).map(binding); }
  /* Order by monotonic rowid (durable insertion order), not created_at/UUID, so quality-gate latest-verdict matches the DONE gate's total ordering. */
  listEvidence(taskId: string): MazzyEvidence[] { return (this.db.prepare("SELECT * FROM evidence WHERE task_id=? ORDER BY rowid").all(taskId) as Row[]).map((r) => this.evidence(r)); }
  /* Order by monotonic rowid (durable insertion order), not mutable updated_at + random UUID, so a same-millisecond report tie resolves to the truly latest report and attestCompletion's idempotent replay keeps matching the current worker (consistent with listEvidence/requireCurrentReviewerPass). */
  getReviewReport(taskId: string, acceptanceRevision?: number): MazzyReviewReport | undefined { const r = this.db.prepare(`SELECT * FROM review_reports WHERE task_id=? ${acceptanceRevision === undefined ? "" : "AND acceptance_revision=?"} ORDER BY rowid DESC LIMIT 1`).get(...(acceptanceRevision === undefined ? [taskId] : [taskId, acceptanceRevision])) as Row | undefined; return r ? this.rowReport(r) : undefined; }
  getTaskDetail(taskId: string): MazzyTaskDetail | undefined { const t = this.getTask(taskId); if (!t) return undefined; const bindings = this.listBindings(taskId), report = this.getReviewReport(taskId, t.acceptanceRevision); const inconsistentWorker = bindings.find((b) => b.role === "worker" && b.state === "active" && ["REVIEW", "DONE"].includes(t.state)); const inconsistencies = inconsistentWorker ? [`Active worker ${inconsistentWorker.runId} is inconsistent with terminal/review task state ${t.state}; completion reconciliation is pending.`] : []; const evidence = this.listEvidence(taskId).map((x) => ({ ...x, freshness: (x.acceptanceRevision === t.acceptanceRevision ? "current" : "stale") as "current" | "stale" })); return { task: t, comments: this.listComments(taskId), bindings, requests: this.listControlRequests(taskId), report, reportStatus: report ? "present" : this.getReviewReport(taskId) ? "stale" : "report missing", evidence, events: this.eventsFor(taskId), inconsistencies, qualityGates: this.computeQualityGates(t, bindings, evidence, report) }; }
  /** Pure read-only audit projection: derive each gate's status from durable facts only. No gate is stored or writable, so comments/self-reports can never satisfy one. */
  private computeQualityGates(t: MazzyTask, bindings: MazzyRunBinding[], evidence: Array<MazzyEvidence & { freshness: "current" | "stale" }>, report?: MazzyReviewReport): QualityGateSummary {
    const currentWorker = bindings.filter((b) => b.role === "worker" && b.state === "completed" && b.acceptanceRevision === t.acceptanceRevision && b.acceptanceDigest === t.acceptanceDigest).sort((a, b) => b.taskRevision - a.taskRevision)[0];
    /* Only count reviewer evidence whose reviewer binding was assigned at/after the current worker (immutable task_revision), so a stale prior-cycle PASS is never reported as current — matching the authoritative requireCurrentReviewerPass gate. listEvidence is in durable rowid order, so the LAST match is the authoritative latest verdict (a later FAIL wins). */
    const currentReviewerEvidence = evidence.filter((e) => e.freshness === "current" && e.runId && bindings.some((b) => b.id === e.bindingId && b.role === "reviewer" && (!currentWorker || b.taskRevision >= currentWorker.taskRevision)));
    const latestVerdict = currentReviewerEvidence[currentReviewerEvidence.length - 1]?.verdict;
    const staleReview = evidence.some((e) => e.freshness === "stale" && bindings.some((b) => b.id === e.bindingId && b.role === "reviewer"));
    const activeWorkerInTerminal = bindings.some((b) => b.role === "worker" && b.state === "active" && ["REVIEW", "DONE"].includes(t.state));
    const gate = (id: string, label: string, category: QualityGateCategory, required: boolean, status: QualityGateStatus, detail: string): QualityGate => ({ id, label, category, required, status, detail });
    const gates: QualityGate[] = [
      gate("worker-completed", "Completed worker for current acceptance", "worker-report", true,
        currentWorker ? "PASS" : ["DONE", "REVIEW"].includes(t.state) ? "MISSING" : "PENDING",
        currentWorker ? `Worker ${currentWorker.runId} completed at acceptance revision ${t.acceptanceRevision}.` : "No completed worker bound to the current acceptance yet."),
      gate("review-report", "Structured review report present", "worker-report", true,
        report ? "PASS" : this.getReviewReport(t.id) ? "STALE" : currentWorker ? "MISSING" : "PENDING",
        report ? "A report exists for the current acceptance." : this.getReviewReport(t.id) ? "Only a stale report from an earlier acceptance exists." : "No structured report has been imported."),
      gate("independent-review", "Independent reviewer PASS (current acceptance)", "independent-review", true,
        latestVerdict === "PASS" ? "PASS" : latestVerdict === "FAIL" ? "FAIL" : latestVerdict === "UNCERTAIN" ? "PENDING" : staleReview ? "STALE" : currentWorker ? "MISSING" : "PENDING",
        latestVerdict ? `Latest current-acceptance reviewer verdict: ${latestVerdict}.` : staleReview ? "Only stale-acceptance reviewer evidence exists." : "No independent reviewer evidence at the current acceptance."),
      gate("acceptance-freshness", "Evidence matches current acceptance revision", "acceptance-freshness", true,
        evidence.length === 0 ? (currentWorker ? "MISSING" : "PENDING") : evidence.some((e) => e.freshness === "current") ? "PASS" : "STALE",
        `Acceptance revision ${t.acceptanceRevision}; ${evidence.filter((e) => e.freshness === "current").length} current / ${evidence.filter((e) => e.freshness === "stale").length} stale evidence rows.`),
      gate("binding-integrity", "No active worker in terminal/review state", "binding-integrity", true,
        activeWorkerInTerminal ? "FAIL" : "PASS",
        activeWorkerInTerminal ? "An active worker binding is inconsistent with the task state; reconciliation pending." : "Run bindings are consistent with the task state."),
      gate("done-eligibility", "Eligible to close as DONE", "orchestration", false,
        t.state === "DONE" ? "PASS" : (currentWorker && report && latestVerdict === "PASS" && !activeWorkerInTerminal) ? "PASS" : "PENDING",
        t.state === "DONE" ? "Task is DONE with a satisfied current-acceptance gate set." : "DONE requires a completed worker, a current report, and a current independent PASS."),
    ];
    const required = gates.filter((g) => g.required);
    const requiredPassed = required.filter((g) => g.status === "PASS" || g.status === "N/A").length;
    const blocking = required.filter((g) => g.status !== "PASS" && g.status !== "N/A").length;
    return { gates, requiredTotal: required.length, requiredPassed, blocking, readyForDone: blocking === 0 };
  }
  listEvents(afterId = 0, limit = 200): MazzyEvent[] { return this.rowsEvents(this.db.prepare("SELECT * FROM events WHERE id>? ORDER BY id LIMIT ?").all(afterId, Math.max(1, Math.min(1000, limit))) as Row[]); }
  latestEventId(): number { return Number((this.db.prepare("SELECT COALESCE(MAX(id),0) AS id FROM events").get() as { id: number }).id); }
  /** Subscribe to committed durable events. The returned function must be called on client disconnect. */
  subscribeEvents(listener: (event: MazzyEvent) => void): () => void { this.eventListeners.add(listener); return () => this.eventListeners.delete(listener); }
  snapshot(): MazzySnapshot { const tasks = this.listTasks(), counts = Object.fromEntries(TASK_STATES.map((s) => [s, 0])) as Record<TaskState, number>; for (const t of tasks) counts[t.state]++; return { tasks, counts, states: TASK_STATES, allowedTransitions: ALLOWED_TRANSITIONS, uiTransitions: UI_TRANSITIONS, latestEventId: this.latestEventId() }; }

  private evidence(r: Row): MazzyEvidence { return { id: String(r.id), taskId: String(r.task_id), taskRevision: Number(r.task_revision), acceptanceRevision: Number(r.acceptance_revision), kind: String(r.kind), verdict: r.verdict as MazzyEvidence["verdict"], actor: String(r.actor), payload: JSON.parse(String(r.payload_json)), createdAt: String(r.created_at), runId: r.run_id ? String(r.run_id) : undefined, bindingId: r.binding_id ? String(r.binding_id) : undefined }; }
  private comment(r: Row): MazzyTaskComment { return { id: String(r.id), taskId: String(r.task_id), body: String(r.body), actor: String(r.actor), role: r.role as TaskCommentRole, deliveryState: (r.delivery_state === "acknowledged" || r.delivery_state === "failed" ? r.delivery_state : "sent"), clientMessageId: r.client_message_id ? String(r.client_message_id) : undefined, replyTo: r.reply_to ? String(r.reply_to) : undefined, runId: r.run_id ? String(r.run_id) : undefined, sessionId: r.session_id ? String(r.session_id) : undefined, createdAt: String(r.created_at), acknowledgedAt: r.acknowledged_at ? String(r.acknowledged_at) : undefined, error: r.error ? String(r.error) : undefined }; }
  private tx<T>(fn: () => T): T { this.db.exec("BEGIN IMMEDIATE"); const previous = this.committedEvents; this.committedEvents = []; try { const value = fn(); this.db.exec("COMMIT"); const events = this.committedEvents; this.committedEvents = previous; for (const event of events) for (const listener of this.eventListeners) { try { listener(event); } catch { /* A dashboard subscriber cannot affect committed store state. */ } } return value; } catch (e) { this.db.exec("ROLLBACK"); this.committedEvents = previous; throw e; } }
  private requireRevision(id: string, expected: number | undefined): MazzyTask { if (expected === undefined) throw new Error("expectedRevision is required for task updates"); const t = this.getTask(id); if (!t) throw new Error(`Task not found: ${id}`); if (t.revision !== expected) throw new Error(`Revision conflict: expected ${expected}, current ${t.revision}`); return t; }
  private byRun(runId: string): MazzyRunBinding | undefined { const r = this.db.prepare("SELECT * FROM run_bindings WHERE run_id=?").get(runId) as Row | undefined; return r ? binding(r) : undefined; }
  private byBindingKey(key: string): MazzyRunBinding | undefined { const r = this.db.prepare("SELECT * FROM run_bindings WHERE idempotency_key=?").get(key) as Row | undefined; return r ? binding(r) : undefined; }
  private byRequestKey(key: string): MazzyControlRequest | undefined { const r = this.db.prepare("SELECT * FROM orchestration_requests WHERE idempotency_key=?").get(key) as Row | undefined; return r ? request(r) : undefined; }
  private replayBinding(old: MazzyRunBinding, fp: string): MazzyRunBinding { if (old.operationFingerprint !== fp) throw new Error("Idempotency key conflict: operation fingerprint differs"); return old; }
  private replayRequest(old: MazzyControlRequest, fp: string): MazzyControlRequest { if (old.operationFingerprint !== fp) throw new Error("Idempotency key conflict: operation fingerprint differs"); return old; }
  private activeWorker(id: string): MazzyRunBinding | undefined { const r = this.db.prepare("SELECT * FROM run_bindings WHERE task_id=? AND role='worker' AND state='active'").get(id) as Row | undefined; return r ? binding(r) : undefined; }
  /* Order by the immutable assignment epoch (task_revision), never by mutable updated_at: an update-monitor call on an older completed worker must not make it "latest" and let its stale reviewer PASS close a newer, unreviewed submission. */
  private latestCompletedWorker(t: MazzyTask): MazzyRunBinding | undefined { const r = this.db.prepare("SELECT * FROM run_bindings WHERE task_id=? AND role='worker' AND state='completed' AND acceptance_revision=? AND acceptance_digest=? ORDER BY task_revision DESC,created_at DESC,id DESC LIMIT 1").get(t.id, t.acceptanceRevision, t.acceptanceDigest) as Row | undefined; return r ? binding(r) : undefined; }
  private insertBinding(input: { task: MazzyTask; revision: number; runId: string; agent: string; role: RunRole; key: string; fp: string; parentSessionId?: string; childSessionId?: string; model?: string; cycle?: number }): MazzyRunBinding { const id = randomUUID(), at = now(); this.db.prepare("INSERT INTO run_bindings(id,task_id,task_revision,acceptance_revision,acceptance_digest,run_id,agent,role,state,idempotency_key,operation_fingerprint,parent_session_id,child_session_id,lifecycle,model,cycle,last_activity_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,'active',?,?,?,?,?,'queued',?,?,?,?)").run(id, input.task.id, input.revision, input.task.acceptanceRevision, input.task.acceptanceDigest, input.runId, input.agent, input.role, input.key, input.fp, input.parentSessionId ?? null, input.childSessionId ?? null, input.model?.trim() || null, input.cycle ?? null, at, at, at); return this.byRun(input.runId)!; }
  private assignmentTask(t: MazzyTask, state: TaskState, actor: string, revision: number): void { const result = this.db.prepare("UPDATE tasks SET state=?,executor_actor=?,revision=?,updated_at=? WHERE id=? AND revision=? AND state=?").run(state, actor, revision, now(), t.id, t.revision, t.state); if (Number(result.changes) !== 1) throw new Error("Concurrent task update conflict"); }
  private release(id: string, state: Exclude<RunBindingState, "active" | "completed">, at: string): void { this.db.prepare("UPDATE run_bindings SET state=?,updated_at=? WHERE id=? AND state='active'").run(state, at, id); }
  private releaseActiveReviewers(id: string, at: string): number { return Number(this.db.prepare("UPDATE run_bindings SET state='superseded',updated_at=? WHERE task_id=? AND role='reviewer' AND state='active'").run(at, id).changes); }
  private releaseActiveBindings(id: string, state: Exclude<RunBindingState, "active" | "completed">, at: string): number { return Number(this.db.prepare("UPDATE run_bindings SET state=?,updated_at=? WHERE task_id=? AND state='active'").run(state, at, id).changes); }
  private controlApplicable(t: MazzyTask, command: MazzyControlRequest["command"], target?: string): void { if (command === "GO") { if (!["BACKLOG", "READY", "BLOCKED"].includes(t.state)) throw new Error(`GO is not applicable from ${t.state}`); /* A GO-applicable task that still holds an active worker (e.g. BLOCKED->READY preserves the binding) cannot fulfil a fresh GO: assignRun rejects a second worker and transferRun requires RUNNING. Reject the unrealizable request in every GO state, not only BLOCKED. */ if (this.activeWorker(t.id)) throw new Error(`GO is not applicable while ${t.state} still holds an active worker; resume or release it first`); return; } const b = this.activeWorker(t.id); // taskRevision is the immutable assignment epoch. Lifecycle/metadata revisions
    // must not invalidate control of the still-current accepted worker run.
    if (t.state !== "RUNNING" || !target || !b || b.taskId !== t.id || b.runId !== target || b.acceptanceRevision !== t.acceptanceRevision || b.acceptanceDigest !== t.acceptanceDigest) throw new Error(`${command} requires the current active worker binding`); }
  private requestHasMatchingRun(r: MazzyControlRequest): boolean { if (r.command !== "GO") { const current = this.getTask(r.taskId); if (!current) return false; try { this.controlApplicable(current, r.command, r.targetRunId); return true; } catch { return false; } } return Boolean((this.db.prepare("SELECT 1 FROM run_bindings WHERE task_id=? AND parent_session_id=? AND task_revision>=? LIMIT 1").get(r.taskId, r.parentSessionId ?? "", r.expectedTaskRevision) as Row | undefined)); }
  private requireCurrentReviewerPass(t: MazzyTask): void { const worker = this.latestCompletedWorker(t); if (!worker) throw new Error("DONE requires a completed worker for the current acceptance"); /* The reviewer must have been assigned to review THIS worker submission, not an earlier same-acceptance one. A reviewer that reviewed this worker was necessarily assigned after it, so its immutable binding task_revision >= worker.taskRevision. We compare that immutable assignment epoch rather than wall-clock created_at, because millisecond timestamps can tie (all sequential ops in one ms would let a stale prior-cycle W1 PASS reach DONE for an unreviewed W2). Verdicts are ordered by monotonic rowid (durable insertion order) so a later FAIL sharing a millisecond with an earlier PASS stays authoritative. */ const latest = this.db.prepare("SELECT e.verdict FROM evidence e JOIN run_bindings b ON b.id=e.binding_id WHERE e.task_id=? AND e.acceptance_revision=? AND b.role='reviewer' AND b.acceptance_revision=? AND b.acceptance_digest=? AND b.run_id<>? AND b.agent<>? AND b.task_revision>=? ORDER BY e.rowid DESC LIMIT 1").get(t.id, t.acceptanceRevision, t.acceptanceRevision, t.acceptanceDigest, worker.runId, worker.agent, worker.taskRevision) as { verdict?: string } | undefined; if (latest?.verdict !== "PASS") throw new Error("DONE requires the latest conclusive independent reviewer PASS at the current acceptance revision"); }
  private sameReport(existing: MazzyReviewReport, b: MazzyRunBinding, report: ReviewReportInput): boolean { return existing.workerRunId === b.runId && existing.agent === (report.agent?.trim() || b.agent) && existing.summary === report.summary?.trim() && existing.whatChanged === report.whatChanged?.trim() && existing.checks === report.checks?.trim() && existing.howToUse === report.howToUse?.trim() && JSON.stringify(existing.acceptanceCriteria) === JSON.stringify(report.acceptanceCriteria) && JSON.stringify(existing.results) === JSON.stringify(report.results) && existing.limitations === report.limitations?.trim() && existing.model === (report.model?.trim() || undefined) && existing.sessionId === (report.sessionId?.trim() || undefined) && existing.runId === (report.runId?.trim() || undefined) && existing.cycle === report.cycle; }
  private upsertReport(t: MazzyTask, b: MazzyRunBinding, report: ReviewReportInput, at: string): MazzyReviewReport { const strings: Array<[string, string]> = [["summary", report.summary], ["whatChanged", report.whatChanged], ["checks", report.checks], ["howToUse", report.howToUse], ["limitations", report.limitations]]; for (const [label, value] of strings) { if (typeof value !== "string" || !value.trim() || value.length > MAX_REPORT_FIELD_LENGTH) throw new Error(`Review report ${label} must be a non-empty bounded string`); } if (report.model && report.model.length > 500) throw new Error("Review report model is too long"); const existing = this.db.prepare("SELECT id FROM review_reports WHERE task_id=? AND acceptance_revision=? AND worker_run_id=?").get(t.id, t.acceptanceRevision, b.runId) as { id: string } | undefined; if (existing) this.db.prepare("UPDATE review_reports SET summary=?,what_changed=?,checks=?,how_to_use=?,acceptance_criteria_json=?,results_json=?,limitations=?,model=?,session_id=?,run_id=?,cycle=?,updated_at=? WHERE id=?").run(report.summary.trim(), report.whatChanged.trim(), report.checks.trim(), report.howToUse.trim(), JSON.stringify(report.acceptanceCriteria), JSON.stringify(report.results), report.limitations.trim(), report.model?.trim() || null, report.sessionId?.trim() || null, report.runId?.trim() || null, report.cycle ?? null, at, existing.id); else this.db.prepare("INSERT INTO review_reports(id,task_id,acceptance_revision,worker_run_id,agent,parent_session_id,child_session_id,summary,what_changed,checks,how_to_use,acceptance_criteria_json,results_json,limitations,model,session_id,run_id,cycle,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)").run(randomUUID(), t.id, t.acceptanceRevision, b.runId, report.agent?.trim() || b.agent, b.parentSessionId ?? null, b.childSessionId ?? null, report.summary.trim(), report.whatChanged.trim(), report.checks.trim(), report.howToUse.trim(), JSON.stringify(report.acceptanceCriteria), JSON.stringify(report.results), report.limitations.trim(), report.model?.trim() || null, report.sessionId?.trim() || null, report.runId?.trim() || null, report.cycle ?? null, at, at); return this.getReviewReport(t.id, t.acceptanceRevision)!; }
  private rowReport(r: Row): MazzyReviewReport { return { id: String(r.id), taskId: String(r.task_id), acceptanceRevision: Number(r.acceptance_revision), workerRunId: String(r.worker_run_id), agent: String(r.agent), parentSessionId: r.parent_session_id ? String(r.parent_session_id) : undefined, childSessionId: r.child_session_id ? String(r.child_session_id) : undefined, summary: String(r.summary), whatChanged: String(r.what_changed), checks: String(r.checks), howToUse: String(r.how_to_use), acceptanceCriteria: JSON.parse(String(r.acceptance_criteria_json)), results: JSON.parse(String(r.results_json)), limitations: String(r.limitations), model: r.model ? String(r.model) : undefined, sessionId: r.session_id ? String(r.session_id) : undefined, runId: r.run_id ? String(r.run_id) : undefined, cycle: r.cycle === null ? undefined : Number(r.cycle), createdAt: String(r.created_at), updatedAt: String(r.updated_at) }; }
  private event(taskId: string, type: string, payload: unknown, actor: string): void { this.db.prepare("INSERT INTO events(task_id,type,payload_json,actor,created_at) VALUES(?,?,?,?,?)").run(taskId, type, JSON.stringify(payload), actor, now()); const row = this.db.prepare("SELECT * FROM events WHERE id=last_insert_rowid()").get() as Row; const emitted = this.rowsEvents([row])[0]!; if (this.committedEvents) this.committedEvents.push(emitted); else for (const listener of this.eventListeners) { try { listener(emitted); } catch { /* Subscribers are observational only. */ } } }
  private eventsFor(taskId: string): MazzyEvent[] { return this.rowsEvents(this.db.prepare("SELECT * FROM events WHERE task_id=? ORDER BY id").all(taskId) as Row[]); }
  private rowsEvents(rows: Row[]): MazzyEvent[] { return rows.map((r) => ({ id: Number(r.id), taskId: String(r.task_id), type: String(r.type), payload: JSON.parse(String(r.payload_json)), actor: String(r.actor), createdAt: String(r.created_at) })); }
}