// Red/green unit coverage for the agent-answerable escalation completer (epic #156, slice U6;
// ADR 0046). Drives the canonical attributed completer + the agent/revert entry points against an
// in-memory data layer and a stub engine, asserting the three things U6 promises:
//
//   1. an AGENT assignee completes the SAME typed form a human would — the engine `completeUserTask`
//      is called with the EXACT variables (no parallel path), and the process resume is driven;
//   2. attribution is recorded — a `task_completions` ledger row captures actor_kind=agent + the
//      agent id + the submitted variables;
//   3. the completion is reversible — a human can revert an agent completion (recording who + when),
//      while a human completion is NOT reversible and a completion can be reverted only once.
//
// The taxonomy/spine slices proved the form + resume round-trip end to end (the e2e does too, on the
// real process); this suite pins the host-side attribution + reversibility contract in isolation.

import { test } from "node:test";
import { assert, assertEquals, assertRejects } from "#test-assert";
import {
  completeEscalationAsAgent,
  completeEscalationAsHuman,
  completeUserTaskAttributed,
  latestCompletion,
  revertAgentCompletion,
  type TaskCompletion,
  validateEscalationVariables,
} from "./agentCompletion.ts";

/** A minimal in-memory `Table<T>`: AUTOINCREMENT ids on insert, structural `find`, `get`, `update`. */
function memTable(rows: any[], key: string) {
  let seq = rows.reduce((m, r) => Math.max(m, Number(r[key]) || 0), 0);
  return {
    insert: (row: any) => {
      const id = ++seq;
      const stored = key === "id" ? { ...row, id } : { ...row };
      rows.push(stored);
      return Promise.resolve(key === "id" ? id : stored[key]);
    },
    get: (k: any) => Promise.resolve(rows.find((r) => r[key] === k)),
    find: (q: any = {}) =>
      Promise.resolve(rows.filter((r) => Object.entries(q).every(([f, v]) => r[f] === v))),
    findOne: (q: any = {}) =>
      Promise.resolve(rows.find((r) => Object.entries(q).every(([f, v]) => r[f] === v))),
    update: (k: any, patch: any) => {
      const r = rows.find((x) => x[key] === k);
      if (r) Object.assign(r, patch);
      return Promise.resolve(r ? 1 : 0);
    },
    delete: (k: any) => {
      const i = rows.findIndex((r) => r[key] === k);
      if (i >= 0) rows.splice(i, 1);
      return Promise.resolve(i >= 0 ? 1 : 0);
    },
    count: (q: any = {}) =>
      Promise.resolve(rows.filter((r) => Object.entries(q).every(([f, v]) => r[f] === v)).length),
    all: () => Promise.resolve(rows.slice()),
  };
}

