everyinc__compound-engineering-plugin/tests/ce-babysit-pr-snapshot.test.ts

import { describe, expect, test, beforeEach, setDefaultTimeout } from "bun:test"
import { spawn, spawnSync } from "node:child_process"
import { existsSync, mkdtempSync, writeFileSync, readFileSync, renameSync } from "node:fs"
import { tmpdir } from "node:os"
import path from "node:path"

setDefaultTimeout(30_000)

// Regression tests for the ce-babysit-pr pr-snapshot claim->act->confirm engine.
// Exercised via --fetch-file (no live PR), following the tests/*-validator.test.ts
// spawnSync pattern. Locks in the ce-code-review fixes: crash-safety, needs-human
// silencing + open_needs_human visibility, checks_terminal, key-collision, null-head.
const SCRIPT = path.join(import.meta.dir, "..", "skills", "ce-babysit-pr", "scripts", "pr-snapshot")
const ORDINARY_TEST_BUDGET_SECONDS = "28800"
const EXPIRING_TEST_INVOCATION = ["--start-invocation", "--invocation-budget-seconds", "1"]

function fetchFile(dir: string, name: string, obj: unknown): string {
  const p = path.join(dir, name)
  writeFileSync(p, JSON.stringify(obj))
  return p
}

function persistedInvocationArgs(stateDir: string): string[] {
  if (!existsSync(path.join(stateDir, "state.json"))) return []
  const state = JSON.parse(readFileSync(path.join(stateDir, "state.json"), "utf8"))
  if (!state.invocation_id || !state.started_at || !state.invocation_budget_seconds) return []
  return ["--invocation-id", state.invocation_id, "--session-started-at", state.started_at,
    "--invocation-budget-seconds", String(state.invocation_budget_seconds)]
}

