From 585fb6027cb1ddde79b67f3bb460094f541174cf Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Wed, 26 Aug 2026 23:42:53 -0400
Subject: [PATCH 1/6] fix: start crosscheck on PR registration

---
 bin/fm-crosscheck-autostart.py           | 815 +++++++++++++++++++++++
 bin/fm-crosscheck.py                     |  32 +-
 bin/fm-crosscheck.sh                     |   6 +-
 bin/fm-pr-check.sh                       |  51 +-
 docs/crosscheck.md                       |  46 +-
 tests/behavior-test-durations.tsv        |   1 +
 tests/fm-crosscheck.test.sh              |  26 +
 tests/fm-pr-crosscheck-autostart.test.sh | 429 ++++++++++++
 tests/fm-pr-merge.test.sh                |   1 +
 tests/fm-secondmate-safety.test.sh       |   1 +
 tests/fm-teardown-suite.sh               |   1 +
 tests/test-capabilities.tsv              |   1 +
 12 files changed, 1385 insertions(+), 25 deletions(-)
 create mode 100755 bin/fm-crosscheck-autostart.py
 create mode 100755 tests/fm-pr-crosscheck-autostart.test.sh

diff --git a/bin/fm-crosscheck-autostart.py b/bin/fm-crosscheck-autostart.py
new file mode 100755
index 00000000000..b1481f5eb1a
--- /dev/null
+++ b/bin/fm-crosscheck-autostart.py
@@ -0,0 +1,815 @@
+#!/usr/bin/env python3
+"""Coordinate prompt-return Crosscheck launches for PR-ready registration.
+
+`fm-pr-check.sh` owns the public registration surface.
+This helper persists the latest requested exact head, keeps one coordinator per task, and runs Crosscheck outside the caller.
+Coordinator and operation locks are task-local, while the existing Azure lane and cost admission remain the spending authority.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+from typing import Any, Dict, NoReturn, Optional, Tuple
+
+
+SCHEMA = "firstmate.crosscheck-autostart.v1"
+REQUEST_SCHEMA = "firstmate.crosscheck-autostart-request.v1"
+ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
+SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+PR_RE = re.compile(r"^https://github\.com/[^/\s]+/[^/\s]+/pull/[1-9][0-9]*$")
+MAX_RECORD_BYTES = 64 * 1024
+MAX_FLEET_ENV_BYTES = 1024 * 1024
+MAX_COMMAND_OUTPUT_BYTES = 256 * 1024
+MAX_LOG_BYTES = 2 * 1024 * 1024
+DEFAULT_ACTIVE_WAIT_SECONDS = 4 * 60 * 60
+DEFAULT_COMMAND_TIMEOUT_SECONDS = 4 * 60 * 60
+MAX_HEAD_RESTARTS = 16
+
+
+class AutostartError(RuntimeError):
+    """Raised when the task-local launcher cannot honestly start or clear review."""
+
+
+def fail(message: str) -> NoReturn:
+    raise AutostartError(message)
+
+
+def utc_now() -> str:
+    return (
+        dt.datetime.now(dt.timezone.utc)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def bounded_text(value: str, limit: int = 500) -> str:
+    flattened = " ".join(value.replace("\r", "\n").splitlines())
+    if len(flattened) <= limit:
+        return flattened
+    return flattened[: limit - 3] + "..."
+
+
+def positive_int_environment(name: str, default: int) -> int:
+    raw = os.environ.get(name, str(default))
+    if not raw.isdigit() or int(raw) <= 0:
+        fail(f"{name} must be a positive integer")
+    return int(raw)
+
+
+def absolute_without_symlink_resolution(value: str) -> Path:
+    return Path(os.path.abspath(os.path.expanduser(value)))
+
+
+def runtime_paths() -> Tuple[Path, Path, Path]:
+    script = Path(__file__).resolve()
+    default_root = script.parent.parent
+    root = Path(os.environ.get("FM_ROOT_OVERRIDE", str(default_root))).resolve()
+    home = Path(os.environ.get("FM_HOME", str(root))).resolve()
+    state = absolute_without_symlink_resolution(
+        os.environ.get("FM_STATE_OVERRIDE", str(home / "state"))
+    )
+    try:
+        metadata = state.lstat()
+    except FileNotFoundError:
+        state.mkdir(parents=True, mode=0o700)
+        metadata = state.lstat()
+    except OSError as exc:
+        fail(f"Crosscheck autostart state inspection failed at {state}: {exc}")
+    if not stat.S_ISDIR(metadata.st_mode) or state.is_symlink():
+        fail(f"Crosscheck autostart state is not a real directory: {state}")
+    return root, home, state
+
+
+def validate_identity(task_id: str, url: str, head: str, generation: str) -> None:
+    if ID_RE.fullmatch(task_id) is None:
+        fail(f"invalid task id: {task_id!r}")
+    if PR_RE.fullmatch(url) is None:
+        fail(f"invalid full GitHub PR URL: {url!r}")
+    if SHA_RE.fullmatch(head) is None:
+        fail(f"invalid exact PR head: {head!r}")
+    if (
+        not generation
+        or len(generation.encode("utf-8")) > 512
+        or any(character in generation for character in "\0\r\n")
+    ):
+        fail("invalid task generation identity")
+
+
+def task_paths(state: Path, task_id: str) -> Dict[str, Path]:
+    return {
+        "request": state / f"{task_id}.crosscheck-autostart.request.json",
+        "state": state / f"{task_id}.crosscheck-autostart.json",
+        "log": state / f"{task_id}.crosscheck-autostart.log",
+        "coordinator_lock": state / f".{task_id}.crosscheck-autostart.lock",
+        "crosscheck_lock": state / f".{task_id}.crosscheck.lock",
+    }
+
+
+def atomic_json(path: Path, value: Dict[str, Any]) -> None:
+    encoded = (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
+    if len(encoded) > MAX_RECORD_BYTES:
+        fail(f"Crosscheck autostart JSON exceeds its bound at {path}")
+    descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    temporary = Path(temporary_name)
+    try:
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "wb", closefd=True) as stream:
+            stream.write(encoded)
+            stream.flush()
+            os.fsync(stream.fileno())
+        if path.exists() or path.is_symlink():
+            metadata = path.lstat()
+            if not stat.S_ISREG(metadata.st_mode) or path.is_symlink():
+                fail(f"unsafe Crosscheck autostart destination: {path}")
+        os.replace(temporary, path)
+    finally:
+        try:
+            temporary.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def load_json(path: Path, schema: str) -> Optional[Dict[str, Any]]:
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except FileNotFoundError:
+        return None
+    except OSError as exc:
+        fail(f"cannot open Crosscheck autostart record {path}: {exc}")
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            fail(f"unsafe Crosscheck autostart record: {path}")
+        if metadata.st_size > MAX_RECORD_BYTES:
+            fail(f"Crosscheck autostart record exceeds its bound: {path}")
+        raw = os.read(descriptor, MAX_RECORD_BYTES + 1)
+    finally:
+        os.close(descriptor)
+    if len(raw) > MAX_RECORD_BYTES:
+        fail(f"Crosscheck autostart record exceeds its bound: {path}")
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeError, json.JSONDecodeError) as exc:
+        fail(f"cannot read Crosscheck autostart record {path}: {exc}")
+    if not isinstance(value, dict) or value.get("schema") != schema:
+        fail(f"invalid Crosscheck autostart record at {path}")
+    return value
+
+
+def request_record(
+    task_id: str, url: str, head: str, generation: str
+) -> Dict[str, Any]:
+    return {
+        "schema": REQUEST_SCHEMA,
+        "task_id": task_id,
+        "pull_request": url,
+        "head_sha": head,
+        "generation_id": generation,
+        "requested_at": utc_now(),
+    }
+
+
+def write_state(
+    path: Path,
+    task_id: str,
+    url: str,
+    head: str,
+    generation: str,
+    lifecycle: str,
+    attempt: int,
+    message: str,
+    pid: int,
+    log: Path,
+) -> None:
+    if lifecycle not in {"starting", "running", "clear", "failed"}:
+        fail(f"invalid Crosscheck autostart state: {lifecycle}")
+    atomic_json(
+        path,
+        {
+            "schema": SCHEMA,
+            "task_id": task_id,
+            "pull_request": url,
+            "head_sha": head,
+            "generation_id": generation,
+            "state": lifecycle,
+            "attempt": attempt,
+            "pid": pid,
+            "updated_at": utc_now(),
+            "message": bounded_text(message),
+            "log": log.name,
+        },
+    )
+
+
+def open_lock(path: Path) -> int:
+    flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags, 0o600)
+    except OSError as exc:
+        fail(f"Crosscheck autostart lock open failed at {path}: {exc}")
+    metadata = os.fstat(descriptor)
+    if not stat.S_ISREG(metadata.st_mode):
+        os.close(descriptor)
+        fail(f"unsafe Crosscheck autostart lock: {path}")
+    return descriptor
+
+
+def fleet_environment_path() -> Path:
+    configured = os.environ.get("FM_CROSSCHECK_FLEET_ENV", "")
+    if configured:
+        path = Path(os.path.expanduser(configured))
+    else:
+        path = Path.home() / ".fm-azure" / "fleet.env"
+    if not path.is_absolute():
+        fail("Crosscheck fleet environment path must be absolute")
+    return path
+
+
+def open_fleet_environment(path: Path) -> int:
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except FileNotFoundError:
+        fail(f"Crosscheck fleet environment is missing: {path}")
+    except OSError as exc:
+        fail(f"Crosscheck fleet environment cannot be opened safely at {path}: {exc}")
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            fail(f"Crosscheck fleet environment must be a regular non-symlink file: {path}")
+        if metadata.st_uid != os.getuid():
+            fail(f"Crosscheck fleet environment is not owned by the current operator: {path}")
+        if stat.S_IMODE(metadata.st_mode) & 0o022:
+            fail(f"Crosscheck fleet environment is group/world writable: {path}")
+        if metadata.st_size > MAX_FLEET_ENV_BYTES:
+            fail(f"Crosscheck fleet environment exceeds its byte bound: {path}")
+        return descriptor
+    except Exception:
+        os.close(descriptor)
+        raise
+
+
+def crosscheck_command(root: Path) -> Path:
+    command = root / "bin" / "fm-crosscheck.sh"
+    try:
+        metadata = command.lstat()
+    except FileNotFoundError:
+        fail(f"Crosscheck command is missing: {command}")
+    except OSError as exc:
+        fail(f"Crosscheck command inspection failed at {command}: {exc}")
+    if (
+        not stat.S_ISREG(metadata.st_mode)
+        or command.is_symlink()
+        or not os.access(command, os.X_OK)
+    ):
+        fail(f"Crosscheck command is not a real executable file: {command}")
+    return command
+
+
+def next_attempt(state_path: Path, generation: str) -> int:
+    existing = load_json(state_path, SCHEMA)
+    if existing is None or existing.get("generation_id") != generation:
+        return 1
+    value = existing.get("attempt")
+    if not isinstance(value, int) or isinstance(value, bool) or value < 0:
+        fail(f"invalid Crosscheck autostart attempt at {state_path}")
+    return value + 1
+
+
+def open_log_descriptor(path: Path) -> int:
+    flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags, 0o600)
+    except OSError as exc:
+        fail(f"Crosscheck autostart log open failed at {path}: {exc}")
+    metadata = os.fstat(descriptor)
+    if not stat.S_ISREG(metadata.st_mode):
+        os.close(descriptor)
+        fail(f"unsafe Crosscheck autostart log: {path}")
+    return descriptor
+
+
+def append_log(path: Path, label: str, output: bytes) -> None:
+    if len(output) > MAX_COMMAND_OUTPUT_BYTES:
+        output = output[:MAX_COMMAND_OUTPUT_BYTES] + b"\n[output clipped]\n"
+    descriptor = open_log_descriptor(path)
+    try:
+        metadata = os.fstat(descriptor)
+        if metadata.st_size > MAX_LOG_BYTES:
+            os.ftruncate(descriptor, 0)
+        os.write(descriptor, f"\n[{utc_now()}] {label}\n".encode())
+        if output:
+            os.write(descriptor, output)
+            if not output.endswith(b"\n"):
+                os.write(descriptor, b"\n")
+    finally:
+        os.close(descriptor)
+
+
+def command_environment() -> Dict[str, str]:
+    return os.environ.copy()
+
+
+def run_crosscheck_command(
+    fleet_env: Path,
+    command: Path,
+    verb: str,
+    task_id: str,
+    url: str,
+    expected_head: str,
+    log: Path,
+) -> Tuple[int, str]:
+    environment_descriptor: Optional[int] = None
+    command_arguments = [str(command), verb, task_id, url]
+    if verb == "run":
+        command_arguments.extend(["--expected-head", expected_head])
+        environment_descriptor = open_fleet_environment(fleet_env)
+        # Source only the validated, already-open descriptor and suppress output
+        # from the private file itself. Crosscheck output is captured separately;
+        # environment values never enter argv or Python-owned durable records.
+        shell = (
+            'set -a; if ! . "/dev/fd/$1" >/dev/null 2>&1; then '
+            'printf "Crosscheck fleet environment could not be loaded safely\\n" >&2; '
+            'exit 78; fi; set +a; shift; exec "$@"'
+        )
+        arguments = [
+            "/bin/bash",
+            "--noprofile",
+            "--norc",
+            "-c",
+            shell,
+            "fm-crosscheck-autostart-env",
+            str(environment_descriptor),
+            *command_arguments,
+        ]
+        pass_fds = (environment_descriptor,)
+    else:
+        # Exact-head verification is read-only and needs no fleet credential.
+        # Doing it first lets a prior CLEAR result deduplicate even while the
+        # private Azure launch configuration is temporarily unavailable.
+        arguments = command_arguments
+        pass_fds = ()
+    timeout = positive_int_environment(
+        "FM_CROSSCHECK_AUTOSTART_COMMAND_TIMEOUT_SECONDS",
+        DEFAULT_COMMAND_TIMEOUT_SECONDS,
+    )
+    try:
+        completed = subprocess.run(
+            arguments,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=False,
+            timeout=timeout,
+            env=command_environment(),
+            pass_fds=pass_fds,
+        )
+        output = completed.stdout or b""
+        append_log(
+            log,
+            f"{verb} expected_head={expected_head} exit={completed.returncode}",
+            output,
+        )
+        return completed.returncode, output.decode("utf-8", errors="replace")
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout if isinstance(exc.stdout, bytes) else b""
+        stderr = exc.stderr if isinstance(exc.stderr, bytes) else b""
+        append_log(log, f"{verb} expected_head={expected_head} timed out", stdout + stderr)
+        return 124, "Crosscheck command timed out"
+    except OSError as exc:
+        append_log(
+            log,
+            f"{verb} expected_head={expected_head} launch failed",
+            str(exc).encode(),
+        )
+        return 125, f"Crosscheck command launch failed: {exc}"
+    finally:
+        if environment_descriptor is not None:
+            os.close(environment_descriptor)
+
+
+def lock_is_active(lock_path: Path) -> bool:
+    descriptor = open_lock(lock_path)
+    try:
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return True
+        fcntl.flock(descriptor, fcntl.LOCK_UN)
+        return False
+    finally:
+        os.close(descriptor)
+
+
+def latest_request(path: Path) -> Tuple[str, str, str, str]:
+    value = load_json(path, REQUEST_SCHEMA)
+    if value is None:
+        fail(f"Crosscheck autostart request disappeared: {path}")
+    task_id = value.get("task_id")
+    url = value.get("pull_request")
+    head = value.get("head_sha")
+    generation = value.get("generation_id")
+    if not all(isinstance(item, str) for item in (task_id, url, head, generation)):
+        fail(f"invalid Crosscheck autostart request identity at {path}")
+    validate_identity(task_id, url, head, generation)
+    return task_id, url, head, generation
+
+
+def validate_task_generation(state: Path, task_id: str, generation: str) -> None:
+    metadata_path = state / f"{task_id}.meta"
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(metadata_path, flags)
+    except FileNotFoundError:
+        fail(f"task metadata disappeared before Crosscheck autostart: {metadata_path}")
+    except OSError as exc:
+        fail(f"task metadata cannot be opened before Crosscheck autostart: {exc}")
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            fail(f"task metadata is unsafe before Crosscheck autostart: {metadata_path}")
+        if metadata.st_size > MAX_RECORD_BYTES:
+            fail(f"task metadata exceeds the Crosscheck autostart bound: {metadata_path}")
+        raw = os.read(descriptor, MAX_RECORD_BYTES + 1)
+    finally:
+        os.close(descriptor)
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeError as exc:
+        fail(f"task metadata cannot be read before Crosscheck autostart: {exc}")
+    observed = ""
+    for line in text.splitlines():
+        if line.startswith("generation_id="):
+            observed = line.split("=", 1)[1]
+    if observed != generation:
+        fail("task generation changed before Crosscheck autostart")
+
+
+def matching_verified_head(output: str, expected_head: str) -> bool:
+    return output.strip() == expected_head
+
+
+def worker(lock_descriptor: int, task_id: str, starting_attempt: int) -> int:
+    root, _home, state = runtime_paths()
+    os.set_inheritable(lock_descriptor, False)
+    paths = task_paths(state, task_id)
+    attempt = starting_attempt
+    processed = 0
+    try:
+        while processed < MAX_HEAD_RESTARTS:
+            request = latest_request(paths["request"])
+            current_task, url, head, generation = request
+            if current_task != task_id:
+                fail("Crosscheck autostart request changed task identity")
+            validate_task_generation(state, task_id, generation)
+            command = crosscheck_command(root)
+            fleet_env = fleet_environment_path()
+            write_state(
+                paths["state"],
+                task_id,
+                url,
+                head,
+                generation,
+                "running",
+                attempt,
+                "exact-head Crosscheck coordinator is running",
+                os.getpid(),
+                paths["log"],
+            )
+            active_wait = positive_int_environment(
+                "FM_CROSSCHECK_AUTOSTART_ACTIVE_WAIT_SECONDS",
+                DEFAULT_ACTIVE_WAIT_SECONDS,
+            )
+            active_deadline = time.monotonic() + active_wait
+            if lock_is_active(paths["crosscheck_lock"]):
+                write_state(
+                    paths["state"],
+                    task_id,
+                    url,
+                    head,
+                    generation,
+                    "running",
+                    attempt,
+                    "matching task Crosscheck operation is active; waiting to deduplicate",
+                    os.getpid(),
+                    paths["log"],
+                )
+            while lock_is_active(paths["crosscheck_lock"]):
+                if time.monotonic() >= active_deadline:
+                    fail("timed out waiting for the matching task Crosscheck operation")
+                time.sleep(0.1)
+            if latest_request(paths["request"]) != request:
+                attempt += 1
+                processed += 1
+                continue
+            verify_status, verify_output = run_crosscheck_command(
+                fleet_env,
+                command,
+                "verify",
+                task_id,
+                url,
+                head,
+                paths["log"],
+            )
+            if verify_status == 0 and matching_verified_head(verify_output, head):
+                result_state = "clear"
+                result_message = "matching exact-head CLEAR review already exists"
+            else:
+                run_status, run_output = run_crosscheck_command(
+                    fleet_env,
+                    command,
+                    "run",
+                    task_id,
+                    url,
+                    head,
+                    paths["log"],
+                )
+                if run_status == 0:
+                    post_status, post_output = run_crosscheck_command(
+                        fleet_env,
+                        command,
+                        "verify",
+                        task_id,
+                        url,
+                        head,
+                        paths["log"],
+                    )
+                    if post_status == 0 and matching_verified_head(post_output, head):
+                        result_state = "clear"
+                        result_message = "exact-head Crosscheck completed CLEAR"
+                    else:
+                        result_state = "failed"
+                        result_message = (
+                            "Crosscheck returned success but exact-head verification failed: "
+                            + bounded_text(post_output)
+                        )
+                else:
+                    result_state = "failed"
+                    result_message = "Crosscheck run failed: " + bounded_text(run_output)
+            write_state(
+                paths["state"],
+                task_id,
+                url,
+                head,
+                generation,
+                result_state,
+                attempt,
+                result_message,
+                os.getpid(),
+                paths["log"],
+            )
+            if latest_request(paths["request"]) == request:
+                return 0 if result_state == "clear" else 1
+            attempt += 1
+            processed += 1
+        fail(f"Crosscheck autostart exceeded {MAX_HEAD_RESTARTS} queued head restarts")
+    except Exception as exc:
+        message = str(exc) if isinstance(exc, AutostartError) else (
+            f"unexpected {type(exc).__name__}: {exc}"
+        )
+        try:
+            failed_task, url, head, generation = latest_request(paths["request"])
+            if failed_task == task_id:
+                write_state(
+                    paths["state"],
+                    task_id,
+                    url,
+                    head,
+                    generation,
+                    "failed",
+                    attempt,
+                    message,
+                    os.getpid(),
+                    paths["log"],
+                )
+                append_log(paths["log"], "coordinator failure", message.encode())
+        except Exception:
+            pass
+        return 1
+    finally:
+        os.close(lock_descriptor)
+
+
+def start(task_id: str, url: str, head: str, generation: str) -> int:
+    validate_identity(task_id, url, head, generation)
+    root, _home, state = runtime_paths()
+    validate_task_generation(state, task_id, generation)
+    paths = task_paths(state, task_id)
+    descriptor = open_lock(paths["coordinator_lock"])
+    try:
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            atomic_json(paths["request"], request_record(task_id, url, head, generation))
+            existing = load_json(paths["state"], SCHEMA)
+            active_head = existing.get("head_sha") if existing else "unknown"
+            active_generation = existing.get("generation_id") if existing else ""
+            if active_head == head and active_generation == generation:
+                print(
+                    f"crosscheck autostart: matching review already active for {task_id} at {head}"
+                )
+            else:
+                print(f"crosscheck autostart: queued new head {head} for active task {task_id}")
+            return 0
+        atomic_json(paths["request"], request_record(task_id, url, head, generation))
+        attempt = next_attempt(paths["state"], generation)
+        try:
+            crosscheck_command(root)
+            write_state(
+                paths["state"],
+                task_id,
+                url,
+                head,
+                generation,
+                "starting",
+                attempt,
+                "starting exact-head Crosscheck coordinator",
+                0,
+                paths["log"],
+            )
+            environment = os.environ.copy()
+            log_descriptor = open_log_descriptor(paths["log"])
+            try:
+                process = subprocess.Popen(
+                    [
+                        sys.executable,
+                        str(Path(__file__).resolve()),
+                        "worker",
+                        str(descriptor),
+                        task_id,
+                        str(attempt),
+                    ],
+                    stdin=subprocess.DEVNULL,
+                    stdout=log_descriptor,
+                    stderr=subprocess.STDOUT,
+                    close_fds=True,
+                    pass_fds=(descriptor,),
+                    start_new_session=True,
+                    env=environment,
+                    cwd=str(root),
+                )
+            finally:
+                os.close(log_descriptor)
+            print(f"crosscheck autostart: started {task_id} at {head} (pid {process.pid})")
+            return 0
+        except Exception as exc:
+            message = str(exc) if isinstance(exc, AutostartError) else (
+                f"unexpected {type(exc).__name__}: {exc}"
+            )
+            write_state(
+                paths["state"],
+                task_id,
+                url,
+                head,
+                generation,
+                "failed",
+                attempt,
+                "Crosscheck autostart launch failed: " + message,
+                0,
+                paths["log"],
+            )
+            append_log(paths["log"], "launcher failure", message.encode())
+            print(
+                "UNREVIEWED: Crosscheck autostart launch failed: "
+                + bounded_text(message),
+                file=sys.stderr,
+            )
+            # PR registration already succeeded. Launcher faults are visible in
+            # the task-local state and retry with the same command; they never
+            # turn registration into a global or operator-blocking refusal.
+            return 0
+    finally:
+        os.close(descriptor)
+
+
+def status(task_id: str, url: str, head: str, generation: str) -> int:
+    validate_identity(task_id, url, head, generation)
+    _root, _home, state = runtime_paths()
+    paths = task_paths(state, task_id)
+    record = load_json(paths["state"], SCHEMA)
+    record_matches = record is not None and (
+        record.get("pull_request") == url
+        and record.get("head_sha") == head
+        and record.get("generation_id") == generation
+    )
+    if not record_matches:
+        request = load_json(paths["request"], REQUEST_SCHEMA)
+        request_matches = request is not None and (
+            request.get("task_id") == task_id
+            and request.get("pull_request") == url
+            and request.get("head_sha") == head
+            and request.get("generation_id") == generation
+        )
+        if not request_matches or lock_is_active(paths["coordinator_lock"]):
+            return 0
+        attempt = next_attempt(paths["state"], generation)
+        message = "Crosscheck coordinator stopped before the requested head started; rerun fm-pr-check.sh"
+        write_state(
+            paths["state"],
+            task_id,
+            url,
+            head,
+            generation,
+            "failed",
+            attempt,
+            message,
+            0,
+            paths["log"],
+        )
+        print(f"{message}; log={paths['log'].name}")
+        return 1
+    assert record is not None
+    lifecycle = record.get("state")
+    if lifecycle in {"starting", "running"} and not lock_is_active(
+        paths["coordinator_lock"]
+    ):
+        attempt = record.get("attempt")
+        if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
+            fail(f"invalid Crosscheck autostart attempt at {paths['state']}")
+        message = "Crosscheck coordinator is no longer active; rerun fm-pr-check.sh"
+        write_state(
+            paths["state"],
+            task_id,
+            url,
+            head,
+            generation,
+            "failed",
+            attempt,
+            message,
+            0,
+            paths["log"],
+        )
+        lifecycle = "failed"
+        record["message"] = message
+    if lifecycle == "failed":
+        message = bounded_text(str(record.get("message", "Crosscheck autostart failed")))
+        print(f"{message}; log={record.get('log', '')}")
+        return 1
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    start_parser = subparsers.add_parser("start")
+    status_parser = subparsers.add_parser("status")
+    for command in (start_parser, status_parser):
+        command.add_argument("task_id")
+        command.add_argument("pr_url")
+        command.add_argument("head_sha")
+        command.add_argument("generation_id")
+    worker_parser = subparsers.add_parser("worker")
+    worker_parser.add_argument("lock_descriptor", type=int)
+    worker_parser.add_argument("task_id")
+    worker_parser.add_argument("attempt", type=int)
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        if args.command == "start":
+            return start(
+                args.task_id,
+                args.pr_url,
+                args.head_sha,
+                args.generation_id,
+            )
+        if args.command == "status":
+            return status(
+                args.task_id,
+                args.pr_url,
+                args.head_sha,
+                args.generation_id,
+            )
+        return worker(args.lock_descriptor, args.task_id, args.attempt)
+    except AutostartError as exc:
+        print(
+            f"UNREVIEWED: Crosscheck autostart failed: {bounded_text(str(exc))}",
+            file=sys.stderr,
+        )
+        return 1
+    except Exception as exc:
+        print(
+            "UNREVIEWED: Crosscheck autostart failed unexpectedly: "
+            f"{type(exc).__name__}: {bounded_text(str(exc))}",
+            file=sys.stderr,
+        )
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/bin/fm-crosscheck.py b/bin/fm-crosscheck.py
index 17d6f4b681b..0708a009608 100755
--- a/bin/fm-crosscheck.py
+++ b/bin/fm-crosscheck.py
@@ -1,5 +1,9 @@
 #!/usr/bin/env python3
-"""Fail-closed independent review ledger bound to an exact pull-request head."""
+"""Fail-closed independent review ledger bound to an exact pull-request head.
+
+The public `run TASK URL` surface resolves the live head itself.
+The PR-registration coordinator additionally passes `--expected-head SHA` so a head change between registration and launch refuses before reviewer or Azure spending.
+"""
 
 from __future__ import annotations
 
@@ -6954,7 +6958,13 @@ def write_ledger(path: Path, ledger: dict[str, Any]) -> None:
     atomic_write(path, encoded)
 
 
-def run_crosscheck(root: Path, home: Path, task_id: str, url: str) -> int:
+def run_crosscheck(
+    root: Path,
+    home: Path,
+    task_id: str,
+    url: str,
+    expected_head: str | None = None,
+) -> int:
     # C1 (docs/azure-requirements.md): the invocation's clock starts here, so
     # the recorded `total` covers everything the caller waits for, including
     # the unattributed gaps between the named phases.
@@ -6985,6 +6995,11 @@ def run_crosscheck(root: Path, home: Path, task_id: str, url: str) -> int:
             snapshot_value = github_snapshot(root, url)
         except CrosscheckError as exc:
             tool_fail(f"GitHub snapshot preflight failed: {exc}")
+        if expected_head is not None and snapshot_value["head_sha"] != expected_head:
+            tool_fail(
+                "registered PR head changed before Crosscheck launch: expected "
+                f"{expected_head}, observed {snapshot_value['head_sha']}"
+            )
     with timer.phase("ledger"):
         try:
             ledger = load_ledger(ledger_path, task_id, url)
@@ -7796,6 +7811,8 @@ def build_parser() -> argparse.ArgumentParser:
         command = subparsers.add_parser(name)
         command.add_argument("task_id")
         command.add_argument("pr_url")
+        if name == "run":
+            command.add_argument("--expected-head")
     timings = subparsers.add_parser("timings")
     timings.add_argument("task_id")
     economics = subparsers.add_parser("economics")
@@ -7904,7 +7921,16 @@ def main() -> int:
             except BlockingIOError:
                 tool_fail("another crosscheck operation already owns this task")
             if args.command == "run":
-                return run_crosscheck(root, home, args.task_id, args.pr_url)
+                expected_head = args.expected_head
+                if expected_head is not None and SHA_RE.fullmatch(expected_head) is None:
+                    tool_fail("expected registered PR head must be one 40-hex SHA")
+                return run_crosscheck(
+                    root,
+                    home,
+                    args.task_id,
+                    args.pr_url,
+                    expected_head,
+                )
             if args.command == "verify":
                 return verify_crosscheck(root, home, args.task_id, args.pr_url)
             return merge_crosschecked(
diff --git a/bin/fm-crosscheck.sh b/bin/fm-crosscheck.sh
index d58bb481eab..37de4229ceb 100755
--- a/bin/fm-crosscheck.sh
+++ b/bin/fm-crosscheck.sh
@@ -2,7 +2,7 @@
 # Run or verify the independent exact-head crosscheck ledger for a task PR.
 #
 # Usage:
-#   fm-crosscheck.sh run <task-id> <full GitHub PR URL>
+#   fm-crosscheck.sh run <task-id> <full GitHub PR URL> [--expected-head <SHA>]
 #   fm-crosscheck.sh verify <task-id> <full GitHub PR URL>
 #   fm-crosscheck.sh status
 #   fm-crosscheck.sh timings <task-id>
@@ -10,7 +10,9 @@
 #   fm-crosscheck.sh merge <task-id> <full GitHub PR URL> <reviewed SHA> <method> [--allow-queue]
 #
 # `run` is intentionally independent of no-mistakes so both reviews can be in
-# flight together once a PR exists. `verify` is the merge-gate operation: it
+# flight together once a PR exists. The task-local PR-registration coordinator
+# uses `--expected-head` to refuse a moved head before reviewer or Azure spend.
+# `verify` is the merge-gate operation: it
 # re-reads live GitHub state, requires the latest attempt for that exact head
 # and claims document to be clear, and prints only the reviewed SHA.
 # `timings` is the read-only C1 breakdown: it prints the per-phase duration
diff --git a/bin/fm-pr-check.sh b/bin/fm-pr-check.sh
index acfda113993..08420d8f634 100755
--- a/bin/fm-pr-check.sh
+++ b/bin/fm-pr-check.sh
@@ -1,11 +1,13 @@
 #!/usr/bin/env bash
-# Record a PR-ready task: appends pr=<url> and GitHub's pr_head=<sha> to
-# state/<id>.meta when available, then arms the watcher's merge poll by writing
-# state/<id>.check.sh, which prints one line when the PR is merged or its lookup
-# fails (the watcher's check contract: output = wake, silence = keep sleeping).
-# With central Slack config installed, then binds the live PR head to the signed
-# launch record created before the task agent started. Issuance failure exits
-# nonzero after poll setup.
+# Register a PR-ready task: append pr=<url> and GitHub's pr_head=<sha> to
+# state/<id>.meta, arm the watcher's merge poll, and asynchronously start the
+# independent exact-head Crosscheck review. Registration returns after the
+# task-local coordinator is requested; review latency never parks the caller.
+# Matching active or CLEAR heads deduplicate, failed/dead coordinators remain
+# visible and retryable, and unrelated tasks never share a launcher lock.
+# With central Slack config installed, binds the live PR head to the signed
+# launch record created before the task agent started before requesting review.
+# Issuance failure exits nonzero after poll setup.
 # Usage: fm-pr-check.sh <task-id> <pr-url>
 set -eu
 
@@ -26,6 +28,22 @@ META="$STATE/$ID.meta"
 LOOKUP_WT=
 LOOKUP_GENERATION=
 PR_HEAD=
+CROSSCHECK_AUTOSTART="$SCRIPT_DIR/fm-crosscheck-autostart.py"
+CROSSCHECK_AUTOSTART_ENABLED=1
+case "${FM_CROSSCHECK_AUTOSTART_TEST_DISABLE:-}" in
+  '') ;;
+  firstmate-pr-check-nonautostart-test-v1)
+    [ "${FM_TEST_RUNNER_ACTIVE:-}" = firstmate-test-runner-v1 ] || {
+      echo "error: the Crosscheck autostart test bypass is available only inside the sealed behavior-test runner" >&2
+      exit 1
+    }
+    CROSSCHECK_AUTOSTART_ENABLED=0
+    ;;
+  *)
+    echo "error: invalid FM_CROSSCHECK_AUTOSTART_TEST_DISABLE value" >&2
+    exit 1
+    ;;
+esac
 META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
 if [ ! -f "$META" ]; then
   fm_account_meta_lock_release "$META_LOCK"
@@ -85,8 +103,17 @@ else
 fi
 CHECK_TMP=$(mktemp "$STATE/.$ID.check.XXXXXX") || exit 1
 printf -v PR_ADAPTER_Q '%q' "$FM_ROOT/bin/fm-github-pr.py"
+printf -v CROSSCHECK_AUTOSTART_Q '%q' "$CROSSCHECK_AUTOSTART"
+printf -v ID_Q '%q' "$ID"
 printf -v URL_Q '%q' "$URL"
+printf -v PR_HEAD_Q '%q' "$PR_HEAD"
+printf -v GENERATION_Q '%q' "$LOOKUP_GENERATION"
 cat > "$CHECK_TMP" <<EOF
+if ! crosscheck_state=\$($CROSSCHECK_AUTOSTART_Q status $ID_Q $URL_Q $PR_HEAD_Q $GENERATION_Q 2>&1); then
+  diagnostic=\$(printf '%s' "\$crosscheck_state" | tr '\r\n' '  ')
+  printf 'UNREVIEWED: Crosscheck autostart failed: %.500s\n' "\$diagnostic"
+  exit 0
+fi
 if ! state=\$($PR_ADAPTER_Q state $URL_Q 2>&1); then
   diagnostic=\$(printf '%s' "\$state" | tr '\r\n' '  ')
   printf 'UNREVIEWED: PR state lookup failed: %.500s\n' "\$diagnostic"
@@ -111,3 +138,13 @@ if [ -f "$SLACK_CONFIG" ]; then
     }
 fi
 echo "armed: state/$ID.check.sh polls $URL"
+if [ "$CROSSCHECK_AUTOSTART_ENABLED" = 1 ]; then
+  if CROSSCHECK_AUTOSTART_OUT=$("$CROSSCHECK_AUTOSTART" start \
+    "$ID" "$URL" "$PR_HEAD" "$LOOKUP_GENERATION" 2>&1); then
+    printf '%s\n' "$CROSSCHECK_AUTOSTART_OUT"
+  else
+    CROSSCHECK_AUTOSTART_DIAGNOSTIC=$(printf '%s' "$CROSSCHECK_AUTOSTART_OUT" | tr '\r\n' '  ')
+    printf 'UNREVIEWED: Crosscheck autostart launcher failed: %.500s\n' \
+      "$CROSSCHECK_AUTOSTART_DIAGNOSTIC" >&2
+  fi
+fi
diff --git a/docs/crosscheck.md b/docs/crosscheck.md
index 9a9240dadd4..7743a08c37a 100644
--- a/docs/crosscheck.md
+++ b/docs/crosscheck.md
@@ -1,16 +1,37 @@
 # Crosscheck
 
-Crosscheck is an on-demand, exact-head PR reviewer. It is independent of
-Firstmate task orchestration: any agent or operator that can run the supported
-wrapper and read the configured Firstmate home can use it.
+Crosscheck is an exact-head PR reviewer that starts automatically when Firstmate registers a PR-ready task and remains available on demand.
+It is independent of Firstmate task orchestration: any agent or operator that can run the supported wrapper and read the configured Firstmate home can use it.
 
-Crosscheck does one job. It reviews the current PR head, returns `CLEAR` or
-`BLOCKING`, and records cited findings and suspicions against that exact SHA.
+Crosscheck does one job.
+It reviews the current PR head, returns `CLEAR` or `BLOCKING`, and records cited findings and suspicions against that exact SHA.
 It does not rerun CI, manufacture proof scripts, or launch verifier VMs.
 
 ## Run it
 
-Use a unique task ID and the full public GitHub PR URL:
+The normal Firstmate path is PR-ready registration:
+
+```sh
+FM_HOME=/Users/dongkeun/firstmate-home \
+  bin/fm-pr-check.sh <task-id> \
+  https://github.com/OWNER/REPO/pull/NUMBER
+```
+
+Registration records the live PR head, arms the merge poll, durably requests Crosscheck, starts one task-local coordinator, and returns without waiting for the review.
+A matching active request is reused, and a matching exact-head and exact-claims `CLEAR` result is verified without another review.
+Registering a new head replaces the queued request so the coordinator reviews that head next.
+A dead or failed coordinator releases its task-local lock and retries when the same registration command runs again.
+Unrelated task coordinators share no launcher lock, so the Azure lane-capacity and cost-admission controls remain the only review spending authority.
+
+Before launching a review, the coordinator loads the authoritative operator-private fleet environment from `~/.fm-azure/fleet.env` by default.
+`FM_CROSSCHECK_FLEET_ENV` may select another absolute file.
+The launcher opens that file without following symlinks and requires a current-operator-owned regular file that is not group or world writable.
+It sources the already-open file only inside the Crosscheck child, suppresses output from the source operation, and never copies environment values into argv, prompts, logs, repository files, or launcher records.
+
+Missing, unsafe, or incomplete fleet configuration does not undo or fail PR registration.
+The task remains honestly uncleared, the actionable failure is recorded in `state/<task-id>.crosscheck-autostart.json` and `state/<task-id>.crosscheck-autostart.log`, and the task check surfaces it for repair and retry.
+
+For an explicit on-demand run, use a unique task ID and the full public GitHub PR URL:
 
 ```sh
 set -a
@@ -21,15 +42,14 @@ FM_HOME=/Users/dongkeun/firstmate-home \
   https://github.com/OWNER/REPO/pull/NUMBER
 ```
 
-The fleet environment is operator-private Azure configuration. Load it into the
-process environment; never paste its values into a prompt or command.
+The fleet environment is operator-private Azure configuration.
+Load it into the process environment; never paste its values into a prompt or command.
 
-A new task ID needs no pre-created metadata file. Existing state must match the
-same task and PR identity or the run fails closed.
+A new task ID needs no pre-created metadata file.
+Existing state must match the same task and PR identity or the run fails closed.
 
-The command exits zero only for a valid `CLEAR` verdict on the live head. A
-finding, unresolved suspicion, stale head, provider failure, malformed verdict,
-or infrastructure failure exits nonzero and is never presented as clearance.
+The command exits zero only for a valid `CLEAR` verdict on the live head.
+A finding, unresolved suspicion, stale head, provider failure, malformed verdict, or infrastructure failure exits nonzero and is never presented as clearance.
 
 Results are written to:
 
diff --git a/tests/behavior-test-durations.tsv b/tests/behavior-test-durations.tsv
index a5217e85125..16f6e25fca1 100644
--- a/tests/behavior-test-durations.tsv
+++ b/tests/behavior-test-durations.tsv
@@ -88,6 +88,7 @@
 4000	tests/fm-pi-refresh.test.sh
 3750	tests/fm-pi-retry-continuity.test.sh
 3170	tests/fm-pi-watch-extension.test.sh
+14000	tests/fm-pr-crosscheck-autostart.test.sh
 1574	tests/fm-pr-merge.test.sh
 6030	tests/fm-process-tree.test.sh
 802	tests/fm-prompt-exec.test.sh
diff --git a/tests/fm-crosscheck.test.sh b/tests/fm-crosscheck.test.sh
index cb4c35e050c..0487f7da87b 100755
--- a/tests/fm-crosscheck.test.sh
+++ b/tests/fm-crosscheck.test.sh
@@ -3462,6 +3462,30 @@ assert run["state"] == "clear"
   pass "a pipeline-updated PR is reviewed at its exact remote head while the author worktree remains behind"
 }
 
+test_registered_expected_head_refuses_a_moved_head_before_spend() {
+  local record case_dir base head expected rc
+  record=$(make_case registered-head-moved)
+  IFS=$'\t' read -r case_dir base head <<< "$record"
+  expected=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
+  [ "$expected" != "$head" ] || fail "expected-head fixture did not move"
+  set +e
+  run_case "$case_dir" "$base" "$head" clear run --expected-head "$expected" \
+    > "$case_dir/out" 2> "$case_dir/err"
+  rc=$?
+  set -e
+  expect_code 1 "$rc" "moved registered head"
+  assert_grep "registered PR head changed before Crosscheck launch: expected $expected, observed $head" \
+    "$case_dir/err" \
+    "moved registered head did not produce the exact pre-spend diagnostic"
+  assert_absent "$case_dir/codex.log" \
+    "reviewer launched after the registered exact head changed"
+  assert_absent "$case_dir/pi.log" \
+    "Pi reviewer launched after the registered exact head changed"
+  assert_absent "$case_dir/data/task-x1/crosscheck-ledger.json" \
+    "head mismatch fabricated a durable review attempt"
+  pass "a moved registered head refuses before reviewer or Azure spending"
+}
+
 test_missing_pr_head_ref_fails_closed() {
   local record case_dir base head rc
   record=$(make_case missing-pr-head-ref)
@@ -6691,6 +6715,7 @@ if [ -n "${FM_TEST_CASE:-}" ]; then
     test_missing_metadata_for_existing_task_fails_closed|\
     test_existing_task_metadata_identity_collision_fails_closed|\
     test_review_fetches_exact_pr_head_when_author_worktree_is_behind|\
+    test_registered_expected_head_refuses_a_moved_head_before_spend|\
     test_missing_pr_head_ref_fails_closed|\
     test_codex_reviewer_requires_bound_auth_and_clears_ambient_credentials|\
     test_launcher_requires_supported_python|\
@@ -6845,6 +6870,7 @@ test_mismatched_state_without_metadata_fails_closed
 test_missing_metadata_for_existing_task_fails_closed
 test_existing_task_metadata_identity_collision_fails_closed
 test_review_fetches_exact_pr_head_when_author_worktree_is_behind
+test_registered_expected_head_refuses_a_moved_head_before_spend
 test_missing_pr_head_ref_fails_closed
 test_codex_reviewer_requires_bound_auth_and_clears_ambient_credentials
 test_null_ledger_fails_without_normalization
diff --git a/tests/fm-pr-crosscheck-autostart.test.sh b/tests/fm-pr-crosscheck-autostart.test.sh
new file mode 100755
index 00000000000..a9faed9dd9b
--- /dev/null
+++ b/tests/fm-pr-crosscheck-autostart.test.sh
@@ -0,0 +1,429 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+# End-to-end scratch-home coverage for PR-ready Crosscheck autostart.
+set -u
+
+# shellcheck source=tests/lib.sh
+. "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
+
+PR_CHECK="$ROOT/bin/fm-pr-check.sh"
+SECRET_VALUE='fixture-private-fleet-value'
+HEAD_ONE=1111111111111111111111111111111111111111
+HEAD_TWO=2222222222222222222222222222222222222222
+fm_test_tmproot_into TMP_ROOT fm-pr-crosscheck-autostart
+
+cleanup_autostart_workers() {
+  local record pid state
+  while IFS= read -r record; do
+    [ -f "$record" ] || continue
+    read -r state pid <<EOF
+$(python3 - "$record" <<'PY' 2>/dev/null || true
+import json
+import sys
+try:
+    value = json.load(open(sys.argv[1], encoding="utf-8"))
+except Exception:
+    raise SystemExit
+print(value.get("state", ""), value.get("pid", 0))
+PY
+)
+EOF
+    case "$state:$pid" in
+      starting:[1-9]*|running:[1-9]*) /bin/kill -TERM "-$pid" >/dev/null 2>&1 || true ;;
+    esac
+  done < <(find "$TMP_ROOT" -name '*.crosscheck-autostart.json' -type f 2>/dev/null)
+  fm_test_cleanup
+}
+trap cleanup_autostart_workers EXIT
+
+make_case() {
+  local name=$1 case_dir root home control
+  case_dir="$TMP_ROOT/$name"
+  root="$case_dir/root"
+  home="$case_dir/home"
+  control="$case_dir/control"
+  mkdir -p "$root/bin" "$home/state" "$home/data" "$home/config" "$control"
+  cat > "$root/bin/fm-guard.sh" <<'SH'
+#!/usr/bin/env bash
+exit 0
+SH
+  cat > "$root/bin/fm-github-pr.py" <<'SH'
+#!/usr/bin/env bash
+set -u
+command=$1
+url=$2
+number=${url##*/}
+case "$command" in
+  head) cat "$FM_TEST_CONTROL/head-$number" ;;
+  state) printf 'OPEN\n' ;;
+  *) exit 97 ;;
+esac
+SH
+  cat > "$root/bin/fm-crosscheck.sh" <<'SH'
+#!/usr/bin/env bash
+set -u
+verb=$1
+task=$2
+url=$3
+number=${url##*/}
+head=$(cat "$FM_TEST_CONTROL/head-$number")
+shift 3
+case "$verb" in
+  run)
+    [ "${FM_TEST_FLEET_SECRET:-}" = fixture-private-fleet-value ] || {
+      printf 'fleet environment is incomplete\n'
+      exit 96
+    }
+    [ "${1:-}" = --expected-head ] && [ "${2:-}" = "$head" ] || {
+      printf 'registered exact head was not enforced\n'
+      exit 95
+    }
+    ;;
+  verify) [ "$#" -eq 0 ] || exit 94 ;;
+  *) exit 97 ;;
+esac
+control=${FM_TEST_CONTROL:?}
+printf '%s\t%s\t%s\t%s\n' "$verb" "$task" "$head" "$url" >> "$control/calls"
+case "$verb" in
+  verify)
+    if [ -f "$control/clear-$task-$head" ]; then
+      printf '%s\n' "$head"
+      exit 0
+    fi
+    printf 'no clear review for %s\n' "$head"
+    exit 1
+    ;;
+  run)
+    touch "$control/started-$task-$head"
+    while [ -f "$control/block-$task-$head" ] \
+      && [ ! -f "$control/release-$task-$head" ]; do
+      sleep 0.05
+    done
+    touch "$control/clear-$task-$head"
+    printf 'crosscheck clear: %s at %s\n' "$url" "$head"
+    ;;
+esac
+SH
+  chmod +x "$root/bin/fm-guard.sh" "$root/bin/fm-github-pr.py" "$root/bin/fm-crosscheck.sh"
+  cat > "$case_dir/fleet.env" <<EOF
+FM_TEST_FLEET_SECRET='$SECRET_VALUE'
+EOF
+  chmod 600 "$case_dir/fleet.env"
+  : > "$control/calls"
+  printf '%s\n' "$case_dir"
+}
+
+seed_task() {
+  local case_dir=$1 task=$2
+  mkdir -p "$case_dir/home/data/$task" "$case_dir/worktrees/$task" "$case_dir/projects/$task"
+  fm_write_meta "$case_dir/home/state/$task.meta" \
+    "window=fm-$task" \
+    "worktree=$case_dir/worktrees/$task" \
+    "project=$case_dir/projects/$task" \
+    'kind=ship' \
+    'mode=no-mistakes' \
+    "generation_id=generation-$task"
+}
+
+set_head() {
+  local case_dir=$1 pull=$2 head=$3
+  printf '%s\n' "$head" > "$case_dir/control/head-$pull"
+}
+
+run_pr_check() {
+  local case_dir=$1 task=$2 pull=$3
+  FM_ROOT_OVERRIDE="$case_dir/root" \
+  FM_HOME="$case_dir/home" \
+  FM_STATE_OVERRIDE="$case_dir/home/state" \
+  FM_DATA_OVERRIDE="$case_dir/home/data" \
+  FM_CROSSCHECK_FLEET_ENV="$case_dir/fleet.env" \
+  FM_CROSSCHECK_AUTOSTART_ACTIVE_WAIT_SECONDS=10 \
+  FM_CROSSCHECK_AUTOSTART_COMMAND_TIMEOUT_SECONDS=20 \
+  FM_TEST_CONTROL="$case_dir/control" \
+    "$PR_CHECK" "$task" "https://github.com/example/repo/pull/$pull"
+}
+
+json_field() {
+  python3 - "$1" "$2" <<'PY'
+import json
+import sys
+value = json.load(open(sys.argv[1], encoding="utf-8"))
+result = value.get(sys.argv[2], "")
+print(result)
+PY
+}
+
+wait_for_state() {
+  local record=$1 expected_state=$2 expected_head=$3 minimum_attempt=${4:-1} i=0 state head attempt
+  while [ "$i" -lt 400 ]; do
+    if [ -f "$record" ]; then
+      state=$(json_field "$record" state 2>/dev/null || true)
+      head=$(json_field "$record" head_sha 2>/dev/null || true)
+      attempt=$(json_field "$record" attempt 2>/dev/null || true)
+      if [ "$state" = "$expected_state" ] && [ "$head" = "$expected_head" ] \
+        && case "$attempt" in ''|*[!0-9]*) false ;; *) [ "$attempt" -ge "$minimum_attempt" ] ;; esac; then
+        return 0
+      fi
+    fi
+    sleep 0.05
+    i=$((i + 1))
+  done
+  [ ! -f "$record" ] || cat "$record" >&2
+  return 1
+}
+
+count_run_calls() {
+  local case_dir=$1 task=$2 head=${3:-}
+  if [ -n "$head" ]; then
+    awk -F '\t' -v task="$task" -v head="$head" \
+      '$1 == "run" && $2 == task && $3 == head { count++ } END { print count + 0 }' \
+      "$case_dir/control/calls"
+  else
+    awk -F '\t' -v task="$task" \
+      '$1 == "run" && $2 == task { count++ } END { print count + 0 }' \
+      "$case_dir/control/calls"
+  fi
+}
+
+monotonic_now() {
+  python3 -c 'import time; print(time.monotonic())'
+}
+
+elapsed_seconds() {
+  python3 - "$1" "$2" <<'PY'
+import sys
+print(float(sys.argv[2]) - float(sys.argv[1]))
+PY
+}
+
+assert_less_than() {
+  python3 - "$1" "$2" <<'PY' || fail "$3"
+import sys
+raise SystemExit(0 if float(sys.argv[1]) < float(sys.argv[2]) else 1)
+PY
+}
+
+test_prompt_return_active_and_clear_dedupe() {
+  local case_dir task=preturn pull=1 started ended elapsed state_file out calls
+  case_dir=$(make_case prompt-return)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  touch "$case_dir/control/block-$task-$HEAD_ONE"
+
+  started=$(monotonic_now)
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "PR-ready registration failed before asynchronous review: $out"
+  ended=$(monotonic_now)
+  elapsed=$(elapsed_seconds "$started" "$ended")
+  assert_less_than "$elapsed" 3 \
+    "PR-ready registration waited ${elapsed}s for the blocked Crosscheck review"
+  assert_contains "$out" 'crosscheck autostart: started' \
+    "PR-ready registration did not report the asynchronous launch"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_ONE" '' 0.02 \
+    || fail "asynchronous Crosscheck review never started"
+
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "matching active registration failed: $out"
+  assert_contains "$out" 'matching review already active' \
+    "matching active review was not deduplicated"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "matching active Crosscheck run count"
+
+  touch "$case_dir/control/release-$task-$HEAD_ONE"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+  wait_for_state "$state_file" clear "$HEAD_ONE" 1 \
+    || fail "first Crosscheck review did not reach exact-head CLEAR"
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "matching clear registration failed: $out"
+  wait_for_state "$state_file" clear "$HEAD_ONE" 2 \
+    || fail "matching CLEAR deduplication did not finish exact-head verification"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "matching CLEAR Crosscheck run count"
+  rm "$case_dir/fleet.env"
+  run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "matching CLEAR registration depended on repaired launch configuration"
+  wait_for_state "$state_file" clear "$HEAD_ONE" 3 \
+    || fail "matching CLEAR result was not reusable while launch configuration was absent"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "configuration-independent matching CLEAR run count"
+  if grep -F "$SECRET_VALUE" \
+    "$case_dir/home/state/$task.crosscheck-autostart.json" \
+    "$case_dir/home/state/$task.crosscheck-autostart.request.json" \
+    "$case_dir/home/state/$task.crosscheck-autostart.log" >/dev/null; then
+    fail "operator fleet environment value was copied into autostart state or logs"
+  fi
+  pass "PR-ready registration returns promptly and deduplicates active and CLEAR exact heads"
+}
+
+test_configuration_failures_are_visible_and_retryable() {
+  local case_dir task=retry pull=2 state_file out wake calls message
+  case_dir=$(make_case retry)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+
+  rm "$case_dir/fleet.env"
+  out=$(run_pr_check "$case_dir" "$task" "$pull" 2>&1) \
+    || fail "missing fleet environment failed PR registration"
+  assert_contains "$out" 'crosscheck autostart: started' \
+    "missing fleet environment prevented prompt-return coordinator launch"
+  wait_for_state "$state_file" failed "$HEAD_ONE" 1 \
+    || fail "missing fleet environment did not persist task-local failure"
+  message=$(json_field "$state_file" message)
+  assert_contains "$message" 'fleet environment is missing' \
+    "missing fleet environment did not leave an actionable task diagnostic"
+
+  printf "FM_TEST_FLEET_SECRET='%s'\n" "$SECRET_VALUE" > "$case_dir/fleet.env"
+  chmod 666 "$case_dir/fleet.env"
+  out=$(run_pr_check "$case_dir" "$task" "$pull" 2>&1) \
+    || fail "unsafe fleet environment failed PR registration"
+  assert_contains "$out" 'crosscheck autostart: started' \
+    "unsafe fleet environment prevented prompt-return coordinator launch"
+  wait_for_state "$state_file" failed "$HEAD_ONE" 2 \
+    || fail "unsafe fleet environment did not persist retryable state"
+  message=$(json_field "$state_file" message)
+  assert_contains "$message" 'group/world writable' \
+    "unsafe fleet environment did not leave an actionable task diagnostic"
+
+  printf 'FM_TEST_UNRELATED=present\n' > "$case_dir/fleet.env"
+  chmod 600 "$case_dir/fleet.env"
+  run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "incomplete fleet environment failed PR registration"
+  wait_for_state "$state_file" failed "$HEAD_ONE" 3 \
+    || fail "incomplete fleet environment did not become an asynchronous task failure"
+  wake=$(FM_HOME="$case_dir/home" FM_STATE_OVERRIDE="$case_dir/home/state" \
+    FM_TEST_CONTROL="$case_dir/control" \
+    bash "$case_dir/home/state/$task.check.sh") \
+    || fail "failed-autostart task check exited nonzero"
+  assert_contains "$wake" 'UNREVIEWED: Crosscheck autostart failed' \
+    "failed background launch was not visible to supervision"
+
+  printf "FM_TEST_FLEET_SECRET='%s'\n" "$SECRET_VALUE" > "$case_dir/fleet.env"
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "retry after restoring the fleet environment failed: $out"
+  wait_for_state "$state_file" clear "$HEAD_ONE" 4 \
+    || fail "retry did not reach exact-head CLEAR"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "retry Crosscheck run count"
+  pass "missing, unsafe, and incomplete configuration stay visible and retry with one task identity"
+}
+
+test_dead_coordinator_is_visible_and_retryable() {
+  local case_dir task=dead pull=3 state_file pid wake calls out
+  case_dir=$(make_case dead-coordinator)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  touch "$case_dir/control/block-$task-$HEAD_ONE"
+  run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "dead-coordinator fixture did not register"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_ONE" '' 0.02 \
+    || fail "dead-coordinator fixture never entered the review"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+  wait_for_state "$state_file" running "$HEAD_ONE" 1 \
+    || fail "dead-coordinator fixture never recorded a running owner"
+  pid=$(json_field "$state_file" pid)
+
+  set_head "$case_dir" "$pull" "$HEAD_TWO"
+  touch "$case_dir/control/block-$task-$HEAD_TWO"
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "dead-coordinator new-head queue failed: $out"
+  assert_contains "$out" "queued new head $HEAD_TWO" \
+    "dead-coordinator fixture did not queue its successor head"
+  /bin/kill -TERM "-$pid" >/dev/null 2>&1 \
+    || fail "could not terminate the isolated coordinator fixture"
+
+  sleep 0.1
+  wake=$(FM_HOME="$case_dir/home" FM_STATE_OVERRIDE="$case_dir/home/state" \
+    FM_TEST_CONTROL="$case_dir/control" \
+    bash "$case_dir/home/state/$task.check.sh") \
+    || fail "dead coordinator task check exited nonzero"
+  assert_contains "$wake" 'stopped before the requested head started' \
+    "dead coordinator did not surface its queued head as retryable"
+  wait_for_state "$state_file" failed "$HEAD_TWO" 2 \
+    || fail "dead coordinator did not record failure for its queued head"
+
+  touch "$case_dir/control/release-$task-$HEAD_TWO"
+  run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "dead coordinator did not retry under the same task identity"
+  wait_for_state "$state_file" clear "$HEAD_TWO" 3 \
+    || fail "dead coordinator retry did not reach exact-head CLEAR"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "dead coordinator abandoned-head Crosscheck run count"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_TWO")
+  expect_code 1 "$calls" "dead coordinator retry Crosscheck run count"
+  pass "a dead task-local coordinator surfaces and retries its queued exact head"
+}
+
+test_new_head_restarts_without_prompt_wait() {
+  local case_dir task=newhead pull=4 state_file out calls
+  case_dir=$(make_case new-head)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  touch "$case_dir/control/block-$task-$HEAD_ONE"
+  run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "first-head registration failed"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_ONE" '' 0.02 \
+    || fail "first-head review did not start"
+
+  set_head "$case_dir" "$pull" "$HEAD_TWO"
+  touch "$case_dir/control/block-$task-$HEAD_TWO"
+  out=$(run_pr_check "$case_dir" "$task" "$pull") \
+    || fail "new-head registration failed: $out"
+  assert_contains "$out" "queued new head $HEAD_TWO" \
+    "new head did not replace the active coordinator request"
+  touch "$case_dir/control/release-$task-$HEAD_ONE"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_TWO" '' 0.02 \
+    || fail "coordinator did not restart Crosscheck for the new head"
+  touch "$case_dir/control/release-$task-$HEAD_TWO"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+  wait_for_state "$state_file" clear "$HEAD_TWO" 2 \
+    || fail "new-head review did not reach exact-head CLEAR"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_ONE")
+  expect_code 1 "$calls" "first-head Crosscheck run count"
+  calls=$(count_run_calls "$case_dir" "$task" "$HEAD_TWO")
+  expect_code 1 "$calls" "new-head Crosscheck run count"
+  pass "a newly registered PR head restarts Crosscheck after the active head returns"
+}
+
+test_unrelated_prs_start_concurrently() {
+  local case_dir task_a=parallel-a task_b=parallel-b state_a state_b calls
+  case_dir=$(make_case concurrency)
+  seed_task "$case_dir" "$task_a"
+  seed_task "$case_dir" "$task_b"
+  set_head "$case_dir" 5 "$HEAD_ONE"
+  set_head "$case_dir" 6 "$HEAD_TWO"
+  touch "$case_dir/control/block-$task_a-$HEAD_ONE" \
+    "$case_dir/control/block-$task_b-$HEAD_TWO"
+
+  run_pr_check "$case_dir" "$task_a" 5 >/dev/null \
+    || fail "first unrelated PR registration failed"
+  run_pr_check "$case_dir" "$task_b" 6 >/dev/null \
+    || fail "second unrelated PR registration failed"
+  fm_test_wait_for_file "$case_dir/control/started-$task_a-$HEAD_ONE" '' 0.02 \
+    || fail "first unrelated Crosscheck did not start"
+  fm_test_wait_for_file "$case_dir/control/started-$task_b-$HEAD_TWO" '' 0.02 \
+    || fail "second unrelated Crosscheck was serialized behind a fleet-global lock"
+  calls=$(count_run_calls "$case_dir" "$task_a")
+  expect_code 1 "$calls" "first unrelated Crosscheck run count"
+  calls=$(count_run_calls "$case_dir" "$task_b")
+  expect_code 1 "$calls" "second unrelated Crosscheck run count"
+
+  touch "$case_dir/control/release-$task_a-$HEAD_ONE" \
+    "$case_dir/control/release-$task_b-$HEAD_TWO"
+  state_a="$case_dir/home/state/$task_a.crosscheck-autostart.json"
+  state_b="$case_dir/home/state/$task_b.crosscheck-autostart.json"
+  wait_for_state "$state_a" clear "$HEAD_ONE" 1 \
+    || fail "first concurrent Crosscheck did not clear"
+  wait_for_state "$state_b" clear "$HEAD_TWO" 1 \
+    || fail "second concurrent Crosscheck did not clear"
+  pass "unrelated PR Crosschecks run concurrently with task-local coordination only"
+}
+
+test_prompt_return_active_and_clear_dedupe
+test_configuration_failures_are_visible_and_retryable
+test_dead_coordinator_is_visible_and_retryable
+test_new_head_restarts_without_prompt_wait
+test_unrelated_prs_start_concurrently
+
+echo '# all fm-pr-crosscheck-autostart tests passed'
diff --git a/tests/fm-pr-merge.test.sh b/tests/fm-pr-merge.test.sh
index fb05ac715b5..38c268fd693 100755
--- a/tests/fm-pr-merge.test.sh
+++ b/tests/fm-pr-merge.test.sh
@@ -8,6 +8,7 @@
 # merge API. The gh-axi double emits only recorded 0.1.25 TOON shapes and
 # rejects the unsupported raw-gh and unguarded `pr merge` surfaces.
 set -u
+export FM_CROSSCHECK_AUTOSTART_TEST_DISABLE=firstmate-pr-check-nonautostart-test-v1
 
 # shellcheck source=tests/lib.sh
 . "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
diff --git a/tests/fm-secondmate-safety.test.sh b/tests/fm-secondmate-safety.test.sh
index 18ee75ef0bc..cd85cfa3224 100755
--- a/tests/fm-secondmate-safety.test.sh
+++ b/tests/fm-secondmate-safety.test.sh
@@ -8,6 +8,7 @@
 # operator flow lives in fm-secondmate-lifecycle-e2e.test.sh; this file keeps the
 # destructive-invariant coverage that an e2e run cannot deterministically reach.
 set -u
+export FM_CROSSCHECK_AUTOSTART_TEST_DISABLE=firstmate-pr-check-nonautostart-test-v1
 
 # shellcheck source=tests/secondmate-helpers.sh disable=SC1091
 . "$(dirname "${BASH_SOURCE[0]}")/secondmate-helpers.sh"
diff --git a/tests/fm-teardown-suite.sh b/tests/fm-teardown-suite.sh
index bf796d28a51..fe422de417a 100644
--- a/tests/fm-teardown-suite.sh
+++ b/tests/fm-teardown-suite.sh
@@ -70,6 +70,7 @@
 # The task metadata needs a session-scoped handle for the same reason - see
 # docs/tmux-backend.md "Proving absence needs a session-scoped handle".
 set -u
+export FM_CROSSCHECK_AUTOSTART_TEST_DISABLE=firstmate-pr-check-nonautostart-test-v1
 
 # shellcheck source=tests/lib.sh disable=SC1091
 . "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
diff --git a/tests/test-capabilities.tsv b/tests/test-capabilities.tsv
index b327928fc10..167d0ca9e94 100644
--- a/tests/test-capabilities.tsv
+++ b/tests/test-capabilities.tsv
@@ -81,6 +81,7 @@ fm-pi-primary-live-e2e.test.sh	hermetic
 fm-pi-primary-types.test.sh	hermetic
 fm-pi-refresh.test.sh	hermetic
 fm-pi-watch-extension.test.sh	hermetic
+fm-pr-crosscheck-autostart.test.sh	hermetic
 fm-pr-merge.test.sh	hermetic
 fm-process-tree.test.sh	hermetic
 fm-prompt-exec.test.sh	hermetic

From 0de7c5795cab91fafc773f985f6fb37282a3e5cc Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Thu, 27 Aug 2026 00:22:10 -0400
Subject: [PATCH 2/6] no-mistakes(review): Fix Crosscheck retirement handoff
 and live merge detection

---
 bin/fm-crosscheck-autostart.py           | 76 ++++++++++++++++--------
 bin/fm-pr-check.sh                       | 14 +++--
 docs/crosscheck.md                       |  2 +
 tests/fm-pr-crosscheck-autostart.test.sh | 71 +++++++++++++++++++++-
 4 files changed, 132 insertions(+), 31 deletions(-)

diff --git a/bin/fm-crosscheck-autostart.py b/bin/fm-crosscheck-autostart.py
index b1481f5eb1a..6aad4305efb 100755
--- a/bin/fm-crosscheck-autostart.py
+++ b/bin/fm-crosscheck-autostart.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import argparse
+from contextlib import contextmanager
 import datetime as dt
 import fcntl
 import json
@@ -112,6 +113,7 @@ def task_paths(state: Path, task_id: str) -> Dict[str, Path]:
         "request": state / f"{task_id}.crosscheck-autostart.request.json",
         "state": state / f"{task_id}.crosscheck-autostart.json",
         "log": state / f"{task_id}.crosscheck-autostart.log",
+        "handoff_lock": state / f".{task_id}.crosscheck-autostart-handoff.lock",
         "coordinator_lock": state / f".{task_id}.crosscheck-autostart.lock",
         "crosscheck_lock": state / f".{task_id}.crosscheck.lock",
     }
@@ -227,6 +229,16 @@ def open_lock(path: Path) -> int:
     return descriptor
 
 
+@contextmanager
+def task_handoff(paths: Dict[str, Path]):
+    descriptor = open_lock(paths["handoff_lock"])
+    try:
+        fcntl.flock(descriptor, fcntl.LOCK_EX)
+        yield
+    finally:
+        os.close(descriptor)
+
+
 def fleet_environment_path() -> Path:
     configured = os.environ.get("FM_CROSSCHECK_FLEET_ENV", "")
     if configured:
@@ -571,36 +583,43 @@ def worker(lock_descriptor: int, task_id: str, starting_attempt: int) -> int:
                 os.getpid(),
                 paths["log"],
             )
-            if latest_request(paths["request"]) == request:
-                return 0 if result_state == "clear" else 1
+            with task_handoff(paths):
+                if latest_request(paths["request"]) == request:
+                    os.close(lock_descriptor)
+                    lock_descriptor = -1
+                    return 0 if result_state == "clear" else 1
             attempt += 1
             processed += 1
         fail(f"Crosscheck autostart exceeded {MAX_HEAD_RESTARTS} queued head restarts")
     except Exception as exc:
-        message = str(exc) if isinstance(exc, AutostartError) else (
-            f"unexpected {type(exc).__name__}: {exc}"
-        )
-        try:
-            failed_task, url, head, generation = latest_request(paths["request"])
-            if failed_task == task_id:
-                write_state(
-                    paths["state"],
-                    task_id,
-                    url,
-                    head,
-                    generation,
-                    "failed",
-                    attempt,
-                    message,
-                    os.getpid(),
-                    paths["log"],
-                )
-                append_log(paths["log"], "coordinator failure", message.encode())
-        except Exception:
-            pass
-        return 1
+        with task_handoff(paths):
+            message = str(exc) if isinstance(exc, AutostartError) else (
+                f"unexpected {type(exc).__name__}: {exc}"
+            )
+            try:
+                failed_task, url, head, generation = latest_request(paths["request"])
+                if failed_task == task_id:
+                    write_state(
+                        paths["state"],
+                        task_id,
+                        url,
+                        head,
+                        generation,
+                        "failed",
+                        attempt,
+                        message,
+                        os.getpid(),
+                        paths["log"],
+                    )
+                    append_log(paths["log"], "coordinator failure", message.encode())
+            except Exception:
+                pass
+            os.close(lock_descriptor)
+            lock_descriptor = -1
+            return 1
     finally:
-        os.close(lock_descriptor)
+        if lock_descriptor >= 0:
+            os.close(lock_descriptor)
 
 
 def start(task_id: str, url: str, head: str, generation: str) -> int:
@@ -608,6 +627,13 @@ def start(task_id: str, url: str, head: str, generation: str) -> int:
     root, _home, state = runtime_paths()
     validate_task_generation(state, task_id, generation)
     paths = task_paths(state, task_id)
+    with task_handoff(paths):
+        return start_locked(root, paths, task_id, url, head, generation)
+
+
+def start_locked(
+    root: Path, paths: Dict[str, Path], task_id: str, url: str, head: str, generation: str
+) -> int:
     descriptor = open_lock(paths["coordinator_lock"])
     try:
         try:
diff --git a/bin/fm-pr-check.sh b/bin/fm-pr-check.sh
index 08420d8f634..22f949fb1b5 100755
--- a/bin/fm-pr-check.sh
+++ b/bin/fm-pr-check.sh
@@ -109,16 +109,20 @@ printf -v URL_Q '%q' "$URL"
 printf -v PR_HEAD_Q '%q' "$PR_HEAD"
 printf -v GENERATION_Q '%q' "$LOOKUP_GENERATION"
 cat > "$CHECK_TMP" <<EOF
-if ! crosscheck_state=\$($CROSSCHECK_AUTOSTART_Q status $ID_Q $URL_Q $PR_HEAD_Q $GENERATION_Q 2>&1); then
-  diagnostic=\$(printf '%s' "\$crosscheck_state" | tr '\r\n' '  ')
-  printf 'UNREVIEWED: Crosscheck autostart failed: %.500s\n' "\$diagnostic"
-  exit 0
-fi
 if ! state=\$($PR_ADAPTER_Q state $URL_Q 2>&1); then
   diagnostic=\$(printf '%s' "\$state" | tr '\r\n' '  ')
   printf 'UNREVIEWED: PR state lookup failed: %.500s\n' "\$diagnostic"
   exit 0
 fi
+if [ "\$state" = MERGED ]; then
+  echo "merged"
+  exit 0
+fi
+if ! crosscheck_state=\$($CROSSCHECK_AUTOSTART_Q status $ID_Q $URL_Q $PR_HEAD_Q $GENERATION_Q 2>&1); then
+  diagnostic=\$(printf '%s' "\$crosscheck_state" | tr '\r\n' '  ')
+  printf 'UNREVIEWED: Crosscheck autostart failed: %.500s\n' "\$diagnostic"
+  exit 0
+fi
 case "\$state" in
   OPEN) ;;
   MERGED) echo "merged" ;;
diff --git a/docs/crosscheck.md b/docs/crosscheck.md
index 7743a08c37a..7df0d49728c 100644
--- a/docs/crosscheck.md
+++ b/docs/crosscheck.md
@@ -20,7 +20,9 @@ FM_HOME=/Users/dongkeun/firstmate-home \
 Registration records the live PR head, arms the merge poll, durably requests Crosscheck, starts one task-local coordinator, and returns without waiting for the review.
 A matching active request is reused, and a matching exact-head and exact-claims `CLEAR` result is verified without another review.
 Registering a new head replaces the queued request so the coordinator reviews that head next.
+A short task-local handoff lock couples request publication with coordinator retirement; it is never held during review execution.
 A dead or failed coordinator releases its task-local lock and retries when the same registration command runs again.
+The merge poll observes live GitHub merge state before reporting launcher failures, so manual completion and merge still trigger cleanup without granting merge authorization.
 Unrelated task coordinators share no launcher lock, so the Azure lane-capacity and cost-admission controls remain the only review spending authority.
 
 Before launching a review, the coordinator loads the authoritative operator-private fleet environment from `~/.fm-azure/fleet.env` by default.
diff --git a/tests/fm-pr-crosscheck-autostart.test.sh b/tests/fm-pr-crosscheck-autostart.test.sh
index a9faed9dd9b..d51a647c4b4 100755
--- a/tests/fm-pr-crosscheck-autostart.test.sh
+++ b/tests/fm-pr-crosscheck-autostart.test.sh
@@ -56,7 +56,13 @@ url=$2
 number=${url##*/}
 case "$command" in
   head) cat "$FM_TEST_CONTROL/head-$number" ;;
-  state) printf 'OPEN\n' ;;
+  state)
+    if [ -f "$FM_TEST_CONTROL/merged-$number" ]; then
+      printf 'MERGED\n'
+    else
+      printf 'OPEN\n'
+    fi
+    ;;
   *) exit 97 ;;
 esac
 SH
@@ -299,6 +305,16 @@ test_configuration_failures_are_visible_and_retryable() {
   assert_contains "$wake" 'UNREVIEWED: Crosscheck autostart failed' \
     "failed background launch was not visible to supervision"
 
+  touch "$case_dir/control/merged-$pull"
+  wake=$(FM_HOME="$case_dir/home" FM_STATE_OVERRIDE="$case_dir/home/state" \
+    FM_TEST_CONTROL="$case_dir/control" \
+    bash "$case_dir/home/state/$task.check.sh") \
+    || fail "merged task check exited nonzero"
+  [ "$wake" = merged ] || fail "persisted launcher failure hid live merge: $wake"
+  [ "$(json_field "$state_file" state)" = failed ] \
+    || fail "merge observation erased the actionable launcher failure"
+  rm "$case_dir/control/merged-$pull"
+
   printf "FM_TEST_FLEET_SECRET='%s'\n" "$SECRET_VALUE" > "$case_dir/fleet.env"
   out=$(run_pr_check "$case_dir" "$task" "$pull") \
     || fail "retry after restoring the fleet environment failed: $out"
@@ -420,6 +436,59 @@ test_unrelated_prs_start_concurrently() {
   pass "unrelated PR Crosschecks run concurrently with task-local coordination only"
 }
 
+test_retirement_handoff() {
+  local case_dir task=retire pull=7
+  case_dir=$(make_case retirement)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  mkdir "$case_dir/hook"
+  cat > "$case_dir/hook/sitecustomize.py" <<'PYHOOK'
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+if len(sys.argv) > 3 and sys.argv[1] == "worker":
+    original_close = os.close
+    coordinator = int(sys.argv[2])
+    control = Path(os.environ["FM_TEST_CONTROL"])
+
+    def close(descriptor):
+        if descriptor == coordinator and not (control / "retirement-entered").exists():
+            (control / "retirement-entered").touch()
+            (control / "head-7").write_text("2" * 40 + "\n")
+            environment = os.environ.copy()
+            environment.pop("PYTHONPATH", None)
+            with (control / "registration-output").open("w") as output:
+                process = subprocess.Popen(
+                    [environment["FM_TEST_PR_CHECK"], "retire",
+                     "https://github.com/example/repo/pull/7"],
+                    env=environment, stdout=output, stderr=output,
+                )
+                try:
+                    process.wait(timeout=2)
+                except subprocess.TimeoutExpired:
+                    pass
+        return original_close(descriptor)
+
+    os.close = close
+PYHOOK
+  PYTHONPATH="$case_dir/hook" FM_TEST_PR_CHECK="$PR_CHECK" \
+    run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "retirement fixture registration failed"
+  fm_test_wait_for_file "$case_dir/control/retirement-entered" '' 0.02 \
+    || fail "coordinator did not reach the retirement boundary"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_TWO" '' 0.02 \
+    || fail "registration during retirement lost its successor review"
+  wait_for_state "$case_dir/home/state/$task.crosscheck-autostart.json" clear "$HEAD_TWO" 2 \
+    || fail "retirement successor did not clear the new exact head"
+  expect_code 1 "$(count_run_calls "$case_dir" "$task" "$HEAD_TWO")" \
+    "retirement successor Crosscheck count"
+  pass "registration at coordinator retirement hands off the new exact head"
+}
+
+test_retirement_handoff
+
 test_prompt_return_active_and_clear_dedupe
 test_configuration_failures_are_visible_and_retryable
 test_dead_coordinator_is_visible_and_retryable

From a7bc41dfe44ab15e90dbbd35189d775364acdfed Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Thu, 27 Aug 2026 00:28:09 -0400
Subject: [PATCH 3/6] no-mistakes(review): Preserve Crosscheck registration
 order and completion state

---
 bin/fm-crosscheck-autostart.py           |   7 +
 bin/fm-pr-check.sh                       |  14 +-
 docs/crosscheck.md                       |   3 +-
 tests/fm-pr-crosscheck-autostart.test.sh | 238 +++++++++++++++++++++--
 4 files changed, 235 insertions(+), 27 deletions(-)

diff --git a/bin/fm-crosscheck-autostart.py b/bin/fm-crosscheck-autostart.py
index 6aad4305efb..fff4e5935a8 100755
--- a/bin/fm-crosscheck-autostart.py
+++ b/bin/fm-crosscheck-autostart.py
@@ -725,6 +725,13 @@ def status(task_id: str, url: str, head: str, generation: str) -> int:
     validate_identity(task_id, url, head, generation)
     _root, _home, state = runtime_paths()
     paths = task_paths(state, task_id)
+    with task_handoff(paths):
+        return status_locked(paths, task_id, url, head, generation)
+
+
+def status_locked(
+    paths: Dict[str, Path], task_id: str, url: str, head: str, generation: str
+) -> int:
     record = load_json(paths["state"], SCHEMA)
     record_matches = record is not None and (
         record.get("pull_request") == url
diff --git a/bin/fm-pr-check.sh b/bin/fm-pr-check.sh
index 22f949fb1b5..8b5cf769723 100755
--- a/bin/fm-pr-check.sh
+++ b/bin/fm-pr-check.sh
@@ -45,6 +45,10 @@ case "${FM_CROSSCHECK_AUTOSTART_TEST_DISABLE:-}" in
     ;;
 esac
 META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
+release_meta_lock() {
+  fm_account_meta_lock_release "$META_LOCK" >/dev/null 2>&1 || true
+}
+trap release_meta_lock EXIT
 if [ ! -f "$META" ]; then
   fm_account_meta_lock_release "$META_LOCK"
   echo "error: no task metadata for $ID" >&2
@@ -72,18 +76,12 @@ if [ -z "$LOOKUP_GENERATION" ]; then
     exit 1
   fi
 fi
-fm_account_meta_lock_release "$META_LOCK"
 if ! PR_HEAD_LOOKUP=$("$FM_ROOT/bin/fm-github-pr.py" head "$URL" 2>&1); then
   PR_HEAD_DIAGNOSTIC=$(printf '%s' "$PR_HEAD_LOOKUP" | tr '\r\n' '  ')
   printf 'UNREVIEWED: PR head lookup failed: %.500s\n' "$PR_HEAD_DIAGNOSTIC" >&2
   exit 1
 fi
 PR_HEAD=$PR_HEAD_LOOKUP
-META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
-release_meta_lock() {
-  fm_account_meta_lock_release "$META_LOCK" >/dev/null 2>&1 || true
-}
-trap release_meta_lock EXIT
 if [ -f "$META" ]; then
   CURRENT_WT=$(fm_account_meta_value "$META" worktree)
   CURRENT_GENERATION=$(fm_account_meta_value "$META" generation_id)
@@ -131,8 +129,6 @@ esac
 EOF
 chmod +x "$CHECK_TMP"
 mv "$CHECK_TMP" "$STATE/$ID.check.sh"
-fm_account_meta_lock_release "$META_LOCK"
-trap - EXIT
 SLACK_CONFIG=${FM_CROSSCHECK_SLACK_CONFIG:-$FM_HOME/config/crosscheck-slack.json}
 if [ -f "$SLACK_CONFIG" ]; then
   "$FM_ROOT/bin/fm-crosscheck-slack.sh" attest-task \
@@ -152,3 +148,5 @@ if [ "$CROSSCHECK_AUTOSTART_ENABLED" = 1 ]; then
       "$CROSSCHECK_AUTOSTART_DIAGNOSTIC" >&2
   fi
 fi
+fm_account_meta_lock_release "$META_LOCK"
+trap - EXIT
diff --git a/docs/crosscheck.md b/docs/crosscheck.md
index 7df0d49728c..60a61295899 100644
--- a/docs/crosscheck.md
+++ b/docs/crosscheck.md
@@ -20,7 +20,8 @@ FM_HOME=/Users/dongkeun/firstmate-home \
 Registration records the live PR head, arms the merge poll, durably requests Crosscheck, starts one task-local coordinator, and returns without waiting for the review.
 A matching active request is reused, and a matching exact-head and exact-claims `CLEAR` result is verified without another review.
 Registering a new head replaces the queued request so the coordinator reviews that head next.
-A short task-local handoff lock couples request publication with coordinator retirement; it is never held during review execution.
+Registration holds the task metadata lock from head capture through poll emission and request publication, so an older capture cannot replace a newer registration.
+A short task-local handoff lock couples request publication, status reconciliation, and coordinator retirement; it is never held during review execution.
 A dead or failed coordinator releases its task-local lock and retries when the same registration command runs again.
 The merge poll observes live GitHub merge state before reporting launcher failures, so manual completion and merge still trigger cleanup without granting merge authorization.
 Unrelated task coordinators share no launcher lock, so the Azure lane-capacity and cost-admission controls remain the only review spending authority.
diff --git a/tests/fm-pr-crosscheck-autostart.test.sh b/tests/fm-pr-crosscheck-autostart.test.sh
index d51a647c4b4..a42d5c187e4 100755
--- a/tests/fm-pr-crosscheck-autostart.test.sh
+++ b/tests/fm-pr-crosscheck-autostart.test.sh
@@ -37,6 +37,16 @@ EOF
 }
 trap cleanup_autostart_workers EXIT
 
+if [ -n "${FM_TEST_AUTOSTART_REVISION:-}" ]; then
+  mkdir -p "$TMP_ROOT/implementation"
+  cp -R "$ROOT/bin" "$TMP_ROOT/implementation/bin"
+  for implementation in fm-pr-check.sh fm-crosscheck-autostart.py; do
+    git -C "$ROOT" show "$FM_TEST_AUTOSTART_REVISION:bin/$implementation" \
+      > "$TMP_ROOT/implementation/bin/$implementation" || fail "cannot load pre-repair implementation"
+  done
+  PR_CHECK="$TMP_ROOT/implementation/bin/fm-pr-check.sh"
+fi
+
 make_case() {
   local name=$1 case_dir root home control
   case_dir="$TMP_ROOT/$name"
@@ -443,32 +453,59 @@ test_retirement_handoff() {
   set_head "$case_dir" "$pull" "$HEAD_ONE"
   mkdir "$case_dir/hook"
   cat > "$case_dir/hook/sitecustomize.py" <<'PYHOOK'
+import fcntl
 import os
 from pathlib import Path
 import subprocess
 import sys
+import time
+
+control = Path(os.environ["FM_TEST_CONTROL"])
+state = Path(os.environ["FM_STATE_OVERRIDE"])
+original_close = os.close
+original_flock = fcntl.flock
+original_replace = os.replace
+
+
+def wait_for(path):
+    deadline = time.monotonic() + 15
+    while not path.exists():
+        if time.monotonic() > deadline:
+            raise RuntimeError("retirement barrier timed out")
+        time.sleep(0.01)
+
+
+if len(sys.argv) > 3 and sys.argv[1] == "start":
+    def flock(descriptor, operation):
+        handoff = state / ".retire.crosscheck-autostart-handoff.lock"
+        if handoff.exists() and os.fstat(descriptor).st_ino == handoff.stat().st_ino:
+            (control / "successor-contending").touch()
+        return original_flock(descriptor, operation)
+
+    def replace(source, destination, *args, **kwargs):
+        result = original_replace(source, destination, *args, **kwargs)
+        if Path(destination).name == "retire.crosscheck-autostart.request.json":
+            (control / "successor-contending").touch()
+        return result
+
+    fcntl.flock = flock
+    os.replace = replace
 
 if len(sys.argv) > 3 and sys.argv[1] == "worker":
-    original_close = os.close
     coordinator = int(sys.argv[2])
-    control = Path(os.environ["FM_TEST_CONTROL"])
 
     def close(descriptor):
         if descriptor == coordinator and not (control / "retirement-entered").exists():
             (control / "retirement-entered").touch()
             (control / "head-7").write_text("2" * 40 + "\n")
-            environment = os.environ.copy()
-            environment.pop("PYTHONPATH", None)
+            (control / "successor-contending").unlink(missing_ok=True)
             with (control / "registration-output").open("w") as output:
-                process = subprocess.Popen(
-                    [environment["FM_TEST_PR_CHECK"], "retire",
+                subprocess.Popen(
+                    [os.environ["FM_TEST_PR_CHECK"], "retire",
                      "https://github.com/example/repo/pull/7"],
-                    env=environment, stdout=output, stderr=output,
+                    stdout=output, stderr=output,
                 )
-                try:
-                    process.wait(timeout=2)
-                except subprocess.TimeoutExpired:
-                    pass
+            wait_for(control / "successor-contending")
         return original_close(descriptor)
 
     os.close = close
@@ -487,12 +524,177 @@ PYHOOK
   pass "registration at coordinator retirement hands off the new exact head"
 }
 
-test_retirement_handoff
+test_registration_capture_order() {
+  local case_dir task=ordered pull=8 first second state_file
+  case_dir=$(make_case capture-order)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  cat > "$case_dir/root/bin/fm-github-pr.py" <<'SH'
+#!/usr/bin/env bash
+if [ "$1" != head ]; then
+  echo OPEN
+  exit 0
+fi
+head=$(cat "$FM_TEST_CONTROL/head-8")
+if [ "${FM_TEST_CAPTURE_FIRST:-}" = 1 ]; then
+  touch "$FM_TEST_CONTROL/captured-first"
+  while [ ! -f "$FM_TEST_CONTROL/release-capture" ]; do sleep 0.01; done
+fi
+printf '%s\n' "$head"
+SH
+  touch "$case_dir/control/block-$task-$HEAD_TWO"
+  FM_TEST_CAPTURE_FIRST=1 run_pr_check "$case_dir" "$task" "$pull" \
+    > "$case_dir/first.out" 2>&1 &
+  first=$!
+  fm_test_wait_for_file "$case_dir/control/captured-first" '' 0.02 \
+    || fail "first registration did not capture its head"
+  set_head "$case_dir" "$pull" "$HEAD_TWO"
+  (
+    FM_ACCOUNT_ROUTING_TEST_LAB=firstmate-account-routing-test-lab-v1 \
+    FM_ACCOUNT_TEST_HOOKS=firstmate-account-tests-v1 \
+    FM_ACCOUNT_LOCK_WAIT_TEST_OBSERVED="$case_dir/control/second-waiting" \
+      run_pr_check "$case_dir" "$task" "$pull" > "$case_dir/second.out" 2>&1
+    result=$?
+    touch "$case_dir/control/second-finished"
+    exit "$result"
+  ) &
+  second=$!
+  python3 - "$case_dir/control" <<'PYWAIT' || fail "second registration did not reach the ordering barrier"
+from pathlib import Path
+import sys
+import time
+control = Path(sys.argv[1])
+deadline = time.monotonic() + 15
+while not any((control / name).exists() for name in ("second-waiting", "second-finished")):
+    if time.monotonic() > deadline:
+        raise SystemExit(1)
+    time.sleep(0.01)
+PYWAIT
+  touch "$case_dir/control/release-capture"
+  wait "$first" || fail "first registration failed"
+  wait "$second" || fail "second registration failed"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.request.json"
+  [ "$(json_field "$state_file" head_sha)" = "$HEAD_TWO" ] \
+    || fail "older captured head replaced the newer registration"
+  touch "$case_dir/control/release-$task-$HEAD_TWO"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+  wait_for_state "$state_file" clear "$HEAD_TWO" \
+    || fail "latest captured head never reached CLEAR"
+  expect_code 1 "$(count_run_calls "$case_dir" "$task" "$HEAD_TWO")" \
+    "latest captured head review count"
+  pass "head capture and publication preserve task-local registration order"
+}
+
+test_status_completion_race() {
+  local case_dir task=statusrace pull=9 state_file out
+  case_dir=$(make_case status-race)
+  seed_task "$case_dir" "$task"
+  set_head "$case_dir" "$pull" "$HEAD_ONE"
+  touch "$case_dir/control/block-$task-$HEAD_ONE"
+  mkdir "$case_dir/hook"
+  cat > "$case_dir/hook/sitecustomize.py" <<'PYHOOK'
+import fcntl
+import json
+import os
+from pathlib import Path
+import sys
+import time
+
+control = Path(os.environ["FM_TEST_CONTROL"])
+state = Path(os.environ["FM_STATE_OVERRIDE"])
+original_flock = fcntl.flock
+original_close = os.close
+original_loads = json.loads
+
+
+def same_file(descriptor, name):
+    path = state / name
+    return path.exists() and os.fstat(descriptor).st_ino == path.stat().st_ino
+
+
+def wait_for(path):
+    deadline = time.monotonic() + 15
+    while not path.exists():
+        if time.monotonic() > deadline:
+            raise RuntimeError("status barrier timed out")
+        time.sleep(0.01)
+
+
+if len(sys.argv) > 3 and sys.argv[1] == "worker":
+    coordinator = int(sys.argv[2])
+
+    def flock(descriptor, operation):
+        if same_file(descriptor, ".statusrace.crosscheck-autostart-handoff.lock"):
+            (control / "worker-retiring").touch()
+        return original_flock(descriptor, operation)
+
+    def close(descriptor):
+        result = original_close(descriptor)
+        if descriptor == coordinator:
+            (control / "worker-closed").touch()
+        return result
+
+    fcntl.flock = flock
+    os.close = close
+
+if len(sys.argv) > 3 and sys.argv[1] == "status":
+    coordinated = False
+
+    def flock(descriptor, operation):
+        global coordinated
+        result = original_flock(descriptor, operation)
+        if same_file(descriptor, ".statusrace.crosscheck-autostart-handoff.lock"):
+            coordinated = True
+        return result
+
+    fcntl.flock = flock
+
+    def loads(raw, *args, **kwargs):
+        value = original_loads(raw, *args, **kwargs)
+        if isinstance(value, dict) and value.get("state") == "running":
+            (control / "status-read-running").touch()
+            (control / ("release-statusrace-" + "1" * 40)).touch()
+            wait_for(control / ("worker-retiring" if coordinated else "worker-closed"))
+        return value
 
-test_prompt_return_active_and_clear_dedupe
-test_configuration_failures_are_visible_and_retryable
-test_dead_coordinator_is_visible_and_retryable
-test_new_head_restarts_without_prompt_wait
-test_unrelated_prs_start_concurrently
+    json.loads = loads
+PYHOOK
+  PYTHONPATH="$case_dir/hook" run_pr_check "$case_dir" "$task" "$pull" >/dev/null \
+    || fail "status-race registration failed"
+  fm_test_wait_for_file "$case_dir/control/started-$task-$HEAD_ONE" '' 0.02 \
+    || fail "status-race review never started"
+  out=$(PYTHONPATH="$case_dir/hook" FM_HOME="$case_dir/home" \
+    FM_STATE_OVERRIDE="$case_dir/home/state" FM_TEST_CONTROL="$case_dir/control" \
+    "$(dirname "$PR_CHECK")/fm-crosscheck-autostart.py" status "$task" \
+      "https://github.com/example/repo/pull/$pull" "$HEAD_ONE" "generation-$task" 2>&1) \
+    || fail "status replaced worker completion with failure: $out"
+  [ -f "$case_dir/control/status-read-running" ] || fail "status race was not exercised"
+  state_file="$case_dir/home/state/$task.crosscheck-autostart.json"
+  wait_for_state "$state_file" clear "$HEAD_ONE" \
+    || fail "status overwrote durable CLEAR"
+  out=$(FM_HOME="$case_dir/home" FM_STATE_OVERRIDE="$case_dir/home/state" \
+    FM_TEST_CONTROL="$case_dir/control" bash "$case_dir/home/state/$task.check.sh") \
+    || fail "follow-up poll failed"
+  [ -z "$out" ] || fail "CLEAR unexpectedly emitted a merge or failure wake: $out"
+  pass "status cannot replace concurrent worker completion with a stale failure"
+}
+
+case "${FM_TEST_AUTOSTART_CASE:-all}" in
+  retirement) test_retirement_handoff ;;
+  registration) test_registration_capture_order ;;
+  status) test_status_completion_race ;;
+  consumer) test_configuration_failures_are_visible_and_retryable ;;
+  all)
+    test_retirement_handoff
+    test_registration_capture_order
+    test_status_completion_race
+    test_prompt_return_active_and_clear_dedupe
+    test_configuration_failures_are_visible_and_retryable
+    test_dead_coordinator_is_visible_and_retryable
+    test_new_head_restarts_without_prompt_wait
+    test_unrelated_prs_start_concurrently
+    ;;
+  *) fail "unknown autostart regression selection" ;;
+esac
 
-echo '# all fm-pr-crosscheck-autostart tests passed'
+echo '# all selected fm-pr-crosscheck-autostart tests passed'

From f1165f8f09a1c6165f12964014f2e147a861b0ef Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Thu, 27 Aug 2026 00:42:38 -0400
Subject: [PATCH 4/6] no-mistakes(document): Replace stale registration
 description with authoritative documentation pointer

---
 docs/scripts.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/scripts.md b/docs/scripts.md
index 61fb21e4b6d..bfdd8e8dabe 100644
--- a/docs/scripts.md
+++ b/docs/scripts.md
@@ -106,7 +106,7 @@ The shared no-mistakes gate refusal used by every directly invocable mutating co
 | `fm-crosscheck-slack.sh` | Run, preflight, selftest, or issue exact-head provenance for the Slack Crosscheck lane |
 | `fm-crosscheck-slack.py` | Serve allowlisted, metered, exact-head Slack reviews through the shared core lanes |
 | `fm-crosscheck-slack-service.sh` | Install and operate the credential-free macOS launchd wrapper for the central listener |
-| `fm-pr-check.sh`         | Record `pr=` and `pr_head=` for a PR-ready task, then arm the watcher's merge poll   |
+| `fm-pr-check.sh`         | Register a PR-ready task; see [Crosscheck operator flow](crosscheck.md#run-it) |
 | `fm-pr-merge.sh`         | Require exact-head crosscheck, record PR metadata, and atomically merge or enqueue the reviewed SHA |
 | `fm-promote.sh`          | Promote a scout task in place to a protected ship task                               |
 | `fm-report-contract-lib.sh` | Render the shared ship completion-report contract inserted into briefs and continuation prompts |

From 1f3d6f2949bfcd379428b1a5947c321ce1e0d4b7 Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Thu, 27 Aug 2026 01:16:21 -0400
Subject: [PATCH 5/6] no-mistakes: apply CI fixes

---
 bin/fm-pr-check.sh                       | 18 +++++++++++--
 tests/fm-crosscheck-slack.test.sh        | 32 ++++++++++++++++++++++++
 tests/fm-pr-crosscheck-autostart.test.sh |  4 +--
 tests/fm-spawn-dispatch-profile.test.sh  |  4 ++-
 4 files changed, 53 insertions(+), 5 deletions(-)

diff --git a/bin/fm-pr-check.sh b/bin/fm-pr-check.sh
index 8b5cf769723..b776c4179db 100755
--- a/bin/fm-pr-check.sh
+++ b/bin/fm-pr-check.sh
@@ -5,6 +5,8 @@
 # task-local coordinator is requested; review latency never parks the caller.
 # Matching active or CLEAR heads deduplicate, failed/dead coordinators remain
 # visible and retryable, and unrelated tasks never share a launcher lock.
+# A task-local registration lock orders head capture and publication without
+# holding the account metadata lock across the remote lookup.
 # With central Slack config installed, binds the live PR head to the signed
 # launch record created before the task agent started before requesting review.
 # Issuance failure exits nonzero after poll setup.
@@ -44,11 +46,17 @@ case "${FM_CROSSCHECK_AUTOSTART_TEST_DISABLE:-}" in
     exit 1
     ;;
 esac
-META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
+REGISTRATION_LOCK=$(fm_account_lock_acquire "$STATE" "$ID" pr-registration \
+  "PR registration" "${FM_ACCOUNT_META_LOCK_WAIT_SECONDS:-10}") || exit 1
+META_LOCK=
 release_meta_lock() {
-  fm_account_meta_lock_release "$META_LOCK" >/dev/null 2>&1 || true
+  if [ -n "$META_LOCK" ]; then
+    fm_account_meta_lock_release "$META_LOCK" >/dev/null 2>&1 || true
+  fi
+  fm_account_meta_lock_release "$REGISTRATION_LOCK" >/dev/null 2>&1 || true
 }
 trap release_meta_lock EXIT
+META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
 if [ ! -f "$META" ]; then
   fm_account_meta_lock_release "$META_LOCK"
   echo "error: no task metadata for $ID" >&2
@@ -76,12 +84,17 @@ if [ -z "$LOOKUP_GENERATION" ]; then
     exit 1
   fi
 fi
+# Serialize head capture/publication only against other registrations, not
+# account-session updates or task retirement during a remote lookup.
+fm_account_meta_lock_release "$META_LOCK"
+META_LOCK=
 if ! PR_HEAD_LOOKUP=$("$FM_ROOT/bin/fm-github-pr.py" head "$URL" 2>&1); then
   PR_HEAD_DIAGNOSTIC=$(printf '%s' "$PR_HEAD_LOOKUP" | tr '\r\n' '  ')
   printf 'UNREVIEWED: PR head lookup failed: %.500s\n' "$PR_HEAD_DIAGNOSTIC" >&2
   exit 1
 fi
 PR_HEAD=$PR_HEAD_LOOKUP
+META_LOCK=$(fm_account_meta_lock_acquire "$STATE" "$ID") || exit 1
 if [ -f "$META" ]; then
   CURRENT_WT=$(fm_account_meta_value "$META" worktree)
   CURRENT_GENERATION=$(fm_account_meta_value "$META" generation_id)
@@ -149,4 +162,5 @@ if [ "$CROSSCHECK_AUTOSTART_ENABLED" = 1 ]; then
   fi
 fi
 fm_account_meta_lock_release "$META_LOCK"
+fm_account_meta_lock_release "$REGISTRATION_LOCK"
 trap - EXIT
diff --git a/tests/fm-crosscheck-slack.test.sh b/tests/fm-crosscheck-slack.test.sh
index 9fd44be0868..4def26213c8 100755
--- a/tests/fm-crosscheck-slack.test.sh
+++ b/tests/fm-crosscheck-slack.test.sh
@@ -1276,6 +1276,38 @@ with Path({str(launch_log)!r}).open('a') as handle:
 sys.exit(1 if sys.argv[1] == 'print' else 0)
 """)
 launchctl.chmod(0o755)
