// Behavioural coverage for the sub-process merge-loop (issue #466).
//
// The merge-loop was refactored from a flat state machine into sub-processes
// (`SP_cifix`, `SP_rebase`) whose outcomes are propagated to the top level via
// end-event `ciOutcome`/`rebaseOutcome` output mappings and re-discriminated by
// `gw-ci-outcome`/`gw-rebase-outcome`. The previous merge guards were *structural
// text assertions* over the flat topology; they broke by construction under any
// re-shaping and re-encoded the model's shape rather than its behaviour.
//
// Per direction (issue #466) these are replaced with **behavioural** tests that
// deploy the committed model into the real WASM engine (`@nanobpm/urban-testkit`)
// and drive tokens through it, asserting the observable invariant — activated
// jobs, taken outcomes, terminal state, escalations, budget counters — so they
// protect what the loop *does*, not how it is drawn. The invariants preserved
// here are exactly those the retired guards protected:
//   - mergeRetryArm (#334): transient-retry arm re-arms within budget, escalates
//     when exhausted, advances the attempt counter only on a retry, no agent.
//   - mergeCiReattempt (#134): fix-ci `reattempt`/no-verdict re-arms (never pages
//     a human); blocked reconciles once from ground truth before escalating.
//   - mergeRebaseArm: conflict → bounded rebase agent → re-arm / escalate /
//     reconcile / wait-on-PR.
//   - mergeEscalationQuestion (#329/#454): the four blocked/SLA triggers and the
//     draft verdict each produce a distinct, human-actionable question; a
//     persist-escalation `escalated:false` re-enters the poller instead of
//     parking a dead user task.
//   - mergeEscalationUserTask (#256): escalation parks on the native
//     `wait-merge-answer` user task and the answer reconciles then re-arms.
// Plus the terminate semantics the refactor had to preserve: `MergeAbandoned`
// terminates the whole instance (it stays at the root, not inside a sub-process).
import { after, test } from "node:test";
import { assert, assertStringIncludes } from "#test-assert";
import { readFileSync } from "node:fs";
import {
  assertThatInstance,
  assertThatUserTask,
  byProcessId,
  createWasmEngineClient,
  type WasmEngineClient,
} from "@nanobpm/urban-testkit";

const MODEL = readFileSync("resources/processes/merge-loop.bpmn", "utf8");

const AGENT_SLA_MS = 30 * 60 * 1000; // matches the PT30M we start instances with
const LANDED_WAIT_MS = 30 * 60 * 1000; // matches the PT30M landedWaitTimeout we start instances with
const MERGEABLE_WAIT_MS = 30 * 60 * 1000; // matches the PT30M mergeableWaitTimeout we start instances with
const MERGEABLE_REPOLL_MS = 2 * 60 * 1000; // matches the PT2M mergeableRepollInterval we start instances with

type Output = Record<string, unknown>;
type Responder = Output | Output[] | ((job: { variables: Record<string, unknown> }) => Output);

const ALL_JOB_TYPES = [
  "pr.arm-merge",
  "pr.merge",
  "pr.merge-stall-probe",
  "pr.mark-merged",
  "senior:fix-ci",
  "senior:rebase",
  "pr.persist-escalation",
  "pr.answer-escalation",
  "pr.record-dependency",
] as const;

const DEFAULT_RESPONSES: Record<string, Responder> = {
  "pr.arm-merge": {},
  "pr.mark-merged": {},
  "pr.record-dependency": {},
  "pr.answer-escalation": {},
  "pr.persist-escalation": { escalated: true },
};

// Every FEEL expression in the model references these; start them defined (null,
// or a typed zero where the model compares/arithmetics the value, e.g.
// `failingChecks > 0`) so a missing-variable access can never raise a spurious
// incident in a test.
const DEFAULT_VARS: Record<string, unknown> = {
  prKey: "pr-1",
  repo: "acme/app",
  prNumber: 1,
  prUrl: "https://example.test/pr/1",
  ciFixMax: 3,
  rebaseMax: 3,
  mergeRetryMax: 3,
  ciFixRound: 0,
  rebaseRound: 0,
  mergeRetryRound: 0,
  agentSlaTimeout: "PT30M",
  landedWaitTimeout: "PT30M",
  mergeableWaitTimeout: "PT30M",
  mergeableRepollInterval: "PT2M",
  mergeStallRounds: 0,
  mergeStallMax: 3,
  abandonBrief: null,
  failingChecksList: null,
  status: null,
  mergeState: null,
  mergeStatus: null,
  ciBlockedReconciled: null,
  failingChecks: 0,
};