function memData(
  stores: Record<string, { rows: any[]; key: string }>,
  opts: { failExec?: (sql: string) => boolean; failUpdate?: (table: string) => boolean } = {},
) {
  // Minimal `open().exec` emulating ONLY the tombstone `UPDATE "pr_adjudications" SET "invalidated_at"
  // = ? WHERE "id" = ? AND "invalidated_at" IS NULL` that `revertAgentCompletion` issues via
  // `invalidateAdjudication` (Copilot review of #806). A revert TOMBSTONES the source adjudication (it
  // does NOT delete it) so a redelivered `record-answer` cannot re-insert the same fingerprint and
  // resurrect the reverted decision. The SQL itself is validated against real SQLite in
  // app/adjudications.test.ts; here it need only mutate the in-memory store so a revert-invalidation
  // assertion can observe the row being tombstoned. `open().tx(fn)` runs `fn` against the same store and
  // ROLLS BACK (restores a pre-tx snapshot) on throw, mirroring the real SQLite transaction the revert
  // now commits atomically (issue #806 review — atomicity of tombstone + `reverted` flip). `opts` lets a
  // test inject a transient write failure at a chosen point (exec or a table update) to exercise rollback.
  const rawExec = async (sql: string, params: unknown[] = []) => {
    const m = /UPDATE "pr_adjudications" SET "invalidated_at" = \? WHERE "id" = \? AND "invalidated_at" IS NULL/.exec(sql);
    if (m) {
      const store = stores.pr_adjudications;
      if (store) {
        const r = store.rows.find((r) => r[store.key] === params[1]);
        if (r && (r.invalidated_at == null || String(r.invalidated_at).trim() === "")) {
          r.invalidated_at = params[0];
          return { changed: 1 };
        }
      }
      return { changed: 0 };
    }
    // The FIRST-HAND agent-revert tombstone keyed on `source_completion_id` (Copilot review of #806) —
    // `invalidateAdjudicationByCompletion`. Tombstones EVERY live row this completion produced (a
    // completion settles one question, so at most one) so reverting a first-hand agent answer (which has
    // no `source_adjudication_id`) still stops the poller re-auto-applying it.
    const c = /UPDATE "pr_adjudications" SET "invalidated_at" = \? WHERE "source_completion_id" = \? AND "invalidated_at" IS NULL/.exec(sql);
    if (c) {
      const store = stores.pr_adjudications;
      let changed = 0;
      if (store) {
        for (const r of store.rows) {
          if (r.source_completion_id === params[1] && (r.invalidated_at == null || String(r.invalidated_at).trim() === "")) {
            r.invalidated_at = params[0];
            changed++;
          }
        }
      }
      return { changed };
    }
    // The conditional ledger flip the revert now issues (Copilot review of #806): the `reverted = 0`
    // fence is what serialises two concurrent reverts of the SAME completion — the loser's guarded
    // UPDATE changes ZERO rows, so the revert throws to roll the whole transaction (its tombstones
    // included) back, leaving the winner's one-time audit metadata unclobbered. Emulate the fence so
    // `res.changed` is honest.
    const rev = /UPDATE "task_completions" SET "reverted" = 1, "reverted_by" = \?, "reverted_note" = \?, "reverted_at" = \? WHERE "id" = \? AND "reverted" = 0/.exec(sql);
    if (rev) {
      const store = stores.task_completions;
      let changed = 0;
      if (store) {
        const r = store.rows.find((r) => r[store.key] === params[3]);
        if (r && (r.reverted === 0 || r.reverted == null)) {
          r.reverted = 1;
          r.reverted_by = params[0];
          r.reverted_note = params[1];
          r.reverted_at = params[2];
          changed = 1;
        }
      }
      return { changed };
    }
    throw new Error(`unexpected exec sql: ${sql}`);
  };
  const exec = async (sql: string, params: unknown[] = []) => {
    if (opts.failExec?.(sql)) throw new Error("transient write failure");
    return rawExec(sql, params);
  };
  const table = (name: string, key: string) => {
    const base = memTable(stores[name]?.rows ?? [], stores[name]?.key ?? key);
    if (opts.failUpdate?.(name)) {
      return { ...base, update: () => Promise.reject(new Error("transient write failure")) };
    }
    return base;
  };
  const source: any = {
    exec,
    table,
    tx: async (fn: (t: any) => Promise<unknown>) => {
      const snap = Object.fromEntries(
        Object.entries(stores).map(([n, s]) => [n, JSON.parse(JSON.stringify(s.rows))]),
      );
      try {
        return await fn(source);
      } catch (e) {
        for (const [n, s] of Object.entries(stores)) {
          s.rows.length = 0;
          s.rows.push(...snap[n]);
        }
        throw e;
      }
    },
  };
  return {
    open: () => source,
    table,
  } as any;
}

/** A stub engine recording every `completeUserTask`, with a seeded set of open user tasks. The
 *  completers resolve tasks via `openUserTasks` (CREATED only), so the fixture exposes it; a legacy
 *  `searchUserTasks` (ANY state) is also present to prove the completer does NOT reach for it. */
function fakeEngine(openTasks: Array<{ userTaskKey: string; elementId?: string }>) {
  const completed: Array<{ userTaskKey: string; variables?: Record<string, unknown> }> = [];
  const engine = {
    openUserTasks: (_filter?: Record<string, unknown>) => Promise.resolve(openTasks),
    searchUserTasks: (_filter?: Record<string, unknown>) =>
      Promise.reject(new Error("completer must resolve via openUserTasks, not searchUserTasks")),
    completeUserTask: (userTaskKey: string, variables?: Record<string, unknown>) => {
      completed.push({ userTaskKey, variables });
      return Promise.resolve();
    },
  } as any;
  return { engine, completed };
}

test("agent completion resumes with the exact typed vars a human submits AND records agent attribution", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([
    { userTaskKey: "ut-1", elementId: "feature-escalation" },
  ]);

  const r = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-1",
    agentId: "senior:answer-bot",
    variables: { resolution: "answer", answer: "use v2" },
  });

  assertEquals(r.ok, true);
  assertEquals(r.elementId, "feature-escalation");

  // Same resume path a human drives: completeUserTask called with the identical typed variables.
  assertEquals(completed.length, 1);
  assertEquals(completed[0].userTaskKey, "ut-1");
  assertEquals(completed[0].variables, { resolution: "answer", answer: "use v2", completedUserTaskKey: "ut-1", completedCompletionId: 1 });

  // Attribution recorded: an agent completion, its id, and the submitted variables.
  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.actor_kind, "agent");
  assertEquals(row.actor_id, "senior:answer-bot");
  assertEquals(row.element_id, "feature-escalation");
  assertEquals(JSON.parse(row.variables_json), { resolution: "answer", answer: "use v2" });
  assertEquals(row.reversible, 1, "an agent completion is reversible");
  assertEquals(row.reverted, 0);
});