+# Model the macOS plist editor on every host; the emitted plist is parsed below
+# and its actual service command is executed with the emitted environment.
+plutil = bin_dir / "plutil"
+plutil.write_text(f"#!{sys.executable}\n" + """
+import plistlib
+from pathlib import Path
+import sys
+args = sys.argv[1:]
+path = Path(args[-1])
+if args[0] == '-create':
+    value = {}
+else:
+    value = plistlib.loads(path.read_bytes())
+    parts = args[1].split('.')
+    parent = value
+    for part in parts[:-1]:
+        parent = parent[int(part)] if isinstance(parent, list) else parent[part]
+    kind = args[2]
+    item = {'-array': [], '-dictionary': {}}.get(kind)
+    if kind == '-string':
+        item = args[3]
+    elif kind == '-bool':
+        item = args[3] == 'true'
+    elif kind == '-integer':
+        item = int(args[3])
+    if isinstance(parent, list):
+        parent.insert(int(parts[-1]), item)
+    else:
+        parent[parts[-1]] = item
+path.write_bytes(plistlib.dumps(value))
+""")
+plutil.chmod(0o755)
 environment = dict(os.environ, HOME=str(home), FM_HOME=str(fm_home), FM_ROOT_OVERRIDE=str(fixture), FM_CROSSCHECK_SLACK_CONFIG=str(config_path), FM_CROSSCHECK_PYTHON=sys.executable, PATH=str(bin_dir) + os.pathsep + os.environ['PATH'])
 for name in ('app_token_env', 'bot_token_env', 'github_token_env'):
     environment[config[name]] = 'fixture-inherited-secret'