/**
 * Deploy the committed merge-loop and start one instance, wired to a per-job-type
 * responder. A responder may be a fixed output, a queue consumed per activation,
 * or a function of the job. A job type mapped to `null` registers **no** worker,
 * so its token parks on the task — used to let an agent SLA boundary fire.
 */
async function startMergeLoop(opts: {
  responses?: Record<string, Responder | null>;
  vars?: Record<string, unknown>;
} = {}): Promise<WasmEngineClient> {
  const engine = await createWasmEngineClient();
  await engine.deployResources([{ name: "merge-loop.bpmn", content: MODEL, contentType: "text/xml" }]);
  const responses: Record<string, Responder | null> = { ...DEFAULT_RESPONSES, ...(opts.responses ?? {}) };
  for (const jobType of ALL_JOB_TYPES) {
    const responder = jobType in responses ? responses[jobType] : undefined;
    if (responder === null) continue; // park the token (e.g. to let the SLA timer fire)
    const queue = Array.isArray(responder) ? [...responder] : null;
    await engine.registerWorker(jobType, (job) => {
      // The escalation `question`/`status` are job-LOCAL input mappings on
      // `merge-esc-*` (fed to `pr.persist-escalation`), so they never surface as
      // instance variables — capture them off the job the worker sees instead.
      if (jobType === "pr.persist-escalation") {
        lastEscalationByEngine.set(engine, (job as { variables?: Record<string, unknown> }).variables ?? {});
      }
      if (queue) return queue.length > 1 ? queue.shift()! : queue[0] ?? {};
      if (typeof responder === "function") return responder(job as { variables: Record<string, unknown> });
      return (responder as Output | undefined) ?? {};
    });
  }
  await engine.createInstance({
    processDefinitionId: "merge-loop",
    awaitCompletion: false,
    variables: { ...DEFAULT_VARS, ...(opts.vars ?? {}) },
  });
  return engine;
}

// Positive checks use the engine-testkit `assertThat*` DSL below. The DSL has no
// *negative* element matcher ("element X did NOT complete") and no *substring*
// variable matcher (`hasVariable` is deep-equal), so the two readers below cover
// exactly those gaps. They read via the **same canonical snapshot accessors the
// DSL uses internally** (`instance.js`): completions from the snapshot-global
// `elementStats` (`{ elementId, completed }`) and live vars from
// `instances[].variables`. Sound because every test runs one isolated instance
// per engine — the single-instance precondition the DSL's own aggregate read
// relies on.

/** Element ids completed by the single instance — mirrors the DSL's `completedElementIds`. */
function completedElementIds(engine: WasmEngineClient): Set<string> {
  const snap = engine.snapshot() as { elementStats?: { elementId: string; completed: number }[] };
  return new Set((snap.elementStats ?? []).filter((s) => s.completed > 0).map((s) => s.elementId));
}

/** The single instance's live variables — mirrors the DSL's `variablesOf`. */
function instanceVars(engine: WasmEngineClient): Record<string, unknown> {
  const snap = engine.snapshot() as { instances?: { variables?: Record<string, unknown> }[] };
  return snap.instances?.[0]?.variables ?? {};
}

/**
 * The variables the `pr.persist-escalation` worker was last activated with, captured
 * in `startMergeLoop`. The escalation `question`/`status` are job-local `zeebe:input`
 * mappings on `merge-esc-*`, so they are only observable on the persist-escalation job
 * — not as instance variables — this is the canonical read for the escalation payload.
 */
const lastEscalationByEngine = new WeakMap<WasmEngineClient, Record<string, unknown>>();
function escalation(engine: WasmEngineClient): Record<string, unknown> {
  return lastEscalationByEngine.get(engine) ?? {};
}

const engines: WasmEngineClient[] = [];

