test: preserve recovery fixtures and pin CLI comparison ancestry

apps/presentation/dashboard/smoke/team-evidence-return-smoke.mjs

@@ -99,7 +99,8 @@ try {
   assert.doesNotMatch(await checkObservation.innerText(), /任务验收规则|Task validation rules/,
     "An independently validated original request must not claim a Todo validation basis");
   // Another version's witness must never appear as a check of this report.
-  validation = {...validation, output_versions: [{ref: "report.json", sha256: "f".repeat(64)}, outputVersions[1]]};
+  // Restore the original Todo subject before the linked recovery journey.
+  validation = {...currentValidation, output_versions: [{ref: "report.json", sha256: "f".repeat(64)}, outputVersions[1]]};
   await recheck.click();
   await original().waitFor();
   await evidence.getByText("本次验收依据", {exact: true}).click();

apps/presentation/dashboard/smoke/team-feedback-http-fixture.py

@@ -6,6 +6,7 @@
 import sys
 import tempfile
 import threading
+from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[4]))
 
@@ -17,11 +18,12 @@
 def main() -> None:
     with tempfile.TemporaryDirectory(prefix="loopx-team-feedback-") as directory:
         root = Path(directory)
+        runtime_root = root / "runtime"
         registry = root / "registry.json"
-        registry.write_text(json.dumps({"schema_version": "0.1", "goals": [
+        registry.write_text(json.dumps({"schema_version": "0.1", "common_runtime_root": str(runtime_root), "goals": [
             {"id": "research", "repo": str(root), "status": "active"},
         ]}), encoding="utf-8")
-        store = ChatSessionStore(root / "runtime")
+        store = ChatSessionStore(runtime_root)
         runtime = ChatRuntimeController(store=store, codex_bin="missing-codex", registry_path=registry)
         store.create_session(goal_id="research", agent_id="codex", channel_id="goal.research",
                              upstream_thread_id="fixture", session_id="feedback",
@@ -34,7 +36,8 @@ def main() -> None:
         server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
         server.verbose = False
         server.registry_path = registry
-        server.runtime_root = root / "runtime"
+        server.runtime_root = runtime_root
+        server.runtime_root_override = str(runtime_root)
         server.chat_store = store
         server.runtime_controller = runtime
         thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -59,4 +62,9 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    main()
+    # A missing route must fail even on a clean CI host, before reading or
+    # writing any ambient runtime. Inbox and Chat operations remain real.
+    with patch("loopx.paths.select_default_runtime_root", side_effect=AssertionError(
+        "The feedback fixture must use only its temporary runtime"
+    )):
+        main()

apps/presentation/dashboard/smoke/team-feedback-recovery-smoke.mjs

@@ -38,7 +38,10 @@ try {
   await page.getByRole("button", {name: "开启 LoopX 模式", exact: true}).click();
   await page.getByLabel("已注册的协调身份").selectOption("lead");
   await page.getByLabel("协调员总 token 额度").fill("100000");
+  const configurationAccepted = page.waitForResponse(response => response.request().method() === "POST"
+    && response.request().postDataJSON()?.operation === "configure");
   await page.getByRole("button", {name: "保存设置", exact: true}).click();
+  assert.equal((await configurationAccepted).status(), 200);
   const configured = api.loopxModeRequests.findLast(row => row.operation === "configure");
   const mode = page.__loopxRuntime.loopxModes.get(configured.sessionId);
   Object.assign(mode, {enabled: true, paused: false, active_turn_id: "fixture-loopx-turn",

examples/control_plane/cli-output-base-head-differential-smoke.py

@@ -155,11 +155,33 @@ def _is_ancestor(ancestor: str, descendant: str) -> bool:
     )
 
 
+def _pin_probe_base_ref(base_ref: str, *, main_ref: str) -> str:
+    # CI may fetch a newer main after its candidate merge commit was created.
+    # Compare that candidate's ancestry, not unrelated later output changes.
+    command = (
+        ["git", "merge-base", "--all", "--", base_ref, "HEAD"]
+        if base_ref == main_ref else
+        ["git", "rev-parse", "--verify", "--end-of-options", f"{base_ref}^{{commit}}"]
+    )
+    completed = subprocess.run(
+        command, cwd=REPO_ROOT, check=False, stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace",
+    )
+    revisions = completed.stdout.splitlines()
+    if completed.returncode != 0 or len(revisions) != 1:
+        raise RuntimeError(
+            f"cannot resolve one CLI output base for {base_ref}; "
+            f"select an explicit --base-ref commit: {completed.stderr.strip()}"
+        )
+    return revisions[0]
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument(
         "--base-ref",
         default=os.environ.get("LOOPX_CLI_OUTPUT_BASE_REF", "origin/main"),
+        help="Comparison base; the selected main branch uses its common ancestor with HEAD.",
     )
     parser.add_argument(
         "--main-ref",
@@ -171,6 +193,7 @@ def main() -> int:
         main_ref=args.main_ref,
         is_ancestor=_is_ancestor,
     )
+    base_ref = _pin_probe_base_ref(base_ref, main_ref=args.main_ref)
     if not PROBE_TEST.is_file():
         raise RuntimeError(f"missing probe source: {PROBE_TEST.relative_to(REPO_ROOT)}")
     if not PROBE_RUNNER.is_file():
@@ -271,7 +294,7 @@ def main() -> int:
                 if row["status"] == "failed"
             ]
             raise AssertionError(
-                "agent-facing CLI base/head differential failed\n"
+                f"agent-facing CLI base/head differential failed (base_ref={base_ref})\n"
                 + json.dumps(failed, ensure_ascii=False, indent=2, sort_keys=True)
             )
         if comparison["review_required"]:

tests/control_plane/test_cli_output_differential.py

@@ -162,6 +162,7 @@ def probe_transport(**kwargs):
     monkeypatch.setitem(namespace, "_run", git_transport)
     monkeypatch.setitem(namespace, "_run_probe", probe_transport)
     monkeypatch.setitem(namespace, "_is_ancestor", lambda *_: True)
+    monkeypatch.setitem(namespace, "_pin_probe_base_ref", lambda ref, **_: ref)
     monkeypatch.setattr(sys, "argv", ["probe", "--base-ref", "synthetic-base"])
     with pytest.raises(AssertionError, match="baseline ceiling"):
         main()
@@ -477,6 +478,130 @@ def test_sync_commit_uses_main_as_cli_output_base() -> None:
     assert selected == "origin/main"
 
 
+@pytest.mark.parametrize("growth,explicit_base,advance_main", [
+    (0, False, False), (0, False, True), (1000, False, True), (0, True, True),
+])
+def test_probe_compares_pinned_candidate_ancestry_when_main_advances(
+    tmp_path, monkeypatch, growth, explicit_base, advance_main,
+):
+    import os
+    import runpy
+    import subprocess
+    from pathlib import Path
+
+    from loopx.control_plane.testing.cli_output_budget import (
+        CLI_OUTPUT_BUDGET_BY_ID, assert_cli_output_baseline,
+    )
+
+    source = Path(__file__).resolve().parents[2]
+    smoke = runpy.run_path(str(
+        source / "examples/control_plane/cli-output-base-head-differential-smoke.py"
+    ))
+    main = smoke["main"]
+    namespace = main.__globals__
+    repo = tmp_path / "repo"
+    repo.mkdir()
+
+    def git(*args):
+        return subprocess.run(
+            ["git", "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
+             "-c", f"core.hooksPath={os.devnull}", "-c", "commit.gpgsign=false",
+             *args], cwd=repo, check=True, text=True, encoding="utf-8",
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        ).stdout.strip()
+
+    git("init", "-b", "main")
+    packet = repo / "packet.json"
+    common_payload = {
+        "interaction_contract": {}, "scheduler_hint": {}, "selected_todo": {},
+        "note": "x" * 4000,
+    }
+    packet.write_text(json.dumps(common_payload), encoding="utf-8")
+    probe_test = repo / "tests" / "probe.py"
+    probe_test.parent.mkdir()
+    probe_test.write_text("# Public synthetic probe\n", encoding="utf-8")
+    git("add", "packet.json", "tests/probe.py")
+    git("commit", "-m", "common baseline")
+    common = git("rev-parse", "HEAD")
+    git("checkout", "-b", "candidate")
+    packet.write_text(json.dumps({**common_payload, "note": "x" * (4000 + growth)}), encoding="utf-8")
+    git("add", "packet.json")
+    git("commit", "--allow-empty", "-m", "candidate")
+    candidate = git("rev-parse", "HEAD")
+    git("checkout", "main")
+    if advance_main:
+        packet.write_text(json.dumps({**common_payload, "note": "short"}), encoding="utf-8")
+        git("add", "packet.json")
+        git("commit", "-m", "later upstream compaction")
+    upstream = git("rev-parse", "HEAD")
+    git("update-ref", "refs/remotes/origin/main", upstream)
+    git("checkout", "candidate")
+    probes = []
+    pinned = []
+    run = namespace["_run"]
+
+    def git_transport(command, **kwargs):
+        if "worktree" in command and "add" in command:
+            pinned.append(command[-1])
+            # Advance the ref again after selection but before checkout.
+            git("update-ref", "refs/remotes/origin/main", candidate)
+        run(command, **kwargs)
+
+    def probe_transport(**kwargs):
+        text = (kwargs["source_root"] / "packet.json").read_text(encoding="utf-8")
+        measured = measure_cli_output(text, output_format="json")
+        row = _row(
+            row_id="surface/quota_should_run/crowded/json",
+            surface_id="quota_should_run", scenario="crowded",
+            **{key: measured[key] for key in (
+                "chars", "utf8_bytes", "lines", "compact_payload_chars",
+            )},
+        )
+        if kwargs["enforce_budget"]:
+            assert_cli_output_baseline(
+                CLI_OUTPUT_BUDGET_BY_ID["quota_should_run"], scenario="crowded",
+                output_format="json", text=text, measurement=measured,
+            )
+        probes.append(kwargs["enforce_budget"])
+        kwargs["receipt_path"].write_text(json.dumps(_receipt(row)), encoding="utf-8")
+
+    monkeypatch.setitem(namespace, "REPO_ROOT", repo)
+    monkeypatch.setitem(namespace, "PROBE_TEST", probe_test)
+    monkeypatch.setitem(namespace, "_run", git_transport)
+    monkeypatch.setitem(namespace, "_run_probe", probe_transport)
+    # An explicit historical commit remains authoritative even off this branch.
+    args = ["probe", "--base-ref", upstream] if explicit_base else ["probe"]
+    monkeypatch.setattr(sys, "argv", args)
+    if growth or explicit_base:
+        delta = 3995 if explicit_base else growth
+        with pytest.raises(AssertionError, match=f"chars grew by {delta}") as error:
+            main()
+        assert f"base_ref={upstream if explicit_base else common}" in str(error.value)
+    else:
+        assert main() == 0
+    assert probes == [False, True]
+    assert pinned == [upstream if explicit_base else common]
+    assert git("worktree", "list", "--porcelain").count("worktree ") == 1
+    assert git("rev-parse", "HEAD") == candidate
+    assert git("show", f"{common}:packet.json") == json.dumps(common_payload)
+
+
+@pytest.mark.parametrize("base_ref", ["origin/main", "missing-base"])
+def test_probe_rejects_an_unresolvable_base(tmp_path, monkeypatch, base_ref):
+    import runpy
+    import subprocess
+    from pathlib import Path
+
+    pin = runpy.run_path(str(
+        Path(__file__).resolve().parents[2]
+        / "examples/control_plane/cli-output-base-head-differential-smoke.py"
+    ))["_pin_probe_base_ref"]
+    subprocess.run(["git", "init", str(tmp_path)], check=True, capture_output=True)
+    monkeypatch.setitem(pin.__globals__, "REPO_ROOT", tmp_path)
+    with pytest.raises(RuntimeError, match="select an explicit --base-ref commit"):
+        pin(base_ref, main_ref="origin/main")
+
+
 @pytest.mark.parametrize("row_kind", ["surface", "variant"])
 @pytest.mark.parametrize("mode", ["thin", "brief", "compact"])
 def test_host_safety_restoration_budget_is_one_time_bounded_and_prompt_only(