fix(local-api): identify the recorded server and serialize launches

backend/scripts/local_capture_api.py

@@ -33,6 +33,8 @@
 from __future__ import annotations
 
 import argparse
+import fcntl
+import hashlib
 import json
 import os
 import secrets
@@ -157,10 +159,9 @@ def _pid_alive(pid: object) -> bool:
 def _live_session(session_path: Path) -> dict | None:
     """The session at `session_path` if the server that wrote it still serves.
 
-    Live means its pid is alive and its loopback port accepts a connection.
-    Either alone misleads: a crashed server's pid can be reused, and its port
-    can be taken by another process. A missing or unreadable file is not live;
-    no client could use its credentials either.
+    A live pid and an accepting port can both belong to unrelated processes
+    after a crash. The local-only identity endpoint must also return the hash
+    of this session's reviewer token. The probe never sends that token.
     """
     try:
         session = json.loads(session_path.read_text())
@@ -175,11 +176,18 @@ def _live_session(session_path: Path) -> dict | None:
         return None
     if host not in LOOPBACK_HOSTS or port is None:
         return None
-    try:
-        with socket.create_connection((host, port), timeout=1):
-            return session
-    except OSError:
+    reviewer_token = session.get("reviewerToken")
+    if not isinstance(reviewer_token, str) or not reviewer_token:
         return None
+    identity = hashlib.sha256(reviewer_token.encode()).hexdigest()
+    try:
+        with _DIRECT.open(f"{_public_url(host, port)}/v1/local-session", timeout=1) as response:
+            proof = json.loads(response.read(256))
+            if response.status == 200 and isinstance(proof, dict) and proof.get("identity") == identity:
+                return session
+    except (OSError, ValueError):
+        pass
+    return None
 
 
 # The readiness probe must reach the listener itself. urlopen honors
@@ -307,8 +315,52 @@ def main(argv: list[str] | None = None) -> int:
         return 2
 
     session_dir = args.session_dir
-    volume_root = session_dir / "volume"
     session_path = session_dir / "session.json"
+    # Keep this lock file in place: unlinking it could let another launcher lock
+    # a different inode while this server still owns the original one. The OS
+    # releases flock on exit, including a crash, so no stale-lock recovery is needed.
+    session_dir.mkdir(parents=True, exist_ok=True)
+    fd = os.open(session_dir / "session.lock", os.O_RDWR | os.O_CREAT, 0o600)
+    with os.fdopen(fd, "a") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            print(
+                f"refusing to start: {session_path} is locked by a starting or live local capture API. "
+                "Stop it first, or pass another --session-dir.",
+                file=sys.stderr,
+            )
+            return 1
+        return _serve(args, session_path)
+
+
+def _local_app(reviewer_token: str):
+    """Add identity proof only to the loopback launcher's app, not deployed APIs."""
+    from fastapi.responses import JSONResponse
+
+    from server.api.app import create_app
+
+    identity = hashlib.sha256(reviewer_token.encode()).hexdigest()
+
+    class _Identity:
+        def __init__(self, app):
+            self.app = app
+
+        async def __call__(self, scope, receive, send):
+            if scope["type"] == "http" and scope["method"] == "GET" and scope["path"] == "/v1/local-session":
+                await JSONResponse({"identity": identity})(scope, receive, send)
+            else:
+                await self.app(scope, receive, send)
+
+    app = create_app()
+    # Wrap the API's bearer gate: the probe reads a hash, never a credential,
+    # and must not send the session's token to an unverified listener.
+    app.add_middleware(_Identity)
+    return app
+
+
+def _serve(args: argparse.Namespace, session_path: Path) -> int:
+    volume_root = session_path.parent / "volume"
     # A second launch on a free port would bind and replace a running server's
     # credentials, so the failed-bind guard below never fires; refuse instead.
     live = _live_session(session_path)
@@ -371,14 +423,16 @@ def main(argv: list[str] | None = None) -> int:
     # HSS_LOCAL_OUTCOME and HSS_LOCAL_STAGE_DELAY_S pass through untouched.
 
     config = uvicorn.Config(
-        "server.api.app:create_app",
-        factory=True,
+        _local_app(reviewer_token),
         host=args.host,
         port=port,
         log_level="info",
     )
-    _announce(base, session_path, args.health_timeout)
-    uvicorn.Server(config).run(sockets=[listener])
+    try:
+        _announce(base, session_path, args.health_timeout)
+        uvicorn.Server(config).run(sockets=[listener])
+    finally:
+        listener.close()
     return 0
 
 

backend/tests/test_local_capture_api.py

@@ -492,6 +492,125 @@ def test_second_launch_refuses_a_live_session(tmp_path: Path) -> None:
         _stop(first)
 
 