test("agent completer refuses a non-escalation user task (scoped to the migrated escalation tasks)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-x", elementId: "decide" }]);

  const r = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-x",
    agentId: "bot",
    variables: { decision: "approve" },
  });

  assertEquals(r.ok, false);
  assertEquals(r.reason, "not a completable task");
  assertEquals(completed.length, 0, "a non-escalation task is never completed");
  assertEquals(stores.task_completions.rows.length, 0, "and no attribution row is written");
});

test("agent completer is a no-op for an unknown userTaskKey", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([
    { userTaskKey: "ut-1", elementId: "feature-escalation" },
  ]);

  const r = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-missing",
    agentId: "bot",
    variables: { resolution: "abandon" },
  });

  assertEquals(r.ok, false);
  assertEquals(r.reason, "no open completable task");
  assertEquals(completed.length, 0);
});

test("completer resolves only OPEN tasks — a completed/canceled task's key is a 404-style no-op, not a doomed re-completion", async () => {
  // A looping instance keeps COMPLETED/CANCELED user tasks alongside the live one. If the completer
  // matched by key against ANY-state tasks (`searchUserTasks`), a stale key would drive a re-completion
  // that the engine rejects with a 5xx instead of the intended "no open task" no-op. The completer must
  // query `openUserTasks` (CREATED only), so a key that exists only as a non-open task does not match.
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const completed: Array<{ userTaskKey: string }> = [];
  const engine = {
    // No open tasks…
    openUserTasks: () => Promise.resolve([]),
    // …even though a completed task with this key exists in the full (any-state) search.
    searchUserTasks: () => Promise.resolve([{ userTaskKey: "ut-done", elementId: "feature-escalation" }]),
    completeUserTask: (userTaskKey: string) => {
      completed.push({ userTaskKey });
      return Promise.resolve();
    },
  } as any;

  const r = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-done",
    operatorId: "alice",
    variables: { resolution: "abandon" },
  });

  assertEquals(r.ok, false);
  assertEquals(r.reason, "no open completable task");
  assertEquals(completed.length, 0, "a non-open key never drives a doomed re-completion");
});

test("a HUMAN operator completes a feature escalation via the SAME attributed resume path (issue #210)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const r = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-1",
    operatorId: "alice",
    variables: { resolution: "answer", answer: "use v2" },
  });

  assertEquals(r.ok, true);
  assertEquals(r.elementId, "feature-escalation");

  // Identical resume path to the agent/task-inbox: completeUserTask with the exact typed variables.
  assertEquals(completed.length, 1);
  assertEquals(completed[0].variables, { resolution: "answer", answer: "use v2", completedUserTaskKey: "ut-1", completedCompletionId: 1 });

  // Attribution recorded as a HUMAN completion — the authority, so NOT reversible.
  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.actor_kind, "human");
  assertEquals(row.actor_id, "alice");
  assertEquals(row.reversible, 0, "a human completion is the authority (not reversible)");
});

test("feature-blocked is HUMAN-completable but NOT agent-completable (issue #332)", async () => {
  // Issue #332 folded the bespoke `acknowledge-blocked` door onto the canonical human completer, so a
  // HUMAN operator retires a blocked run through `completeEscalationAsHuman`. It stays OUTSIDE the agent
  // surface (`ESCALATION_TASK_ELEMENTS`) — an agent must never acknowledge a blocked run on a human's
  // behalf — so the agent completer refuses it.
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-b", elementId: "feature-blocked" }]);

  const asAgent = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-b",
    agentId: "bot",
    variables: { note: "n" },
  });
  assertEquals(asAgent.ok, false, "the agent completer refuses feature-blocked");
  assertEquals(asAgent.reason, "not a completable task");
  assertEquals(completed.length, 0);

  const asHuman = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-b",
    operatorId: "alice",
    variables: { note: "reassigned to a human" },
  });
  assertEquals(asHuman.ok, true, "the human completer retires feature-blocked");
  assertEquals(asHuman.elementId, "feature-blocked");
  assertEquals(completed.length, 1);
  assertEquals(completed[0].variables, { note: "reassigned to a human", completedUserTaskKey: "ut-b", completedCompletionId: 1 });
});

test("conformance-escalation is HUMAN-completable but NOT agent-completable (issue #216)", async () => {
  // The retro conformance ack mirrors feature-blocked: a HUMAN operator retires it via
  // `completeEscalationAsHuman`, but it stays OUTSIDE the agent surface (`ESCALATION_TASK_ELEMENTS`) —
  // an agent must never acknowledge a conformance review on a human's behalf.
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-c", elementId: "conformance-escalation" }]);

  const asAgent = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-c",
    agentId: "bot",
    variables: { note: "n" },
  });
  assertEquals(asAgent.ok, false, "the agent completer refuses conformance-escalation");
  assertEquals(asAgent.reason, "not a completable task");
  assertEquals(completed.length, 0);

  const asHuman = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-c",
    operatorId: "alice",
    variables: { note: "filed follow-up" },
  });
  assertEquals(asHuman.ok, true, "the human completer retires conformance-escalation");
  assertEquals(asHuman.elementId, "conformance-escalation");
  assertEquals(completed.length, 1);
  assertEquals(completed[0].variables, { note: "filed follow-up", completedUserTaskKey: "ut-c", completedCompletionId: 1 });
});