/**
 * urban-testkit's `assertThat*` DSL reads through a booted-app port: instance
 * matchers use `app.snapshot()`, user-task matchers use
 * `app.engine.{search,open}UserTasks`. These tests drive the WASM engine
 * directly (no full app boot), so expose the client as its own read-model app:
 * `snapshot()` is native, and `.engine` self-references so the task reads land
 * on the same client.
 */
function asReadModelApp(engine: WasmEngineClient): WasmEngineClient {
  (engine as unknown as { engine: WasmEngineClient }).engine = engine;
  return engine;
}

async function boot(opts?: Parameters<typeof startMergeLoop>[0]): Promise<WasmEngineClient> {
  const engine = asReadModelApp(await startMergeLoop(opts));
  engines.push(engine);
  return engine;
}

/** The key of the single open `wait-merge-answer` task, via the typed read model. */
async function mergeAnswerTaskKey(engine: WasmEngineClient): Promise<string> {
  const tasks = await engine.searchUserTasks({});
  const row = tasks.find((t) => t.elementId === "wait-merge-answer");
  assert(row, "expected an open wait-merge-answer user task");
  return row.userTaskKey;
}
after(async () => {
  await Promise.all(engines.map((e) => e.close()));
});

// ---------------------------------------------------------------------------
// Happy paths
// ---------------------------------------------------------------------------

test("a ready PR merges and the instance completes via mark-merged", async () => {
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "merged" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompleted().hasNoIncident().hasCompletedElements("mark-merged");
});

test("a queued merge parks on the event gateway; the landed message marks it merged", async () => {
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "queued" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElements("wait-landed", "wait-evicted", "wait-landed-timeout");
  await engine.publishMessage({ name: "merge-landed", correlationKey: "pr-1" });
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompleted().hasCompletedElements("mark-merged");
});