diff --git a/tests/fm-pr-crosscheck-autostart.test.sh b/tests/fm-pr-crosscheck-autostart.test.sh
index a42d5c187e4..5a0485cb6bc 100755
--- a/tests/fm-pr-crosscheck-autostart.test.sh
+++ b/tests/fm-pr-crosscheck-autostart.test.sh
@@ -30,7 +30,7 @@ PY
 )
 EOF
     case "$state:$pid" in
-      starting:[1-9]*|running:[1-9]*) /bin/kill -TERM "-$pid" >/dev/null 2>&1 || true ;;
+      starting:[1-9]*|running:[1-9]*) /bin/kill -TERM -- "-$pid" >/dev/null 2>&1 || true ;;
     esac
   done < <(find "$TMP_ROOT" -name '*.crosscheck-autostart.json' -type f 2>/dev/null)
   fm_test_cleanup
@@ -356,7 +356,7 @@ test_dead_coordinator_is_visible_and_retryable() {
     || fail "dead-coordinator new-head queue failed: $out"
   assert_contains "$out" "queued new head $HEAD_TWO" \
     "dead-coordinator fixture did not queue its successor head"
-  /bin/kill -TERM "-$pid" >/dev/null 2>&1 \
+  /bin/kill -TERM -- "-$pid" >/dev/null 2>&1 \
     || fail "could not terminate the isolated coordinator fixture"
 
   sleep 0.1
diff --git a/tests/fm-spawn-dispatch-profile.test.sh b/tests/fm-spawn-dispatch-profile.test.sh
index 7f382def88d..e0376dcd640 100755
--- a/tests/fm-spawn-dispatch-profile.test.sh
+++ b/tests/fm-spawn-dispatch-profile.test.sh
@@ -655,7 +655,9 @@ print(value["issuer"], value["task"]["task_id"], value["author"]["harness"], val
   git -C "$leased_worktree" checkout -qb codex/crosscheck-provenance-fixture
   printf 'authored after launch\n' > "$leased_worktree/crosscheck-provenance-fixture.txt"
   git -C "$leased_worktree" add crosscheck-provenance-fixture.txt
-  git -C "$leased_worktree" commit -qm 'test: author after launch'
+  git -C "$leased_worktree" -c user.name='Firstmate Tests' \
+    -c user.email='tests@example.invalid' commit -qm 'test: author after launch' \
+    || fail "could not commit the post-launch authorship fixture"
   exact_head=$(git -C "$leased_worktree" rev-parse HEAD)
   printf 'pr=https://github.com/ruby-labs/firstmate/pull/999\npr_head=%s\n' \
     "$exact_head" >> "$HOME_DIR/state/$id.meta"

From b9a28213424777f061e2275097b6e58a548cddfc Mon Sep 17 00:00:00 2001
From: Dongkeun Lee <dongkeun@rubydata.ai>
Date: Thu, 27 Aug 2026 01:41:40 -0400
Subject: [PATCH 6/6] no-mistakes: apply CI fixes

---
 tests/fm-report-stack-suite.sh | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/tests/fm-report-stack-suite.sh b/tests/fm-report-stack-suite.sh
index f37083f7112..a5fdc760284 100644
--- a/tests/fm-report-stack-suite.sh
+++ b/tests/fm-report-stack-suite.sh
@@ -277,8 +277,13 @@ test_legacy_cutover_preserves_fresh_reports_and_retires_expired_raw_paths() {
     FM_REPORT_LEGACY_CUTOVER_TEST_READY="$ready" FM_REPORT_LEGACY_CUTOVER_TEST_PROCEED="$proceed" \
     "$SCRIPT" render > "$output" 2>&1 &
   pid=$!
-  for _ in $(seq 1 100); do [ -e "$ready" ] && break; sleep 0.02; done
-  [ -e "$ready" ] || { kill -TERM "$pid" 2>/dev/null || true; fail "legacy cutover preparation gate did not open"; }
+  # Preparation runs filesystem helpers; synchronize on readiness, not a two-second speed budget.
+  if ! fm_test_wait_for_file "$ready" "$pid"; then
+    kill -TERM "$pid" 2>/dev/null || true
+    wait "$pid" 2>/dev/null || true
+    cat "$output" >&2
+    fail "legacy cutover preparation gate did not open"
+  fi
   assert_grep 'fresh bytes' "$stack/entries/legacy-fresh/report.md" \
     "bounded legacy migration hid an unstaged fresh report"
   assert_grep 'second fresh bytes' "$stack/entries/legacy-fresh-two/report.md" \