test("empty-plan-escalation is HUMAN-completable but NOT agent-completable (issues #623/#624)", async () => {
  // The empty-plan operator decision mirrors feature-blocked/conformance: a HUMAN operator adjudicates
  // whether an empty plan is a legitimate no-op (accept) or needs re-planning (revise), through
  // `completeEscalationAsHuman`. It stays OUTSIDE the agent surface (`ESCALATION_TASK_ELEMENTS`) — the
  // fleet must never silently auto-resolve the very "no work was produced" case a human must attend.
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-e", elementId: "empty-plan-escalation" }]);

  const asAgent = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-e",
    agentId: "bot",
    variables: { directive: "accept" },
  });
  assertEquals(asAgent.ok, false, "the agent completer refuses empty-plan-escalation");
  assertEquals(asAgent.reason, "not a completable task");
  assertEquals(completed.length, 0);

  const asHuman = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-e",
    operatorId: "alice",
    variables: { directive: "revise", notes: "look again" },
  });
  assertEquals(asHuman.ok, true, "the human completer retires empty-plan-escalation");
  assertEquals(asHuman.elementId, "empty-plan-escalation");
  assertEquals(completed.length, 1);
  assertEquals(completed[0].variables, { directive: "revise", notes: "look again", completedUserTaskKey: "ut-e", completedCompletionId: 1 });
});

test("readiness-escalation(-pf) is HUMAN-completable but NOT agent-completable (issue #674)", async () => {
  // A readiness/preflight gate adjudicates whether upstream is ACTUALLY ready (proceed) or the gate
  // should be abandoned. Like feature-blocked/conformance/empty-plan it is a HUMAN operator decision —
  // an agent must never auto-answer it, or the fleet would silently defeat the very readiness gate the
  // task exists to enforce. So both ids stay OUTSIDE `ESCALATION_TASK_ELEMENTS` (agent-refused) but are
  // retired by the HUMAN completer via the one canonical `complete-user-task` door.
  for (const elementId of ["readiness-escalation-pf", "readiness-escalation"] as const) {
    const stores = { task_completions: { rows: [] as any[], key: "id" } };
    const data = memData(stores);
    const { engine, completed } = fakeEngine([{ userTaskKey: "ut-r", elementId }]);

    const asAgent = await completeEscalationAsAgent(data, engine, {
      userTaskKey: "ut-r",
      agentId: "bot",
      variables: { resolution: "acknowledge" },
    });
    assertEquals(asAgent.ok, false, `the agent completer refuses ${elementId}`);
    assertEquals(asAgent.reason, "not a completable task");
    assertEquals(completed.length, 0);

    const asHuman = await completeEscalationAsHuman(data, engine, {
      userTaskKey: "ut-r",
      operatorId: "alice",
      variables: { resolution: "abandon", answer: "upstream never published" },
    });
    assertEquals(asHuman.ok, true, `the human completer retires ${elementId}`);
    assertEquals(asHuman.elementId, elementId);
    assertEquals(completed.length, 1);
    assertEquals(completed[0].variables, { resolution: "abandon", answer: "upstream never published", completedUserTaskKey: "ut-r", completedCompletionId: 1 });
  }
});

test("human completer refuses a non-escalation user task and is a no-op for an unknown key", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-x", elementId: "decide" }]);

  const notEsc = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-x",
    operatorId: "alice",
    variables: { resolution: "abandon" },
  });
  assertEquals(notEsc.ok, false);
  assertEquals(notEsc.reason, "not a completable task");

  const missing = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-missing",
    operatorId: "alice",
    variables: { resolution: "abandon" },
  });
  assertEquals(missing.ok, false);
  assertEquals(missing.reason, "no open completable task");

  assertEquals(completed.length, 0, "neither refusal completes a task");
  assertEquals(stores.task_completions.rows.length, 0, "and no attribution row is written");
});

test("a human can revert/override an agent completion (recording who + when + corrective note)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "guess" } },
    { kind: "agent", id: "bot" },
  );

  const r = await revertAgentCompletion(data, completionId, { kind: "human", id: "alice" }, "wrong — use v3");
  assertEquals(r.ok, true);

  const row = (await latestCompletion(data, "ut-1"))!;
  assertEquals(row.reverted, 1);
  assertEquals(row.reverted_by, "alice");
  assertEquals(row.reverted_note, "wrong — use v3", "the human's correction is captured");
  assert(typeof row.reverted_at === "string" && row.reverted_at.length > 0, "reverted_at is stamped");
});