test("a queued merge that never lands escalates when the landing timeout fires (#556)", async () => {
  // The Mergify wedge: `attempt-merge` classifies the merge `queued` on an ambiguous signal, but the
  // repo never actually enqueued the PR, so `merge-landed` can never be published. Without the timer
  // arm the token would park at `wait-landed` forever (ACTIVE, no incident). The timer bounds it.
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "queued" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-landed");
  await engine.advanceTime(LANDED_WAIT_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-landed");
  assert(!completedElementIds(engine).has("mark-merged"), "a never-landed queued merge must not mark-merged");
  assertStringIncludes(String(escalation(engine).question ?? ""), "did not land within the merge-queue landing timeout", "the escalation must name the landing-timeout trigger");
});

test("answering the landing-timeout escalation re-arms the merge poller (#556)", async () => {
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "queued" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  await engine.advanceTime(LANDED_WAIT_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  await engine.completeUserTask(await mergeAnswerTaskKey(engine), { answer: "queued it manually, retry" });
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
});

// ---------------------------------------------------------------------------
// Mergeable-wait bounded-wait backstop (#636) — a stalled poller must never
// wedge the instance at `waiting_merge` forever. The `gw-merge-wait` event-based
// gateway races the poller's `merge-ready` message against a `mergeableWaitTimeout`
// timer; on timeout `merge-stall-probe` re-derives mergeability and the existing
// `gw-mergeable` routes to the correct arm — or, once `mergeStallMax` is exhausted,
// to human escalation.
// ---------------------------------------------------------------------------

test("wait-mergeable races merge-ready against a timer (no bare catch) (#636)", async () => {
  // After the poller arms the merge, the token forks the event-based gateway: it parks at BOTH the
  // `merge-ready` catch and the timeout timer. Before the fix, `wait-mergeable` was a bare catch —
  // no timer sibling — so a dead poller wedged it forever.
  const engine = await boot();
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElements("wait-mergeable", "wait-mergeable-timeout");
});

test("a dead poller's PR is probed on timeout and advances to merge — never wedges (#636)", async () => {
  // No `merge-ready` is EVER published (the poller is dead). RED (before the fix): the instance sits
  // at `wait-mergeable` forever. GREEN: the timer fires, `merge-stall-probe` re-derives `ready`, and
  // the existing `ready` arm merges the PR.
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "ready", failingChecks: 0, failingChecksList: "" },
      "pr.merge": { mergeStatus: "merged" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  await engine.advanceTime(MERGEABLE_WAIT_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .hasCompleted()
    .hasNoIncident()
    .hasCompletedElements("merge-stall-probe", "attempt-merge", "mark-merged");
});

test("a CONFLICTING PR whose poller never signalled reaches the rebase arm via the timeout path (#636)", async () => {
  // The incident that motivated #636: a `CONFLICTING`/`DIRTY` PR (base advanced under it) parked at
  // `waiting_merge` because the poller never signalled — the auto-rebase arm was never reached. The
  // timer path re-derives `conflict` and routes it to the bounded rebase agent.
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "conflict", failingChecks: 0, failingChecksList: "" },
      "senior:rebase": { status: "rebased" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.advanceTime(MERGEABLE_WAIT_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("merge-stall-probe", "rebase") // reached the rebase arm via the timeout path
    .hasVariable("rebaseRound", 1);
});

test("a blocked PR probed on timeout reaches the CI-fix arm via the timeout path (#636)", async () => {
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "blocked", failingChecks: 1, failingChecksList: "build" },
      "senior:fix-ci": { status: "fixed" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.advanceTime(MERGEABLE_WAIT_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("merge-stall-probe", "fix-ci")
    .hasVariable("ciFixRound", 1);
});

test("the probe advances mergeStallRounds and re-arms within budget (#636)", async () => {
  // A re-derived `conflict` re-arms after the rebase; the stall counter advanced so a still-dead
  // poller cannot loop the probe forever — it is bounded by mergeStallMax.
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "conflict", failingChecks: 0, failingChecksList: "" },
      "senior:rebase": { status: "rebased" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.advanceTime(MERGEABLE_WAIT_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop")).hasVariable("mergeStallRounds", 1);
  assert(!completedElementIds(engine).has("merge-esc-conflict"), "a within-budget stall must NOT escalate");
});

test("the stall-round cap escalates to a human instead of looping forever (#636)", async () => {
  // `mergeStallMax: 0` — escalate on the first stall. The probe runs once, `gw-merge-stall` finds the
  // budget exhausted, and routes to `merge-esc-conflict` → the native `wait-merge-answer` user task
  // rather than re-arming into another timeout round.
  const engine = await boot({
    vars: { mergeStallMax: 0 },
    responses: {
      "pr.merge-stall-probe": { mergeState: "conflict", failingChecks: 0, failingChecksList: "" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.advanceTime(MERGEABLE_WAIT_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-stall-probe", "merge-esc-conflict");
  assert(!completedElementIds(engine).has("mark-merged"), "an exhausted stall must not mark-merged");
});

// ---------------------------------------------------------------------------
// Async-UNKNOWN "waiting" verdict → bounded re-poll, NOT a human (issue #774)
// ---------------------------------------------------------------------------
// GitHub returns `mergeable=UNKNOWN` transiently while it computes mergeability in the background,
// which `classifyMergeability` maps to `"waiting"`. Before the fix, a `"waiting"` verdict at
// `gw-mergeable` hit the DEFAULT arm and escalated straight to the `wait-merge-answer` human task —
// so a PR whose real state was still settling (e.g. to `"conflict"`, which would auto-fire
// `senior:rebase`) paged a human instead of self-healing. The fix routes `"waiting"` to a bounded
// re-poll (`wait-mergeable-repoll` timer → `merge-stall-probe` → the existing `gw-mergeable` arms),
// bounded by the shared `mergeStallMax` budget, and routes the true DEFAULT to auto-rebase.

test("a 'waiting' (async UNKNOWN) verdict re-polls instead of escalating to a human (#774)", async () => {
  // RED (before the fix): `waiting` → DEFAULT → `merge-esc-conflict` → `wait-merge-answer`. GREEN:
  // `waiting` parks on the re-poll timer; when it fires, `merge-stall-probe` re-derives `ready` and
  // the PR merges — no human ever touched.
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "ready", failingChecks: 0, failingChecksList: "" },
      "pr.merge": { mergeStatus: "merged" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "waiting" } });
  // Parks on the bounded re-poll timer — NOT on a human task, NOT escalated.
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable-repoll");
  assert(!completedElementIds(engine).has("merge-esc-conflict"), "a waiting verdict must NOT escalate to a human");
  const openTasks = await engine.searchUserTasks({});
  assert(!openTasks.some((t) => t.elementId === "wait-merge-answer"), "a waiting verdict must not park a user task");
  // The timer fires → the probe re-derives ground truth (`ready`) → the PR merges.
  await engine.advanceTime(MERGEABLE_REPOLL_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .hasCompleted()
    .hasNoIncident()
    .hasCompletedElements("wait-mergeable-repoll", "merge-stall-probe", "attempt-merge", "mark-merged");
});

test("a 'waiting' verdict that settles to conflict on re-poll reaches the rebase arm (#774)", async () => {
  // The incident (Magikcraft/nano-bpm#1166): the PR was really CONFLICTING, but GitHub reported
  // UNKNOWN transiently. The re-poll must let it settle to `conflict` → `senior:rebase`, the arm that
  // would have fixed it, rather than paging a human on the transient UNKNOWN.
  const engine = await boot({
    responses: {
      "pr.merge-stall-probe": { mergeState: "conflict", failingChecks: 0, failingChecksList: "" },
      "senior:rebase": { status: "rebased" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "waiting" } });
  await engine.advanceTime(MERGEABLE_REPOLL_MS + 1);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasCompletedElements("wait-mergeable-repoll", "merge-stall-probe", "rebase")
    .hasVariable("rebaseRound", 1);
  assert(!completedElementIds(engine).has("merge-esc-conflict"), "settling to conflict must rebase, not escalate");
});

test("the 'waiting' re-poll is bounded — exhausting mergeStallMax escalates to a human (#774)", async () => {
  // A permanently-UNKNOWN PR cannot re-poll forever: the re-poll shares the `mergeStallMax` budget.
  // `mergeStallMax: 0` — escalate on the first re-poll. The probe runs once, `gw-merge-stall` finds
  // the budget exhausted, and routes to the human `wait-merge-answer` (last resort, not first).
  const engine = await boot({
    vars: { mergeStallMax: 0 },
    responses: {
      "pr.merge-stall-probe": { mergeState: "waiting", failingChecks: 0, failingChecksList: "" },
    },
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "waiting" } });
  assert(!completedElementIds(engine).has("merge-esc-conflict"), "must re-poll before ever escalating");
  await engine.advanceTime(MERGEABLE_REPOLL_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("wait-mergeable-repoll", "merge-stall-probe", "merge-esc-conflict");
});

test("an unclassified verdict routes to auto-rebase (DEFAULT), not to a human (#774)", async () => {
  // The DEFAULT arm is a last-resort self-heal, not a page: an unexpected/garbage `mergeState` routes
  // to the bounded rebase agent (which reconciles from ground truth), never straight to a human.
  const engine = await boot({ responses: { "senior:rebase": { status: "rebased" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "bogus" } });
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasCompletedElements("rebase")
    .hasVariable("rebaseRound", 1);
  const openTasks = await engine.searchUserTasks({});
  assert(!openTasks.some((t) => t.elementId === "wait-merge-answer"), "the default arm must not page a human");
});

test("an evicted queued merge re-arms the poller rather than completing", async () => {
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "queued" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  await engine.publishMessage({ name: "merge-evicted", correlationKey: "pr-1" });
  // back at the poller's wait, not merged
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  assert(!completedElementIds(engine).has("mark-merged"), "an evicted merge must not mark-merged");
});

// ---------------------------------------------------------------------------
// Transient-retry arm (mergeRetryArm, #334)
// ---------------------------------------------------------------------------

test("a transient retry re-arms the poller within budget and advances the retry counter only on retry", async () => {
  const engine = await boot({ responses: { "pr.merge": [{ mergeStatus: "retry" }, { mergeStatus: "merged" }] } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  // 1st attempt: retry (base moved) → within budget → re-arm → 2nd attempt: merged
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  // the retry re-armed and re-polled; feed a second mergeable so the 2nd attempt runs
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  assert(!completedElementIds(engine).has("merge-esc-attempt"), "a within-budget retry must NOT escalate");
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompleted();
});

test("a retry that exhausts the budget escalates as a repeated race, not a generic refusal", async () => {
  const engine = await boot({
    responses: { "pr.merge": { mergeStatus: "retry" } },
    vars: { mergeRetryMax: 0 }, // first retry → mergeRetryRound 1 > 0 → exhausted
  });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-attempt");
  assertStringIncludes(String(escalation(engine).question ?? ""), "retry budget", "the retry escalation must read as a repeated race");
});

// ---------------------------------------------------------------------------
// CI-fix sub-process (SP_cifix) — mergeCiReattempt (#134), #329
// ---------------------------------------------------------------------------

async function driveToCiFix(engine: WasmEngineClient): Promise<void> {
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({
    name: "merge-ready",
    correlationKey: "pr-1",
    variables: { mergeState: "blocked", failingChecks: 1, failingChecksList: "build" },
  });
}

test("a fixed CI verdict runs the agent, advances the fix counter, and re-arms the poller", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "fixed" } } });
  await driveToCiFix(engine);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("fix-ci")
    .hasVariable("ciFixRound", 1); // the fix counter advanced across the sub-process boundary
  const done = completedElementIds(engine);
  assert(!done.has("merge-esc-attempt") && !done.has("merge-esc-conflict"), "a fixed verdict must never escalate");
});

test("a reattempt CI verdict re-arms the poller and never pages a human (#134)", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "reattempt" } } });
  await driveToCiFix(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  const done = completedElementIds(engine);
  assert(!done.has("merge-esc-attempt") && !done.has("merge-esc-conflict"), "a reattempt must not escalate");
});

test("a no-verdict CI result reconciles from ground truth (re-arm), not escalation (#134)", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { summary: "unclear" } } }); // no `status`
  await driveToCiFix(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  assert(!completedElementIds(engine).has("merge-esc-attempt"), "a missing status must not escalate");
});

test("a blocked CI verdict with nothing pushed reconciles once before escalating", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "blocked", pushed: false } } });
  await driveToCiFix(engine);
  // ci-reconcile re-arms and re-checks mergeable; it does not escalate on the first block
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("ci-reconcile")
    .hasVariable("ciBlockedReconciled", true); // the one-shot reconcile is marked spent
  assert(!completedElementIds(engine).has("merge-esc-attempt"), "the first block episode must reconcile, not page a human");
});