function snapshot(stateDir: string, fetch: string, extra: string[] = []): any {
  const hasInvocationMode = extra.includes("--start-invocation")
    || extra.includes("--reset-session")
    || extra.includes("--continue-invocation")
    || extra.includes("--invocation-id")
  const persistedArgs = !hasInvocationMode ? persistedInvocationArgs(stateDir) : []
  const startsInvocation = (!hasInvocationMode && persistedArgs.length === 0)
    || extra.includes("--start-invocation")
    || extra.includes("--reset-session")
  const budgetArgs = startsInvocation && !extra.includes("--invocation-budget-seconds")
    ? ["--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS]
    : []
  const r = spawnSync(
    "python3",
    [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", stateDir, "--fetch-file", fetch,
      ...(hasInvocationMode ? [] : persistedArgs.length > 0 ? persistedArgs : ["--start-invocation"]),
      ...budgetArgs, ...extra],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout)
}

function currentInvocationArgs(stateDir: string, fetch: string): string[] {
  const persistedArgs = persistedInvocationArgs(stateDir)
  if (persistedArgs.length > 0) return persistedArgs
  const started = snapshot(stateDir, fetch, ["--start-invocation",
    "--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS])
  return ["--invocation-id", started.invocation_id, "--session-started-at", started.invocation_started_at,
    "--invocation-budget-seconds", String(started.invocation_budget_seconds)]
}

function mark(stateDir: string, args: string[]): void {
  // Default the at-mark baseline fetch to empty threads (-> lazy first-observation baseline, no gh
  // call); a test exercising at-mark capture passes its own --fetch-file, which we don't override.
  const extra = args.includes("--fetch-file")
    ? []
    : ["--fetch-file", fetchFile(path.dirname(stateDir), "mark-empty.json", { threads: [] })]
  const r = spawnSync("python3", [SCRIPT, "mark", "--state-dir", stateDir,
    ...persistedInvocationArgs(stateDir), ...args, ...extra], { encoding: "utf8" })
  expect(r.status, r.stderr).toBe(0)
}

function markCurrency(stateDir: string, key: string, disposition: string, fingerprint?: string): void {
  const args = ["--currency-key", key, "--currency-disposition", disposition]
  if (fingerprint) args.push("--semantic-conflict-fingerprint", fingerprint)
  mark(stateDir, args)
}

function markCurrencyOutcome(stateDir: string, key: string, outcome: string): void {
  mark(stateDir, ["--currency-key", key, "--currency-outcome", outcome])
}

function markCurrencyInspection(stateDir: string, key: string, fingerprint: string): void {
  mark(stateDir, ["--currency-key", key, "--currency-inspected-fingerprint", fingerprint])
}

function watch(stateDir: string, fetch: string, extra: string[] = []): any {
  const invocationArgs = extra.includes("--invocation-id") ? [] : currentInvocationArgs(stateDir, fetch)
  const r = spawnSync(
    "python3",
    [SCRIPT, "watch", "--pr", "1", "--repo", "o/r", "--state-dir", stateDir, "--fetch-file", fetch,
      "--interval", "0.1", ...invocationArgs, ...extra],
    { encoding: "utf8", timeout: 5000 },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim().split("\n").pop()!) // the wake sentinel is the final line
}

function startWatch(stateDir: string, fetch: string, extra: string[] = []) {
  const invocationArgs = extra.includes("--invocation-id") ? [] : currentInvocationArgs(stateDir, fetch)
  const child = spawn(
    "python3",
    [SCRIPT, "watch", "--pr", "1", "--repo", "o/r", "--state-dir", stateDir, "--fetch-file", fetch,
      "--interval", "0.05", ...invocationArgs, ...extra],
    { stdio: ["ignore", "pipe", "pipe"] },
  )
  let stdout = ""
  let stderr = ""
  child.stdout.on("data", (chunk) => { stdout += chunk })
  child.stderr.on("data", (chunk) => { stderr += chunk })
  const result = new Promise<{ code: number | null; stdout: string; stderr: string }>((resolve) => {
    child.on("close", (code) => resolve({ code, stdout, stderr }))
  })
  return { child, result }
}

async function waitForWatchGeneration(stateDir: string, previous: string | null = null): Promise<string> {
  const deadline = Date.now() + 5000
  while (Date.now() < deadline) {
    try {
      const generation = JSON.parse(readFileSync(path.join(stateDir, "state.json"), "utf8")).watch_generation
      if (typeof generation === "string" && generation !== previous) return generation
    } catch {
      // The first watcher may still be creating state.json.
    }
    await Bun.sleep(20)
  }
  throw new Error(`watch generation did not advance from ${previous}`)
}

function wakeReason(snapshotValue: unknown, settleSeconds = 0): string | null {
  const r = spawnSync(
    "python3",
    [
      "-c",
      `import json; from importlib.machinery import SourceFileLoader; ` +
        `m=SourceFileLoader('prs', ${JSON.stringify(SCRIPT)}).load_module(); ` +
        `print(json.dumps(m._wake_reason(json.loads(${JSON.stringify(JSON.stringify(snapshotValue))}), ${settleSeconds})))`,
    ],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

function extractFeedback(view: unknown): any[] {
  const r = spawnSync(
    "python3",
    [
      "-c",
      `import json; from importlib.machinery import SourceFileLoader; ` +
        `m=SourceFileLoader('prs', ${JSON.stringify(SCRIPT)}).load_module(); ` +
        `print(json.dumps(m._extract_feedback(json.loads(${JSON.stringify(JSON.stringify(view))}))))`,
    ],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

function eyesReactionIdentities(pages: unknown): string[] {
  const r = spawnSync(
    "python3",
    [
      "-c",
      `import json; from importlib.machinery import SourceFileLoader; ` +
        `m=SourceFileLoader('prs', ${JSON.stringify(SCRIPT)}).load_module(); ` +
        `print(json.dumps(m._eyes_reaction_identities(json.loads(${JSON.stringify(JSON.stringify(pages))}))))`,
    ],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

function probeHostBranchUpdateCapability(response: { status: number; stdout?: unknown }): {
  capability: boolean | "unknown"
  call: string[]
} {
  const payload = JSON.stringify(response)
  const r = spawnSync(
    "python3",
    [
      "-c",
      `import json
from importlib.machinery import SourceFileLoader
m=SourceFileLoader('prs', ${JSON.stringify(SCRIPT)}).load_module()
p=json.loads(${JSON.stringify(payload)})
calls=[]
class Result: pass
def fake(cmd):
    calls.append(cmd)
    result=Result()
    result.returncode=p["status"]
    result.stderr="probe failed" if result.returncode else ""
    value=p.get("stdout")
    result.stdout=value if isinstance(value, str) else json.dumps(value)
    return result
m._run=fake
capability=m.fetch_host_branch_update_capability(7, "o", "r", "ghe.acme.test")
print(json.dumps({"capability": capability, "call": calls[0]}))`,
    ],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

function probeAwaitingApproval(response: { status: number; stdout?: string }): number | null {
  const r = spawnSync(
    "python3",
    [
      "-c",
      `import json
from importlib.machinery import SourceFileLoader
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
class Result: pass
result = Result()
result.returncode = ${JSON.stringify(response.status)}
result.stdout = ${JSON.stringify(response.stdout ?? "")}
m._run = lambda _cmd: result
print(json.dumps(m.fetch_awaiting_approval("o", "r", "head")))`,
    ],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

function probeChain(options: {
  pr?: number
  url?: string
  baseRef?: string
  headRef?: string
  stackView: { status: number; stdout?: unknown; stderr?: string }
  graphql: { status: number; stdout?: unknown; stderr?: string }
  defaultBranch?: { status: number; stdout?: unknown; stderr?: string }
  openPrs?: unknown[]
}): { chain: any; calls: string[] } {
  const payload = {
    pr: options.pr ?? 42,
    url: options.url ?? "https://github.com/o/r/pull/42",
    base_ref: options.baseRef ?? "main",
    head_ref: options.headRef ?? "feature",
    stack_view: options.stackView,
    graphql: options.graphql,
    default_branch: options.defaultBranch ?? { status: 0, stdout: "main\n" },
    open_prs: options.openPrs ?? [],
  }
  const python = `
import json
from importlib.machinery import SourceFileLoader

m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
p = json.loads(${JSON.stringify(JSON.stringify(payload))})
calls = []

class Result:
    pass

def fake(cmd):
    calls.append(" ".join(cmd))
    if cmd[:4] == ["gh", "stack", "view", "--json"]:
        cfg = p["stack_view"]
    elif cmd[:3] == ["gh", "api", "graphql"]:
        cfg = p["graphql"]
    elif cmd[:2] == ["gh", "api"]:
        cfg = p["default_branch"]
    else:
        cfg = {"status": 0, "stdout": p["open_prs"]}
    result = Result()
    result.returncode = cfg["status"]
    value = cfg.get("stdout")
    result.stdout = value if isinstance(value, str) else json.dumps(value)
    result.stderr = cfg.get("stderr", "")
    return result

m._run = fake
chain = m.fetch_pr_chain(p["pr"], "o/r", p["url"], p["base_ref"], p["head_ref"], "o", "r", None)
print(json.dumps({"chain": chain, "calls": calls}))
`
  const r = spawnSync(
    "python3",
    ["-c", python],
    { encoding: "utf8" },
  )
  expect(r.status, r.stderr).toBe(0)
  return JSON.parse(r.stdout.trim())
}

const CODEX_WRAPPER = `
### 💡 Codex Review

Here are some automated review suggestions for this pull request.

**Reviewed commit:** \`50ffb4dd99\`

<details> <summary>ℹ️ About Codex in GitHub</summary>
<br/>

[Your team has set up Codex to review pull requests in this repo](https://chatgpt.com/codex/cloud/settings/general). Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

</details>`

const FAILING = {
  pr_state: "OPEN",
  mergeable: "MERGEABLE",
  merge_state_status: "BLOCKED",
  review_decision: "REVIEW_REQUIRED",
  head_sha: "s1",
  base: {
    host: "github.com",
    repository: "o/r",
    ref: "main",
    oid: "base-1",
    pr_oid: "base-1",
    freshness: "current",
  },
  url: "http://x/1",
  checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }],
  threads: [{ thread_id: "T1", last_comment_id: "C1", last_comment_at: "t1" }],
}

function currencyFixture(overrides: Record<string, unknown> = {}): Record<string, unknown> {
  const fixture = {
    ...FAILING,
    mergeable: "MERGEABLE",
    merge_state_status: "BEHIND",
    base: {
      host: "github.com",
      repository: "o/r",
      ref: "main",
      oid: "base-1",
      pr_oid: "base-1",
      freshness: "current",
    },
    host_branch_update_capability: true,
    pr_chain: {
      manager_status: "absent",
      manager_source: null,
      relationship_status: "independent",
      default_branch: "main",
      parent_prs: [],
      dependent_prs: [],
    },
    ...overrides,
  }
  if (overrides.base && typeof overrides.base === "object") {
    const base = overrides.base as Record<string, unknown>
    fixture.base = {
      pr_oid: base.oid,
      freshness: "current",
      ...base,
    } as typeof fixture.base
  }
  return fixture
}

function quietCurrencyFixture(overrides: Record<string, unknown> = {}): Record<string, unknown> {
  return currencyFixture({
    review_decision: "APPROVED",
    checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
    threads: [],
    ...overrides,
  })
}

describe("ce-babysit-pr pr-snapshot engine", () => {
  let dir: string
  let state: string
  beforeEach(() => {
    dir = mkdtempSync(path.join(tmpdir(), "prsnap-"))
    state = path.join(dir, "state")
  })

  test("branch currency: complete remote base identity creates one stable normal-base observation", () => {
    const fetch = fetchFile(dir, "currency-stable.json", currencyFixture())
    const first = snapshot(state, fetch)
    const second = snapshot(state, fetch)

    expect(first.branch_currency).toMatchObject({
      disposition: "open",
      route: "normal-base",
      host: "github.com",
      base_repository: "o/r",
      base_ref: "main",
      base_oid: "base-1",
      head_sha: "s1",
      status: "BEHIND",
      host_branch_update_capability: true,
    })
    expect(second.branch_currency.key).toBe(first.branch_currency.key)
    expect(second.branch_currency.disposition).toBe("open")
    expect(second.mergeability_certain).toBe(true)
    expect(second.branch_currency_blocker).toEqual({
      key: second.branch_currency.key,
      disposition: "open",
      recovery_state: null,
    })
  })

  test("branch currency: an old state file gains safe defaults without consuming the unseen item", () => {
    const fetch = fetchFile(dir, "currency-migration.json", currencyFixture())
    snapshot(state, fetch)
    const statePath = path.join(state, "state.json")
    const legacy = JSON.parse(readFileSync(statePath, "utf8"))
    delete legacy.branch_currency_state
    writeFileSync(statePath, JSON.stringify(legacy))

    const migrated = snapshot(state, fetch)
    expect(migrated.branch_currency.disposition).toBe("open")
    const persisted = JSON.parse(readFileSync(statePath, "utf8"))
    expect(persisted.branch_currency_state.current_key).toBe(migrated.branch_currency.key)
  })

  test("branch currency: UNKNOWN mergeability re-polls without creating or consuming an item", () => {
    const uncertain = currencyFixture({
      mergeable: "UNKNOWN",
      merge_state_status: "UNKNOWN",
    })
    const value = snapshot(state, fetchFile(dir, "currency-unknown.json", uncertain))
    expect(value.mergeability_certain).toBe(false)
    expect(value.branch_currency).toBeNull()
    expect(value.branch_currency_blocker).toBeNull()
    const persisted = JSON.parse(readFileSync(path.join(state, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.current_key).toBeNull()
    expect(persisted.branch_currency_state.items).toEqual({})
  })

  test("branch currency: wake precedence favors review and failing CI, while passive checks do not delay maintenance", () => {
    const open = snapshot(state, fetchFile(dir, "currency-wake.json", quietCurrencyFixture({
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
      review_in_progress: true,
    })))
    expect(open.branch_currency.attention).toBe("claim")
    expect(wakeReason(open)).toBe("branch-currency")
    expect(wakeReason({ ...open, counts: { ...open.counts, threads: 1 } })).toBe("actionable")
    expect(wakeReason({
      ...open,
      open_needs_human: 1,
      needs_human_ids: ["parked-review-decision"],
    })).toBe("branch-currency")
    expect(wakeReason({ ...open, blocked_external: true })).toBe("branch-currency")
    expect(wakeReason({
      ...open,
      counts: { ...open.counts, ci: 0 },
      has_failing_checks: true,
      checks_terminal: true,
    })).toBe("blocked-failing")
  }, 15000)

  test("branch currency: claim re-entry reconciles, permits one proven-no-mutation retry, and never retries ambiguity", () => {
    const fetch = fetchFile(dir, "currency-lifecycle.json", quietCurrencyFixture())
    const first = snapshot(state, fetch)
    markCurrency(state, first.branch_currency.key, "claimed")

    const sameInvocation = snapshot(state, fetch)
    expect(sameInvocation.branch_currency).toMatchObject({
      disposition: "claimed",
      attention: null,
      retry_count: 0,
      mutation_consumed: false,
    })

    const resumed = snapshot(state, fetch, ["--start-invocation",
      "--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS])
    expect(resumed.branch_currency.attention).toBe("reconcile")
    expect(wakeReason(resumed)).toBe("branch-currency")

    markCurrencyOutcome(state, resumed.branch_currency.key, "proven-no-mutation")
    const backingOff = snapshot(state, fetch)
    expect(backingOff.branch_currency).toMatchObject({
      disposition: "open",
      attention: null,
      retry_count: 1,
      recovery_state: "retry-authorized",
    })
    expect(backingOff.branch_currency.retry_wait_seconds).toBeGreaterThan(0)
    const earlyRetry = spawnSync("python3", [SCRIPT, "mark", "--state-dir", state,
      ...persistedInvocationArgs(state), "--currency-key", backingOff.branch_currency.key,
      "--currency-disposition", "claimed"], { encoding: "utf8" })
    expect(earlyRetry.status).not.toBe(0)

    const statePath = path.join(state, "state.json")
    const afterBackoff = JSON.parse(readFileSync(statePath, "utf8"))
    afterBackoff.branch_currency_state.items[backingOff.branch_currency.key].retry_not_before = "2000-01-01T00:00:00Z"
    writeFileSync(statePath, JSON.stringify(afterBackoff))
    const retry = snapshot(state, fetch)
    expect(retry.branch_currency.attention).toBe("claim")

    markCurrency(state, retry.branch_currency.key, "claimed")
    markCurrencyOutcome(state, retry.branch_currency.key, "proven-no-mutation")
    const exhausted = snapshot(state, fetch)
    expect(exhausted.branch_currency).toMatchObject({
      disposition: "needs-human",
      attention: null,
      retry_count: 1,
      recovery_state: "retry-exhausted",
    })

    const ambiguousState = path.join(dir, "currency-ambiguous")
    const ambiguous = snapshot(ambiguousState, fetch)
    markCurrency(ambiguousState, ambiguous.branch_currency.key, "claimed")
    markCurrencyOutcome(ambiguousState, ambiguous.branch_currency.key, "ambiguous")
    expect(snapshot(ambiguousState, fetch).branch_currency).toMatchObject({
      disposition: "claimed",
      attention: null,
      recovery_state: "ambiguous",
      mutation_consumed: false,
    })
    const ambiguousResume = snapshot(ambiguousState, fetch, ["--start-invocation",
      "--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS])
    expect(ambiguousResume.branch_currency.attention).toBe("reconcile")
    markCurrency(ambiguousState, ambiguous.branch_currency.key, "claimed")
    const persisted = JSON.parse(readFileSync(path.join(ambiguousState, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.items[ambiguous.branch_currency.key].recovery_state).toBe("ambiguous")
  }, 20000)

  test("branch currency: mutation observation consumes the attempt and remains reconciliation-only", () => {
    const fetch = fetchFile(dir, "currency-consumed.json", quietCurrencyFixture())
    const observed = snapshot(state, fetch)
    markCurrency(state, observed.branch_currency.key, "claimed")
    markCurrencyOutcome(state, observed.branch_currency.key, "mutation-observed")
    const current = snapshot(state, fetch)
    expect(current.branch_currency).toMatchObject({
      disposition: "claimed",
      attention: null,
      mutation_consumed: true,
      recovery_state: "mutation-observed",
    })
    expect(wakeReason({
      ...current,
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
      checks_terminal: true,
      has_failing_checks: false,
      review_in_progress: false,
      quiet_seconds: 2000,
    }, 0)).toBeNull()

    const invalidRetry = spawnSync("python3", [SCRIPT, "mark", "--state-dir", state,
      ...persistedInvocationArgs(state), "--currency-key", observed.branch_currency.key,
      "--currency-outcome", "proven-no-mutation"], { encoding: "utf8" })
    expect(invalidRetry.status).not.toBe(0)

    const directReopen = spawnSync("python3", [SCRIPT, "mark", "--state-dir", state,
      ...persistedInvocationArgs(state), "--currency-key", observed.branch_currency.key,
      "--currency-disposition", "open"], { encoding: "utf8" })
    expect(directReopen.status).not.toBe(0)
    expect(snapshot(state, fetch).branch_currency).toMatchObject({
      disposition: "claimed",
      mutation_consumed: true,
      recovery_state: "mutation-observed",
    })

    const resumed = snapshot(state, fetch, ["--start-invocation",
      "--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS])
    expect(resumed.branch_currency.attention).toBe("reconcile")
  }, 20000)

  test("branch currency: async evidence movement re-wakes a same-invocation mutation claim once", () => {
    const behind = fetchFile(dir, "currency-async-mutation-behind.json", quietCurrencyFixture())
    const observed = snapshot(state, behind)
    markCurrency(state, observed.branch_currency.key, "claimed")
    markCurrencyOutcome(state, observed.branch_currency.key, "mutation-observed")

    const unchanged = snapshot(state, behind)
    expect(unchanged.branch_currency).toMatchObject({
      key: observed.branch_currency.key,
      disposition: "claimed",
      attention: null,
      reconciliation_only: true,
    })

    const updated = fetchFile(dir, "currency-async-mutation-clean.json", quietCurrencyFixture({
      head_sha: "s2",
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
    }))
    const moved = snapshot(state, updated)
    expect(moved.branch_currency).toMatchObject({
      key: observed.branch_currency.key,
      disposition: "claimed",
      attention: "reconcile",
      reconciliation_only: true,
      mutation_consumed: true,
    })
    expect(wakeReason(moved)).toBe("branch-currency")

    markCurrency(state, observed.branch_currency.key, "confirmed")
    const confirmed = snapshot(state, updated)
    expect(confirmed.branch_currency).toBeNull()
    expect(confirmed.branch_currency_blocker).toBeNull()
    expect(wakeReason({ ...confirmed, quiet_seconds: 0 }, 300)).toBeNull()
  }, 15000)

  test("branch currency: async evidence movement also re-wakes an ambiguous same-invocation claim", () => {
    const behind = fetchFile(dir, "currency-async-ambiguous-behind.json", quietCurrencyFixture())
    const observed = snapshot(state, behind)
    markCurrency(state, observed.branch_currency.key, "claimed")
    markCurrencyOutcome(state, observed.branch_currency.key, "ambiguous")

    expect(snapshot(state, behind).branch_currency.attention).toBeNull()

    const moved = snapshot(state, fetchFile(dir, "currency-async-ambiguous-clean.json", quietCurrencyFixture({
      head_sha: "s2",
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
    })))
    expect(moved.branch_currency).toMatchObject({
      key: observed.branch_currency.key,
      disposition: "claimed",
      attention: "reconcile",
      recovery_state: "ambiguous",
      reconciliation_only: true,
    })
    expect(wakeReason(moved)).toBe("branch-currency")
  }, 15000)

  test("branch currency: a BEHIND capability park reopens when the same observation becomes updateable", () => {
    for (const capability of [false, "unknown"] as const) {
      const capabilityState = path.join(dir, `currency-capability-${capability}`)
      const unavailable = fetchFile(dir, `currency-capability-${capability}-off.json`, quietCurrencyFixture({
        host_branch_update_capability: capability,
      }))
      const observed = snapshot(capabilityState, unavailable)
      markCurrency(capabilityState, observed.branch_currency.key, "needs-human")
      expect(snapshot(capabilityState, unavailable).branch_currency).toMatchObject({
        disposition: "needs-human",
        attention: null,
      })

      const enabled = snapshot(capabilityState, fetchFile(dir, `currency-capability-${capability}-on.json`, quietCurrencyFixture({
        host_branch_update_capability: true,
      })))
      expect(enabled.branch_currency).toMatchObject({
        key: observed.branch_currency.key,
        disposition: "open",
        attention: "claim",
        host_branch_update_capability: true,
      })
      expect(wakeReason(enabled)).toBe("branch-currency")
    }
  }, 15000)

  test("branch currency: standing confirmed and needs-human residuals stay quiet, and max-runtime outranks new work", () => {
    const fetch = fetchFile(dir, "currency-standing.json", quietCurrencyFixture())
    for (const disposition of ["confirmed", "needs-human"]) {
      const residualState = path.join(dir, `currency-standing-${disposition}`)
      const observed = snapshot(residualState, fetch)
      markCurrency(residualState, observed.branch_currency.key, "claimed")
      markCurrency(residualState, observed.branch_currency.key, disposition,
        disposition === "needs-human" ? "semantic-v1" : undefined)
      const residualPath = path.join(residualState, "state.json")
      const expiredResidual = JSON.parse(readFileSync(residualPath, "utf8"))
      expiredResidual.started_at = "2000-01-01T00:00:00Z"
      expiredResidual.invocation_budget_seconds = 1
      writeFileSync(residualPath, JSON.stringify(expiredResidual))
      expect(watch(residualState, fetch).reason).toBe("max-runtime")
    }

    const budgetState = path.join(dir, "currency-budget")
    snapshot(budgetState, fetch)
    const statePath = path.join(budgetState, "state.json")
    const expired = JSON.parse(readFileSync(statePath, "utf8"))
    expired.started_at = "2000-01-01T00:00:00Z"
    expired.invocation_budget_seconds = 1
    writeFileSync(statePath, JSON.stringify(expired))
    expect(watch(budgetState, fetch).reason).toBe("max-runtime")
    const lateClaim = spawnSync("python3", [SCRIPT, "mark", "--state-dir", budgetState,
      ...persistedInvocationArgs(budgetState), "--currency-key",
      expired.branch_currency_state.current_key, "--currency-disposition", "claimed"],
    { encoding: "utf8" })
    expect(lateClaim.status).not.toBe(0)
  }, 20000)

  // Active-watch-capability budget: the 8h cap is spent in active time, not raw wall-clock.
  // A suspended machine (laptop asleep) is excluded; the 3-day backstop stays wall-clock.
  const FAILING_ACTIONABLE = {
    pr_state: "OPEN", mergeable: "MERGEABLE", merge_state_status: "UNSTABLE", review_decision: null,
    head_sha: "h1", url: "https://github.com/o/r/pull/1", threads: [],
    checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }],
    feedback: [], awaiting_approval: 0,
  }
  const isoAgo = (seconds: number) =>
    new Date(Date.now() - seconds * 1000).toISOString().replace(/\.\d+Z$/, "Z")
  function patchState(stateDir: string, patch: Record<string, unknown>): void {
    const p = path.join(stateDir, "state.json")
    writeFileSync(p, JSON.stringify({ ...JSON.parse(readFileSync(p, "utf8")), ...patch }))
  }
  function readState(stateDir: string): any {
    return JSON.parse(readFileSync(path.join(stateDir, "state.json"), "utf8"))
  }

  test("active-time budget: a suspended span (stale activity heartbeat) is excluded from the 8h cap", () => {
    // Covers AE1. started_at and last_activity both ~6h stale (machine was suspended), budget 8h.
    const fetch = fetchFile(dir, "active-suspend.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { started_at: isoAgo(6 * 3600), last_activity_at: isoAgo(6 * 3600),
      dead_time_seconds: 0 })
    // The watch arms, measures the ~6h activity gap, charges it to dead time, and does NOT max-runtime.
    expect(watch(state, fetch).reason).toBe("actionable")
    const after = readState(state)
    // ~6h minus the 15-min threshold is charged to dead time (excluded from the active budget).
    expect(after.dead_time_seconds).toBeGreaterThan(6 * 3600 - 15 * 60 - 60)
    expect(after.dead_time_seconds).toBeLessThan(6 * 3600)
  }, 20000)

  test("active-time budget: steady sub-threshold polling accrues no dead time", () => {
    // Covers AE2. A recent heartbeat (well under the 15-min threshold) never registers as suspend.
    const fetch = fetchFile(dir, "active-steady.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { started_at: isoAgo(2 * 3600), last_activity_at: isoAgo(30),
      dead_time_seconds: 0 })
    expect(watch(state, fetch).reason).toBe("actionable")
    expect(readState(state).dead_time_seconds).toBe(0)
  }, 20000)

  test("3-day backstop: raw wall-clock expiry fires even when active elapsed is ~0", () => {
    // Covers AE4. started_at 4 days ago, dead_time ~4 days (active ~0), heartbeat fresh so the poll
    // adds nothing. The 8h active cap is nowhere near hit, but the wall-clock backstop terminates.
    const fetch = fetchFile(dir, "backstop.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { started_at: isoAgo(4 * 86400), last_activity_at: isoAgo(10),
      dead_time_seconds: 4 * 86400 })
    const wake = watch(state, fetch)
    expect(wake.reason).toBe("max-runtime")
    expect(wake.max_runtime_ceiling).toBe("backstop")
  }, 20000)

  test("active cap: active elapsed past the 8h budget fires max-runtime with the active-budget ceiling", () => {
    // started_at 9h ago, no dead time, wall-clock < 3-day backstop -> the active budget is the ceiling.
    const fetch = fetchFile(dir, "active-cap.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { started_at: isoAgo(9 * 3600), last_activity_at: isoAgo(10),
      dead_time_seconds: 0 })
    const wake = watch(state, fetch)
    expect(wake.reason).toBe("max-runtime")
    expect(wake.max_runtime_ceiling).toBe("active-budget")
  }, 20000)

  test("legacy state without active-time fields migrates on load", () => {
    // Covers U1. A pre-existing state file lacking the new fields gains them with safe defaults.
    const fetch = fetchFile(dir, "migrate.json", quietCurrencyFixture())
    snapshot(state, fetch)
    const legacy = readState(state)
    delete legacy.last_activity_at
    delete legacy.dead_time_seconds
    delete legacy.invocation_backstop_seconds
    writeFileSync(path.join(state, "state.json"), JSON.stringify(legacy))
    snapshot(state, fetch)
    const migrated = readState(state)
    expect(migrated.dead_time_seconds).toBe(0)
    expect(migrated.last_activity_at).toBeTruthy()
    expect(migrated.invocation_backstop_seconds).toBe(3 * 24 * 60 * 60)
  }, 20000)

  test("legacy migration does not refund pre-migration time: an expired old invocation still max-runtimes", () => {
    // Regression (Cursor/Codex): seeding last_activity to the OLD started_at made the first poll
    // charge the whole historical invocation as one suspend gap, so a 9h-old 8h run read as ~15 min
    // active and never expired. Seeding to load time keeps it on wall-clock -> it must still expire.
    const fetch = fetchFile(dir, "legacy-expire.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    const legacy = readState(state)
    legacy.started_at = isoAgo(9 * 3600)
    legacy.invocation_budget_seconds = 8 * 3600
    delete legacy.last_activity_at
    delete legacy.dead_time_seconds
    delete legacy.invocation_backstop_seconds
    writeFileSync(path.join(state, "state.json"), JSON.stringify(legacy))
    const wake = watch(state, fetch)
    expect(wake.reason).toBe("max-runtime")
    expect(wake.max_runtime_ceiling).toBe("active-budget")
    // The historical span was not laundered into dead time.
    expect(readState(state).dead_time_seconds).toBe(0)
  }, 20000)

  test("re-arm preserves accumulated dead time and the backstop (no reset, no extend)", () => {
    // Covers AE5. A continue-invocation re-arm keeps the accumulated dead time rather than resetting.
    const fetch = fetchFile(dir, "rearm.json", quietCurrencyFixture())
    snapshot(state, fetch)
    patchState(state, { dead_time_seconds: 1234, last_activity_at: isoAgo(10) })
    const inv = persistedInvocationArgs(state)
    const r = spawnSync("python3", [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r",
      "--state-dir", state, "--fetch-file", fetch, "--continue-invocation", ...inv],
      { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    const after = readState(state)
    expect(after.dead_time_seconds).toBe(1234)
    expect(after.invocation_backstop_seconds).toBe(3 * 24 * 60 * 60)
  }, 20000)

  test("checkpoint mode: an agent snapshot with a stale heartbeat never accumulates dead time", () => {
    // KTD4 scope guard: only the in-session watch (watch_generation) accumulates. A plain agent
    // snapshot bumps the heartbeat with accumulate=False, so checkpoint/durable runs stay wall-clock.
    const fetch = fetchFile(dir, "checkpoint.json", quietCurrencyFixture())
    snapshot(state, fetch)
    patchState(state, { started_at: isoAgo(6 * 3600), last_activity_at: isoAgo(6 * 3600),
      dead_time_seconds: 0 })
    const observed = snapshot(state, fetch) // plain (non-watch) agent snapshot
    expect(readState(state).dead_time_seconds).toBe(0)
    // Wall-clock retained: elapsed reflects the full ~6h with nothing refunded.
    expect(observed.invocation_elapsed_seconds).toBeGreaterThan(6 * 3600 - 120)
  }, 20000)

  test("an agent mark bumps the activity heartbeat without accumulating dead time", () => {
    // A long tick that only marks keeps the heartbeat fresh, so the next watch poll charges nothing.
    const fetch = fetchFile(dir, "markbump.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { last_activity_at: isoAgo(6 * 3600), dead_time_seconds: 0 })
    mark(state, ["--check", "CI/test"])
    const after = readState(state)
    expect(after.dead_time_seconds).toBe(0)
    expect(new Date(after.last_activity_at).getTime()).toBeGreaterThan(Date.now() - 60 * 1000)
  }, 20000)

  test("clock-backward safety: a future heartbeat/anchor never produces negative accounting", () => {
    const fetch = fetchFile(dir, "clockback.json", FAILING_ACTIONABLE)
    snapshot(state, fetch)
    patchState(state, { started_at: new Date(Date.now() + 3600 * 1000).toISOString(),
      last_activity_at: new Date(Date.now() + 3600 * 1000).toISOString(), dead_time_seconds: 0 })
    const wake = watch(state, fetch)
    expect(wake.reason).not.toBe("max-runtime")
    const after = readState(state)
    expect(after.dead_time_seconds).toBeGreaterThanOrEqual(0)
  }, 20000)

  test("continue-invocation adopting a new id into a used state dir resets the active-time clock", () => {
    // Regression: adopting a fresh invocation must not inherit a prior invocation's dead time.
    const fetch = fetchFile(dir, "adopt.json", quietCurrencyFixture())
    snapshot(state, fetch)
    patchState(state, { dead_time_seconds: 5000, last_activity_at: isoAgo(6 * 3600) })
    const prior = readState(state)
    const r = spawnSync("python3", [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r",
      "--state-dir", state, "--fetch-file", fetch, "--continue-invocation",
      "--invocation-id", "adopted-new-id", "--session-started-at", prior.started_at,
      "--invocation-budget-seconds", String(prior.invocation_budget_seconds)],
      { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    const after = readState(state)
    expect(after.invocation_id).toBe("adopted-new-id")
    expect(after.dead_time_seconds).toBe(0)
  }, 20000)

  test("managed-stack continuation carries accumulated dead time to the next layer's state dir", () => {
    // Codex P2: the shared active-time budget spans stack layers, but dead time is per-state-dir.
    // --continue-dead-time-seconds threads the prior layer's excluded-suspend total into the new
    // layer so it is not re-counted as active. Absent the arg, an adopt still resets to 0.
    const fetch = fetchFile(dir, "stack-carry.json", quietCurrencyFixture())
    snapshot(state, fetch)
    const prior = readState(state)
    const cont = (stateDir: string, extra: string[]) => spawnSync("python3",
      [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", stateDir, "--fetch-file", fetch,
        "--continue-invocation", "--invocation-id", prior.invocation_id,
        "--session-started-at", prior.started_at,
        "--invocation-budget-seconds", String(prior.invocation_budget_seconds), ...extra],
      { encoding: "utf8" })
    const layer2 = path.join(dir, "layer2")
    const r = cont(layer2, ["--continue-dead-time-seconds", "4200"])
    expect(r.status, r.stderr).toBe(0)
    expect(readState(layer2).dead_time_seconds).toBe(4200)
    const layer3 = path.join(dir, "layer3")
    const r2 = cont(layer3, [])
    expect(r2.status, r2.stderr).toBe(0)
    expect(readState(layer3).dead_time_seconds).toBe(0)
  }, 20000)

  test("carry arg on a same-id re-continue raises the dead-time floor without clobbering accumulation", () => {
    // Bugbot: the carry must be honored on a later same-id re-continue (early-return path), not only
    // the first adopt. It sets a monotonic floor — raises 0 to the carry, never lowers a larger value.
    const fetch = fetchFile(dir, "recont.json", quietCurrencyFixture())
    snapshot(state, fetch)
    const prior = readState(state)
    const layer = path.join(dir, "recont-layer")
    const cont = (extra: string[]) => spawnSync("python3",
      [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", layer, "--fetch-file", fetch,
        "--continue-invocation", "--invocation-id", prior.invocation_id,
        "--session-started-at", prior.started_at,
        "--invocation-budget-seconds", String(prior.invocation_budget_seconds), ...extra],
      { encoding: "utf8" })
    expect(cont([]).status).toBe(0) // first continue (adopt), no carry -> 0
    expect(readState(layer).dead_time_seconds).toBe(0)
    expect(cont(["--continue-dead-time-seconds", "3000"]).status).toBe(0) // re-continue raises floor
    expect(readState(layer).dead_time_seconds).toBe(3000)
    expect(cont(["--continue-dead-time-seconds", "1000"]).status).toBe(0) // lower carry never clobbers
    expect(readState(layer).dead_time_seconds).toBe(3000)
  }, 20000)

  test("branch currency: a carried semantic park wakes only for inspection and unchanged evidence stays parked", () => {
    const dirty = quietCurrencyFixture({ mergeable: "CONFLICTING", merge_state_status: "DIRTY" })
    const original = snapshot(state, fetchFile(dir, "currency-inspect-1.json", dirty))
    markCurrency(state, original.branch_currency.key, "claimed")
    markCurrency(state, original.branch_currency.key, "needs-human", "conflict-v1")

    const moved = snapshot(state, fetchFile(dir, "currency-inspect-2.json", quietCurrencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(moved.branch_currency.attention).toBe("inspect")
    expect(wakeReason(moved)).toBe("branch-currency")

    const prematureClaim = spawnSync("python3", [SCRIPT, "mark", "--state-dir", state,
      ...persistedInvocationArgs(state), "--currency-key", moved.branch_currency.key,
      "--currency-disposition", "claimed"], { encoding: "utf8" })
    expect(prematureClaim.status).not.toBe(0)

    markCurrencyInspection(state, moved.branch_currency.key, "conflict-v1")
    expect(snapshot(state, fetchFile(dir, "currency-inspect-same.json", quietCurrencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    }))).branch_currency).toMatchObject({
      disposition: "needs-human",
      attention: null,
      recovery_state: "semantic-unchanged",
    })

    const changed = snapshot(state, fetchFile(dir, "currency-inspect-3.json", quietCurrencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-3" },
    })))
    expect(changed.branch_currency.attention).toBe("inspect")
    markCurrencyInspection(state, changed.branch_currency.key, "conflict-v2")
    expect(snapshot(state, fetchFile(dir, "currency-inspect-changed.json", quietCurrencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-3" },
    }))).branch_currency).toMatchObject({
      disposition: "open",
      attention: "claim",
      inspection_result: "changed",
    })
    const persisted = JSON.parse(readFileSync(path.join(state, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.semantic_parks).toEqual({})

    const nextBase = snapshot(state, fetchFile(dir, "currency-inspect-4.json", quietCurrencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-4" },
    })))
    expect(nextBase.branch_currency).toMatchObject({
      disposition: "open",
      attention: "claim",
      parked_semantic_fingerprints: [],
    })
  }, 20000)

  test("branch currency: unresolved claims survive base and expected head movement until reconciled", () => {
    const first = snapshot(state, fetchFile(dir, "currency-key-1.json", currencyFixture()))
    markCurrency(state, first.branch_currency.key, "claimed")
    expect(snapshot(state, fetchFile(dir, "currency-key-1b.json", currencyFixture())).branch_currency.disposition).toBe("claimed")

    const movedWhileClaimed = snapshot(state, fetchFile(dir, "currency-key-2.json", currencyFixture({
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(movedWhileClaimed.branch_currency).toMatchObject({
      key: first.branch_currency.key,
      disposition: "claimed",
      reconciliation_only: true,
    })
    expect(movedWhileClaimed.branch_currency_blocker.key).toBe(first.branch_currency.key)

    const resumed = snapshot(state, fetchFile(dir, "currency-key-2-resumed.json", currencyFixture({
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })), ["--start-invocation", "--invocation-budget-seconds", ORDINARY_TEST_BUDGET_SECONDS])
    expect(resumed.branch_currency.attention).toBe("reconcile")
    markCurrencyOutcome(state, first.branch_currency.key, "proven-no-mutation")
    const statePath = path.join(state, "state.json")
    const retryState = JSON.parse(readFileSync(statePath, "utf8"))
    retryState.branch_currency_state.items[first.branch_currency.key].retry_not_before = "2000-01-01T00:00:00Z"
    writeFileSync(statePath, JSON.stringify(retryState))

    const movedBase = snapshot(state, fetchFile(dir, "currency-key-2-open.json", currencyFixture({
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(movedBase.branch_currency.key).not.toBe(first.branch_currency.key)
    expect(movedBase.branch_currency.disposition).toBe("open")
    markCurrency(state, movedBase.branch_currency.key, "claimed")
    markCurrencyOutcome(state, movedBase.branch_currency.key, "mutation-observed")

    const movedHead = snapshot(state, fetchFile(dir, "currency-key-3.json", currencyFixture({
      head_sha: "s2",
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(movedHead.branch_currency).toMatchObject({
      key: movedBase.branch_currency.key,
      disposition: "claimed",
      mutation_consumed: true,
      reconciliation_only: true,
    })
    expect(movedHead.branch_currency_blocker.key).toBe(movedBase.branch_currency.key)
    markCurrency(state, movedBase.branch_currency.key, "confirmed")
    const confirmed = snapshot(state, fetchFile(dir, "currency-key-3-confirmed.json", currencyFixture({
      head_sha: "s2",
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(confirmed.branch_currency).toBeNull()
    expect(confirmed.branch_currency_blocker).toBeNull()
    const persisted = JSON.parse(readFileSync(path.join(state, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.current_key).toBeNull()
  }, 20000)

  test("branch currency: invocation-fenced transitions preserve an unchanged semantic park", () => {
    const observed = snapshot(state, fetchFile(dir, "currency-park.json", currencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
    })))
    markCurrency(state, observed.branch_currency.key, "claimed")
    markCurrency(state, observed.branch_currency.key, "needs-human", "conflict-v1")
    const parked = snapshot(state, fetchFile(dir, "currency-park-again.json", currencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
    })))
    expect(parked.branch_currency).toMatchObject({
      key: observed.branch_currency.key,
      disposition: "needs-human",
      semantic_conflict_fingerprint: "conflict-v1",
    })

    const stale = spawnSync("python3", [SCRIPT, "mark", "--state-dir", state,
      "--invocation-id", "stale", "--session-started-at", parked.invocation_started_at,
      "--invocation-budget-seconds", String(parked.invocation_budget_seconds),
      "--currency-key", observed.branch_currency.key, "--currency-disposition", "open"],
    { encoding: "utf8" })
    expect(stale.status).not.toBe(0)
    const stillParked = snapshot(state, fetchFile(dir, "currency-park-still.json", currencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
    })))
    expect(stillParked.branch_currency.disposition).toBe("needs-human")
    markCurrency(state, stillParked.branch_currency.key, "open")
    expect(snapshot(state, fetchFile(dir, "currency-park-reopened.json", currencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
    }))).branch_currency.disposition).toBe("open")
  }, 15000)

  test("branch currency: base-only movement retains the semantic park for later inspection", () => {
    const dirty = currencyFixture({ mergeable: "CONFLICTING", merge_state_status: "DIRTY" })
    const parkedObservation = snapshot(state, fetchFile(dir, "currency-park-base-1.json", dirty))
    markCurrency(state, parkedObservation.branch_currency.key, "claimed")
    markCurrency(state, parkedObservation.branch_currency.key, "needs-human", "conflict-v1")

    const moved = snapshot(state, fetchFile(dir, "currency-park-base-2.json", currencyFixture({
      mergeable: "CONFLICTING",
      merge_state_status: "DIRTY",
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(moved.branch_currency.key).not.toBe(parkedObservation.branch_currency.key)
    expect(moved.branch_currency.disposition).toBe("open")
    expect(moved.branch_currency.parked_semantic_fingerprints).toEqual(["conflict-v1"])

    const persisted = JSON.parse(readFileSync(path.join(state, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.semantic_parks["conflict-v1"]).toMatchObject({
      head_sha: "s1",
      status: "DIRTY",
      route: "normal-base",
      observation_key: parkedObservation.branch_currency.key,
    })
    expect(persisted.branch_currency_state.semantic_parks["conflict-v1"]).not.toHaveProperty("base_oid")
    expect(persisted.branch_currency_state.items[parkedObservation.branch_currency.key].disposition).toBe("needs-human")
  }, 15000)

  test("branch currency: a carried DIRTY semantic park does not divert a later BEHIND update into inspection", () => {
    const dirty = currencyFixture({ mergeable: "CONFLICTING", merge_state_status: "DIRTY" })
    const parkedObservation = snapshot(state, fetchFile(dir, "currency-dirty-to-behind-1.json", dirty))
    markCurrency(state, parkedObservation.branch_currency.key, "claimed")
    markCurrency(state, parkedObservation.branch_currency.key, "needs-human", "conflict-v1")

    const behind = snapshot(state, fetchFile(dir, "currency-dirty-to-behind-2.json", currencyFixture({
      base: { host: "github.com", repository: "o/r", ref: "main", oid: "base-2" },
    })))
    expect(behind.branch_currency).toMatchObject({
      status: "BEHIND",
      disposition: "open",
      attention: "claim",
      inspection_required: false,
      parked_semantic_fingerprints: ["conflict-v1"],
    })

    markCurrency(state, behind.branch_currency.key, "claimed")
    const persisted = JSON.parse(readFileSync(path.join(state, "state.json"), "utf8"))
    expect(persisted.branch_currency_state.semantic_parks).toHaveProperty("conflict-v1")
  }, 15000)

  test("branch currency: dependents do not block a normal-base root, but managed, open-parent, and uncertain routes do", () => {
    const dependentRoot = snapshot(path.join(dir, "currency-dependent-root"), fetchFile(dir, "currency-dependent-root.json", currencyFixture({
      pr_chain: {
        manager_status: "absent",
        relationship_status: "dependent",
        default_branch: "main",
        parent_prs: [],
        dependent_prs: [{ number: 2, state: "OPEN", baseRefName: "feature", headRefName: "child" }],
      },
    })))
    expect(dependentRoot.branch_currency.route).toBe("normal-base")

    const exclusions = [
      { manager_status: "confirmed", relationship_status: "independent", default_branch: "main", parent_prs: [], dependent_prs: [] },
      { manager_status: "absent", relationship_status: "dependent", default_branch: "main", parent_prs: [{ number: 3, state: "OPEN" }], dependent_prs: [] },
      { manager_status: "probe-error", relationship_status: "independent", default_branch: "main", parent_prs: [], dependent_prs: [] },
      { manager_status: "absent", relationship_status: "probe-error", default_branch: "main", parent_prs: [], dependent_prs: [] },
      { manager_status: "absent", relationship_status: "independent", default_branch: null, parent_prs: [], dependent_prs: [] },
    ]
    for (const [index, prChain] of exclusions.entries()) {
      const value = snapshot(path.join(dir, `currency-excluded-${index}`), fetchFile(dir, `currency-excluded-${index}.json`, currencyFixture({ pr_chain: prChain })))
      expect(value.branch_currency).toBeNull()
    }
  }, 15000)

  test("host branch-update capability reads the operation-specific true, false, and unknown outcomes", () => {
    const positive = probeHostBranchUpdateCapability({
      status: 0,
      stdout: { data: { repository: { pullRequest: { viewerCanUpdateBranch: true } } } },
    })
    expect(positive.capability).toBe(true)
    expect(positive.call.join(" ")).toContain("viewerCanUpdateBranch")
    expect(positive.call).toContain("--hostname")
    expect(positive.call).toContain("ghe.acme.test")

    expect(probeHostBranchUpdateCapability({
      status: 0,
      stdout: { data: { repository: { pullRequest: { viewerCanUpdateBranch: false } } } },
    }).capability).toBe(false)
    expect(probeHostBranchUpdateCapability({ status: 1 }).capability).toBe("unknown")
    expect(probeHostBranchUpdateCapability({
      status: 0,
      stdout: { data: { repository: { pullRequest: {} } } },
    }).capability).toBe("unknown")
  })

  test("live fetch requires the host-qualified current base ref before probing an eligible BEHIND update", () => {
    const python = `
import json
from importlib.machinery import SourceFileLoader
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
calls = []
capability_calls = []
current_base_oid = "1111111111111111111111111111111111111111"
ref_probe_status = 0
class Result: pass
def checked(cmd, label):
    calls.append(cmd)
    result = Result()
    result.returncode = 0
    result.stderr = ""
    result.stdout = json.dumps({
        "state": "OPEN", "mergeable": "MERGEABLE", "mergeStateStatus": "BEHIND",
        "reviewDecision": "APPROVED", "headRefOid": "head-1",
        "baseRefOid": "1111111111111111111111111111111111111111",
        "baseRefName": "main", "headRefName": "feature", "number": 7,
        "url": "https://ghe.acme.test/o/r/pull/7", "statusCheckRollup": [],
        "author": {"login": "author"}, "comments": [], "reviews": []})
    return result
def run(cmd):
    calls.append(cmd)
    result = Result()
    result.returncode = ref_probe_status
    result.stderr = "base ref probe failed" if result.returncode else ""
    result.stdout = current_base_oid + "\\n" if result.returncode == 0 else ""
    return result
m._run_checked = checked
m._run = run
m.fetch_eyes_reactors = lambda *args: []
m.fetch_threads = lambda *args: []
m.fetch_awaiting_approval = lambda *args: 0
m.fetch_pr_chain = lambda *args: {"manager_status": "absent", "relationship_status": "independent",
                                  "default_branch": "main", "parent_prs": [], "dependent_prs": []}
def capability(*args):
    capability_calls.append(args)
    return True
m.fetch_host_branch_update_capability = capability
current = m.fetch(7, "ghe.acme.test/o/r")
current_base_oid = "2222222222222222222222222222222222222222"
stale = m.fetch(7, "ghe.acme.test/o/r")
ref_probe_status = 1
probe_error = m.fetch(7, "ghe.acme.test/o/r")
print(json.dumps({"current": current, "stale": stale, "probe_error": probe_error,
                  "calls": calls, "capability_calls": len(capability_calls)}))
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    const result = JSON.parse(r.stdout)
    expect(result.calls[0].join(" ")).toContain("baseRefOid")
    const refCalls = result.calls.filter((call: string[]) => call.includes("repos/o/r/git/ref/heads/main"))
    expect(refCalls).toHaveLength(3)
    for (const call of refCalls) {
      expect(call).toContain("--hostname")
      expect(call).toContain("ghe.acme.test")
    }
    expect(result.current.base).toEqual({
      host: "ghe.acme.test",
      repository: "o/r",
      ref: "main",
      oid: "1111111111111111111111111111111111111111",
      pr_oid: "1111111111111111111111111111111111111111",
      freshness: "current",
    })
    expect(result.current.host_branch_update_capability).toBe(true)
    expect(result.stale.base).toEqual({
      host: "ghe.acme.test",
      repository: "o/r",
      ref: "main",
      oid: "2222222222222222222222222222222222222222",
      pr_oid: "1111111111111111111111111111111111111111",
      freshness: "stale",
    })
    expect(result.stale.host_branch_update_capability).toBe("unknown")
    expect(result.probe_error.base).toEqual({
      host: "ghe.acme.test",
      repository: "o/r",
      ref: "main",
      oid: null,
      pr_oid: "1111111111111111111111111111111111111111",
      freshness: "probe-error",
    })
    expect(result.probe_error.host_branch_update_capability).toBe("unknown")
    expect(result.capability_calls).toBe(1)
  })

  test("base-ref freshness blocks readiness, resets quiet on current-to-stale, and fails closed on probe error", () => {
    const clean = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
    }
    const currentFile = fetchFile(dir, "base-current.json", clean)
    snapshot(state, currentFile)
    const statePath = path.join(state, "state.json")
    const settled = JSON.parse(readFileSync(statePath, "utf8"))
    settled.last_change_at = "2026-07-17T12:00:00+00:00"
    writeFileSync(statePath, JSON.stringify(settled))

    const current = snapshot(state, currentFile)
    expect(current.base_ref_blocker).toBeNull()
    expect(current.mergeability_certain).toBe(true)
    expect(current.changed_this_tick).toBe(false)
    expect(current.quiet_seconds).toBeGreaterThan(60)
    expect(wakeReason(current, 0)).toBe("merge-ready")

    const stale = snapshot(state, fetchFile(dir, "base-stale.json", {
      ...clean,
      base: {
        host: "github.com",
        repository: "o/r",
        ref: "main",
        oid: "base-2",
        pr_oid: "base-1",
        freshness: "stale",
      },
    }))
    expect(stale.base_ref_blocker).toBe("stale")
    expect(stale.mergeability_certain).toBe(false)
    expect(stale.changed_this_tick).toBe(true)
    expect(stale.quiet_seconds).toBeLessThan(2)
    expect(stale.branch_currency).toBeNull()
    expect(wakeReason(stale, 0)).toBe("base-ref-blocked")

    const probeError = snapshot(state, fetchFile(dir, "base-probe-error.json", {
      ...clean,
      base: {
        host: "github.com",
        repository: "o/r",
        ref: "main",
        oid: null,
        pr_oid: "base-1",
        freshness: "probe-error",
      },
    }))
    expect(probeError.base_ref_blocker).toBe("probe-error")
    expect(probeError.mergeability_certain).toBe(false)
    expect(probeError.branch_currency).toBeNull()
    expect(wakeReason(probeError, 0)).toBe("base-ref-blocked")
  })

  test("watch keeps polling an already-surfaced base-ref blocker instead of busy-waking or declaring ready", () => {
    const stale = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      base: {
        host: "github.com",
        repository: "o/r",
        ref: "main",
        oid: "base-2",
        pr_oid: "base-1",
        freshness: "stale",
      },
    }
    const staleFile = fetchFile(dir, "base-stale-standing.json", stale)
    const staleState = path.join(dir, "base-stale-standing")
    snapshot(staleState, staleFile, EXPIRING_TEST_INVOCATION)
    expect(watch(staleState, staleFile, ["--settle-seconds", "0"]).reason).toBe("max-runtime")
  }, 15000)

  test("first snapshot: thread + failing check are actionable; checks terminal", () => {
    const d = snapshot(state, fetchFile(dir, "a.json", FAILING))
    expect(d.counts.threads).toBe(1)
    expect(d.counts.ci).toBe(1)
    expect(d.has_failing_checks).toBe(true)
    expect(d.checks_terminal).toBe(true)
  })

  test("crash-safety: un-marked items stay actionable on the next tick", () => {
    const f = fetchFile(dir, "a.json", FAILING)
    const first = snapshot(state, f)
    const second = snapshot(state, f)
    expect(second.counts.threads).toBe(first.counts.threads)
    expect(second.counts.ci).toBe(first.counts.ci)
  })

  test("needs-human thread: silenced despite the resolver's own reply moving identity, but stays visible via open_needs_human", () => {
    snapshot(state, fetchFile(dir, "a.json", FAILING))
    mark(state, ["--thread", "T1", "--disposition", "needs-human"])
    // The resolver posts decision_context, moving the thread's last-comment identity.
    const replied = { ...FAILING, threads: [{ thread_id: "T1", last_comment_id: "C2", last_comment_at: "t2" }] }
    const d = snapshot(state, fetchFile(dir, "b.json", replied))
    expect(d.counts.threads).toBe(0) // no re-actionize (the P1 fix)
    expect(d.open_needs_human).toBe(1) // still blocks merge-ready
  })

  test("mark --check silences it; a new head SHA re-actionizes", () => {
    const f = fetchFile(dir, "a.json", FAILING)
    snapshot(state, f)
    mark(state, ["--check", "CI/test"])
    expect(snapshot(state, f).counts.ci).toBe(0)
    const newHead = { ...FAILING, head_sha: "s2" }
    expect(snapshot(state, fetchFile(dir, "c.json", newHead)).counts.ci).toBe(1)
  })

  test("checks_terminal is false while a check is IN_PROGRESS; all_checks_ok stays false", () => {
    const inprog = {
      ...FAILING,
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const d = snapshot(state, fetchFile(dir, "ip.json", inprog))
    expect(d.checks_terminal).toBe(false)
    expect(d.all_checks_ok).toBe(false)
    expect(d.has_failing_checks).toBe(false)
  })

  test("clean + terminal + approved: all_checks_ok true, mergeStateStatus passthrough, no open needs-human", () => {
    const clean = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
    }
    const d = snapshot(state, fetchFile(dir, "cl.json", clean))
    expect(d.all_checks_ok).toBe(true)
    expect(d.checks_terminal).toBe(true)
    expect(d.merge_state_status).toBe("CLEAN")
    expect(d.open_needs_human).toBe(0)
  })

  test("gh stack view is the first probe and a target match supplies managed freshness without GraphQL", () => {
    const { chain, calls } = probeChain({
      stackView: {
        status: 0,
        stdout: {
          trunk: "main",
          currentBranch: "feature",
          branches: [
            { name: "parent", needsRebase: false, pr: { number: 41, url: "https://github.com/o/r/pull/41", state: "OPEN" } },
            { name: "feature", isCurrent: true, needsRebase: true, pr: { number: 42, url: "https://GITHUB.COM/O/R/pull/42/", state: "OPEN" } },
            { name: "child", needsRebase: false, pr: { number: 43, url: "https://github.com/o/r/pull/43", state: "OPEN", isDraft: true } },
          ],
        },
      },
      graphql: { status: 1, stderr: "must not run" },
    })
    expect(calls[0]).toBe("gh stack view --json")
    expect(calls.some((call) => call.includes("api graphql"))).toBe(false)
    expect(chain.manager_status).toBe("confirmed")
    expect(chain.manager_source).toBe("gh-stack")
    expect(chain.relationship_status).toBe("dependent")
    expect(chain.target_position).toBe(2)
    expect(chain.target_needs_rebase).toBe(true)
    expect(chain.entries[2].isDraft).toBe(true)
    expect(chain.dependent_prs[0].isDraft).toBe(true)
  })

  test("a successful view of another local stack falls back to GraphQL instead of misclassifying the target", () => {
    const { chain, calls } = probeChain({
      stackView: {
        status: 0,
        stdout: { trunk: "main", currentBranch: "other", branches: [{ name: "other", pr: { number: 7 } }] },
      },
      graphql: {
        status: 0,
        stdout: {
          data: { repository: { pullRequest: {
            stackEntry: { position: 2 },
            stack: {
              id: "STACK_1", number: 99, size: 3, baseRefName: "main",
              entries: { nodes: [
                { position: 1, pullRequest: { number: 41, url: "https://github.com/o/r/pull/41", state: "OPEN", headRefName: "parent" } },
                { position: 2, pullRequest: { number: 42, url: "https://github.com/o/r/pull/42", state: "OPEN", headRefName: "feature" } },
                { position: 3, pullRequest: { number: 43, url: "https://github.com/o/r/pull/43", state: "OPEN", headRefName: "child" } },
              ] },
            },
          } } },
        },
      },
    })
    expect(calls.some((call) => call.includes("api graphql"))).toBe(true)
    expect(chain.manager_status).toBe("confirmed")
    expect(chain.manager_source).toBe("graphql")
    expect(chain.target_position).toBe(2)
    expect(chain.target_needs_rebase).toBeNull()
  })

  test("a local stack entry with the target number in another repository falls back to GraphQL", () => {
    const { chain, calls } = probeChain({
      stackView: {
        status: 0,
        stdout: {
          trunk: "main",
          currentBranch: "feature",
          branches: [
            {
              name: "feature",
              isCurrent: true,
              needsRebase: true,
              pr: { number: 42, url: "https://github.com/another/repository/pull/42", state: "OPEN" },
            },
          ],
        },
      },
      graphql: {
        status: 0,
        stdout: {
          data: { repository: {
            defaultBranchRef: { name: "main" },
            pullRequest: {
              stackEntry: { position: 1 },
              stack: {
                id: "STACK_2", number: 100, size: 1, baseRefName: "main",
                entries: { nodes: [
                  { position: 1, pullRequest: { number: 42, url: "https://github.com/o/r/pull/42", state: "OPEN", headRefName: "feature" } },
                ] },
              },
            },
          } },
        },
      },
    })
    expect(calls.some((call) => call.includes("api graphql"))).toBe(true)
    expect(chain.manager_status).toBe("confirmed")
    expect(chain.manager_source).toBe("graphql")
    expect(chain.target_needs_rebase).toBeNull()
  })

  test("successful null GraphQL stack classifies an ordinary manual dependency chain", () => {
    const { chain, calls } = probeChain({
      baseRef: "parent",
      headRef: "feature",
      stackView: { status: 1, stderr: "no current stack" },
      graphql: { status: 0, stdout: { data: { repository: {
        defaultBranchRef: { name: "main" },
        pullRequest: { stackEntry: null, stack: null },
      } } } },
      openPrs: [
        { number: 41, url: "https://github.com/o/r/pull/41", state: "MERGED", baseRefName: "main", headRefName: "parent" },
        { number: 42, url: "https://github.com/o/r/pull/42", state: "OPEN", baseRefName: "parent", headRefName: "feature" },
        { number: 43, url: "https://github.com/o/r/pull/43", state: "OPEN", baseRefName: "feature", headRefName: "child" },
      ],
    })
    expect(chain.manager_status).toBe("absent")
    expect(calls.some((call) => call.includes("gh pr list") && call.includes("--state all") && call.includes("--head parent"))).toBe(true)
    expect(calls.some((call) => call.includes("gh pr list") && call.includes("--state open") && call.includes("--base feature"))).toBe(true)
    expect(chain.relationship_status).toBe("dependent")
    expect(chain.parent_prs.map((pr: any) => pr.number)).toEqual([41])
    expect(chain.dependent_prs.map((pr: any) => pr.number)).toEqual([43])
  })

  test("a PR based on the default branch ignores unrelated PRs whose head has the default-branch name", () => {
    const { chain, calls } = probeChain({
      baseRef: "main",
      headRef: "feature",
      stackView: { status: 1, stderr: "no current stack" },
      graphql: { status: 0, stdout: { data: { repository: {
        defaultBranchRef: { name: "main" },
        pullRequest: { stackEntry: null, stack: null },
      } } } },
      openPrs: [
        { number: 500, url: "https://github.com/o/r/pull/500", state: "OPEN", baseRefName: "main", headRefName: "main" },
        { number: 320, url: "https://github.com/o/r/pull/320", state: "OPEN", baseRefName: "main", headRefName: "main" },
      ],
    })
    expect(calls.some((call) => call.includes("gh pr list") && call.includes("--head main"))).toBe(false)
    expect(calls.some((call) => call.includes("gh pr list") && call.includes("--base feature"))).toBe(true)
    expect(chain.manager_status).toBe("absent")
    expect(chain.relationship_status).toBe("independent")
    expect(chain.parent_prs).toEqual([])
    expect(chain.dependent_prs).toEqual([])
  })

  test("manager probe failure remains unknown and never collapses to absent", () => {
    const { chain, calls } = probeChain({
      stackView: { status: 1, stderr: "no current stack" },
      graphql: { status: 1, stderr: "gh: HTTP 401: Bad credentials" },
    })
    expect(chain.manager_status).toBe("probe-error")
    expect(chain.manager_status).not.toBe("absent")
    expect(calls.filter((call) => call.startsWith("gh api ")).length).toBe(1)
  })

  test("unavailable stack fields fall back to the default branch and manual-chain classification", () => {
    const { chain, calls } = probeChain({
      baseRef: "parent",
      headRef: "feature",
      stackView: { status: 1, stderr: "no current stack" },
      graphql: { status: 1, stderr: "gh: Field 'stackEntry' doesn't exist on type 'PullRequest'" },
      defaultBranch: { status: 0, stdout: "main\n" },
      openPrs: [
        { number: 41, url: "https://github.com/o/r/pull/41", state: "OPEN", baseRefName: "main", headRefName: "parent" },
        { number: 43, url: "https://github.com/o/r/pull/43", state: "OPEN", baseRefName: "feature", headRefName: "child" },
      ],
    })
    expect(calls).toContain("gh api repos/o/r --jq .default_branch")
    expect(chain.manager_status).toBe("absent")
    expect(chain.relationship_status).toBe("dependent")
    expect(chain.parent_prs.map((pr: any) => pr.number)).toEqual([41])
    expect(chain.dependent_prs.map((pr: any) => pr.number)).toEqual([43])
  })

  test("stack fields unavailable with no default-branch fallback remains a manager probe error", () => {
    const { chain } = probeChain({
      stackView: { status: 1, stderr: "no current stack" },
      graphql: { status: 1, stderr: "GraphQL: Cannot query field \"stack\" on type \"PullRequest\"." },
      defaultBranch: { status: 1, stderr: "network unavailable" },
    })
    expect(chain.manager_status).toBe("probe-error")
  })

  test("colliding check keys are disambiguated (both failing checks surface, neither shadows)", () => {
    const collide = {
      ...FAILING,
      checks: [
        { key: "test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u1" },
        { key: "test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u2" },
      ],
    }
    expect(snapshot(state, fetchFile(dir, "co.json", collide)).counts.ci).toBe(2)
  })

  test("transient null head falls back to the last known head — no ci_dispatched wipe / re-dispatch thrash", () => {
    const f = fetchFile(dir, "a.json", FAILING)
    snapshot(state, f)
    mark(state, ["--check", "CI/test"])
    const nullHead = { ...FAILING, head_sha: null }
    const d = snapshot(state, fetchFile(dir, "nh.json", nullHead))
    expect(d.head_changed).toBe(false)
    expect(d.counts.ci).toBe(0) // still silenced
  })

  // --- trajectory: deterministic cross-tick facts for non-convergence detection ---
  const GREEN_CHECK = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
  const RED_CHECK = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }

  test("check recurrence: fail -> clear -> fail on a NEW head increments recur (ping-pong signal)", () => {
    snapshot(state, fetchFile(dir, "r1.json", { ...FAILING, head_sha: "s1", checks: [RED_CHECK] }))
    snapshot(state, fetchFile(dir, "r2.json", { ...FAILING, head_sha: "s2", checks: [GREEN_CHECK] }))
    const d = snapshot(state, fetchFile(dir, "r3.json", { ...FAILING, head_sha: "s3", checks: [RED_CHECK] }))
    expect(d.trajectory.check_recur_max).toBe(1)
    expect(d.trajectory.recurring_checks).toEqual([{ key: "CI/test", recur: 1 }])
  })

  test("same-head flapping is NOT recurrence (flaky, not ping-pong)", () => {
    const f = { ...FAILING, head_sha: "s1" }
    snapshot(state, fetchFile(dir, "f1.json", { ...f, checks: [RED_CHECK] }))
    snapshot(state, fetchFile(dir, "f2.json", { ...f, checks: [GREEN_CHECK] }))
    const d = snapshot(state, fetchFile(dir, "f3.json", { ...f, checks: [RED_CHECK] }))
    expect(d.trajectory.check_recur_max).toBe(0)
  })

  test("review backlog trend rises and new-thread arrivals are counted (treadmill signal)", () => {
    const th = (ids: string[]) => ids.map((id) => ({ thread_id: id, last_comment_id: `c-${id}`, last_comment_at: id }))
    snapshot(state, fetchFile(dir, "t1.json", { ...FAILING, checks: [], threads: th(["T1"]) }))
    snapshot(state, fetchFile(dir, "t2.json", { ...FAILING, checks: [], threads: th(["T1", "T2"]) }))
    const d = snapshot(state, fetchFile(dir, "t3.json", { ...FAILING, checks: [], threads: th(["T1", "T2", "T3", "T4"]) }))
    expect(d.trajectory.unresolved_trend).toBe("rising")
    expect(d.trajectory.new_threads_this_tick).toBe(2) // T3, T4 are new this tick
    expect(d.trajectory.unresolved_threads).toBe(4)
  })

  test("check_recur_max does not stay elevated after the recurring check leaves CI (stale-key prune)", () => {
    snapshot(state, fetchFile(dir, "p1.json", { ...FAILING, head_sha: "s1", checks: [RED_CHECK] }))
    snapshot(state, fetchFile(dir, "p2.json", { ...FAILING, head_sha: "s2", checks: [GREEN_CHECK] }))
    expect(snapshot(state, fetchFile(dir, "p3.json", { ...FAILING, head_sha: "s3", checks: [RED_CHECK] })).trajectory.check_recur_max).toBe(1)
    // CI/test is gone from the run (renamed/removed); its recurrence must not linger.
    const other = { key: "CI/other", name: "other", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    const d = snapshot(state, fetchFile(dir, "p4.json", { ...FAILING, head_sha: "s4", checks: [other] }))
    expect(d.trajectory.check_recur_max).toBe(0)
  })

  test("heads_since_progress climbs on a persistent failure across heads, but resets on progressive migration", () => {
    // Same check red across three new heads with nothing clearing = a stall.
    snapshot(state, fetchFile(dir, "s1.json", { ...FAILING, head_sha: "h1", checks: [RED_CHECK], threads: [] }))
    expect(snapshot(state, fetchFile(dir, "s2.json", { ...FAILING, head_sha: "h2", checks: [RED_CHECK], threads: [] })).trajectory.heads_since_progress).toBe(1)
    expect(snapshot(state, fetchFile(dir, "s3.json", { ...FAILING, head_sha: "h3", checks: [RED_CHECK], threads: [] })).trajectory.heads_since_progress).toBe(2)
    // A different check now fails (A cleared, B appeared) = progressive migration, not a stall -> reset.
    const other = { key: "CI/other", name: "other", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }
    expect(snapshot(state, fetchFile(dir, "s4.json", { ...FAILING, head_sha: "h4", checks: [other], threads: [] })).trajectory.heads_since_progress).toBe(0)
  })

  test("parking a thread counts as progress: it leaves the non-parked problem set, so no-progress resets", () => {
    const withThread = (headSha: string) => ({
      ...FAILING,
      head_sha: headSha,
      checks: [RED_CHECK],
      threads: [{ thread_id: "T1", last_comment_id: "c1", last_comment_at: "t1" }],
    })
    snapshot(state, fetchFile(dir, "pk1.json", withThread("h1"))) // problems: {CI/test, T1}
    mark(state, ["--thread", "T1", "--disposition", "needs-human"])
    // New head, CI/test still red, T1 now parked (excluded from problems) -> total drops 2->1 = a new low.
    const d = snapshot(state, fetchFile(dir, "pk2.json", withThread("h2")))
    expect(d.open_needs_human).toBe(1)
    expect(d.trajectory.heads_since_progress).toBe(0) // progress was made (a problem left the set), despite the head change
  })

  test("a rerun (IN_PROGRESS) is not a clear — no false recurrence when it fails again", () => {
    snapshot(state, fetchFile(dir, "ir1.json", { ...FAILING, head_sha: "s1", checks: [RED_CHECK] }))
    const rerun = { ...FAILING, head_sha: "s2", checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }] }
    snapshot(state, fetchFile(dir, "ir2.json", rerun))
    const d = snapshot(state, fetchFile(dir, "ir3.json", { ...FAILING, head_sha: "s3", checks: [RED_CHECK] }))
    expect(d.trajectory.check_recur_max).toBe(0)
  })

  test("mark --disposition open re-actionizes a parked needs-human thread (the re-open path)", () => {
    const f = fetchFile(dir, "ro.json", FAILING)
    snapshot(state, f)
    mark(state, ["--thread", "T1", "--disposition", "needs-human"])
    expect(snapshot(state, f).open_needs_human).toBe(1) // parked, not actionable
    mark(state, ["--thread", "T1", "--disposition", "open"])
    const d = snapshot(state, f)
    expect(d.counts.threads).toBe(1) // re-opened -> actionable again
    expect(d.open_needs_human).toBe(0)
  })

  test("a dispatched thread reactivates when a later reviewer comment moves its identity, but not on our own reply (acted_identity baseline)", () => {
    // The false-green fix: a dispatched-but-unresolved thread with fresh reviewer activity must
    // return to actionable, or it stays hidden from counts.threads and lets merge-ready fire.
    const sd = path.join(dir, "react")
    const thr = (cid: string) => ({
      pr_state: "OPEN", mergeable: "MERGEABLE", merge_state_status: "CLEAN", review_decision: null,
      head_sha: "s1", url: "http://x/1", checks: [],
      threads: [{ thread_id: "T1", last_comment_id: cid, last_comment_at: cid }],
    })
    snapshot(sd, fetchFile(dir, "r1.json", thr("C1"))) // open -> actionable
    mark(sd, ["--thread", "T1", "--disposition", "dispatched"])
    // first post-action observation adopts the current identity (our reply) as baseline -> silenced
    expect(snapshot(sd, fetchFile(dir, "r2.json", thr("C1"))).counts.threads).toBe(0)
    // same identity on a later tick -> still silenced (our own reply does not re-trigger)
    expect(snapshot(sd, fetchFile(dir, "r3.json", thr("C1"))).counts.threads).toBe(0)
    // a genuine reviewer reply moves the identity to C2 -> reactivated
    expect(snapshot(sd, fetchFile(dir, "r4.json", thr("C2"))).counts.threads).toBe(1)
  })

  test("a needs-human thread reactivates when a human answers it (a later reply past the baseline), not on our own decision_context reply", () => {
    const sd = path.join(dir, "nhreact")
    const thr = (cid: string) => ({
      ...FAILING, checks: [], threads: [{ thread_id: "T1", last_comment_id: cid, last_comment_at: cid }],
    })
    snapshot(sd, fetchFile(dir, "nh1.json", thr("C1")))
    mark(sd, ["--thread", "T1", "--disposition", "needs-human"])
    // first observation after our decision_context reply (C2) -> adopt as baseline, stays parked
    const d1 = snapshot(sd, fetchFile(dir, "nh2.json", thr("C2")))
    expect(d1.counts.threads).toBe(0)
    expect(d1.open_needs_human).toBe(1) // still parked, blocks merge-ready
    // a human replies past the baseline (C3) -> reactivated to actionable, no longer parked
    const d2 = snapshot(sd, fetchFile(dir, "nh3.json", thr("C3")))
    expect(d2.counts.threads).toBe(1) // reopened -> the loop reprocesses with the human's input
    expect(d2.open_needs_human).toBe(0)
  })

  test("blocked_external waits for other running checks — does not fire while a check is still IN_PROGRESS", () => {
    const RUNNING = { key: "CI/b", name: "b", status: "IN_PROGRESS", conclusion: null, details_url: "u" }
    const GREEN = { key: "CI/a", name: "a", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    // awaiting approval + a still-running check -> NOT blocked_external yet (that check could fail)
    const running = { ...FAILING, threads: [], checks: [RUNNING], awaiting_approval: 1 }
    expect(snapshot(path.join(dir, "be1"), fetchFile(dir, "be1.json", running)).blocked_external).toBe(false)
    // awaiting approval + all other checks terminal -> blocked_external
    const terminal = { ...FAILING, threads: [], checks: [GREEN], awaiting_approval: 1 }
    expect(snapshot(path.join(dir, "be2"), fetchFile(dir, "be2.json", terminal)).blocked_external).toBe(true)
  })

  test("a dispatched (handled) top-level comment does not inflate heads_since_progress across heads", () => {
    // A handled comment never drops out of the fetch, so counting it as an open problem would keep
    // heads_since_progress climbing forever and falsely trip non-convergence on unrelated later work.
    const sd = path.join(dir, "stall")
    const fb = (head: string) => ({
      ...FAILING, head_sha: head, checks: [], threads: [], feedback: [{ id: "IC_1", kind: "comment", author: "r", edit_id: "h" }],
    })
    snapshot(sd, fetchFile(dir, "st1.json", fb("s1"))) // IC_1 open -> a problem
    mark(sd, ["--comment", "IC_1", "--disposition", "dispatched", "--acted-edit-id", "h"])
    const d = snapshot(sd, fetchFile(dir, "st2.json", fb("s2"))) // dispatched + head moved -> handled, progress
    expect(d.trajectory.heads_since_progress).toBe(0)
  })

  test("a watch poll does not consume new_threads_this_tick — the agent's tick still sees the new arrival", () => {
    // The watch's waking poll persists change-detection state but must NOT roll the trajectory, or it
    // marks the just-arrived thread "seen" and the agent's real tick reads 0 new arrivals — hiding a
    // review-bot treadmill from the non-convergence trigger.
    const sd = path.join(dir, "trajwatch")
    const noThreads = { ...FAILING, checks: [], threads: [] }
    snapshot(sd, fetchFile(dir, "tw1.json", noThreads)) // agent tick: baseline, no threads
    const withThread = { ...FAILING, checks: [], threads: [{ thread_id: "T1", last_comment_id: "C1", last_comment_at: "C1" }] }
    expect(watch(sd, fetchFile(dir, "tw2.json", withThread)).reason).toBe("actionable") // a poll wakes on the new thread
    // the agent's real tick then still counts T1 as newly arrived (the poll didn't mark it seen)
    expect(snapshot(sd, fetchFile(dir, "tw3.json", withThread)).trajectory.new_threads_this_tick).toBe(1)
  }, 15000)

  test("heads_since_progress counts head moves across AGENT ticks even when a poll observed the new head first (C2)", () => {
    const sd = path.join(dir, "hspwatch")
    const failAt = (head: string) => ({ ...FAILING, head_sha: head, threads: [], checks: [{ key: "CI/x", name: "x", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }] })
    snapshot(sd, fetchFile(dir, "hw1.json", failAt("s1"))) // agent tick: persistent failure at head s1
    watch(sd, fetchFile(dir, "hw2.json", failAt("s2"))) // a poll observes+persists head s2 (no trajectory roll)
    const d = snapshot(sd, fetchFile(dir, "hw3.json", failAt("s2"))) // agent tick at s2
    expect(d.trajectory.heads_since_progress).toBe(1) // head moved s1->s2 between agent ticks; not starved by the poll
  }, 15000)

  test("check recurrence catches a CLEAR observed only on a watch poll (C1)", () => {
    const sd = path.join(dir, "recurwatch")
    const RED = { key: "CI/x", name: "x", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }
    const GREEN = { key: "CI/x", name: "x", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    snapshot(sd, fetchFile(dir, "rw1.json", { ...FAILING, head_sha: "s1", threads: [], checks: [RED] }),
      EXPIRING_TEST_INVOCATION) // fail h1; the watch below deliberately expires
    watch(sd, fetchFile(dir, "rw2.json", { ...FAILING, head_sha: "s2", threads: [], checks: [GREEN] })) // a poll observes the CLEAR
    const d = snapshot(sd, fetchFile(dir, "rw3.json", { ...FAILING, head_sha: "s3", threads: [], checks: [RED] })) // fail h3
    expect(d.trajectory.check_recur_max).toBe(1) // fail -> clear(seen only on a poll) -> fail = recurrence
  }, 15000)

  test("snapshot refuses to inherit an old budget without an explicit invocation boundary", () => {
    const sd = path.join(dir, "sess")
    snapshot(sd, fetchFile(dir, "se1.json", FAILING))
    // simulate resuming days later against persisted state: backdate started_at
    const statePath = path.join(sd, "state.json")
    const st = JSON.parse(readFileSync(statePath, "utf8"))
    st.started_at = "2020-01-01T00:00:00Z"
    writeFileSync(statePath, JSON.stringify(st))
    const bare = spawnSync("python3", [SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", sd,
      "--fetch-file", fetchFile(dir, "se2.json", FAILING)], { encoding: "utf8" })
    expect(bare.status).not.toBe(0)
    expect(bare.stderr).toContain("requires --start-invocation or --invocation-id")

    const fresh = snapshot(sd, fetchFile(dir, "se3.json", FAILING), ["--start-invocation"])
    expect(fresh.invocation_elapsed_seconds).toBeLessThan(10)
  })

  test("a new invocation clock starts after a slow first fetch", () => {
    const sd = path.join(dir, "slow-first-fetch")
    const python = `
import json
from datetime import datetime, timedelta, timezone
from importlib.machinery import SourceFileLoader
from types import SimpleNamespace
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
before_fetch = datetime(2026, 1, 1, tzinfo=timezone.utc)
after_fetch = before_fetch + timedelta(seconds=61)
clock_reads = 0
def fake_now():
    global clock_reads
    clock_reads += 1
    return before_fetch if clock_reads == 1 else after_fetch
m._now = fake_now
m._fetch_snapshot = lambda args: json.loads(${JSON.stringify(JSON.stringify(FAILING))})
args = SimpleNamespace(state_dir=${JSON.stringify(sd)}, pr=1, repo="o/r", fetch_file=None,
                       reset_session=False, start_invocation=True, continue_invocation=False,
                       invocation_id=None, session_started_at=None,
                       invocation_budget_seconds=28800)
m.cmd_snapshot(args)
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    const value = JSON.parse(r.stdout)
    expect(value.invocation_started_at).toBe("2026-01-01T00:01:01+00:00")
    expect(value.invocation_elapsed_seconds).toBe(0)
  })

  test("a new invocation defaults to one fixed eight-hour budget", () => {
    const sd = path.join(dir, "default-invocation-budget")
    const r = spawnSync("python3", [
      SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", sd,
      "--fetch-file", fetchFile(dir, "default-invocation-budget.json", FAILING),
      "--start-invocation",
    ], { encoding: "utf8" })

    expect(r.status, r.stderr).toBe(0)
    const value = JSON.parse(r.stdout)
    expect(value.invocation_budget_seconds).toBe(28_800)
    expect(value.invocation_remaining_seconds).toBeGreaterThan(28_790)
  })

  test("a new invocation preserves PR history but receives one fresh fixed eight-hour budget", () => {
    const sd = path.join(dir, "invocation-boundary")
    const current = {
      ...FAILING,
      threads: [{ thread_id: "T1", last_comment_id: "C1", last_comment_at: "C1" }],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const initial = snapshot(sd, fetchFile(dir, "ib1.json", current), [
      "--start-invocation", "--invocation-budget-seconds", "28800",
    ])
    mark(sd, ["--thread", "T1", "--disposition", "dispatched"])

    const statePath = path.join(sd, "state.json")
    const old = JSON.parse(readFileSync(statePath, "utf8"))
    old.state_created_at = new Date(Date.now() - 8 * 60 * 60 * 1000).toISOString()
    old.started_at = new Date(Date.now() - 6 * 60 * 60 * 1000).toISOString()
    old.trajectory.stream_series = ["review", "ci"]
    writeFileSync(statePath, JSON.stringify(old))

    // A later explicit skill invocation does not opt into the old clock. Durable review and
    // trajectory state survives, while the invocation clock starts near zero by default.
    const fresh = snapshot(sd, fetchFile(dir, "ib2.json", current), [
      "--start-invocation", "--invocation-budget-seconds", "28800",
    ])
    expect(fresh.invocation_id).not.toBe(initial.invocation_id)
    expect(fresh.invocation_elapsed_seconds).toBeLessThanOrEqual(1)
    expect(fresh.persisted_state_age_seconds).toBeGreaterThan(28_700)
    expect(fresh.counts.threads).toBe(0)
    expect(JSON.parse(readFileSync(statePath, "utf8")).trajectory.stream_series).toEqual(["review", "ci"])

    // Re-arms present the invocation token and preserve its fixed anchor. Put that anchor one
    // second before the real eight-hour cap so the watch proves it stops against the same budget.
    const almostExpired = new Date(Date.now() - 28_799_000).toISOString().replace("Z", "+00:00")
    const persisted = JSON.parse(readFileSync(statePath, "utf8"))
    persisted.started_at = almostExpired
    writeFileSync(statePath, JSON.stringify(persisted))

    const rearmed = snapshot(sd, fetchFile(dir, "ib3.json", current), [
      "--invocation-id", fresh.invocation_id,
      "--session-started-at", almostExpired,
      "--invocation-budget-seconds", "28800",
    ])
    expect(rearmed.invocation_id).toBe(fresh.invocation_id)
    expect(rearmed.invocation_started_at).toBe(almostExpired)

    const wake = watch(sd, fetchFile(dir, "ib4.json", current), [
      "--invocation-id", fresh.invocation_id,
      "--session-started-at", almostExpired,
      "--invocation-budget-seconds", "28800",
    ])
    expect(wake.reason).toBe("max-runtime")
    expect(wake.invocation_elapsed_seconds).toBeGreaterThanOrEqual(28_800)
    expect(wake.invocation_budget_seconds).toBe(28_800)
    expect(wake.invocation_started_at).toBe(almostExpired)
  }, 10000)

  test("an invocation session start carries into a new managed-stack layer state dir", () => {
    const started = new Date(Date.now() - 3_600_000).toISOString()
    const d = snapshot(
      path.join(dir, "next-layer"),
      fetchFile(dir, "next-layer.json", FAILING),
      ["--continue-invocation", "--invocation-id", "managed-stack-invocation",
        "--session-started-at", started, "--invocation-budget-seconds", "28800"],
    )

    expect(new Date(d.invocation_started_at).getTime()).toBe(new Date(started).getTime())
    expect(d.invocation_elapsed_seconds).toBeGreaterThan(3_500)
  })

  test("re-arming watch preserves the invocation budget instead of resetting it", () => {
    const sd = path.join(dir, "watch-budget")
    const started = new Date(Date.now() - 10_000).toISOString()
    const waiting = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    snapshot(sd, fetchFile(dir, "watch-budget-snapshot.json", waiting), [
      "--continue-invocation", "--invocation-id", "watch-budget-invocation",
      "--session-started-at", started, "--invocation-budget-seconds", "1",
    ])

    const wake = watch(
      sd,
      fetchFile(dir, "watch-budget-watch.json", waiting),
      ["--invocation-id", "watch-budget-invocation", "--session-started-at", started,
        "--invocation-budget-seconds", "1"],
    )

    expect(wake.reason).toBe("max-runtime")
  }, 15000)

  test("a re-arm cannot extend the fixed invocation budget", () => {
    const sd = path.join(dir, "watch-budget-extension")
    const started = snapshot(sd, fetchFile(dir, "watch-budget-extension-start.json", FAILING), [
      "--start-invocation", "--invocation-budget-seconds", "28800",
    ])
    const r = spawnSync("python3", [
      SCRIPT, "snapshot", "--pr", "1", "--repo", "o/r", "--state-dir", sd,
      "--fetch-file", fetchFile(dir, "watch-budget-extension-resume.json", FAILING),
      "--invocation-id", started.invocation_id,
      "--session-started-at", started.invocation_started_at,
      "--invocation-budget-seconds", "57600",
    ], { encoding: "utf8" })

    expect(r.status).not.toBe(0)
    expect(r.stderr).toContain("does not match the persisted fixed budget")
    expect(JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8")).invocation_budget_seconds).toBe(28800)
  })

  test("the fixed cap outranks newly actionable work instead of allowing another round", () => {
    const sd = path.join(dir, "budget-outranks-actionable")
    const started = new Date(Date.now() - 2_000).toISOString()
    snapshot(sd, fetchFile(dir, "budget-outranks-actionable-start.json", FAILING), [
      "--continue-invocation", "--invocation-id", "expired-actionable-invocation",
      "--session-started-at", started, "--invocation-budget-seconds", "1",
    ])

    const wake = watch(sd, fetchFile(dir, "budget-outranks-actionable-watch.json", FAILING), [
      "--invocation-id", "expired-actionable-invocation",
      "--session-started-at", started, "--invocation-budget-seconds", "1",
    ])
    expect(wake.reason).toBe("max-runtime")
    expect(wake.invocation_elapsed_seconds).toBeGreaterThanOrEqual(2)
  })

  test("the fixed cap fetches terminal PR state before emitting max-runtime", async () => {
    const sd = path.join(dir, "budget-terminal-precedence")
    const waiting = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "budget-terminal-precedence.json", waiting)
    const started = snapshot(sd, fetch, ["--start-invocation", "--invocation-budget-seconds", "1"])
    const watcher = startWatch(sd, fetch, [
      "--interval", "5",
      "--invocation-id", started.invocation_id,
      "--session-started-at", started.invocation_started_at,
      "--invocation-budget-seconds", "1",
    ])
    await waitForWatchGeneration(sd)

    const replacement = fetchFile(dir, "budget-terminal-precedence-closed.json", {
      ...waiting,
      pr_state: "CLOSED",
    })
    renameSync(replacement, fetch)

    const result = await watcher.result
    expect(result.code, result.stderr).toBe(0)
    const wake = JSON.parse(result.stdout.trim().split("\n").pop()!)
    expect(wake.reason).toBe("terminal")
    expect(wake.pr_state).toBe("CLOSED")
  }, 5000)

  test("the fixed cap preserves a merge-ready result from its final refresh", async () => {
    const sd = path.join(dir, "budget-ready-precedence")
    const waiting = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "budget-ready-precedence.json", waiting)
    const started = snapshot(sd, fetch, ["--start-invocation", "--invocation-budget-seconds", "1"])
    const watcher = startWatch(sd, fetch, [
      "--interval", "5", "--settle-seconds", "0",
      "--invocation-id", started.invocation_id,
      "--session-started-at", started.invocation_started_at,
      "--invocation-budget-seconds", "1",
    ])
    await waitForWatchGeneration(sd)

    const replacement = fetchFile(dir, "budget-ready-precedence-clean.json", {
      ...waiting,
      mergeable: "MERGEABLE",
      merge_state_status: "CLEAN",
      checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
    })
    renameSync(replacement, fetch)

    const result = await watcher.result
    expect(result.code, result.stderr).toBe(0)
    expect(JSON.parse(result.stdout.trim()).reason).toBe("merge-ready")
  }, 5000)

  test("watch cannot start or reset an invocation budget", () => {
    const sd = path.join(dir, "watch-cannot-reset")
    const waiting = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const started = snapshot(sd, fetchFile(dir, "watch-cannot-reset-start.json", waiting), ["--start-invocation"])
    const r = spawnSync("python3", [
      SCRIPT, "watch", "--pr", "1", "--repo", "o/r", "--state-dir", sd,
      "--fetch-file", fetchFile(dir, "watch-cannot-reset-watch.json", waiting),
      "--invocation-id", started.invocation_id,
      "--session-started-at", started.invocation_started_at,
      "--invocation-budget-seconds", String(started.invocation_budget_seconds),
      "--reset-session",
    ], { encoding: "utf8" })
    expect(r.status).not.toBe(0)
    expect(r.stderr).toContain("unrecognized arguments: --reset-session")
  })

  test("mark requires the current invocation tuple before mutating dispositions", () => {
    const sd = path.join(dir, "mark-invocation-fence")
    const fetch = fetchFile(dir, "mark-invocation-fence.json", FAILING)
    snapshot(sd, fetch)
    const oldInvocation = persistedInvocationArgs(sd)
    snapshot(sd, fetch, ["--start-invocation"])

    const tokenless = spawnSync("python3", [SCRIPT, "mark", "--state-dir", sd,
      "--thread", "T1", "--disposition", "dispatched"], { encoding: "utf8" })
    expect(tokenless.status).not.toBe(0)

    const stale = spawnSync("python3", [SCRIPT, "mark", "--state-dir", sd, ...oldInvocation,
      "--thread", "T1", "--disposition", "dispatched"], { encoding: "utf8" })
    expect(stale.status).not.toBe(0)
    const persisted = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))
    expect(persisted.threads?.T1?.disposition).not.toBe("dispatched")

    mark(sd, ["--thread", "T1", "--disposition", "dispatched"])
    const updated = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))
    expect(updated.threads.T1.disposition).toBe("dispatched")
  })

  test("clearing a fork approval gate is movement (resets the settle clock so merge-ready waits for check-runs)", () => {
    const sd = path.join(dir, "appr")
    const gated = { ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", checks: [], threads: [], awaiting_approval: 1 }
    snapshot(sd, fetchFile(dir, "ap1.json", gated)) // first tick
    expect(snapshot(sd, fetchFile(dir, "ap2.json", gated)).changed_this_tick).toBe(false) // stable gate, no movement
    // approval clears (no check-runs created yet) -> registered as movement so quiet resets
    expect(snapshot(sd, fetchFile(dir, "ap3.json", { ...gated, awaiting_approval: 0 })).changed_this_tick).toBe(true)
  })

  test("mark --thread captures the acted baseline at mark time (closes the reviewer-reply race)", () => {
    const sd = path.join(dir, "atmark")
    const thr = (cid: string) => ({
      ...FAILING, checks: [], threads: [{ thread_id: "T1", last_comment_id: cid, last_comment_at: cid }],
    })
    snapshot(sd, fetchFile(dir, "am1.json", thr("C1")))
    // our decision_context reply is C2; marking WITH the current fetch captures C2 as the baseline now
    mark(sd, ["--thread", "T1", "--disposition", "needs-human", "--fetch-file", fetchFile(dir, "am2.json", thr("C2"))])
    // a reviewer reply that raced in (C3) before the next snapshot -> reactivated, not swallowed as baseline
    const d = snapshot(sd, fetchFile(dir, "am3.json", thr("C3")))
    expect(d.counts.threads).toBe(1) // C3 != the C2 baseline captured at mark -> reopened
    expect(d.open_needs_human).toBe(0)
  })

  test("mark --comment needs-human with --acted-edit-id captures the baseline at mark time (closes the answered-by-edit race)", () => {
    const sd = path.join(dir, "cmark")
    const fb = (edit: string) => ({
      ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", checks: [], threads: [],
      feedback: [{ id: "IC_1", kind: "comment", author: "reviewer", edit_id: edit }],
    })
    snapshot(sd, fetchFile(dir, "cm1.json", fb("h1")))
    // park needs-human with the snapshot-time edit_id as the explicit baseline
    mark(sd, ["--comment", "IC_1", "--disposition", "needs-human", "--acted-edit-id", "h1"])
    // an edit that races in (h2) before the next snapshot -> reactivated, not swallowed as baseline
    const raced = snapshot(sd, fetchFile(dir, "cm2.json", fb("h2")))
    expect(raced.counts.comments).toBe(1)
    expect(raced.open_needs_human).toBe(0)
    // on a dispatched mark the same flag is stored but never read: an edit stays silenced
    mark(sd, ["--comment", "IC_1", "--disposition", "dispatched", "--acted-edit-id", "h2"])
    expect(snapshot(sd, fetchFile(dir, "cm3.json", fb("h3"))).counts.comments).toBe(0)
  })

  test("a needs-human comment reactivates when a human answers by editing it (lazy baseline), while parked it blocks merge-ready", () => {
    const sd = path.join(dir, "nhedit")
    const fb = (edit: string) => ({
      ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", checks: [], threads: [],
      feedback: [{ id: "IC_q", kind: "comment", author: "reviewer", edit_id: edit }],
    })
    snapshot(sd, fetchFile(dir, "nh1.json", fb("q1")))
    mark(sd, ["--comment", "IC_q", "--disposition", "needs-human"]) // lazy baseline
    const parked = snapshot(sd, fetchFile(dir, "nh2.json", fb("q1")))
    expect(parked.counts.comments).toBe(0)
    expect(parked.open_needs_human).toBe(1) // parked -> blocks merge-ready
    // the human answers by editing the same comment -> reactivated and actionable again
    const answered = snapshot(sd, fetchFile(dir, "nh3.json", fb("q2")))
    expect(answered.counts.comments).toBe(1)
    expect(answered.open_needs_human).toBe(0)
  })

  test("a dispatched thread reactivates when an EARLIER comment is edited (same last_comment_id, bumped last_comment_at)", () => {
    // fetch_threads sets last_comment_at = max edit/create time across the whole thread, so an edit
    // to an earlier comment (last_comment_id unchanged) still moves the identity and reopens it.
    const sd = path.join(dir, "editearlier")
    const thr = (at: string) => ({
      ...FAILING, checks: [], threads: [{ thread_id: "T1", last_comment_id: "R1", last_comment_at: at }],
    })
    snapshot(sd, fetchFile(dir, "ee1.json", thr("t1")))
    mark(sd, ["--thread", "T1", "--disposition", "dispatched"]) // lazy baseline
    expect(snapshot(sd, fetchFile(dir, "ee2.json", thr("t1"))).counts.threads).toBe(0) // baseline (R1,t1) -> silenced
    // reviewer edits an earlier comment: last_comment_id stays R1 but the thread's max edit time bumps
    expect(snapshot(sd, fetchFile(dir, "ee3.json", thr("t2"))).counts.threads).toBe(1) // reactivated
  })

  test("a dispatched top-level comment does NOT reactivate when its body is edited; a new comment id still does (#1309)", () => {
    // Status bots (changeset-bot, CodeRabbit, Codecov) rewrite their own comment bodies on every
    // push. Edit-keyed reactivation re-actionized the handled comment on every rewrite, so
    // counts.comments never reached 0 and merge-ready could never fire. A marked comment stays
    // silenced across edits; a genuinely new request is a new comment id and stays actionable.
    const sd = path.join(dir, "editfb")
    const fb = (feedback: object[]) => ({
      ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", checks: [], threads: [],
      feedback,
    })
    const bot = (edit: string) => ({ id: "IC_1", kind: "comment", author: "changeset-bot", edit_id: edit })
    snapshot(sd, fetchFile(dir, "e1.json", fb([bot("h1")]))) // actionable
    mark(sd, ["--comment", "IC_1", "--disposition", "dispatched"])
    expect(snapshot(sd, fetchFile(dir, "e2.json", fb([bot("h1")]))).counts.comments).toBe(0) // same body -> silenced
    // bot rewrites its status comment on the next push -> STAYS silenced, but the edit is still
    // review activity: it must reset the settle clock so merge-ready cannot fire off an old quiet
    // window right after fresh edits (edit_id is part of _change_sig even though it no longer
    // reopens the item)
    patchState(sd, { last_change_at: isoAgo(60 * 60) })
    const edited = snapshot(sd, fetchFile(dir, "e3.json", fb([bot("h2")])))
    expect(edited.counts.comments).toBe(0)
    expect(edited.changed_this_tick).toBe(true)
    expect(edited.quiet_seconds).toBeLessThan(2)
    // an unchanged tick after the edit settles normally
    patchState(sd, { last_change_at: isoAgo(60 * 60) })
    const settled = snapshot(sd, fetchFile(dir, "e3b.json", fb([bot("h2")])))
    expect(settled.changed_this_tick).toBe(false)
    expect(settled.quiet_seconds).toBeGreaterThan(60)
    expect(snapshot(sd, fetchFile(dir, "e4.json", fb([bot("h3")]))).counts.comments).toBe(0)
    // a brand-new comment is a new id -> actionable; the handled one stays out of the count
    const next = snapshot(sd, fetchFile(dir, "e5.json", fb([bot("h3"), { id: "IC_2", kind: "comment", author: "reviewer", edit_id: "x1" }])))
    expect(next.counts.comments).toBe(1)
    expect(next.actionable.comments.map((c: any) => c.id)).toEqual(["IC_2"])
    // explicit re-open still works
    mark(sd, ["--comment", "IC_1", "--disposition", "open"])
    expect(snapshot(sd, fetchFile(dir, "e6.json", fb([bot("h3"), { id: "IC_2", kind: "comment", author: "reviewer", edit_id: "x1" }]))).counts.comments).toBe(2)
  })

  test("a fork-PR workflow awaiting maintainer approval blocks 'all_checks_ok' and flags blocked_external", () => {
    const gated = {
      ...FAILING,
      merge_state_status: "UNSTABLE",
      review_decision: "",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      awaiting_approval: 1, // real CI hasn't run — awaiting a base-repo maintainer's approval
    }
    const d = snapshot(state, fetchFile(dir, "aa.json", gated))
    expect(d.checks_awaiting_approval).toBe(1)
    expect(d.has_failing_checks).toBe(false)
    expect(d.all_checks_ok).toBe(false) // not "ok" — the gated CI is invisible to the rollup
    expect(d.blocked_external).toBe(true)
  })

  test("an approval probe failure is unknown rather than a proven-clear gate", () => {
    expect(probeAwaitingApproval({ status: 1 })).toBeNull()
    expect(probeAwaitingApproval({ status: 0, stdout: "not-a-count" })).toBeNull()
    expect(probeAwaitingApproval({ status: 0, stdout: "0" })).toBe(0)
  })

  test("approval review-drain clock is head-scoped and resets only on external review movement", () => {
    const sd = path.join(dir, "approval-drain-state")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
    }

    const first = snapshot(sd, fetchFile(dir, "approval-drain-first.json", gated))
    expect(first.blocked_external).toBe(true)
    expect(first.blocked_external_review_quiet_seconds).toBeLessThan(2)
    const startedAt = first.blocked_external_review_last_activity_at

    patchState(sd, { blocked_external_review_last_activity_at: isoAgo(10 * 60) })
    const unchanged = snapshot(sd, fetchFile(dir, "approval-drain-unchanged.json", gated))
    expect(unchanged.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(9 * 60)

    const activityBeforeProbeFailure = unchanged.blocked_external_review_last_activity_at
    const unknown = snapshot(sd, fetchFile(dir, "approval-drain-probe-failed.json", {
      ...gated,
      awaiting_approval: null,
    }))
    expect(unknown.checks_awaiting_approval).toBe(1)
    expect(unknown.blocked_external).toBe(true)
    expect(unknown.blocked_external_review_last_activity_at).toBe(activityBeforeProbeFailure)
    const recovered = snapshot(sd, fetchFile(dir, "approval-drain-probe-recovered.json", gated))
    expect(recovered.blocked_external_review_last_activity_at).toBe(activityBeforeProbeFailure)
    expect(recovered.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(9 * 60)

    const persistentFeedback = {
      ...gated,
      feedback: [{ id: "C1", kind: "comment", author: "reviewer", edit_id: "e1" }],
    }
    snapshot(sd, fetchFile(dir, "approval-drain-feedback-baseline.json", persistentFeedback))
    patchState(sd, { blocked_external_review_last_activity_at: isoAgo(10 * 60) })
    mark(sd, ["--comment", "C1", "--disposition", "dispatched", "--acted-edit-id", "e1"])
    const dispositionOnly = snapshot(sd, fetchFile(dir, "approval-drain-disposition.json", persistentFeedback))
    expect(dispositionOnly.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(9 * 60)

    const newFeedback = {
      ...persistentFeedback,
      feedback: [
        { id: "C1", kind: "comment", author: "reviewer", edit_id: "e1" },
        { id: "C2", kind: "comment", author: "reviewer", edit_id: "e2" },
      ],
    }
    const moved = snapshot(sd, fetchFile(dir, "approval-drain-feedback.json", newFeedback))
    expect(moved.blocked_external).toBe(false)
    expect(moved.blocked_external_review_quiet_seconds).toBeLessThan(2)
    expect(moved.blocked_external_review_last_activity_at).not.toBe(startedAt)

    const newHead = snapshot(sd, fetchFile(dir, "approval-drain-head.json", {
      ...gated,
      head_sha: "gated-h2",
    }))
    expect(newHead.blocked_external_review_quiet_seconds).toBeLessThan(2)

    const cleared = snapshot(sd, fetchFile(dir, "approval-drain-cleared.json", {
      ...gated,
      head_sha: "gated-h2",
      awaiting_approval: 0,
    }))
    expect(cleared.blocked_external_review_last_activity_at).toBeNull()
    expect(cleared.blocked_external_review_quiet_seconds).toBe(0)
  })

  test("external review movement resets a confirmed approval drain when the current probe is unknown", () => {
    const sd = path.join(dir, "approval-drain-unknown-movement")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
    }
    snapshot(sd, fetchFile(dir, "approval-drain-unknown-start.json", gated))
    patchState(sd, { blocked_external_review_last_activity_at: isoAgo(10 * 60) })

    const withExternalMovement = {
      ...gated,
      feedback: [{ id: "C1", kind: "comment", author: "reviewer", edit_id: "e1" }],
      awaiting_approval: null,
    }
    const moved = snapshot(sd, fetchFile(dir, "approval-drain-unknown-moved.json", withExternalMovement))
    expect(moved.checks_awaiting_approval).toBe(1)
    expect(moved.blocked_external_review_moved_this_tick).toBe(true)
    expect(moved.blocked_external_review_quiet_seconds).toBeLessThan(2)
    const resetAt = moved.blocked_external_review_last_activity_at

    const unchanged = snapshot(sd, fetchFile(dir, "approval-drain-unknown-unchanged.json", withExternalMovement))
    expect(unchanged.blocked_external_review_moved_this_tick).toBe(false)
    expect(unchanged.blocked_external_review_last_activity_at).toBe(resetAt)

    const cleared = snapshot(sd, fetchFile(dir, "approval-drain-unknown-cleared.json", {
      ...withExternalMovement,
      awaiting_approval: 0,
    }))
    expect(cleared.checks_awaiting_approval).toBe(0)
    expect(cleared.blocked_external_review_last_activity_at).toBeNull()
  })

  test("approval drain ignores the resolver reply baseline but wakes for a later reviewer reply", () => {
    const gated = (cid: string) => ({
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [{ thread_id: "T1", last_comment_id: cid, last_comment_at: cid }],
      feedback: [],
      awaiting_approval: 1,
    })

    const drainedState = path.join(dir, "approval-drain-resolver-reply")
    snapshot(drainedState, fetchFile(dir, "approval-drain-reviewer-c1.json", gated("C1")))
    const resolverReply = fetchFile(dir, "approval-drain-resolver-c2.json", gated("C2"))
    mark(drainedState, ["--thread", "T1", "--disposition", "dispatched", "--fetch-file", resolverReply])
    patchState(drainedState, { blocked_external_review_last_activity_at: isoAgo(10) })
    const resolverObserved = snapshot(drainedState, resolverReply)
    expect(resolverObserved.blocked_external_review_moved_this_tick).toBe(false)
    expect(resolverObserved.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(1)
    const drained = watch(drainedState, resolverReply, ["--blocked-external-drain-seconds", "1"])
    expect(drained.reason).toBe("blocked-external-drained")
    expect(drained.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(1)

    const wakeState = path.join(dir, "approval-drain-later-reviewer")
    snapshot(wakeState, fetchFile(dir, "approval-drain-reviewer-start.json", gated("C1")))
    mark(wakeState, ["--thread", "T1", "--disposition", "dispatched", "--fetch-file", resolverReply])
    patchState(wakeState, { blocked_external_review_last_activity_at: isoAgo(10) })
    const reviewerReply = fetchFile(dir, "approval-drain-reviewer-c3.json", gated("C3"))
    const reviewerObserved = snapshot(wakeState, reviewerReply)
    expect(reviewerObserved.blocked_external_review_moved_this_tick).toBe(true)
    expect(reviewerObserved.blocked_external_review_quiet_seconds).toBeLessThan(2)
    expect(reviewerObserved.counts.threads).toBe(1)
    expect(wakeReason(reviewerObserved)).toBe("actionable")
  })

  test("approval drain ignores loop-owned thread resolution but wakes for a needs-human removal", () => {
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [{ thread_id: "T1", last_comment_id: "C1", last_comment_at: "C1" }],
      feedback: [],
      awaiting_approval: 1,
    }
    const resolved = fetchFile(dir, "approval-drain-resolved.json", { ...gated, threads: [] })

    const dispatchedState = path.join(dir, "approval-drain-dispatched-resolution")
    snapshot(dispatchedState, fetchFile(dir, "approval-drain-dispatched-start.json", gated))
    mark(dispatchedState, ["--thread", "T1", "--disposition", "dispatched", "--fetch-file",
      fetchFile(dir, "approval-drain-dispatched-baseline.json", gated)])
    patchState(dispatchedState, { blocked_external_review_last_activity_at: isoAgo(10) })
    const loopResolved = snapshot(dispatchedState, resolved)
    expect(loopResolved.blocked_external_review_moved_this_tick).toBe(false)
    expect(loopResolved.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(1)

    const needsHumanState = path.join(dir, "approval-drain-needs-human-removal")
    snapshot(needsHumanState, fetchFile(dir, "approval-drain-needs-human-start.json", gated))
    mark(needsHumanState, ["--thread", "T1", "--disposition", "needs-human", "--fetch-file",
      fetchFile(dir, "approval-drain-needs-human-baseline.json", gated)])
    patchState(needsHumanState, { blocked_external_review_last_activity_at: isoAgo(10) })
    const externallyRemoved = snapshot(needsHumanState, resolved)
    expect(externallyRemoved.blocked_external_review_moved_this_tick).toBe(true)
    expect(externallyRemoved.blocked_external_review_quiet_seconds).toBeLessThan(2)
  })

  test("approval review-drain wakes terminally after its selected quiet bound", () => {
    const sd = path.join(dir, "approval-drain-expiry")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
    }
    snapshot(sd, fetchFile(dir, "approval-drain-expiry-first.json", gated))
    patchState(sd, { blocked_external_review_last_activity_at: isoAgo(10) })

    const expired = watch(sd, fetchFile(dir, "approval-drain-expiry-watch.json", gated), [
      "--blocked-external-drain-seconds", "1",
    ])
    expect(expired.reason).toBe("blocked-external-drained")
    expect(expired.blocked_external_review_quiet_seconds).toBeGreaterThanOrEqual(1)
    expect(expired.blocked_external_drain_seconds).toBe(1)
  })

  test("the invocation budget outranks an expired approval review-drain", () => {
    const sd = path.join(dir, "approval-drain-budget")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
    }
    snapshot(sd, fetchFile(dir, "approval-drain-budget-first.json", gated))
    patchState(sd, {
      started_at: isoAgo(9 * 3600),
      last_activity_at: isoAgo(10),
      dead_time_seconds: 0,
      blocked_external_review_last_activity_at: isoAgo(10),
    })

    const expired = watch(sd, fetchFile(dir, "approval-drain-budget-watch.json", gated), [
      "--blocked-external-drain-seconds", "1",
    ])
    expect(expired.reason).toBe("max-runtime")
  })

  test("new review feedback outranks an expired approval review-drain", () => {
    const sd = path.join(dir, "approval-drain-feedback-wake")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
    }
    snapshot(sd, fetchFile(dir, "approval-drain-feedback-first.json", gated))
    patchState(sd, { blocked_external_review_last_activity_at: isoAgo(10) })
    const withFeedback = {
      ...gated,
      feedback: [{ id: "C1", kind: "comment", author: "reviewer", edit_id: "e1" }],
    }

    const wake = watch(sd, fetchFile(dir, "approval-drain-feedback-watch.json", withFeedback), [
      "--blocked-external-drain-seconds", "1",
    ])
    expect(wake.reason).toBe("feedback-candidate")
  })

  test("a review lifecycle starting during an approval drain wakes for a longer bound", () => {
    const sd = path.join(dir, "approval-drain-signal")
    const gated = {
      ...FAILING,
      head_sha: "gated-h1",
      merge_state_status: "UNSTABLE",
      checks: [{ key: "Track", name: "Track", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }],
      threads: [],
      feedback: [],
      awaiting_approval: 1,
      review_in_progress: false,
      review_signal_count: 0,
      review_signal_identities: [],
    }
    snapshot(sd, fetchFile(dir, "approval-drain-signal-first.json", gated))
    const withSignal = {
      ...gated,
      review_in_progress: true,
      review_signal_count: 1,
      review_signal_identities: ["review-bot"],
    }

    const wake = watch(sd, fetchFile(dir, "approval-drain-signal-watch.json", withSignal), [
      "--blocked-external-drain-seconds", "300",
    ])
    expect(wake.reason).toBe("blocked-external")
  })

  test("an empty statusCheckRollup (no check-runs yet) is not ok — checks_present false blocks a pipeline false-success", () => {
    const noChecks = { ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", checks: [], threads: [] }
    const d = snapshot(state, fetchFile(dir, "nc.json", noChecks))
    expect(d.checks_present).toBe(false)
    expect(d.all_checks_ok).toBe(false) // no observed checks -> not "ok"; the pipeline stop must not exit-success
    expect(d.checks_terminal).toBe(true) // vacuously terminal on an empty set — exactly why checks_present is needed
  })

  test("_resolve_repo_ref parses the host from the PR URL so gh api targets GHE, not github.com", () => {
    const r = spawnSync(
      "python3",
      [
        "-c",
        `from importlib.machinery import SourceFileLoader; ` +
          `m=SourceFileLoader('prs', ${JSON.stringify(SCRIPT)}).load_module(); ` +
          `print(m._resolve_repo_ref('', 'https://ghe.acme.com/o/r/pull/5')); ` +
          `print(m._host_args('ghe.acme.com')); print(m._host_args(None))`,
      ],
      { encoding: "utf8" },
    )
    expect(r.status, r.stderr).toBe(0)
    const lines = r.stdout.trim().split("\n")
    expect(lines[0]).toBe("('o', 'r', 'ghe.acme.com')")
    expect(lines[1]).toBe("['--hostname', 'ghe.acme.com']")
    expect(lines[2]).toBe("[]")
  })

  test("cross-stream alternation: ci-only then review-only then ci-only ticks flip (churn signal)", () => {
    const th = (ids: string[]) => ids.map((id) => ({ thread_id: id, last_comment_id: `c-${id}`, last_comment_at: id }))
    snapshot(state, fetchFile(dir, "a1.json", { ...FAILING, head_sha: "s1", checks: [RED_CHECK], threads: [] }))
    snapshot(state, fetchFile(dir, "a2.json", { ...FAILING, head_sha: "s2", checks: [GREEN_CHECK], threads: th(["T1"]) }))
    const d = snapshot(state, fetchFile(dir, "a3.json", { ...FAILING, head_sha: "s3", checks: [RED_CHECK], threads: [] }))
    expect(d.trajectory.stream_alternations).toBe(2) // ci -> review -> ci
  })

  test("non-thread feedback: a top-level comment / review body is actionable, mark --comment silences it, needs-human blocks ready", () => {
    const withFeedback = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [GREEN_CHECK],
      threads: [],
      feedback: [
        { id: "IC_1", kind: "comment", author: "reviewer" },
        { id: "PRR_1", kind: "review", author: "coderabbit", state: "COMMENTED" },
      ],
    }
    const f = fetchFile(dir, "fb.json", withFeedback)
    const d = snapshot(state, f)
    expect(d.counts.comments).toBe(2) // both surfaced as feedback candidates with no inline thread
    expect(d.actionable.comments.map((c: any) => c.id).sort()).toEqual(["IC_1", "PRR_1"])

    mark(state, ["--comment", "IC_1", "--disposition", "dispatched"])
    mark(state, ["--comment", "PRR_1", "--disposition", "needs-human"])
    const d2 = snapshot(state, f)
    expect(d2.counts.comments).toBe(0) // dispatched item silenced; needs-human item parked, not actionable
    expect(d2.open_needs_human).toBe(1) // parked comment blocks merge-ready just like a parked thread
  })

  test("_extract_feedback surfaces every non-empty external body for agent judgment", () => {
    const v = {
      author: { login: "me" },
      comments: [
        { id: "c-me", author: { login: "me" }, body: "my own note" }, // author -> excluded
        { id: "c-cov", author: { login: "codecov[bot]" }, body: "coverage -0.1%" },
        { id: "c-wrapper", author: { login: "chatgpt-codex-connector" }, body: CODEX_WRAPPER },
        { id: "c-near-match", author: { login: "chatgpt-codex-connector" }, body: `${CODEX_WRAPPER}\n\nP1: Preserve this appended actionable finding.` },
        { id: "c-claude", author: { login: "github-actions" }, body: "<!-- claude-review-summary -->\n## Claude Review\nBLOCKING: regenerate code" },
        { id: "c-ghost", author: null, body: "feedback from an unavailable account" },
        { id: "c-empty", author: { login: "octo-reviewer" }, body: "   " }, // empty -> excluded
      ],
      reviews: [
        { id: "r-wrapper", author: { login: "chatgpt-codex-connector" }, body: CODEX_WRAPPER.replace("50ffb4dd99", "1f95273c71"), state: "COMMENTED" },
        { id: "r-codex", author: { login: "chatgpt-codex-connector" }, body: `### 💡 Codex Review\n\nhttps://github.com/o/r/blob/abc/file.ts#L1-L2\n**P2 Block archiving core questions**\n\nAdd the invariant guard.\n\n<details> <summary>ℹ️ About Codex in GitHub</summary></details>`, state: "COMMENTED" },
        { id: "r-cr", author: { login: "coderabbitai[bot]" }, body: "Actionable comments posted: 1\n\nInline review comments failed to post. Fix the custom agent ID path.", state: "COMMENTED" },
        { id: "r-empty", author: { login: "octo-reviewer" }, body: "", state: "APPROVED" }, // empty body -> excluded
      ],
    }
    expect(extractFeedback(v).map((f: any) => f.id).sort()).toEqual([
      "c-claude", "c-cov", "c-ghost", "c-near-match", "c-wrapper", "r-codex", "r-cr", "r-wrapper",
    ])
  })

  test("watch: wakes on actionable backlog, terminal, and merge-ready-after-settle; times out on clean-not-settled", () => {
    const GREEN = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    // actionable backlog (FAILING has an unresolved thread + a failing check) -> wake
    expect(watch(path.join(dir, "w1"), fetchFile(dir, "wa.json", FAILING)).reason).toBe("actionable")
    // terminal PR -> wake regardless of backlog
    const term = fetchFile(dir, "wt.json", { ...FAILING, pr_state: "CLOSED", threads: [], checks: [] })
    expect(watch(path.join(dir, "w2"), term).reason).toBe("terminal")
    // clean + green but not yet settled (settle 300 > quiet ~0) -> keep watching -> times out
    const clean = { ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", threads: [], checks: [GREEN] }
    const cf = fetchFile(dir, "wc.json", clean)
    const unsettledDir = path.join(dir, "w3")
    snapshot(unsettledDir, cf, EXPIRING_TEST_INVOCATION)
    expect(watch(unsettledDir, cf, ["--settle-seconds", "300"]).reason).toBe("max-runtime")
    // same clean state with a zero settle window -> merge-ready wake
    expect(watch(path.join(dir, "w4"), cf, ["--settle-seconds", "0"]).reason).toBe("merge-ready")
  }, 15000) // spawns 4 watch subprocesses incl. a max-runtime timeout -> explicit timeout over Bun's 5s default

  test("watch: a newer valid watcher supersedes the old watcher and owns the only wake", async () => {
    const sd = path.join(dir, "watch-owner")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "watch-owner.json", running)
    snapshot(sd, fetch)
    const beforeTakeover = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))

    const oldWatch = startWatch(sd, fetch)
    const oldGeneration = await waitForWatchGeneration(sd)
    const newWatch = startWatch(sd, fetch)
    await waitForWatchGeneration(sd, oldGeneration)
    const afterTakeover = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))
    expect(afterTakeover.started_at).toBe(beforeTakeover.started_at)
    expect(afterTakeover.last_change_at).toBe(beforeTakeover.last_change_at)

    const nextFetch = `${fetch}.next`
    writeFileSync(nextFetch, JSON.stringify({
      ...running,
      threads: [{ thread_id: "T-new", last_comment_id: "C1", last_comment_at: "t1" }],
    }))
    renameSync(nextFetch, fetch)

    const [oldResult, newResult] = await Promise.all([oldWatch.result, newWatch.result])
    expect(oldResult.code, oldResult.stderr).toBe(0)
    expect(newResult.code, newResult.stderr).toBe(0)

    const wakes = [oldResult.stdout, newResult.stdout]
      .flatMap((output) => output.trim() ? output.trim().split("\n") : [])
      .map((line) => JSON.parse(line))
      .filter((event) => event.event === "BABYSIT_WAKE")
    expect(wakes).toHaveLength(1)
    expect(wakes[0].reason).toBe("actionable")
    expect(wakes[0].watch_generation).toEqual(expect.any(String))
    expect(oldResult.stdout).toBe("")

    const persisted = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))
    expect(persisted.watch_generation).toBe(wakes[0].watch_generation)
    expect(snapshot(sd, fetch).watch_generation).toBe(wakes[0].watch_generation)
  }, 15000)

  test("watch: a replacement that fails preflight leaves the existing watcher active", async () => {
    const sd = path.join(dir, "watch-preflight")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "watch-preflight.json", running)
    snapshot(sd, fetch)

    const existingWatch = startWatch(sd, fetch)
    const activeGeneration = await waitForWatchGeneration(sd)

    const invalidFetch = path.join(dir, "invalid-watch-preflight.json")
    writeFileSync(invalidFetch, "not json")
    const failedReplacement = startWatch(sd, invalidFetch)
    const failedResult = await failedReplacement.result
    expect(failedResult.code).not.toBe(0)
    expect(JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8")).watch_generation).toBe(activeGeneration)

    const nextFetch = `${fetch}.next`
    writeFileSync(nextFetch, JSON.stringify({
      ...running,
      threads: [{ thread_id: "T-after-failure", last_comment_id: "C1", last_comment_at: "t1" }],
    }))
    renameSync(nextFetch, fetch)

    const existingResult = await existingWatch.result
    expect(existingResult.code, existingResult.stderr).toBe(0)
    const wake = JSON.parse(existingResult.stdout.trim())
    expect(wake.reason).toBe("actionable")
    expect(wake.watch_generation).toBe(activeGeneration)
  }, 15000)

  test("watch: an existing stop file wakes before reservation or preflight", () => {
    const sd = path.join(dir, "watch-stopped-before-arm")
    const fetch = fetchFile(dir, "watch-stopped-before-arm.json", {
      ...FAILING,
      head_sha: "incumbent-head",
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    })
    snapshot(sd, fetch)
    const statePath = path.join(sd, "state.json")
    const incumbent = JSON.parse(readFileSync(statePath, "utf8"))
    incumbent.watch_generation = "incumbent-generation"
    incumbent.watch_pid = 999999
    incumbent.watch_process_identity = "incumbent-identity"
    const before = JSON.stringify(incumbent)
    writeFileSync(statePath, before)

    const stopFile = path.join(dir, "watch-stopped-before-arm.stop")
    writeFileSync(stopFile, "stop")
    const missingFetch = path.join(dir, "watch-stopped-before-arm-must-not-fetch.json")
    const r = spawnSync(
      "python3",
      [SCRIPT, "watch", "--pr", "1", "--repo", "o/r", "--state-dir", sd,
        "--fetch-file", missingFetch, "--stop-file", stopFile,
        "--invocation-id", incumbent.invocation_id,
        "--session-started-at", incumbent.started_at,
        "--invocation-budget-seconds", String(incumbent.invocation_budget_seconds)],
      { encoding: "utf8", timeout: 5000 },
    )

    expect(r.status, r.stderr).toBe(0)
    expect(JSON.parse(r.stdout.trim())).toMatchObject({
      event: "BABYSIT_WAKE",
      reason: "stop-signal",
      watch_generation: "incumbent-generation",
    })
    expect(readFileSync(statePath, "utf8")).toBe(before)
    expect(existsSync(path.join(sd, "watch-candidate.json"))).toBe(false)
  })

  test("watch: a newer invocation supersedes an older candidate with a slow preflight", async () => {
    const sd = path.join(dir, "watch-candidate-order")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const initial = fetchFile(dir, "watch-candidate-initial.json", running)
    snapshot(sd, initial)

    const slowFetch = path.join(dir, "watch-candidate-slow.fifo")
    const mkfifo = spawnSync("mkfifo", [slowFetch], { encoding: "utf8" })
    expect(mkfifo.status, mkfifo.stderr).toBe(0)
    const olderCandidate = startWatch(sd, slowFetch)
    await Bun.sleep(200) // the older invocation is blocked in its first fetch

    const fastFetch = fetchFile(dir, "watch-candidate-fast.json", running)
    const newerCandidate = startWatch(sd, fastFetch)
    const activeGeneration = await waitForWatchGeneration(sd)
    const olderStopped = await Promise.race([
      olderCandidate.result.then(() => true),
      Bun.sleep(1000).then(() => false),
    ])
    if (!olderStopped) {
      olderCandidate.child.kill("SIGKILL")
      newerCandidate.child.kill("SIGTERM")
      await Promise.all([olderCandidate.result, newerCandidate.result])
    }
    expect(olderStopped).toBe(true)

    const nextFetch = `${fastFetch}.next`
    writeFileSync(nextFetch, JSON.stringify({
      ...running,
      threads: [{ thread_id: "T-candidate", last_comment_id: "C1", last_comment_at: "t1" }],
    }))
    renameSync(nextFetch, fastFetch)
    const newerResult = await newerCandidate.result
    expect(newerResult.code, newerResult.stderr).toBe(0)
    const wake = JSON.parse(newerResult.stdout.trim())
    expect(wake.watch_generation).toBe(activeGeneration)
  }, 15000)

  test("watch: an explicit invocation replacement emits a non-action supersession wake", async () => {
    const sd = path.join(dir, "watch-invocation-superseded")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "watch-invocation-superseded.json", running)
    const initial = snapshot(sd, fetch, ["--start-invocation", "--invocation-budget-seconds", "28800"])
    const oldWatch = startWatch(sd, fetch)
    await waitForWatchGeneration(sd)

    const replacement = snapshot(sd, fetch, ["--start-invocation", "--invocation-budget-seconds", "12345"])
    const replacementClock = {
      invocation_id: replacement.invocation_id,
      started_at: replacement.invocation_started_at,
      invocation_budget_seconds: replacement.invocation_budget_seconds,
    }

    const result = await oldWatch.result
    expect(result.code, result.stderr).toBe(0)
    const wake = JSON.parse(result.stdout.trim())
    expect(wake).toEqual({
      event: "BABYSIT_WAKE",
      reason: "invocation-superseded",
      watch_generation: expect.any(String),
      superseded_invocation_id: initial.invocation_id,
      current_invocation_id: replacement.invocation_id,
    })

    const persisted = JSON.parse(readFileSync(path.join(sd, "state.json"), "utf8"))
    expect({
      invocation_id: persisted.invocation_id,
      started_at: persisted.started_at,
      invocation_budget_seconds: persisted.invocation_budget_seconds,
    }).toEqual(replacementClock)
  }, 15000)

  test("watch: an expired old budget cannot emit max-runtime after invocation replacement", async () => {
    const sd = path.join(dir, "watch-expired-invocation-superseded")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "watch-expired-invocation-superseded.json", running)
    const initial = snapshot(sd, fetch, EXPIRING_TEST_INVOCATION)
    const oldWatch = startWatch(sd, fetch, [
      "--interval", "5",
      "--invocation-id", initial.invocation_id,
      "--session-started-at", initial.invocation_started_at,
      "--invocation-budget-seconds", "1",
    ])
    await waitForWatchGeneration(sd)

    const replacement = snapshot(sd, fetch, ["--start-invocation", "--invocation-budget-seconds", "28800"])
    const result = await oldWatch.result
    expect(result.code, result.stderr).toBe(0)
    expect(JSON.parse(result.stdout.trim())).toEqual({
      event: "BABYSIT_WAKE",
      reason: "invocation-superseded",
      watch_generation: expect.any(String),
      superseded_invocation_id: initial.invocation_id,
      current_invocation_id: replacement.invocation_id,
    })
  }, 15000)

  test("watch: takeover interrupts an old watcher blocked in its next fetch", async () => {
    const sd = path.join(dir, "watch-blocked-fetch")
    const running = {
      ...FAILING,
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const oldFetch = fetchFile(dir, "watch-blocked-old.json", running)
    snapshot(sd, oldFetch)

    const oldWatch = startWatch(sd, oldFetch, ["--interval", "0.2"])
    const oldGeneration = await waitForWatchGeneration(sd)
    const fifo = `${oldFetch}.fifo`
    const mkfifo = spawnSync("mkfifo", [fifo], { encoding: "utf8" })
    expect(mkfifo.status, mkfifo.stderr).toBe(0)
    renameSync(fifo, oldFetch)
    await Bun.sleep(300) // the old generation is now blocked opening the FIFO for its next fetch

    const replacementFetch = fetchFile(dir, "watch-blocked-new.json", running)
    const replacement = startWatch(sd, replacementFetch)
    await waitForWatchGeneration(sd, oldGeneration)

    const stoppedPromptly = await Promise.race([
      oldWatch.result.then(() => true),
      Bun.sleep(1000).then(() => false),
    ])
    if (!stoppedPromptly) oldWatch.child.kill("SIGKILL")
    expect(stoppedPromptly).toBe(true)
    replacement.child.kill("SIGTERM")
    await replacement.result
  }, 15000)

  test("watch: an in-flight poll cannot persist after its generation becomes stale", async () => {
    const sd = path.join(dir, "watch-stale-poll")
    const running = {
      ...FAILING,
      head_sha: "current-head",
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    }
    const fetch = fetchFile(dir, "watch-stale-poll.json", running)
    snapshot(sd, fetch)
    const oldWatch = startWatch(sd, fetch, ["--interval", "0.2"])
    await waitForWatchGeneration(sd)

    const fifo = `${fetch}.fifo`
    const mkfifo = spawnSync("mkfifo", [fifo], { encoding: "utf8" })
    expect(mkfifo.status, mkfifo.stderr).toBe(0)
    renameSync(fifo, fetch)
    await Bun.sleep(300)

    const statePath = path.join(sd, "state.json")
    const replacementState = JSON.parse(readFileSync(statePath, "utf8"))
    replacementState.watch_generation = "replacement-generation"
    const nextState = `${statePath}.next`
    writeFileSync(nextState, JSON.stringify(replacementState))
    renameSync(nextState, statePath)
    writeFileSync(fetch, JSON.stringify({ ...running, head_sha: "stale-head" }))

    const result = await oldWatch.result
    expect(result.code, result.stderr).toBe(0)
    const persisted = JSON.parse(readFileSync(statePath, "utf8"))
    expect(persisted.watch_generation).toBe("replacement-generation")
    expect(persisted.head_sha).toBe("current-head")
  }, 15000)

  test("watch: preflight stays read-only until activation fences incumbent persistence", () => {
    const sd = path.join(dir, "watch-preflight-fence")
    const base = fetchFile(dir, "watch-preflight-fence-base.json", {
      ...FAILING,
      head_sha: "base-head",
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    })
    const incumbent = fetchFile(dir, "watch-preflight-fence-incumbent.json", {
      ...FAILING,
      head_sha: "incumbent-head",
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    })
    const successor = fetchFile(dir, "watch-preflight-fence-successor.json", {
      ...FAILING,
      head_sha: "successor-head",
      threads: [],
      checks: [{ key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }],
    })
    snapshot(sd, base)
    const statePath = path.join(sd, "state.json")
    const active = JSON.parse(readFileSync(statePath, "utf8"))
    active.watch_generation = "incumbent-generation"
    writeFileSync(statePath, JSON.stringify(active))

    const python = `
import json, subprocess, time
from importlib.machinery import SourceFileLoader
from types import SimpleNamespace
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
state_dir = ${JSON.stringify(sd)}
state_path = ${JSON.stringify(statePath)}
invocation = json.load(open(state_path))
args = SimpleNamespace(state_dir=state_dir, pr=1, repo="o/r",
                       fetch_file=${JSON.stringify(successor)}, reset_session=False,
                       start_invocation=False, continue_invocation=False,
                       invocation_id=invocation["invocation_id"],
                       session_started_at=invocation["started_at"],
                       invocation_budget_seconds=invocation["invocation_budget_seconds"])
generation = "successor-generation"
m._reserve_watch_candidate(args, generation)
cur = m._fetch_snapshot(args)
assert json.load(open(state_path))["head_sha"] == "base-head", "preflight mutated persisted state"

original_diff = m.diff
child = None
def diff_with_incumbent_race(state, current, now=None, advance_trajectory=True):
    global child
    child_code = '''
import json
from importlib.machinery import SourceFileLoader
from types import SimpleNamespace
m = SourceFileLoader("prs_child", ${JSON.stringify(SCRIPT)}).load_module()
invocation = json.load(open(${JSON.stringify(statePath)}))
args = SimpleNamespace(state_dir=${JSON.stringify(sd)}, pr=1, repo="o/r",
                       fetch_file=${JSON.stringify(incumbent)}, reset_session=False,
                       start_invocation=False, continue_invocation=False,
                       invocation_id=invocation["invocation_id"],
                       session_started_at=invocation["started_at"],
                       invocation_budget_seconds=invocation["invocation_budget_seconds"])
try:
    m._run_snapshot(args, m._now(), advance_trajectory=False,
                    watch_generation="incumbent-generation")
except m._WatchSuperseded:
    pass
else:
    raise SystemExit("stale incumbent persist was not rejected")
'''
    child = subprocess.Popen(["python3", "-c", child_code], stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, text=True)
    time.sleep(0.2)
    assert child.poll() is None, "incumbent was not blocked by atomic activation"
    return original_diff(state, current, now, advance_trajectory=advance_trajectory)

m.diff = diff_with_incumbent_race
previous, actionable = m._activate_watch(args, generation, m._now(), cur)
stdout, stderr = child.communicate(timeout=5)
assert child.returncode == 0, stderr
persisted = json.load(open(state_path))
print(json.dumps({"generation": persisted["watch_generation"], "head": persisted["head_sha"]}))
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8", timeout: 10000 })
    expect(r.status, r.stderr).toBe(0)
    expect(JSON.parse(r.stdout)).toEqual({ generation: "successor-generation", head: "successor-head" })
  })

  test("watch: PID identity must still match before a replaced watcher is signaled", () => {
    const python = `
import json
from importlib.machinery import SourceFileLoader
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
signals = []
m.os.kill = lambda pid, sig: signals.append([pid, sig])
m._process_identity = lambda pid: {123: "different", 124: None, 125: "same"}.get(pid)
for pid, identity in ((123, "old"), (124, "old"), (125, "same")):
    m._terminate_replaced_watch({"pid": pid, "process_identity": identity})
print(json.dumps(signals))
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    expect(JSON.parse(r.stdout)).toEqual([[125, 15]])
  })

  test("watch: takeover interrupts and reaps an active fetch subprocess", () => {
    const childPid = path.join(dir, "watch-fetch-child.pid")
    const python = `
import os, signal, subprocess, threading, time
from importlib.machinery import SourceFileLoader
from types import SimpleNamespace
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
pid_file = ${JSON.stringify(childPid)}
def fake_snapshot(args, now, advance_trajectory=True, watch_generation=None):
    subprocess.run(["sh", "-c", "echo $$ > " + pid_file + "; exec sleep 30"], check=True)
    return {"counts": {}, "pr_state": "OPEN", "session_seconds": 0}
def stop_when_child_starts():
    deadline = time.time() + 5
    while time.time() < deadline and not os.path.exists(pid_file):
        time.sleep(0.01)
    os.kill(os.getpid(), signal.SIGTERM)
m._run_snapshot = fake_snapshot
m._fetch_snapshot = lambda args: {}
m._reserve_watch_candidate = lambda args, generation: {}
m._clear_watch_candidate = lambda args, generation: None
m._activate_watch = lambda args, generation, now, cur: (
    {}, {"counts": {}, "pr_state": "OPEN", "session_seconds": 0})
m._terminate_replaced_watch = lambda previous: None
m._watch_is_current = lambda args, generation: True
m._wake_reason = lambda actionable, settle_seconds: None
threading.Thread(target=stop_when_child_starts, daemon=True).start()
args = SimpleNamespace(reset_session=False, stop_file=None, settle_seconds=300, max_runtime=0,
                       interval=0.01, state_dir=${JSON.stringify(dir)}, pr=1, repo="o/r")
started = time.time()
m.cmd_watch(args)
pid = int(open(pid_file).read())
alive = True
try:
    os.kill(pid, 0)
except ProcessLookupError:
    alive = False
print(f"{alive} {time.time() - started:.3f}")
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8", timeout: 5000 })
    expect(r.status, r.stderr).toBe(0)
    const [alive, elapsed] = r.stdout.trim().split(" ")
    expect(alive).toBe("False")
    expect(Number(elapsed)).toBeLessThan(2)
  })

  test("watch: stale teardown ignores a late takeover SIGTERM and ordinary teardown restores it", () => {
    const python = `
import json, os, signal, threading, time
from importlib.machinery import SourceFileLoader
from types import SimpleNamespace
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
args = SimpleNamespace(reset_session=False, stop_file=None, settle_seconds=300, max_runtime=0,
                       interval=0.01, state_dir=${JSON.stringify(dir)}, pr=1, repo="o/r")
actionable = {"counts": {}, "pr_state": "OPEN", "session_seconds": 0}
m._reserve_watch_candidate = lambda args, generation: {}
m._clear_watch_candidate = lambda args, generation: None
m._fetch_snapshot = lambda args: {}
m._activate_watch = lambda args, generation, now, cur: ({}, actionable)
m._terminate_replaced_watch = lambda previous: None
m._emit_wake_if_current = lambda *args, **kwargs: True

real_signal = signal.signal
signal_calls = 0
final_handler_installed = threading.Event()
def track_signal(signum, handler):
    global signal_calls
    result = real_signal(signum, handler)
    if signum == signal.SIGTERM:
        signal_calls += 1
        if signal_calls == 2:
            final_handler_installed.set()
            time.sleep(0.2)
    return result
def send_late_takeover_signal():
    if not final_handler_installed.wait(2):
        os._exit(2)
    os.kill(os.getpid(), signal.SIGTERM)

m.signal.signal = track_signal
m._watch_is_current = lambda args, generation: False
sender = threading.Thread(target=send_late_takeover_signal)
sender.start()
m.cmd_watch(args)
sender.join(timeout=2)
assert not sender.is_alive()

m.signal.signal = real_signal
def caller_handler(_signum, _frame):
    pass
real_signal(signal.SIGTERM, caller_handler)
m._watch_is_current = lambda args, generation: True
m._wake_reason = lambda actionable, settle_seconds: "actionable"
m.cmd_watch(args)
print(json.dumps({"ordinary_restored": signal.getsignal(signal.SIGTERM) is caller_handler}))
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8", timeout: 5000 })
    expect(r.status, r.stderr).toBe(0)
    expect(JSON.parse(r.stdout)).toEqual({ ordinary_restored: true })
  })

  test("fetch_threads follows every GraphQL page before returning unresolved threads", () => {
    const python = `
import json
from importlib.machinery import SourceFileLoader
m = SourceFileLoader("prs", ${JSON.stringify(SCRIPT)}).load_module()
calls = []
class Result: pass
def fake(args, label):
    calls.append(args)
    second = any(arg == "cursor=page-2" for arg in args)
    node = {"id": "T2" if second else "T1", "isResolved": False, "path": "x", "line": 1,
            "comments": {"nodes": [{"id": "C2" if second else "C1", "createdAt": "t2" if second else "t1", "lastEditedAt": None}]}}
    page = {"nodes": [node], "pageInfo": {"hasNextPage": not second, "endCursor": None if second else "page-2"}}
    result = Result()
    result.returncode = 0
    result.stderr = ""
    result.stdout = json.dumps({"data": {"repository": {"pullRequest": {"reviewThreads": page}}}})
    return result
m._run_checked = fake
threads = m.fetch_threads(1, "o", "r")
print(json.dumps({"ids": [t["thread_id"] for t in threads], "calls": calls}))
`
    const r = spawnSync("python3", ["-c", python], { encoding: "utf8" })
    expect(r.status, r.stderr).toBe(0)
    const result = JSON.parse(r.stdout)
    expect(result.ids).toEqual(["T1", "T2"])
    expect(result.calls).toHaveLength(2)
    expect(result.calls[1]).toContain("cursor=page-2")
  })

  test("watch: managed target freshness blocks ordinary CLEAN merge-ready", () => {
    const GREEN = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    const managedStale = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [GREEN],
      threads: [],
      pr_chain: {
        manager_status: "confirmed",
        manager_source: "gh-stack",
        relationship_status: "dependent",
        target_needs_rebase: true,
        upstack_needs_rebase: [],
      },
    }
    expect(wakeReason(snapshot(path.join(dir, "stack-stale"), fetchFile(dir, "stack-stale.json", managedStale)))).toBe("stack-blocked")
  }, 15000)

  test("watch: unknown managed freshness blocks ready, while stale upstack alone still permits ready-as-next", () => {
    const GREEN = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    const base = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [GREEN],
      threads: [],
    }
    const unknown = {
      ...base,
      pr_chain: {
        manager_status: "confirmed",
        manager_source: "graphql",
        relationship_status: "dependent",
        target_needs_rebase: null,
        upstack_needs_rebase: [],
      },
    }
    expect(wakeReason(snapshot(path.join(dir, "stack-unknown"), fetchFile(dir, "stack-unknown.json", unknown)))).toBe("stack-blocked")

    const readyAsNext = {
      ...base,
      pr_chain: {
        manager_status: "confirmed",
        manager_source: "gh-stack",
        relationship_status: "dependent",
        target_needs_rebase: false,
        upstack_needs_rebase: [{ number: 43, position: 3 }],
      },
    }
    expect(wakeReason(snapshot(path.join(dir, "stack-up"), fetchFile(dir, "stack-up.json", readyAsNext)))).toBe("merge-ready")
  }, 15000)

  test("watch: manager probe error is a residual, not an unmanaged merge-ready fallback", () => {
    const GREEN = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    const probeError = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [GREEN],
      threads: [],
      pr_chain: {
        manager_status: "probe-error",
        manager_source: null,
        relationship_status: "independent",
        target_needs_rebase: null,
        upstack_needs_rebase: [],
      },
    }
    expect(wakeReason(snapshot(path.join(dir, "stack-error"), fetchFile(dir, "stack-error.json", probeError)))).toBe("stack-blocked")
  }, 15000)

  test("watch: unresolved ordinary relationship classification also blocks an independent-readiness claim", () => {
    const GREEN = { key: "CI/test", name: "test", status: "COMPLETED", conclusion: "SUCCESS", details_url: "u" }
    const relationshipError = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      checks: [GREEN],
      threads: [],
      pr_chain: {
        manager_status: "absent",
        manager_source: null,
        relationship_status: "probe-error",
        target_needs_rebase: null,
        upstack_needs_rebase: [],
      },
    }
    expect(wakeReason(snapshot(path.join(dir, "relationship-error"), fetchFile(dir, "relationship-error.json", relationshipError)))).toBe("stack-blocked")
  }, 15000)

  test("watch: labels a comments-only wake as a feedback candidate while CI is running", () => {
    const RUNNING = { key: "CI/test", name: "test", status: "IN_PROGRESS", conclusion: null, details_url: "u" }
    const candidate = {
      ...FAILING,
      threads: [],
      checks: [RUNNING],
      feedback: [{ id: "IC_status", kind: "comment", author: "review-bot", edit_id: "status-v1" }],
    }
    expect(watch(path.join(dir, "wfc"), fetchFile(dir, "wfc.json", candidate)).reason).toBe("feedback-candidate")
  }, 15000)

  test("watch: an in-progress review signal blocks until the 15-minute stale-review check", () => {
    const base = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      threads: [],
      checks: [GREEN_CHECK],
      counts: { threads: 0, ci: 0, comments: 0 },
      checks_terminal: true,
      has_failing_checks: false,
      checks_awaiting_approval: 0,
      open_needs_human: 0,
      stack_blocker: null,
    }
    expect(wakeReason({ ...base, review_in_progress: true, quiet_seconds: 899 }, 0)).toBeNull()
    expect(wakeReason({ ...base, review_in_progress: true, quiet_seconds: 900 }, 0)).toBe("merge-ready")
    expect(wakeReason({ ...base, review_in_progress: false, quiet_seconds: 0 }, 0)).toBe("merge-ready")
  })

  test("snapshot: remembers an incomplete current-head review after the eyes signal disappears", () => {
    const base = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      threads: [],
      checks: [GREEN_CHECK],
    }

    snapshot(state, fetchFile(dir, "signal-absent.json", { ...base, review_in_progress: false }))
    const statePath = path.join(state, "state.json")
    const prior = JSON.parse(readFileSync(statePath, "utf8"))
    prior.last_change_at = "2026-07-17T12:00:00+00:00"
    writeFileSync(statePath, JSON.stringify(prior))

    const started = snapshot(state, fetchFile(dir, "signal-started.json", { ...base, review_in_progress: true }))
    expect(started.review_signal_seen_on_head).toBe(true)
    expect(started.review_signal_first_seen_at).toBe(started.review_signal_last_changed_at)
    expect(started.review_signal_first_seen_at).not.toBe(prior.last_change_at)
    const firstSeenAt = started.review_signal_first_seen_at

    const disappeared = snapshot(state, fetchFile(dir, "signal-disappeared.json", { ...base, review_in_progress: false }))
    expect(disappeared.review_in_progress).toBe(false)
    expect(disappeared.review_signal_seen_on_head).toBe(true)
    expect(disappeared.review_signal_first_seen_at).toBe(firstSeenAt)
    expect(disappeared.review_signal_last_changed_at).not.toBe(firstSeenAt)
    expect(disappeared.changed_this_tick).toBe(true)
    expect(disappeared.quiet_seconds).toBeLessThan(2)

    const nextHead = snapshot(state, fetchFile(dir, "signal-new-head.json", {
      ...base,
      head_sha: "s2",
      review_in_progress: false,
    }))
    expect(nextHead.review_signal_seen_on_head).toBe(false)
    expect(nextHead.review_signal_first_seen_at).toBeNull()
    expect(nextHead.review_signal_last_changed_at).toBeNull()
  })

  test("snapshot: eyes identity changes reset quiet time even when the count stays fixed", () => {
    expect(eyesReactionIdentities([[
      { content: "eyes", user: { node_id: "U_bot_b" } },
      { content: "eyes", user: { node_id: "U_bot_a" } },
      { content: "+1", user: { node_id: "U_other" } },
    ]])).toEqual(["U_bot_a", "U_bot_b"])

    const base = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      threads: [],
      checks: [GREEN_CHECK],
      review_in_progress: true,
    }
    const first = snapshot(state, fetchFile(dir, "signal-count-one.json", {
      ...base,
      review_signal_identities: ["U_bot_a"],
    }))
    expect(first.review_signal_count).toBe(1)
    expect(first.review_signal_identities).toEqual(["U_bot_a"])
    const statePath = path.join(state, "state.json")
    const prior = JSON.parse(readFileSync(statePath, "utf8"))
    prior.last_change_at = "2026-07-17T12:00:00+00:00"
    prior.review_signal_last_changed_at = prior.last_change_at
    writeFileSync(statePath, JSON.stringify(prior))

    const swapped = snapshot(state, fetchFile(dir, "signal-reviewer-swapped.json", {
      ...base,
      review_signal_identities: ["U_bot_b"],
    }))
    expect(swapped.review_in_progress).toBe(true)
    expect(swapped.review_signal_count).toBe(1)
    expect(swapped.review_signal_identities).toEqual(["U_bot_b"])
    expect(swapped.review_signal_last_changed_at).not.toBe(prior.review_signal_last_changed_at)
    expect(swapped.changed_this_tick).toBe(true)
    expect(swapped.quiet_seconds).toBeLessThan(2)

    const persisted = JSON.parse(readFileSync(statePath, "utf8"))
    persisted.last_change_at = "2026-07-17T12:00:00+00:00"
    persisted.review_signal_last_changed_at = persisted.last_change_at
    writeFileSync(statePath, JSON.stringify(persisted))

    const unchanged = snapshot(state, fetchFile(dir, "signal-reviewer-unchanged.json", {
      ...base,
      review_signal_identities: ["U_bot_b"],
    }))
    expect(unchanged.review_in_progress).toBe(true)
    expect(unchanged.review_signal_count).toBe(1)
    expect(unchanged.changed_this_tick).toBe(false)
    expect(unchanged.quiet_seconds).toBeGreaterThan(60)
  })

  test("snapshot: a count-only legacy review signal migrates to reactor identities", () => {
    const base = {
      ...FAILING,
      merge_state_status: "CLEAN",
      review_decision: "APPROVED",
      threads: [],
      checks: [GREEN_CHECK],
      review_in_progress: true,
      review_signal_count: 1,
    }
    snapshot(state, fetchFile(dir, "signal-legacy-count.json", base))
    const statePath = path.join(state, "state.json")
    const legacy = JSON.parse(readFileSync(statePath, "utf8"))
    delete legacy.review_signal_identities
    legacy.last_change_at = "2026-07-17T12:00:00+00:00"
    legacy.review_signal_last_changed_at = legacy.last_change_at
    writeFileSync(statePath, JSON.stringify(legacy))

    const migrated = snapshot(state, fetchFile(dir, "signal-identity-aware.json", {
      ...base,
      review_signal_identities: ["U_bot_a"],
    }))
    expect(migrated.review_in_progress).toBe(true)
    expect(migrated.review_signal_count).toBe(1)
    expect(migrated.review_signal_identities).toEqual(["U_bot_a"])
    expect(migrated.review_signal_seen_on_head).toBe(true)
    expect(migrated.changed_this_tick).toBe(true)
    expect(migrated.quiet_seconds).toBeLessThan(2)
  })

  test("watch: a no-check MERGEABLE/CLEAN PR still reaches merge-ready (the >=1-check guard is pipeline-only)", () => {
    // A repo with no configured checks: all_checks_ok is false (no observed check), but the
    // interactive merge-ready wake must still fire for a CLEAN/MERGEABLE PR with no backlog.
    const nochecks = { ...FAILING, merge_state_status: "CLEAN", review_decision: "APPROVED", threads: [], checks: [] }
    expect(watch(path.join(dir, "nc1"), fetchFile(dir, "nc1.json", nochecks), ["--settle-seconds", "0"]).reason).toBe("merge-ready")
  }, 15000)

  test("watch: a dispatched terminal-red check present at arm is a standing residual — kept watching, not re-woken", () => {
    // A failing check ce-debug marked dispatched leaves counts.ci == 0 while has_failing_checks stays
    // true. It was already surfaced when it was dispatched, so it is in the watch's arm-time baseline
    // and must NOT re-wake the loop (that was the pre-gating behavior); the watch keeps running for
    // other streams. `blocked-failing` only fires on a *later* transition to terminal-red (e.g. a
    // rerun completing red) — the same wake-on-new path the parked-needs-human test exercises.
    const red = { ...FAILING, threads: [], checks: [{ key: "CI/test", name: "test", status: "COMPLETED", conclusion: "FAILURE", details_url: "u" }] }
    const rf = fetchFile(dir, "wbf.json", red)
    const sd = path.join(dir, "wbf")
    snapshot(sd, rf, EXPIRING_TEST_INVOCATION) // this standing-residual watch deliberately expires
    mark(sd, ["--check", "CI/test"]) // now dispatched -> counts.ci == 0, terminal-red residual, already surfaced
    expect(watch(sd, rf).reason).toBe("max-runtime")
  }, 15000)

  test("watch: a parked needs-human does not wake or end the loop — it keeps watching the other streams", () => {
    // The stop-vs-residual fix: a standing needs-human present at arm time must NOT re-wake the
    // detector (that would busy-wake / falsely terminate the self-sustaining watch); the watch keeps
    // polling for new work and only wakes when something genuinely new arrives.
    const sd = path.join(dir, "nhwatch")
    const base = (extra: any[] = []) => ({
      pr_state: "OPEN", mergeable: "MERGEABLE", merge_state_status: "CLEAN", review_decision: null,
      head_sha: "s1", url: "http://x/1", checks: [],
      threads: [{ thread_id: "T1", last_comment_id: "C1", last_comment_at: "C1" }, ...extra],
    })
    snapshot(sd, fetchFile(dir, "nhw1.json", base()), EXPIRING_TEST_INVOCATION)
    mark(sd, ["--thread", "T1", "--disposition", "needs-human"])
    // parked needs-human, nothing else actionable -> keeps watching, times out (does NOT wake needs-human)
    expect(watch(sd, fetchFile(dir, "nhw2.json", base())).reason).toBe("max-runtime")
    // The capped invocation is over. A later explicit invocation preserves the parked disposition,
    // while a new actionable thread still wakes that new invocation.
    const withNew = fetchFile(dir, "nhw3.json", base([{ thread_id: "T2", last_comment_id: "D1", last_comment_at: "D1" }]))
    snapshot(sd, withNew, ["--start-invocation"])
    expect(watch(sd, withNew).reason).toBe("actionable")
  }, 15000)
})