test("an auto-applied completion records the source adjudication id, and reverting it invalidates that adjudication (#806 review)", async () => {
  // The convergence poller auto-resumes an already-answered wait-answer by replaying a
  // `pr_adjudications` row. Marking that completion reverted alone would NOT stop the replay — the
  // poller keeps matching the unchanged adjudication row. Reverting must TOMBSTONE the linked
  // adjudication so the override actually sticks and the next round re-parks a human, while a
  // redelivered `record-answer` cannot resurrect it (Copilot review of #806).
  const stores = {
    task_completions: { rows: [] as any[], key: "id" },
    pr_adjudications: { rows: [{ id: 42, pr_key: "o/r#1", answer: "Cap at 5.", adjudicated_by: "alice", invalidated_at: null }] as any[], key: "id" },
  };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "Cap at 5." } },
    { kind: "human", id: "alice" },
    { autoApplied: true, sourceAdjudicationId: 42 },
  );

  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.auto_applied, 1, "an auto-apply is recorded auto_applied");
  assertEquals(row.reversible, 1, "an auto-apply is always reversible, even when attributed to a human adjudicator");
  assertEquals(row.source_adjudication_id, 42, "the replayed adjudication is linked");

  assertEquals(stores.pr_adjudications.rows.length, 1, "the durable adjudication exists before the revert");
  assertEquals(stores.pr_adjudications.rows[0].invalidated_at, null, "and is live (not tombstoned) before the revert");
  const r = await revertAgentCompletion(data, completionId, { kind: "human", id: "bob" }, "override");
  assertEquals(r.ok, true);
  assertEquals(stores.pr_adjudications.rows.length, 1, "the source adjudication row is TOMBSTONED, not deleted, so a redelivered record-answer cannot re-insert its fingerprint");
  assert(
    typeof stores.pr_adjudications.rows[0].invalidated_at === "string" && stores.pr_adjudications.rows[0].invalidated_at.length > 0,
    "reverting the auto-apply tombstoned its source adjudication so the poller cannot re-apply it",
  );
});

test("revert commits the tombstone and the ledger flip ATOMICALLY — a failed tombstone rolls back and is retryable (#806 review)", async () => {
  // Atomicity: the revert tombstones the source adjudication(s) AND flips the ledger `reverted` flag in
  // ONE transaction. If the tombstone throws, the whole transaction rolls back, so a retry sees an
  // un-reverted, un-tombstoned row and completes cleanly — it never trips the `already reverted` guard
  // with a still-live adjudication (an unrecoverable override, Finding 3).
  const stores = {
    task_completions: { rows: [] as any[], key: "id" },
    pr_adjudications: { rows: [{ id: 42, pr_key: "o/r#1", answer: "Cap at 5.", invalidated_at: null }] as any[], key: "id" },
  };
  let failInvalidate = true;
  // The FIRST tombstone attempt throws (a transient write failure); the transaction must roll back.
  const data = memData(stores, { failExec: (sql) => failInvalidate && /pr_adjudications/.test(sql) });
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "Cap at 5." } },
    { kind: "human", id: "alice" },
    { autoApplied: true, sourceAdjudicationId: 42 },
  );

  await assertRejects(() => revertAgentCompletion(data, completionId, { kind: "human", id: "bob" }, "override"));
  assertEquals(
    (stores.task_completions.rows[0] as TaskCompletion).reverted,
    0,
    "a failed tombstone must NOT have flipped the ledger reverted — otherwise the retry below is permanently rejected",
  );

  // The transient failure clears; a retry now completes and both tombstones + the ledger flip land.
  failInvalidate = false;
  const r = await revertAgentCompletion(data, completionId, { kind: "human", id: "bob" }, "override");
  assertEquals(r.ok, true, "the retry after a transient failure succeeds (the revert is recoverable)");
  assertEquals((stores.task_completions.rows[0] as TaskCompletion).reverted, 1, "the ledger is now reverted");
  assert(
    typeof stores.pr_adjudications.rows[0].invalidated_at === "string" && stores.pr_adjudications.rows[0].invalidated_at.length > 0,
    "the source adjudication is tombstoned after the successful retry",
  );
});