test("a blocked CI verdict that already pushed escalates with a could-not-fix question", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "blocked", pushed: true } } });
  await driveToCiFix(engine);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-attempt");
  assertStringIncludes(String(escalation(engine).question ?? ""), "CI-fix agent could not", "the question must name the could-not-fix trigger");
});

test("a CI-fix that discovers a dependency records it and waits on the other PR", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "waiting-on-pr", dependsOn: "acme/app#2" } } });
  await driveToCiFix(engine);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-deps")
    .hasCompletedElements("record-merge-dep"); // a waiting-on-pr verdict records the dependency
});

test("CI-fix budget exhaustion escalates as not-mergeable without running the agent", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": { status: "fixed" } }, vars: { ciFixMax: 0 } });
  await driveToCiFix(engine);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-conflict");
  assert(!completedElementIds(engine).has("fix-ci"), "budget exhaustion must not run the fix-ci agent");
});

test("the fix-ci agent SLA interrupts the sub-process and escalates", async () => {
  const engine = await boot({ responses: { "senior:fix-ci": null } }); // park on the agent so the SLA fires
  await driveToCiFix(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("fix-ci");
  await engine.advanceTime(AGENT_SLA_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-attempt");
  assertStringIncludes(String(escalation(engine).question ?? ""), "time budget (SLA)", "the SLA escalation must name the SLA trigger");
});

// ---------------------------------------------------------------------------
// Rebase sub-process (SP_rebase) — mergeRebaseArm
// ---------------------------------------------------------------------------

async function driveToRebase(engine: WasmEngineClient): Promise<void> {
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "conflict" } });
}