+def test_reused_pid_and_unrelated_listener_allow_startup(tmp_path: Path) -> None:
+    launcher = _launcher_module()
+    listener = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _AnswerEverything)
+    listener.hits = []
+    thread = threading.Thread(target=listener.serve_forever, daemon=True)
+    thread.start()
+    proc = None
+    try:
+        path = tmp_path / "session.json"
+        path.write_text(
+            json.dumps(
+                {
+                    "baseUrl": f"http://127.0.0.1:{listener.server_port}/v1",
+                    "reviewerToken": "session-only-token",
+                    "pid": os.getpid(),
+                }
+            )
+        )
+        assert launcher._live_session(path) is None
+        proc = _start_server(tmp_path, tmp_path / "new.log")
+        session = json.loads(path.read_text())
+        assert session["pid"] == proc.pid
+        assert launcher._live_session(path) == session
+    finally:
+        _stop(proc)
+        listener.shutdown()
+        listener.server_close()
+        thread.join(timeout=5)
+
+
+def test_live_session_requires_the_running_servers_identity(tmp_path: Path) -> None:
+    launcher = _launcher_module()
+    session_dir = tmp_path / "real"
+    proc = _start_server(session_dir, tmp_path / "real.log")
+    try:
+        session = json.loads((session_dir / "session.json").read_text())
+        copied_path = tmp_path / "session.json"
+        copied_path.write_text(json.dumps(session))
+        assert launcher._live_session(copied_path) == session
+        session["reviewerToken"] = "not-the-running-servers-token"
+        copied_path.write_text(json.dumps(session))
+        assert launcher._live_session(copied_path) is None
+    finally:
+        _stop(proc)
+
+
+def test_concurrent_launches_allow_exactly_one(tmp_path: Path) -> None:
+    # Pause the first launch immediately after its liveness check. Without the
+    # lifetime lock, the second launch can also pass the same check before
+    # either publishes a session. With the lock, it exits before the check.
+    wrapper = """
+import importlib.util
+import os
+from pathlib import Path
+import sys
+import time
+spec = importlib.util.spec_from_file_location("launcher", sys.argv[1])
+launcher = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(launcher)
+original = launcher._live_session
+root = Path(sys.argv[2])
+def paused(path):
+    result = original(path)
+    (root / f"checked-{os.getpid()}").touch()
+    deadline = time.monotonic() + 20
+    while not (root / "release").exists():
+        if time.monotonic() >= deadline:
+            raise RuntimeError("test did not release launch barrier")
+        time.sleep(0.01)
+    return result
+launcher._live_session = paused
+raise SystemExit(launcher.main(["--port", "0", "--session-dir", str(root), "--health-timeout", "20"]))
+"""
+    processes = []
+    logs = []
+    try:
+        for index in range(2):
+            log_path = tmp_path / f"launch-{index}.log"
+            logs.append(log_path)
+            with log_path.open("w") as log:
+                proc = subprocess.Popen(
+                    [sys.executable, "-c", wrapper, str(SCRIPT), str(tmp_path)],
+                    cwd=ROOT,
+                    stdout=log,
+                    stderr=subprocess.STDOUT,
+                    text=True,
+                )
+            processes.append(proc)
+            deadline = time.monotonic() + 20
+            while not (tmp_path / f"checked-{proc.pid}").exists() and proc.poll() is None:
+                assert time.monotonic() < deadline, log_path.read_text()
+                time.sleep(0.01)
+        (tmp_path / "release").touch()
+        deadline = time.monotonic() + 30
+        session_path = tmp_path / "session.json"
+        launcher = _launcher_module()
+        while True:
+            live = launcher._live_session(session_path)
+            if live is not None:
+                break
+            assert time.monotonic() < deadline, [path.read_text() for path in logs]
+            time.sleep(0.05)
+        winners = [proc for proc in processes if proc.poll() is None]
+        assert len(winners) == 1, [path.read_text() for path in logs]
+        assert live["pid"] == winners[0].pid
+        loser = next(proc for proc in processes if proc is not winners[0])
+        assert loser.wait(timeout=5) == 1
+        loser_log = logs[processes.index(loser)].read_text()
+        assert "refusing to start" in loser_log
+        # A dead launch must not leave an exclusive lock behind.
+        _stop(winners[0])
+        restarted = _start_server(tmp_path, tmp_path / "restart.log")
+        processes.append(restarted)
+    finally:
+        (tmp_path / "release").touch()
+        for proc in processes:
+            _stop(proc)
+
+
 def test_live_session_needs_a_live_pid_and_an_answering_port(tmp_path: Path) -> None:
     launcher = _launcher_module()
     path = tmp_path / "session.json"
@@ -508,7 +627,7 @@ def test_live_session_needs_a_live_pid_and_an_answering_port(tmp_path: Path) ->
     exited.wait()
     try:
         cases = [
-            (os.getpid(), f"127.0.0.1:{open_port}", True),
+            (os.getpid(), f"127.0.0.1:{open_port}", False),
             # A crashed server whose pid was reused: nothing listens.
             (os.getpid(), f"127.0.0.1:{refusing.getsockname()[1]}", False),
             # The pid is gone; whatever owns the port now is not this session.