test("revert rolls the tombstone back when the LEDGER flip fails — no reverted=0-but-tombstoned window a redelivered record-answer could revive through (#806 review)", async () => {
  // Finding B (Copilot review of #806): the tombstone and the `reverted` flip are two writes. Were they
  // NOT atomic, a redelivered `record-answer` could interleave AFTER the tombstone but BEFORE `reverted`
  // lands, observe the completion as still live (`reverted = 0`) with a tombstoned decision, and REVIVE
  // it — resurrecting the operator's reverted override. Committing both in one transaction removes that
  // intermediate state entirely: this test forces the SECOND write (the ledger flip) to throw and proves
  // the FIRST (the tombstone) is rolled back, so no `reverted=0`-with-tombstone state is ever left behind.
  const stores = {
    task_completions: { rows: [] as any[], key: "id" },
    pr_adjudications: { rows: [{ id: 42, pr_key: "o/r#1", answer: "Cap at 5.", invalidated_at: null }] as any[], key: "id" },
  };
  let failLedgerFlip = true;
  const data = memData(stores, { failExec: (sql) => failLedgerFlip && /UPDATE "task_completions"/.test(sql) });
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "Cap at 5." } },
    { kind: "human", id: "alice" },
    { autoApplied: true, sourceAdjudicationId: 42 },
  );

  await assertRejects(() => revertAgentCompletion(data, completionId, { kind: "human", id: "bob" }, "override"));
  assertEquals(
    stores.pr_adjudications.rows[0].invalidated_at,
    null,
    "the tombstone is rolled back when the ledger flip fails — the revert is all-or-nothing, so no revivable intermediate state exists",
  );
  assertEquals((stores.task_completions.rows[0] as TaskCompletion).reverted, 0, "the ledger stays un-reverted after the rollback");

  // The transient failure clears; a retry now completes atomically.
  failLedgerFlip = false;
  const r = await revertAgentCompletion(data, completionId, { kind: "human", id: "bob" }, "override");
  assertEquals(r.ok, true, "the retry after a transient failure succeeds");
  assertEquals((stores.task_completions.rows[0] as TaskCompletion).reverted, 1, "the ledger is reverted after the retry");
  assert(
    typeof stores.pr_adjudications.rows[0].invalidated_at === "string" && stores.pr_adjudications.rows[0].invalidated_at.length > 0,
    "and the source adjudication is tombstoned — both writes land together",
  );
});

test("reverting a first-hand completion leaves an UNLINKED adjudication untouched (#806 review)", async () => {
  const stores = {
    task_completions: { rows: [] as any[], key: "id" },
    // An adjudication that this completion did NOT produce (source_completion_id ≠ our id) must not be
    // tombstoned by reverting an unrelated completion.
    pr_adjudications: { rows: [{ id: 7, pr_key: "o/r#1", answer: "keep", source_completion_id: 999, invalidated_at: null }] as any[], key: "id" },
  };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "first-hand" } },
    { kind: "agent", id: "bot" },
  );

  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.auto_applied, 0);
  assertEquals(row.source_adjudication_id, null, "a first-hand completion has no linked adjudication");
  assertEquals((await revertAgentCompletion(data, completionId, { kind: "human", id: "alice" })).ok, true);
  assertEquals(stores.pr_adjudications.rows.length, 1, "no adjudication is deleted for a first-hand revert");
  assertEquals(stores.pr_adjudications.rows[0].invalidated_at, null, "an adjudication this completion did not produce is left live");
});

test("reverting a FIRST-HAND agent completion tombstones the adjudication it produced, via source_completion_id (#806 review)", async () => {
  // A first-hand agent answer to a `wait-answer` records its OWN adjudication (auto_applied=0, no
  // source_adjudication_id) linked back only by `source_completion_id`. Reverting that reversible agent
  // completion must tombstone that decision, or the convergence poller re-auto-applies the overridden
  // answer — the exact gap Finding 1 flagged. There is no `source_adjudication_id` to key on, so the
  // revert finds the decision by the completion that created it.
  const stores = {
    task_completions: { rows: [] as any[], key: "id" },
    pr_adjudications: { rows: [] as any[], key: "id" },
  };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "first-hand" } },
    { kind: "agent", id: "bot" },
  );
  // The first-hand answer's own decision, linked to this completion (what `record-answer` records).
  stores.pr_adjudications.rows.push({ id: 55, pr_key: "o/r#1", answer: "first-hand", source_completion_id: completionId, invalidated_at: null });

  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.auto_applied, 0);
  assertEquals(row.source_adjudication_id, null, "a first-hand completion has no source_adjudication_id — only source_completion_id links it");
  assertEquals((await revertAgentCompletion(data, completionId, { kind: "human", id: "alice" })).ok, true);
  assert(
    typeof stores.pr_adjudications.rows[0].invalidated_at === "string" && stores.pr_adjudications.rows[0].invalidated_at.length > 0,
    "reverting the first-hand agent completion tombstoned the adjudication it produced so the poller cannot re-apply it",
  );
});

test("the ledger rolls back when the engine completion fails (never claims a completion that did not happen)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const engine = {
    searchUserTasks: () => Promise.resolve([{ userTaskKey: "ut-1", elementId: "feature-escalation" }]),
    completeUserTask: () => Promise.reject(new Error("engine rejected the completion")),
  } as any;

  let threw: unknown;
  try {
    await completeUserTaskAttributed(
      data,
      engine,
      { userTaskKey: "ut-1", elementId: "feature-escalation", variables: { resolution: "answer", answer: "x" } },
      { kind: "agent", id: "bot" },
    );
  } catch (err) {
    threw = err;
  }
  assert(threw instanceof Error && /engine rejected/.test(threw.message), "the engine failure propagates");
  assertEquals(stores.task_completions.rows.length, 0, "the attribution row was rolled back on failure");
});