test("a conflict runs the bounded rebase agent and a rebased result re-arms the poller", async () => {
  const engine = await boot({ responses: { "senior:rebase": { status: "rebased" } } });
  await driveToRebase(engine);
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("rebase") // a conflict runs the rebase agent, not page a human
    .hasVariable("rebaseRound", 1); // the rebase counter advanced across the sub-process boundary
});

test("a rebase that cannot resolve escalates with a conflict question", async () => {
  const engine = await boot({ responses: { "senior:rebase": { status: "blocked" } } });
  await driveToRebase(engine);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-attempt");
  assertStringIncludes(String(escalation(engine).question ?? ""), "rebase agent could not resolve", "the question must name the conflict trigger");
});

test("a no-verdict rebase result reconciles from ground truth, not escalation (#134)", async () => {
  const engine = await boot({ responses: { "senior:rebase": { summary: "unclear" } } }); // no `status`
  await driveToRebase(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  assert(!completedElementIds(engine).has("merge-esc-attempt"), "a missing rebase status must not escalate");
});

test("rebase budget exhaustion escalates as not-mergeable without running the agent", async () => {
  const engine = await boot({ responses: { "senior:rebase": { status: "rebased" } }, vars: { rebaseMax: 0 } });
  await driveToRebase(engine);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-conflict");
  assert(!completedElementIds(engine).has("rebase"), "budget exhaustion must not run the rebase agent");
});

test("the rebase agent SLA interrupts the sub-process and escalates as not-mergeable", async () => {
  const engine = await boot({ responses: { "senior:rebase": null } });
  await driveToRebase(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("rebase");
  await engine.advanceTime(AGENT_SLA_MS + 1);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-conflict");
});

// ---------------------------------------------------------------------------
// Escalation user task (mergeEscalationUserTask #256, mergeEscalationQuestion #329/#454)
// ---------------------------------------------------------------------------

test("an escalation parks on the native user task; answering it reconciles then re-arms the poller", async () => {
  const engine = await boot({ responses: { "senior:rebase": { status: "blocked" } } });
  await driveToRebase(engine);
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  await engine.completeUserTask(await mergeAnswerTaskKey(engine), { answer: "rebased manually, retry" });
  assertThatInstance(engine, byProcessId("merge-loop"))
    .isActive()
    .hasActiveElement("wait-mergeable")
    .hasCompletedElements("record-merge-answer"); // answering runs the pr.answer-escalation reconcile
});

test("a persist-escalation that opens nothing (escalated:false) re-enters the poller, not a dead user task", async () => {
  const engine = await boot({
    responses: { "senior:rebase": { status: "blocked" }, "pr.persist-escalation": { escalated: false } },
  });
  await driveToRebase(engine);
  assertThatInstance(engine, byProcessId("merge-loop")).isActive().hasActiveElement("wait-mergeable");
  const openTasks = await engine.searchUserTasks({});
  assert(
    !openTasks.some((t) => t.elementId === "wait-merge-answer"),
    "escalated:false must not park a user task",
  );
});

test("a not-landable gate verdict gives a draft PR an actionable 'mark it ready' question (#454)", async () => {
  const engine = await boot();
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "draft" } });
  await assertThatUserTask(engine, { instance: byProcessId("merge-loop"), elementId: "wait-merge-answer" }).isCreated();
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompletedElements("merge-esc-conflict");
  assertStringIncludes(String(escalation(engine).question ?? ""), "draft", "a draft PR must get a mark-it-ready question");
});