test("a human completion is NOT reversible (it is already the authority)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-2", elementId: "plan-review-decision" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-2", elementId: "plan-review-decision", variables: { directive: "proceed", notes: "" } },
    { kind: "human", id: "operator" },
  );

  const stored = (await latestCompletion(data, "ut-2"))!;
  assertEquals(stored.reversible, 0);

  const r = await revertAgentCompletion(data, completionId, { kind: "human", id: "bob" });
  assertEquals(r.ok, false);
  assertEquals(r.reason, "completion is not reversible");
});

test("an agent completion can be reverted only once", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-3", elementId: "trial-merge-decision" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-3", elementId: "trial-merge-decision", variables: { action: "proceed" } },
    { kind: "agent", id: "bot" },
  );

  assertEquals((await revertAgentCompletion(data, completionId, { kind: "human", id: "alice" })).ok, true);
  const second = await revertAgentCompletion(data, completionId, { kind: "human", id: "bob" });
  assertEquals(second.ok, false);
  assertEquals(second.reason, "completion already reverted");
});

test("reverting an unknown completion id is a no-op", async () => {
  const data = memData({ task_completions: { rows: [], key: "id" } });
  const r = await revertAgentCompletion(data, 999, { kind: "human", id: "alice" });
  assertEquals(r.ok, false);
  assertEquals(r.reason, "no such completion");
});

test("only a human may revert a completion (an agent identity cannot weaken the audit trail)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine } = fakeEngine([{ userTaskKey: "ut-4", elementId: "feature-escalation" }]);

  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "ut-4", elementId: "feature-escalation", variables: { resolution: "answer", answer: "x" } },
    { kind: "agent", id: "bot" },
  );

  const r = await revertAgentCompletion(data, completionId, { kind: "agent", id: "other-bot" });
  assertEquals(r.ok, false);
  assertEquals(r.reason, "only a human may revert a completion");

  const row = (await latestCompletion(data, "ut-4"))!;
  assertEquals(row.reverted, 0, "the completion is left un-reverted when a non-human attempts it");
});

test("the attributed completer normalizes keys and rejects blank attribution", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-5", elementId: "feature-escalation" }]);

  // Whitespace around the key + actor id is trimmed before it reaches the ledger and the engine.
  const { completionId } = await completeUserTaskAttributed(
    data,
    engine,
    { userTaskKey: "  ut-5  ", elementId: "feature-escalation", variables: { resolution: "answer", answer: "x" } },
    { kind: "agent", id: "  bot  " },
  );
  assert(completionId > 0);
  assertEquals(completed[0].userTaskKey, "ut-5", "the engine is completed with the trimmed key");
  const row = stores.task_completions.rows[0] as TaskCompletion;
  assertEquals(row.user_task_key, "ut-5");
  assertEquals(row.actor_id, "bot");

  // A blank key / actor id is rejected upfront — the ledger never records blank attribution.
  await assertRejects(() =>
    completeUserTaskAttributed(
      data,
      engine,
      { userTaskKey: "   ", variables: {} },
      { kind: "human", id: "operator" },
    ),
  );
  await assertRejects(() =>
    completeUserTaskAttributed(
      data,
      engine,
      { userTaskKey: "ut-5", variables: {} },
      { kind: "human", id: "  " },
    ),
  );
  assertEquals(stores.task_completions.rows.length, 1, "no ledger row is written for a rejected completion");
});

test("a rollback failure never masks the original engine error", async () => {
  const engine = {
    searchUserTasks: () => Promise.resolve([{ userTaskKey: "ut-6", elementId: "feature-escalation" }]),
    completeUserTask: () => Promise.reject(new Error("engine rejected the completion")),
  } as any;

  // A data layer whose ledger delete (the rollback) also throws — the engine error must still win.
  const data = {
    table: () => ({
      insert: () => Promise.resolve(1),
      delete: () => Promise.reject(new Error("ledger delete failed")),
    }),
  } as any;

  let threw: unknown;
  try {
    await completeUserTaskAttributed(
      data,
      engine,
      { userTaskKey: "ut-6", elementId: "feature-escalation", variables: { resolution: "answer", answer: "x" } },
      { kind: "agent", id: "bot" },
    );
  } catch (err) {
    threw = err;
  }
  assert(
    threw instanceof Error && /engine rejected/.test(threw.message),
    "the engine failure propagates, not the rollback failure",
  );
});