// ---------------------------------------------------------------------------
// Terminate semantics — the refactor kept MergeAbandoned at the root
// ---------------------------------------------------------------------------

test("an abandoned merge ends the whole instance via the root terminate, without merging", async () => {
  const engine = await boot({ responses: { "pr.merge": { mergeStatus: "abandoned" } } });
  await engine.publishMessage({ name: "deps-cleared", correlationKey: "pr-1" });
  await engine.publishMessage({ name: "merge-ready", correlationKey: "pr-1", variables: { mergeState: "ready" } });
  // `MergeAbandoned` is a terminate end event at the ROOT scope: it ends all tokens, so the whole
  // instance finishes (BPMN: a terminate end event *completes* the instance — `Completed`, not a
  // cancellation `Terminated`). Had it lived inside `SP_cifix`/`SP_rebase`, the terminate would end
  // only that sub-process scope and the outer poller would re-arm, leaving the instance ACTIVE —
  // so `hasCompleted()` (nothing left running) is exactly what proves the terminate stayed at root.
  const done = completedElementIds(engine);
  assert(done.has("MergeAbandoned"), "the abandon terminate end event must fire");
  assert(!done.has("mark-merged"), "an abandoned PR must not mark-merged");
  assertThatInstance(engine, byProcessId("merge-loop")).hasCompleted().hasCompletedElements("MergeAbandoned");
  const open = await engine.searchUserTasks({});
  assert(open.length === 0, "the root terminate must leave nothing running (no re-armed poller, no parked escalation)");
});