test("latestCompletion returns the newest row by id regardless of insertion order", async () => {
  const stores = {
    task_completions: {
      rows: [
        { id: 3, user_task_key: "ut-x", actor_kind: "agent", actor_id: "bot", variables_json: "{}", reversible: 1, reverted: 0, created_at: "t3" },
        { id: 1, user_task_key: "ut-x", actor_kind: "human", actor_id: "alice", variables_json: "{}", reversible: 0, reverted: 0, created_at: "t1" },
        { id: 2, user_task_key: "ut-other", actor_kind: "agent", actor_id: "bot", variables_json: "{}", reversible: 1, reverted: 0, created_at: "t2" },
      ] as any[],
      key: "id",
    },
  };
  const data = memData(stores);

  const newest = (await latestCompletion(data, "ut-x"))!;
  assertEquals(newest.id, 3, "the highest-id row for the key wins, not the first found");
  assertEquals(stores.task_completions.rows[0].id, 3, "the backing array is not reordered");
});

// --- Form-contract enforcement (issue #236 review advisory): a completion must satisfy the linked
// `.form`'s required-field + select allowed-value contract BEFORE the engine resumes, so a missing or
// invalid decision can never park the process in an invalid state. Derived from the canonical `.form`,
// exercised through BOTH completers so agent and human paths reject invalid input identically.

test("completer rejects a completion missing a required form field (no engine resume, no ledger row)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-1", elementId: "wait-answer" }]);

  const r = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-1",
    operatorId: "alice",
    variables: { answer: "   " }, // required, but blank
  });

  assertEquals(r.ok, false);
  assert(String(r.reason).includes("answer"), "the reason names the missing required field");
  assertEquals(completed.length, 0, "an invalid completion never resumes the process");
  assertEquals(stores.task_completions.rows.length, 0, "and no attribution row is written");
});

test("completer rejects a select value outside the form's allowed set", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-2", elementId: "trial-merge-decision" }]);

  const r = await completeEscalationAsAgent(data, engine, {
    userTaskKey: "ut-2",
    agentId: "bot",
    variables: { action: "explode" }, // not one of proceed/rebase/abandon
  });

  assertEquals(r.ok, false);
  assert(String(r.reason).includes("action"), "the reason names the invalid select field");
  assertEquals(completed.length, 0, "an out-of-range decision never resumes the process");
});

test("completer accepts variables that satisfy the form contract (required present + allowed value)", async () => {
  const stores = { task_completions: { rows: [] as any[], key: "id" } };
  const data = memData(stores);
  const { engine, completed } = fakeEngine([{ userTaskKey: "ut-3", elementId: "plan-review-decision" }]);

  const r = await completeEscalationAsHuman(data, engine, {
    userTaskKey: "ut-3",
    operatorId: "alice",
    variables: { directive: "revise", notes: "narrow scope" },
  });

  assertEquals(r.ok, true);
  assertEquals(completed.length, 1, "a contract-valid completion resumes the process");
  assertEquals(completed[0].variables, { directive: "revise", notes: "narrow scope", completedUserTaskKey: "ut-3", completedCompletionId: 1 });
});

test("validateEscalationVariables derives its contract from the canonical .form files", async () => {
  // wait-answer -> pr-escalation.form (answer required)
  assertEquals(validateEscalationVariables("wait-answer", { answer: "ok" }), null);
  assert(validateEscalationVariables("wait-answer", {}) !== null);
  // trial-merge-decision -> action required, allowed proceed/rebase/abandon
  assertEquals(validateEscalationVariables("trial-merge-decision", { action: "abandon" }), null);
  assert(validateEscalationVariables("trial-merge-decision", { action: "nope" }) !== null);
  // plan-review-decision -> directive required, allowed proceed/revise
  assertEquals(validateEscalationVariables("plan-review-decision", { directive: "proceed" }), null);
  assert(validateEscalationVariables("plan-review-decision", { directive: "" }) !== null);
  // empty-plan-escalation -> directive required, allowed accept/revise
  assertEquals(validateEscalationVariables("empty-plan-escalation", { directive: "accept" }), null);
  assertEquals(validateEscalationVariables("empty-plan-escalation", { directive: "revise" }), null);
  assert(validateEscalationVariables("empty-plan-escalation", { directive: "" }) !== null);
  // an element with no linked form contract is not enforced
  assertEquals(validateEscalationVariables("some-other-task", { whatever: 1 }), null);
});

test("feature-escalation demands non-blank answer on the answer path, but not on the hidden abandon path", async () => {
  // resolution=answer shows the conditional `answer` field, which is required → blank/missing rejected.
  assert(
    validateEscalationVariables("feature-escalation", { resolution: "answer" }) !== null,
    "resolution=answer with no guidance is rejected",
  );
  assert(
    validateEscalationVariables("feature-escalation", { resolution: "answer", answer: "   " }) !== null,
    "resolution=answer with whitespace-only guidance is rejected",
  );
  assertEquals(
    validateEscalationVariables("feature-escalation", { resolution: "answer", answer: "use v2" }),
    null,
    "resolution=answer with real guidance passes",
  );
  // resolution=abandon HIDES the `answer` field, so its required-ness is not enforced.
  assertEquals(
    validateEscalationVariables("feature-escalation", { resolution: "abandon" }),
    null,
    "the abandon path does not demand the hidden answer field",
  );
});