From c9a8ae922fbb0849733125dd328a963445490871 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 04:47:26 +0800
Subject: [PATCH 01/10] test(site): validate time machine publication journey

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../dashboard-frontstage-browser-smoke.mjs    | 64 ++++++++++++++++++-
 1 file changed, 62 insertions(+), 2 deletions(-)

diff --git a/examples/dashboard-frontstage-browser-smoke.mjs b/examples/dashboard-frontstage-browser-smoke.mjs
index 86b086f2f2..1a73aa8596 100644
--- a/examples/dashboard-frontstage-browser-smoke.mjs
+++ b/examples/dashboard-frontstage-browser-smoke.mjs
@@ -60,6 +60,64 @@ async function assertAnchorInView(page, id) {
   });
 }
 
+async function assertTimeMachineJourney(page, lang) {
+  await page.goto(`${publicOrigin}/loopx/?lang=${lang}`);
+  const entry = page.locator('#explore .resource-card[href*="/lab/time-machine/"]');
+  await entry.focus();
+  await Promise.all([
+    page.waitForURL(`${publicOrigin}/loopx/lab/time-machine/${lang === "zh" ? "?lang=zh" : ""}`),
+    entry.press("Enter"),
+  ]);
+  // Anchors come from the three public captures, independently of the UI scene copy.
+  const captures = [
+    ["portfolio", 76, "finding_5b6717a721136e001cd0cd18d38ac167431bbe265445e669bac92c72d990aef2", /13\.84/, /20\.53/],
+    ["graph", 134, "finding_b5860e437b2905fbc0b9db2bbdc3ba567783817eee67e0ecc433bf4e65c00b1a", /32%/],
+    ["ad", 105, "finding_c652328314cf9ac2757f85dfcbdb9907bb719cf1a1e36185926640694663a9f2", /32/],
+  ];
+  for (const [index, [id, count, findingId, ...metrics]] of captures.entries()) {
+    if (index) {
+      await page.locator(".tm-cases button").nth(index).click();
+      await page.waitForURL((url) => url.searchParams.get("case") === id);
+    }
+    const capture = JSON.parse(await readFile(resolve(exportSite, `lab/time-machine/data/${id}.json`), "utf8"));
+    assert.equal(capture.events.length, count);
+    const finding = capture.events.find((row) => row.result_id === findingId);
+    assert.equal(finding?.event_kind, "finding");
+    const choice = page.locator(".tm-choices button:not([disabled])").first();
+    await choice.focus();
+    await choice.press("Enter");
+    await page.locator(".tm-revealed").waitFor();
+    for (const metric of metrics) assert.match(await page.locator(".tm-revealed").innerText(), metric);
+    const title = await page.locator(".tm-revealed h3").innerText();
+    await page.locator(".tm-choices button").nth(1).click();
+    assert.equal(await page.locator(".tm-revealed h3").innerText(), title, "either choice reveals the same historical outcome");
+    await page.getByRole("button", { name: lang === "zh" ? "核对这条发现" : "Inspect this finding", exact: true }).click();
+    await page.locator(".tm-original").waitFor({ state: "visible" });
+    assert.equal(await page.locator(".tm-record-body").textContent(), finding.summary, "the outcome opens its original finding");
+    assert.equal(await page.locator(".tm-source").getAttribute("href"), `/loopx/lab/time-machine/data/${id}.json`);
+    const cursor = page.locator("#tm-cursor");
+    assert.equal(await cursor.getAttribute("max"), String(count));
+    await cursor.press("End");
+    assert.equal(await cursor.inputValue(), String(count));
+    assert.equal(await page.locator(".tm-record-body").textContent(), capture.events.at(-1).summary);
+    await page.locator(".tm-return").click();
+    assert.ok(await page.locator("#tm-question").evaluate((heading) => heading === document.activeElement));
+    assert.ok(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth + 1), "Time Machine fits the viewport");
+  }
+  // A real missing HTTP capture must hide outcomes and disable choices; retry
+  // then reads the actual exported bytes through the production verifier.
+  await page.route("**/lab/time-machine/data/portfolio.json", (route) => route.fulfill({ status: 404, body: "Missing capture" }), { times: 1 });
+  await page.reload();
+  await page.locator(".tm-cases button").first().click();
+  await page.locator(".tm-error [role=alert]").waitFor();
+  assert.equal(await page.locator(".tm-revealed").count(), 0);
+  assert.equal(await page.locator(".tm-choices button:disabled").count(), 2);
+  await page.locator(".tm-error button").click();
+  await page.locator(".tm-choices button:not([disabled])").first().click();
+  await page.locator(".tm-revealed").waitFor();
+  assert.match(await page.locator(".tm-revealed").innerText(), /13\.84/);
+}
+
 let browser;
 try {
   for (let i = 0; ; i++) {
@@ -249,7 +307,7 @@ try {
       await page.locator("#explore").waitFor();
       assert.equal(await page.locator("html").getAttribute("lang"), lang === "zh" ? "zh-CN" : "en");
       assert.equal(await page.locator('a[href*="deprecated"], a[href*="frontstage/"]').count(), 0);
-      const expected = ["docs/guides/personal-workspace-user-guide/", `benchmarks/swe-marathon/${lang === "zh" ? "?lang=zh" : ""}`, `benchmarks/lhtb/${lang === "zh" ? "?lang=zh" : ""}`, "benchmarks/deepswe/behavior-discovery/", "benchmarks/deepswe-sol/", `docs/showcases/index${lang === "en" ? ".en" : ""}.html`];
+      const expected = ["docs/guides/personal-workspace-user-guide/", `benchmarks/swe-marathon/${lang === "zh" ? "?lang=zh" : ""}`, `benchmarks/lhtb/${lang === "zh" ? "?lang=zh" : ""}`, "benchmarks/deepswe/behavior-discovery/", "benchmarks/deepswe-sol/", `lab/time-machine/${lang === "zh" ? "?lang=zh" : ""}`];
       assert.deepEqual(await page.locator("#explore .resource-card").evaluateAll((links) => links.map((a) => a.getAttribute("href"))), expected.map((path) => `/loopx/${path}`));
       if (width === 390) {
         await page.getByRole("button", { name: "Open navigation" }).click();
@@ -279,12 +337,14 @@ try {
         await page.goto(`${publicOrigin}/loopx/${expected[i]}`);
         await page.locator("h1").first().waitFor();
       }
+      await assertTimeMachineJourney(page, lang);
+      console.log(`Time Machine discovery, three outcomes, original records and retry: ${width}px, ${lang}: ok`);
     }
   }
   if (process.env.LOOPX_PUBLIC_SITE_DIR) {
     // Check the assembled publication, including the separately built books.
     const checked = new Set();
-    for (const path of ["", "?lang=zh", "benchmarks/swe-marathon/", "benchmarks/lhtb/", "benchmarks/lhtb/?lang=zh", "benchmarks/deepswe/behavior-discovery/", "benchmarks/deepswe-sol/", "docs/showcases/index.html", "docs/showcases/index.en.html", "docs/guides/personal-workspace-user-guide/", "docs/book/", "docs/book/en/", "blog/", "blog/zh/"]) {
+    for (const path of ["", "?lang=zh", "benchmarks/swe-marathon/", "benchmarks/lhtb/", "benchmarks/lhtb/?lang=zh", "benchmarks/deepswe/behavior-discovery/", "benchmarks/deepswe-sol/", "lab/time-machine/", "lab/time-machine/?lang=zh", "docs/showcases/index.html", "docs/showcases/index.en.html", "docs/guides/personal-workspace-user-guide/", "docs/book/", "docs/book/en/", "blog/", "blog/zh/"]) {
       await page.goto(`${publicOrigin}/loopx/${path}`);
       await page.locator("h1").first().waitFor();
       const links = await page.locator("a[href]").evaluateAll((links) => links.map((link) => link.href));

From acc16b42cf459eb389f39dcdaebbaeef795d311e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 05:08:03 +0800
Subject: [PATCH 02/10] test(site): await record focus before keyboard
 navigation

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 examples/dashboard-frontstage-browser-smoke.mjs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/examples/dashboard-frontstage-browser-smoke.mjs b/examples/dashboard-frontstage-browser-smoke.mjs
index 1a73aa8596..dbdbc9384a 100644
--- a/examples/dashboard-frontstage-browser-smoke.mjs
+++ b/examples/dashboard-frontstage-browser-smoke.mjs
@@ -93,13 +93,15 @@ async function assertTimeMachineJourney(page, lang) {
     assert.equal(await page.locator(".tm-revealed h3").innerText(), title, "either choice reveals the same historical outcome");
     await page.getByRole("button", { name: lang === "zh" ? "核对这条发现" : "Inspect this finding", exact: true }).click();
     await page.locator(".tm-original").waitFor({ state: "visible" });
+    // Inspect queues a focus transfer; finish it before sending the next key.
+    await page.waitForFunction(() => document.querySelector(".tm-original h3") === document.activeElement);
     assert.equal(await page.locator(".tm-record-body").textContent(), finding.summary, "the outcome opens its original finding");
     assert.equal(await page.locator(".tm-source").getAttribute("href"), `/loopx/lab/time-machine/data/${id}.json`);
     const cursor = page.locator("#tm-cursor");
     assert.equal(await cursor.getAttribute("max"), String(count));
     await cursor.press("End");
+    await page.waitForFunction((summary) => document.querySelector(".tm-record-body")?.textContent === summary, capture.events.at(-1).summary);
     assert.equal(await cursor.inputValue(), String(count));
-    assert.equal(await page.locator(".tm-record-body").textContent(), capture.events.at(-1).summary);
     await page.locator(".tm-return").click();
     assert.ok(await page.locator("#tm-question").evaluate((heading) => heading === document.activeElement));
     assert.ok(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth + 1), "Time Machine fits the viewport");

From bcb67a8de1d51140505b2f076be75b70252a9a5b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 05:54:25 +0800
Subject: [PATCH 03/10] test(todos): align consumers with administrative lease
 semantics

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_canonical_handoff_invariants.py      | 14 +++++++++++--
 .../test_canonical_handoff_mode.py            | 10 +++++++--
 .../test_canonical_todo_writer_isolation.py   |  4 +++-
 .../test_chat_cold_source_import.py           | 16 ++++++++++----
 .../test_cold_source_import_cli.py            |  5 +++--
 ...test_quota_authority_settlement_journey.py |  4 +++-
 .../test_research_execution_authority.py      | 21 ++++++++++++-------
 .../test_soft_claim_migration_continuation.py | 13 ++++++++----
 .../test_todo_consumer_host_retirement.py     |  4 ++--
 9 files changed, 65 insertions(+), 26 deletions(-)

diff --git a/tests/control_plane/test_canonical_handoff_invariants.py b/tests/control_plane/test_canonical_handoff_invariants.py
index 0f1d9602f1..1159645ece 100644
--- a/tests/control_plane/test_canonical_handoff_invariants.py
+++ b/tests/control_plane/test_canonical_handoff_invariants.py
@@ -540,17 +540,27 @@ def test_metadata_cannot_rewrite_hard_lease_execution_grants(
     _unchanged(registry, state, before, display)
 
 
-def test_metadata_edit_requires_current_execution_proof(tmp_path, canonical_workspace):
+def test_administrative_copy_preserves_explicit_execution_fence(tmp_path, canonical_workspace):
     registry, state = canonical_workspace(tmp_path, handoff_mode="hard_lease")
     todo = _add_todo(registry, claimed_by=AGENT_A)
+    original = _snapshot(registry)
+    updated = update_goal_todo(
+        registry_path=registry, goal_id=GOAL_ID, todo_id=todo["todo_id"],
+        agent_id=AGENT_A, note="Reviewed context",
+    )
+    assert updated["ok"]
+    assert _snapshot(registry)["leases"] == original["leases"]
     before, display = _snapshot(registry), state.read_bytes()
+    # Copy edits grant no execution. Supplying proof retains the strict fence.
     with pytest.raises(LocalCoordinationAuthorityRejection):
         update_goal_todo(
             registry_path=registry,
             goal_id=GOAL_ID,
             todo_id=todo["todo_id"],
             agent_id=AGENT_A,
-            note="Validated progress",
+            note="Unqualified execution",
+            task_lease_idempotency_key="not-acquired",
+            task_lease_expected_version=1,
         )
     _unchanged(registry, state, before, display)
     _acquire(registry, tmp_path, todo["todo_id"], owner=AGENT_A)
diff --git a/tests/control_plane/test_canonical_handoff_mode.py b/tests/control_plane/test_canonical_handoff_mode.py
index 1050a967d8..ddb0ea243e 100644
--- a/tests/control_plane/test_canonical_handoff_mode.py
+++ b/tests/control_plane/test_canonical_handoff_mode.py
@@ -186,9 +186,15 @@ def work(*args, expected_exit=0):
             *args, "--goal-id", "mode-goal", "--todo-id", todo["todo_id"]], capture_output=True, text=True, timeout=90)
         assert child.returncode == expected_exit, child.stdout + child.stderr
         return json.loads(child.stdout)
-    # Policy migration preserves assignment, but does not authorize an execution.
-    rejected = work("todo", "update", "--agent-id", "agent-a", "--text", "Continued work", expected_exit=1)
+    # Ordinary copy is administrative after migration; lease lineage stays intact.
+    assert work("todo", "update", "--agent-id", "agent-a", "--text", "Continued work")["ok"]
+    copied = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True)
+    assert copied["leases"] == after["leases"]
+    rejected = work("todo", "update", "--agent-id", "agent-a", "--note", "Unqualified execution",
+                    "--task-lease-idempotency-key", "not-acquired", "--task-lease-expected-version", "1",
+                    expected_exit=1)
     assert rejected["error_code"] == "handoff_mode_requires_lease"
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True) == copied
     acquired = work("task-lease", "acquire", "--owner", "agent-a", "--idempotency-key", "continued-execution",
                     "--expected-version", "0", "--ttl-seconds", "600", "--write-scope", "src/**")
     assert acquired["acquired"] is True
diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py
index ea12517f8a..4e5cdcc270 100644
--- a/tests/control_plane/test_canonical_todo_writer_isolation.py
+++ b/tests/control_plane/test_canonical_todo_writer_isolation.py
@@ -310,7 +310,9 @@ def guard(turn):
         return run("quota", "should-run", *identity, "--codex-app", "--turn-instance-id", turn)
 
     update = ["todo", "update", *identity, "--note", "Canonical delivery retained"]
-    rejected = run(*update, succeeds=False)
+    # Copy is administrative, while explicitly supplied execution stays fenced.
+    rejected = run(*update, "--task-lease-idempotency-key", "not-acquired",
+                   "--task-lease-expected-version", "1", succeeds=False)
     assert rejected["error_code"] == "handoff_mode_requires_lease"
     acquired = run("task-lease", "acquire", "--goal-id", settlement.GOAL_ID,
                    "--todo-id", settlement.TODO_ID, "--owner", settlement.AGENT_ID,
diff --git a/tests/control_plane/test_chat_cold_source_import.py b/tests/control_plane/test_chat_cold_source_import.py
index e5bedaa830..af755f0892 100644
--- a/tests/control_plane/test_chat_cold_source_import.py
+++ b/tests/control_plane/test_chat_cold_source_import.py
@@ -335,16 +335,24 @@ def test_registered_runtime_alias_retarget_rejects_without_import(cold_api, prov
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_app_enforces_lease_and_missing_provider_without_legacy_fallback(cold_api, provider):
+def test_app_copy_preserves_execution_and_provider_fences_without_legacy_fallback(cold_api, provider):
     call, read, state, _, runtime, _, _, _ = cold_api
     plan = prepare(call, provider)
     assert call("apply", {**carrier(plan), "writers_stopped": True})[1]["status"] == "applied"
     original_state, original_head = state.read_bytes(), read()
+    code, preview = call("/api/actions/preview", {
+        "action_kind": "todo.update", "summary": "Review an administrative edit",
+        "normalized_parameters": {"goal_id": "cold", "todo_id": "todo_current",
+            "operation": "edit", "text": "Reviewed imported work", "agent_id": "agent-a"},
+        "context": {}, "idempotency_key": "administrative-copy"})
+    assert code == 201 and preview["proposal"]["status"] == "preview_ready", preview
+    assert read() == original_head and state.read_bytes() == original_state
+    # Import and copy preview do not authorize completion of the imported work.
     code, refused = call("/api/actions/preview", {
-        "action_kind": "todo.update", "summary": "An edit without execution proof",
+        "action_kind": "todo.update", "summary": "Completion without execution proof",
         "normalized_parameters": {"goal_id": "cold", "todo_id": "todo_current",
-            "operation": "edit", "text": "Forbidden edit", "agent_id": "agent-a"},
-        "context": {}, "idempotency_key": "missing-lease"})
+            "operation": "complete", "agent_id": "agent-a", "no_followup": True},
+        "context": {}, "idempotency_key": "missing-execution"})
     assert code == 400 and refused["error_code"] == "handoff_mode_requires_lease", refused
     assert read() == original_head and state.read_bytes() == original_state
     stores = list(runtime.rglob("authority-store-*.json" if provider == "file" else "authority-*.sqlite"))
diff --git a/tests/control_plane/test_cold_source_import_cli.py b/tests/control_plane/test_cold_source_import_cli.py
index da84677406..99f96bf957 100644
--- a/tests/control_plane/test_cold_source_import_cli.py
+++ b/tests/control_plane/test_cold_source_import_cli.py
@@ -368,8 +368,9 @@ def listing():
         "--agent-id", "agent-a", "--note", "Work retained after migration",
         "--update-operation-id", "later-update",
         "--update-expected-provider-revision", before["authority_read"]["provider_revision"]]
-    # Importing source history grants no lease to mutate the imported work.
-    denied = run(*update, succeeds=False)
+    # Administrative copy is proof-free; import does not qualify explicit proof.
+    denied = run(*update, "--task-lease-idempotency-key", "not-acquired",
+        "--task-lease-expected-version", "1", succeeds=False)
     assert denied["error_code"] == "handoff_mode_requires_lease"
     assert listing()["todos"] == before["todos"]
     acquired = run("task-lease", "acquire", "--goal-id", "cold", "--todo-id", "todo_current",
diff --git a/tests/control_plane/test_quota_authority_settlement_journey.py b/tests/control_plane/test_quota_authority_settlement_journey.py
index 10aec6500e..9e720c0161 100644
--- a/tests/control_plane/test_quota_authority_settlement_journey.py
+++ b/tests/control_plane/test_quota_authority_settlement_journey.py
@@ -294,7 +294,9 @@ def run(*args):
     assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == cli.TODO_ID
     update = ("todo", "update", "--todo-id", cli.TODO_ID, "--agent-id", cli.AGENT_ID,
               "--text", "Continue after policy migration")
-    code, rejected = run(*update)
+    # A migrated owner may edit copy, but migration supplies no execution proof.
+    code, rejected = run(*update, "--task-lease-idempotency-key", "not-acquired",
+                         "--task-lease-expected-version", "1")
     assert code == 1 and rejected["error_code"] == "handoff_mode_requires_lease", rejected
 
     lease_key = "migrated-delivery"
diff --git a/tests/control_plane/test_research_execution_authority.py b/tests/control_plane/test_research_execution_authority.py
index 409afe327b..5c4f61af80 100644
--- a/tests/control_plane/test_research_execution_authority.py
+++ b/tests/control_plane/test_research_execution_authority.py
@@ -242,16 +242,21 @@ def capture_native(method, params, **kwargs):
             from loopx.todos import update_goal_todo
             from loopx.capabilities.explore.composition_frontier import project_live_explore_composition_frontier
             from loopx.control_plane.work_items.semantic_replan_writeback import qualify_replan_writeback
-            from loopx.control_plane.work_items.task_lease import release_task_lease
-            with pytest.raises(LocalCoordinationAuthorityUnavailable, match="Release the active execution lease"):
+            before_pause = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True)
+            with pytest.raises(LocalCoordinationAuthorityUnavailable):
                 update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
-                    status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.")
-            released = release_task_lease(registry_path=registry, runtime_root=runtime, goal_id=goal,
-                todo_id="todo_joint", owner=agent, idempotency_key="native-research-proof",
-                expected_version=acquired["lease"]["version"])
-            assert released["ok"]
-            update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
+                    status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.",
+                    task_lease_idempotency_key="native-research-proof",
+                    task_lease_expected_version=acquired["lease"]["version"] + 1)
+            assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True) == before_pause
+            # The owner's proof-free pause retires its lease in the same CAS.
+            paused = update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
                 status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.")
+            assert paused["blocked_lifecycle_transition"]["lease_retirement"] == "released"
+            retained = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True)
+            old_execution = next(lease for lease in retained["leases"] if lease["todo_id"] == "todo_joint")
+            assert old_execution["status"] == "released"
+            assert old_execution["version"] == acquired["lease"]["version"]
             source = json.loads(registry.read_text())["goals"][0]
             def live_deferred():
                 return project_live_explore_composition_frontier(runtime_root=runtime, goal_id=goal,
diff --git a/tests/control_plane/test_soft_claim_migration_continuation.py b/tests/control_plane/test_soft_claim_migration_continuation.py
index a50dcbd7e0..48c4ed7417 100644
--- a/tests/control_plane/test_soft_claim_migration_continuation.py
+++ b/tests/control_plane/test_soft_claim_migration_continuation.py
@@ -107,11 +107,16 @@ def lost_response(*args, **kwargs):
     assert {k: snapshot()["todos"][0][k] for k in preserved} == preserved
     assert cli(*migration)["status"] == "replayed" and snapshot() == current
     assert registry.read_bytes() == registry_bytes
-    # Returning to hard lease restores its execution requirement; soft copy
-    # admission did not consume, replace or silently upgrade the retained grant.
+    # Ordinary owner copy stays administrative after policy migration; explicit
+    # execution proof still cannot consume or upgrade the released grant.
     plan = registry.parent / "hard-again.json"
     preview = cli("handoff-mode", "plan-migration", "--mode", "hard_lease", "--plan", plan)
     cli("handoff-mode", "migrate", "--plan", plan, "--plan-sha256", preview["plan_sha256"], "--execute")
     hard = snapshot()
-    assert cli(*edit, ok=False)["error_code"] == "handoff_mode_requires_lease"
-    assert snapshot() == hard
+    assert cli(*edit)["status"] == "applied"
+    copied = snapshot()
+    assert copied["leases"] == hard["leases"]
+    assert copied["todos"][0]["note"] == "CLI observation"
+    assert cli(*edit, "--task-lease-idempotency-key", "previous-execution",
+        "--task-lease-expected-version", str(lease["version"]), ok=False)["error_code"] == "handoff_mode_requires_lease"
+    assert snapshot() == copied
diff --git a/tests/control_plane/test_todo_consumer_host_retirement.py b/tests/control_plane/test_todo_consumer_host_retirement.py
index 43b792f062..a80c99a903 100644
--- a/tests/control_plane/test_todo_consumer_host_retirement.py
+++ b/tests/control_plane/test_todo_consumer_host_retirement.py
@@ -80,8 +80,8 @@ def test_app_import_actions_and_restart_without_legacy_consumers(cold_api, provi
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_app_rejects_missing_lease_and_provider_without_legacy_consumers(cold_api, provider):
-    cold.test_app_enforces_lease_and_missing_provider_without_legacy_fallback(cold_api, provider)
+def test_app_copy_preserves_execution_and_provider_fences_without_legacy_consumers(cold_api, provider):
+    cold.test_app_copy_preserves_execution_and_provider_fences_without_legacy_fallback(cold_api, provider)
 
 
 def test_app_rejects_changed_source_and_backup_without_legacy_consumers(cold_api):

From e35b3bf49c70dd5af06f3c2cdb8f8cbfe8845386 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 06:53:55 +0800
Subject: [PATCH 04/10] test(chat): keep research history navigation on its
 frontend

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 examples/personal-workspace-browser/research-history.mjs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/examples/personal-workspace-browser/research-history.mjs b/examples/personal-workspace-browser/research-history.mjs
index e353210bdd..feec44b2fc 100644
--- a/examples/personal-workspace-browser/research-history.mjs
+++ b/examples/personal-workspace-browser/research-history.mjs
@@ -47,7 +47,9 @@ export const researchHistoryScenario = {
     let unavailable = false;
     try {
       context = await openWorkspacePage(browser, url, {collectCoverage, beforeGoto: async (_, page) => {
-        await page.route("**/status.json*", async route => {
+        // The document URL also contains ?statusUrl=/status.json. Only proxy
+        // the status endpoint; navigation must reach the frontend server.
+        await page.route(url => url.pathname === "/status.json", async route => {
           const response = await fetchWorkspaceBackend(route, owner.base);
           assert.equal(response.status(), 200);
           await route.fulfill({response});
@@ -155,7 +157,7 @@ Path(sys.argv[1]).write_text(json.dumps({'goal_id':goal,'display_name':'Syntheti
       const realOwner=owner; owner=collisionOwner;
       await realOwner.close();
       await page.setViewportSize({width:1280,height:900});
-      await page.goto(new URL(`/chat/?goalId=${goal}&view=overview&statusUrl=%2Fstatus.json`,url).href,{waitUntil:'networkidle'});
+      await page.goto(new URL(`?goalId=${goal}&view=overview&statusUrl=%2Fstatus.json`,url).href,{waitUntil:'networkidle'});
       await history.getByRole('status').filter({hasText:'11 / 11'}).waitFor();
       const target = await history.locator(".research-timeline-marker").first().evaluate(button => {
         const box = button.getBoundingClientRect(), dot = getComputedStyle(button, "::after");

From 6435cb4479e1f803826589735cdcd85a835de071 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 06:55:38 +0800
Subject: [PATCH 05/10] fix(benchmark): reuse canonical digest owner for
 trajectory evidence

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../trajectory_assessment_store.py            |  4 +-
 .../capabilities/trajectory_assessment.ts     |  3 +-
 .../test_content_digest_single_owner.py       |  1 +
 .../test_trajectory_assessment_store.py       | 37 +++++++++++++++++++
 .../content_digest_single_owner.test.ts       |  1 +
 .../trajectory_assessment.test.ts             | 19 ++++++++++
 6 files changed, 62 insertions(+), 3 deletions(-)

diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
index 0894d93043..a372a050ef 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
+++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
@@ -10,8 +10,8 @@
 import hashlib
 import json
 from pathlib import Path, PurePosixPath
-import re
 
+from ...control_plane.content_digest import BARE_SHA256_PATTERN
 from ...control_plane.effect_runtime import effect_runtime_result
 from ...file_lock import exclusive_file_lock
 from .trajectory_assessment import _digest, _write, execute, read_job, request_cancel
@@ -28,7 +28,7 @@ def __init__(self, reason_code: str):
 
 
 def _opaque(value: str) -> str:
-    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
+    if not isinstance(value, str) or not BARE_SHA256_PATTERN.fullmatch(value):
         raise PackUnavailable("invalid_identifier")
     return value
 
diff --git a/loopx/control_plane/capabilities/trajectory_assessment.ts b/loopx/control_plane/capabilities/trajectory_assessment.ts
index 96f0e01e62..49a5abae5f 100644
--- a/loopx/control_plane/capabilities/trajectory_assessment.ts
+++ b/loopx/control_plane/capabilities/trajectory_assessment.ts
@@ -1,5 +1,6 @@
 /** Benchmark-toolkit's observer sidecar; never a solver or native-score authority. */
 import {createHash} from 'node:crypto';
+import {BARE_SHA256_PATTERN} from '../content_digest.ts';
 import {EffectRuntimeRequestError} from '../effect_runtime_errors.ts';
 
 export const TRAJECTORY_METHOD = 'research-episodes-v2';
@@ -38,7 +39,7 @@ export function parseTrajectoryInput(raw: unknown): TrajectoryInput {
   const sources = list(x.sources, 'sources').map(rawSource => {
     const r = object(rawSource, 'source');
     const hash = str(r.artifact_sha256, 'artifact_sha256');
-    if (!/^[a-f0-9]{64}$/.test(hash)) throw new EffectRuntimeRequestError('artifact_sha256 must be a sha256 hex digest');
+    if (!BARE_SHA256_PATTERN.test(hash)) throw new EffectRuntimeRequestError('artifact_sha256 must be a sha256 hex digest');
     return {id: str(r.id, 'source.id'), artifact: str(r.artifact, 'artifact'), artifact_sha256: hash,
       first_line: integer(r.first_line, 'first_line', 1), text: str(r.text, 'source.text')};
   });
diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py
index e0a1f61faa..7c6a8d20ad 100644
--- a/tests/architecture/test_content_digest_single_owner.py
+++ b/tests/architecture/test_content_digest_single_owner.py
@@ -161,6 +161,7 @@
     "loopx.capabilities.benchmark_toolkit.integrity",
     "loopx.capabilities.benchmark_toolkit.runtime_continuity",
     "loopx.capabilities.benchmark_toolkit.study_projection",
+    "loopx.capabilities.benchmark_toolkit.trajectory_assessment_store",
     "loopx.capabilities.content_ops.item_lifecycle",
     "loopx.capabilities.deep_research.runtime",
     "loopx.capabilities.issue_fix.outcome_projection",
diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py
index 7bdfa4b0fe..1ab37f89a8 100644
--- a/tests/capabilities/test_trajectory_assessment_store.py
+++ b/tests/capabilities/test_trajectory_assessment_store.py
@@ -147,6 +147,43 @@ def test_owner_isolation_unknown_ids_and_keys_are_rejected(pack):
         cancel_pack(**{**operation, "analysis_key": "a" * 64})
 
 
+def test_digest_rejections_preserve_registration_and_create_no_job(pack):
+    scope, operation, _, _, _, _ = pack
+    root = scope["runtime_root"]
+    before = {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    }
+    malformed = (
+        "a" * 63,
+        "a" * 65,
+        "A" * 64,
+        "g" * 64,
+        "sha256:" + "a" * 64,
+        " " + "a" * 64,
+        "a" * 64 + " ",
+        "\n" + "a" * 64,
+        "a" * 64 + "\n",
+        "a" * 64 + "\r\n",
+        "a" * 64 + "\0",
+        True,
+        64,
+        {"id": "a" * 64},
+        ["a" * 64],
+    )
+    for value in malformed:
+        with pytest.raises(PackUnavailable, match="^invalid_identifier$"):
+            inspect_pack(**scope, pack_id=value)
+    for value in (*malformed, "", None, False, 0, {}, []):
+        with pytest.raises(PackUnavailable, match="^invalid_identifier$"):
+            execute_pack(**{**operation, "analysis_key": value}, codex="must-not-run")
+    for value in ("0" * 64, "0123456789abcdef" * 4, "f" * 64):
+        with pytest.raises(PackUnavailable, match="^pack_not_registered$"):
+            inspect_pack(**scope, pack_id=value)
+    assert {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    } == before
+
+
 def test_registration_requires_real_hash_literal_excerpt_and_contained_path(
     pack, tmp_path
 ):
diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts
index 89abeb040f..f14229be3c 100644
--- a/tests/control_plane_ts/content_digest_single_owner.test.ts
+++ b/tests/control_plane_ts/content_digest_single_owner.test.ts
@@ -95,6 +95,7 @@ const CANONICAL_CONSUMERS = [
   "control_plane/capabilities/explore_result_page.ts",
   "control_plane/capabilities/external_evidence.ts",
   "control_plane/capabilities/goal_capability_organization.ts",
+  "control_plane/capabilities/trajectory_assessment.ts",
   "control_plane/collaboration/chat_mode.ts",
   "control_plane/collaboration/delegation.ts",
   "control_plane/collaboration/delegation_result_use.ts",
diff --git a/tests/control_plane_ts/trajectory_assessment.test.ts b/tests/control_plane_ts/trajectory_assessment.test.ts
index aa47416727..9c04537640 100644
--- a/tests/control_plane_ts/trajectory_assessment.test.ts
+++ b/tests/control_plane_ts/trajectory_assessment.test.ts
@@ -14,6 +14,25 @@ const output = () => ({assessments:[
   {episode_id:'e2',metric:'information_gain',verdict:'unknown',reason:'No valid measurement',citations:[]}],limitations:['Synthetic contract test, not evaluator calibration']});
 const prepared = () => prepareTrajectoryAssessment({input:input(), config});
 
+test('artifact digests retain the exact bare SHA-256 boundary', () => {
+  for (const hash of ['0'.repeat(64), '0123456789abcdef'.repeat(4), 'f'.repeat(64)]) {
+    const x = input(); x.sources[0]!.artifact_sha256 = hash;
+    assert.equal(prepareTrajectoryAssessment({input:x, config}).input.sources[0]!.artifact_sha256, hash);
+  }
+  for (const hash of ['a'.repeat(63), 'a'.repeat(65), 'A'.repeat(64), 'g'.repeat(64),
+    'sha256:' + 'a'.repeat(64), ' ' + 'a'.repeat(64), 'a'.repeat(64) + ' ',
+    '\n' + 'a'.repeat(64), 'a'.repeat(64) + '\n', 'a'.repeat(64) + '\r\n']) {
+    const x = input(); x.sources[0]!.artifact_sha256 = hash;
+    assert.throws(() => prepareTrajectoryAssessment({input:x, config}),
+      {message:'artifact_sha256 must be a sha256 hex digest'});
+  }
+  for (const hash of [null, undefined, true, 64, {}, [], '', ' ', 'a'.repeat(64) + '\0']) {
+    const x = input(); Object.assign(x.sources[0]!, {artifact_sha256:hash});
+    assert.throws(() => prepareTrajectoryAssessment({input:x, config}),
+      {message:'artifact_sha256 must be nonempty text'});
+  }
+});
+
 test('known denominator retains unknown and produces missing-data bounds, not a confidence interval', () => {
   const result=qualifyTrajectoryAssessment({...prepared(),output:output()});
   assert.deepEqual(result.metrics.find(m=>m.metric==='information_gain'),{metric:'information_gain',yes:1,no:0,unknown:1,eligible:2,assessed_fraction:.5,missing_data_bounds:[.5,1]});

From 96a01cf0ef0bc2b6a6bfc6c7b407af4748c214e5 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 07:14:27 +0800
Subject: [PATCH 06/10] fix(recovery): retain the verified backup checkpoint in
 audit readback

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../coordination/authority_backup.ts          |  1 +
 .../control_plane_ts/authority_backup.test.ts | 38 ++++++++++++++-----
 tests/test_state_recovery.py                  |  3 ++
 3 files changed, 32 insertions(+), 10 deletions(-)

diff --git a/loopx/control_plane/coordination/authority_backup.ts b/loopx/control_plane/coordination/authority_backup.ts
index fa94e66bbb..ef12c15d79 100644
--- a/loopx/control_plane/coordination/authority_backup.ts
+++ b/loopx/control_plane/coordination/authority_backup.ts
@@ -159,6 +159,7 @@ export async function inspectAuthorityBackup(dataRoot: string, row: JsonObject):
     if (head === null) throw new Error("authority capture has no head");
     const collection = canonicalTodoCollection(head, goalId, true);
     return {provider: archive.summary.source_provider, head_digest: archive.summary.projection_sha256,
+      provider_revision: archive.summary.source_provider_revision, cursor: archive.summary.commits,
       history_commits: Number(archive.summary.commits), todo_count: collection.projection.todo_ids.length,
       historical_lease_count: collection.leaseIndex?.lease_todo_ids.length ?? 0,
       capture_started_at: row.started_at, capture_completed_at: row.completed_at};
diff --git a/tests/control_plane_ts/authority_backup.test.ts b/tests/control_plane_ts/authority_backup.test.ts
index 1340325b4f..75507ee6b5 100644
--- a/tests/control_plane_ts/authority_backup.test.ts
+++ b/tests/control_plane_ts/authority_backup.test.ts
@@ -1,23 +1,26 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import {existsSync} from "node:fs";
-import {mkdir, mkdtemp, readFile, rm, writeFile} from "node:fs/promises";
+import {mkdir, mkdtemp, readFile, rename, rm, writeFile} from "node:fs/promises";
 import {tmpdir} from "node:os";
 import {dirname, join} from "node:path";
-import {authorityBackupDirectory, captureAuthorityBackup} from "../../loopx/control_plane/coordination/authority_backup.ts";
+import {authorityBackupDirectory, captureAuthorityBackup, inspectAuthorityBackup} from "../../loopx/control_plane/coordination/authority_backup.ts";
 import {verifyAuthorityArchive} from "../../loopx/control_plane/coordination/authority_archive.ts";
 import {openLocalAuthorityStore, localAuthorityProviderPaths, selectLocalSqliteAuthority} from "../../loopx/control_plane/coordination/local_authority_provider.ts";
 import {legacyCoordinationWriterFencePath, COLD_SOURCE_IMPORT_WRITER_FENCE_SCHEMA} from "../../loopx/control_plane/coordination/legacy_writer_fence.ts";
 import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts";
+import {authorityProjectionFixture} from "./authority_projection_fixture.ts";
 
 for (const provider of ["file", "sqlite"] as const) {
-  test(`${provider} owner capture keeps a complete prefix during an actual append`, async () => {
+  test(`${provider} owner capture retains its inspected checkpoint after a later append and source loss`, async () => {
     const root = await mkdtemp(join(tmpdir(), "loopx-authority-backup-"));
     try {
-      if (provider === "sqlite") await selectLocalSqliteAuthority(root, "goal-a", true);
-      const store = await openLocalAuthorityStore(root, "goal-a");
+      const source = join(root, "source");
+      if (provider === "sqlite") await selectLocalSqliteAuthority(source, "goal-a", true);
+      const store = await openLocalAuthorityStore(source, "goal-a");
+      const projection = (value: number) => authorityProjectionFixture("goal-a", [], [], "native", {value});
       const first = await store.commitAuthority({expected_provider_revision: null, operation_id: "first",
-        events: [], receipts: [{operation_id: "first", retained: true}], next_projection: {goal_id: "goal-a", value: 1}});
+        events: [], receipts: [{operation_id: "first", retained: true}], next_projection: projection(1)});
       assert.equal(first.status, "applied");
       let appended = false;
       // Wrap only scheduling of a real provider read. Both commits and the
@@ -28,7 +31,7 @@ for (const provider of ["file", "sqlite"] as const) {
           if (!appended) {
             appended = true;
             const second = await target.commitAuthority({expected_provider_revision: first.provider_revision!,
-              operation_id: "second", events: [], receipts: [], next_projection: {goal_id: "goal-a", value: 2}});
+              operation_id: "second", events: [], receipts: [], next_projection: projection(2)});
             assert.equal(second.status, "applied");
           }
           return page;
@@ -37,13 +40,28 @@ for (const provider of ["file", "sqlite"] as const) {
         return typeof value === "function" ? value.bind(target) : value;
       }});
       const destination = join(root, "capture");
-      const result = await captureAuthorityBackup({runtime_root: root, goal_id: "goal-a", destination},
+      const result = await captureAuthorityBackup({runtime_root: source, goal_id: "goal-a", destination},
         {createStore: () => interleaved});
       assert.equal(result.state, "readable");
-      const exported = await verifyAuthorityArchive(join(destination, authorityBackupDirectory("goal-a"), "authority.jsonl"));
+      const archivePath = join(destination, authorityBackupDirectory("goal-a"), "authority.jsonl");
+      const exported = await verifyAuthorityArchive(archivePath);
       assert.equal(exported.commits, "1");
-      assert.equal((await store.loadAuthority()).cursor, "2");
+      const later = await store.loadAuthority();
+      assert.equal(later.status, "loaded");
+      assert.equal(later.cursor, "2");
       assert.equal(appended, true);
+      await rename(source, join(root, "unavailable-source"));
+      const bytes = await readFile(archivePath);
+      const inspected = await inspectAuthorityBackup(destination, result);
+      assert.equal(inspected.provider_revision, first.provider_revision);
+      assert.equal(inspected.cursor, "1");
+      assert.equal(inspected.history_commits, 1);
+      // A manifest's claimed position cannot replace the archive owner's
+      // verified source position, even when it names a real later revision.
+      await assert.rejects(() => inspectAuthorityBackup(destination, {...result,
+        archive: {...exported, source_provider_revision: later.provider_revision}}), /summary mismatch/);
+      assert.deepEqual(await readFile(archivePath), bytes);
+      assert.equal(existsSync(source), false);
     } finally { await rm(root, {recursive: true, force: true}); }
   });
 }
diff --git a/tests/test_state_recovery.py b/tests/test_state_recovery.py
index 0eef9b61dd..b21a8f6fdb 100644
--- a/tests/test_state_recovery.py
+++ b/tests/test_state_recovery.py
@@ -299,6 +299,9 @@ def test_real_backup_recovers_readable_authority_after_original_sources_disappea
     assert code == 0, captured
     captured_position = captured["authority_read"]
     assert captured_position["cursor"] == "2"
+    [capture] = json.loads(manifest.read_text())["execution"]["authority_captures"]
+    assert capture["archive"]["source_provider_revision"] == captured_position["provider_revision"]
+    assert capture["archive"]["commits"] == captured_position["cursor"]
     # A later real source commit must not become this backup's recovery point.
     code, later = _cli(*source_args, "todo", "add", "--goal-id", "goal-a", "--role", "agent",
                        "--claimed-by", "agent-a", "--operation-id", "after-backup",

From 560fc66b99460aebd25fb2a24b484d9be6d94410 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 07:38:52 +0800
Subject: [PATCH 07/10] fix(benchmark): preserve UTF-8 trajectory evidence
 across locales

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../trajectory_assessment.py                  |  29 ++--
 .../trajectory_assessment_store.py            |  12 +-
 .../test_trajectory_assessment.py             |  71 +++++++---
 .../test_trajectory_assessment_store.py       | 125 +++++++++++++++---
 4 files changed, 189 insertions(+), 48 deletions(-)

diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py
index d1b1523a41..d7c7d38498 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py
+++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py
@@ -33,7 +33,9 @@ def _result_proof(directory: Path, prepared: dict[str, Any]) -> dict[str, str]:
 
 def _write(path: Path, value: Any) -> None:
     temporary = path.with_suffix(".tmp")
-    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n")
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
     temporary.replace(path)
 
 
@@ -43,14 +45,14 @@ def prepare(
     return effect_runtime_result(
         "benchmark.trajectory.prepare",
         {
-            "input": json.loads(input_path.read_text()),
+            "input": json.loads(input_path.read_text(encoding="utf-8")),
             "config": {"model": model, "effort": effort, "service_tier": service_tier},
         },
     )
 
 
 def read_job(directory: Path) -> dict[str, Any]:
-    receipt = json.loads((directory / "receipt.json").read_text())
+    receipt = json.loads((directory / "receipt.json").read_text(encoding="utf-8"))
     if receipt.get("schema_version") != "trajectory_assessment_job_v1":
         raise ValueError("Unsupported trajectory assessment receipt")
     return receipt
@@ -106,7 +108,9 @@ def execute(
                     {
                         **prepared,
                         "output": json.loads(
-                            (directory / "model-output.json").read_text()
+                            (directory / "model-output.json").read_text(
+                                encoding="utf-8"
+                            )
                         ),
                     },
                 )
@@ -138,7 +142,7 @@ def execute(
         (directory / "cancel.request").unlink(missing_ok=True)
         _write(directory / "input.json", prepared["input"])
         _write(directory / "output-schema.json", prepared["output_schema"])
-        (directory / "prompt.txt").write_text(prepared["prompt"])
+        (directory / "prompt.txt").write_text(prepared["prompt"], encoding="utf-8")
         config = prepared["config"]
         command = [
             codex,
@@ -186,9 +190,9 @@ def execute(
         }
         _write(directory / "receipt.json", receipt)
         with (
-            (directory / "prompt.txt").open() as prompt,
-            (directory / "events.jsonl").open("w") as events,
-            (directory / "stderr.txt").open("w") as errors,
+            (directory / "prompt.txt").open(encoding="utf-8") as prompt,
+            (directory / "events.jsonl").open("w", encoding="utf-8") as events,
+            (directory / "stderr.txt").open("w", encoding="utf-8") as errors,
         ):
             process = subprocess.Popen(
                 command,
@@ -197,6 +201,7 @@ def execute(
                 stdout=events,
                 stderr=errors,
                 text=True,
+                encoding="utf-8",
             )
             receipt["process_id"] = process.pid
             _write(directory / "receipt.json", receipt)
@@ -220,7 +225,9 @@ def execute(
                 raise ValueError(
                     f"Codex exited {process.returncode}; inspect stderr.txt and events.jsonl"
                 )
-            output = json.loads((directory / "model-output.json").read_text())
+            output = json.loads(
+                (directory / "model-output.json").read_text(encoding="utf-8")
+            )
             result = effect_runtime_result(
                 "benchmark.trajectory.qualify", {**prepared, "output": output}
             )
@@ -228,7 +235,9 @@ def execute(
             receipt.update(
                 status="succeeded", ok=True, **_result_proof(directory, prepared)
             )
-        for line in (directory / "events.jsonl").read_text().splitlines():
+        for line in (
+            (directory / "events.jsonl").read_text(encoding="utf-8").splitlines()
+        ):
             try:
                 event = json.loads(line)
             except json.JSONDecodeError:
diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
index a372a050ef..aa430d5323 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
+++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
@@ -88,7 +88,7 @@ def _index(owner: Path) -> dict:
     path = owner / "packs.json"
     if not path.exists():
         return {"default_pack_id": None, "pack_ids": []}
-    value = json.loads(path.read_text())
+    value = json.loads(path.read_text(encoding="utf-8"))
     for pack_id in value["pack_ids"]:
         _opaque(pack_id)
     if value["default_pack_id"] not in value["pack_ids"]:
@@ -107,7 +107,7 @@ def _resolve(
     path = owner / "packs" / f"{pack_id}.json"
     if path.is_symlink():
         raise PackUnavailable("invalid_registration")
-    record = json.loads(path.read_text())
+    record = json.loads(path.read_text(encoding="utf-8"))
     if (
         record["owner_goal_id"] != owner_goal_id
         or _pack_id(owner_goal_id, record) != pack_id
@@ -159,7 +159,7 @@ def register_pack(
     prepared = effect_runtime_result(
         "benchmark.trajectory.prepare",
         {
-            "input": json.loads(input_path.read_text()),
+            "input": json.loads(input_path.read_text(encoding="utf-8")),
             "config": config or DEFAULT_CONFIG,
         },
     )
@@ -180,7 +180,7 @@ def register_pack(
         index = _index(owner)
         path = owner / "packs" / f"{pack_id}.json"
         if path.exists():
-            if json.loads(path.read_text()) != record:
+            if json.loads(path.read_text(encoding="utf-8")) != record:
                 raise PackUnavailable("registration_conflict")
         else:
             _write(path, record)
@@ -286,7 +286,9 @@ def inspect_pack(
                     {
                         **prepared,
                         "output": json.loads(
-                            (directory / "model-output.json").read_text()
+                            (directory / "model-output.json").read_text(
+                                encoding="utf-8"
+                            )
                         ),
                     },
                 )
diff --git a/tests/capabilities/test_trajectory_assessment.py b/tests/capabilities/test_trajectory_assessment.py
index 3b40389f88..e94c0fc8fb 100644
--- a/tests/capabilities/test_trajectory_assessment.py
+++ b/tests/capabilities/test_trajectory_assessment.py
@@ -30,9 +30,9 @@ def prepared(tmp_path):
                     case_id="case",
                     run_id="run",
                     revision="r1",
-                    window="one episode",
+                    window="one episode · 观察窗口",
                 ),
-                "objective": "Improve native utility",
+                "objective": "Improve native utility · 原始目标",
                 "selection": dict(
                     rule="synthetic lifecycle fixture",
                     population_episodes=1,
@@ -44,7 +44,7 @@ def prepared(tmp_path):
                         artifact="result.txt",
                         artifact_sha256="a" * 64,
                         first_line=1,
-                        text="No utility measurement is available.",
+                        text="No utility measurement is available · 未测量。",
                     )
                 ],
                 "episodes": [
@@ -55,13 +55,17 @@ def prepared(tmp_path):
                         eligible_metrics=["benefit_test"],
                     )
                 ],
-            }
-        )
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
     )
     return prepare(path, model="fixture", effort="max", service_tier="fast")
 
 
-def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
+def provider(
+    tmp_path: Path, *, delay=0, code=0, invalid=False, invalid_utf8=False
+) -> str:
     script = tmp_path / "codex-fixture"
     output = {
         "assessments": [
@@ -69,7 +73,7 @@ def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
                 episode_id="episode",
                 metric="benefit_test",
                 verdict="unknown",
-                reason="Measurement is unavailable",
+                reason="Measurement is unavailable · 未测量",
                 citations=[],
             )
         ],
@@ -77,15 +81,24 @@ def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
     }
     if invalid:
         output["assessments"] = []
-    script.write_text(f"""#!{sys.executable}
+    payload = (
+        b"\xff"
+        if invalid_utf8
+        else json.dumps(output, ensure_ascii=False).encode("utf-8")
+    )
+    script.write_text(
+        f"""#!{sys.executable}
 import json,sys,time
 from pathlib import Path
-sys.stdin.read()
+assert "\\u539f\\u59cb\\u76ee\\u6807" in sys.stdin.buffer.read().decode("utf-8")
 time.sleep({delay})
 if {code}: sys.exit({code})
-Path(sys.argv[sys.argv.index('-o')+1]).write_text({json.dumps(json.dumps(output))})
-print(json.dumps({{'type':'turn.completed','usage':{{'input_tokens':12,'output_tokens':3}}}}))
-""")
+Path(sys.argv[sys.argv.index('-o')+1]).write_bytes({payload!r})
+sys.stderr.buffer.write("\\u5408\\u6210\\u8bca\\u65ad".encode("utf-8"))
+sys.stdout.buffer.write((json.dumps({{'type':'turn.completed','message':'\\u5206\\u6790\\u5b8c\\u6210','usage':{{'input_tokens':12,'output_tokens':3}}}},ensure_ascii=False)+'\\n').encode("utf-8"))
+""",
+        encoding="utf-8",
+    )
     script.chmod(0o700)
     return str(script)
 
@@ -98,11 +111,13 @@ def test_actual_child_reuse_and_repair_of_derived_output(prepared, tmp_path):
     }
     assert first["served_service_tier"] is None
     directory = Path(first["directory"])
-    (directory / "assessment.json").write_text("{}")
+    (directory / "assessment.json").write_text("{}", encoding="utf-8")
     second = execute(prepared, tmp_path / "jobs", codex="must-not-run")
     assert second["reused"] is True
     assert (
-        json.loads((directory / "assessment.json").read_text())["metrics"][1]["unknown"]
+        json.loads((directory / "assessment.json").read_text(encoding="utf-8"))[
+            "metrics"
+        ][1]["unknown"]
         == 1
     )
     assert second["analysis_wall_seconds"] == first["analysis_wall_seconds"]
@@ -120,11 +135,36 @@ def test_failed_model_output_needs_explicit_retry_and_preserves_failure(
     success = execute(prepared, tmp_path / "jobs", retry=True, codex=provider(tmp_path))
     assert success["status"] == "succeeded" and success["attempt"] == 2
     assert (
-        json.loads((directory / "attempt-1" / "receipt.json").read_text())["status"]
+        json.loads(
+            (directory / "attempt-1" / "receipt.json").read_text(encoding="utf-8")
+        )["status"]
         == "failed"
     )
 
 
+def test_invalid_utf8_output_fails_strictly_and_retry_preserves_bytes(
+    prepared, tmp_path
+):
+    with pytest.raises(UnicodeDecodeError):
+        execute(
+            prepared, tmp_path / "jobs", codex=provider(tmp_path, invalid_utf8=True)
+        )
+    directory = tmp_path / "jobs" / prepared["analysis_key"]
+    assert read_job(directory)["status"] == "failed"
+    assert not (directory / "execution.lock").exists()
+    assert not (directory / "assessment.json").exists()
+    assert execute(prepared, tmp_path / "jobs", codex="must-not-run")["ok"] is False
+    assert (
+        execute(prepared, tmp_path / "jobs", retry=True, codex=provider(tmp_path))[
+            "status"
+        ]
+        == "succeeded"
+    )
+    assert (directory / "attempt-1" / "model-output.json").read_bytes() == b"\xff"
+    result = json.loads((directory / "assessment.json").read_text(encoding="utf-8"))
+    assert result["assessments"][0]["reason"] == "Measurement is unavailable · 未测量"
+
+
 def test_concurrent_execution_and_cancel_do_not_signal_stored_pid(prepared, tmp_path):
     results = []
     worker = threading.Thread(
@@ -170,6 +210,7 @@ def test_preview_cli_is_read_only_and_renders_json(prepared, tmp_path):
         ],
         capture_output=True,
         text=True,
+        encoding="utf-8",
     )
     assert result.returncode == 0, result.stderr
     assert json.loads(result.stdout)["status"] == "preview"
diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py
index 1ab37f89a8..130fe6eaf4 100644
--- a/tests/capabilities/test_trajectory_assessment_store.py
+++ b/tests/capabilities/test_trajectory_assessment_store.py
@@ -4,6 +4,8 @@
 
 import hashlib
 import json
+import os
+import subprocess
 import sys
 
 import pytest
@@ -23,7 +25,9 @@
 def pack(tmp_path):
     sources = tmp_path / "sources"
     sources.mkdir()
-    evidence = b"Prior question\nThe native measurement did not improve utility.\n"
+    evidence = "Prior question\nThe native measurement did not improve utility · 原始测量。\n".encode(
+        "utf-8"
+    )
     (sources / "result.txt").write_bytes(evidence)
     data = {
         "schema_version": "trajectory_assessment_input_v1",
@@ -32,9 +36,9 @@ def pack(tmp_path):
             case_id="case",
             run_id="run",
             revision="fixed",
-            window="one episode",
+            window="one episode · 观察窗口",
         ),
-        "objective": "Improve native utility",
+        "objective": "Improve native utility · 原始目标",
         "selection": dict(
             rule="synthetic complete population", population_episodes=1, exhaustive=True
         ),
@@ -57,10 +61,10 @@ def pack(tmp_path):
         ],
     }
     input_path = tmp_path / "input.json"
-    input_path.write_text(json.dumps(data))
+    input_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
     scope = dict(runtime_root=tmp_path / "runtime", owner_goal_id="research-owner")
     registered = register_pack(
-        **scope, input_path=input_path, source_root=sources, title="Case · one episode"
+        **scope, input_path=input_path, source_root=sources, title="Case · 原始目标"
     )
     operation = {
         **scope,
@@ -73,20 +77,24 @@ def pack(tmp_path):
                 episode_id="episode",
                 metric="benefit_test",
                 verdict="yes",
-                reason="A negative native test still tests benefit.",
+                reason="A negative native test still tests benefit · 负向测量。",
                 citations=[dict(source_id="source", first_line=2, last_line=2)],
             )
         ],
         "limitations": ["Synthetic process; not model quality evidence"],
     }
     executable = tmp_path / "codex-fixture"
-    executable.write_text(f"""#!{sys.executable}
+    executable.write_text(
+        f"""#!{sys.executable}
 import json,sys
 from pathlib import Path
-sys.stdin.read()
-Path(sys.argv[sys.argv.index('-o')+1]).write_text({json.dumps(json.dumps(output))})
-print(json.dumps({{'type':'turn.completed','usage':{{'input_tokens':10,'output_tokens':2}}}}))
-""")
+assert "\\u539f\\u59cb\\u76ee\\u6807" in sys.stdin.buffer.read().decode("utf-8")
+Path(sys.argv[sys.argv.index('-o')+1]).write_bytes({json.dumps(output, ensure_ascii=False).encode("utf-8")!r})
+sys.stderr.buffer.write("\\u5408\\u6210\\u8bca\\u65ad".encode("utf-8"))
+sys.stdout.buffer.write((json.dumps({{'type':'turn.completed','message':'\\u5206\\u6790\\u5b8c\\u6210','usage':{{'input_tokens':10,'output_tokens':2}}}},ensure_ascii=False)+'\\n').encode("utf-8"))
+""",
+        encoding="utf-8",
+    )
     executable.chmod(0o700)
     return scope, operation, registered, input_path, sources, str(executable)
 
@@ -112,10 +120,13 @@ def test_scope_discovery_execute_read_citation_and_changed_source_history(pack):
     assert "directory" not in view["job"] and "process_id" not in view["job"]
     assert execute_pack(**operation, codex="must-not-run")["reused"] is True
     citation = read_citation(**operation, source_id="source", first_line=2, last_line=2)
-    assert citation["text"] == "The native measurement did not improve utility."
+    assert (
+        citation["text"]
+        == "The native measurement did not improve utility · 原始测量。"
+    )
     with pytest.raises(PackUnavailable, match="citation_not_in_result"):
         read_citation(**operation, source_id="source", first_line=1, last_line=2)
-    (sources / "result.txt").write_text("changed")
+    (sources / "result.txt").write_text("changed", encoding="utf-8")
     assert discover(**scope)["packs"][0]["reason_code"] == "source_changed"
     with pytest.raises(PackUnavailable, match="source_changed"):
         execute_pack(**operation, codex="must-not-run")
@@ -132,6 +143,84 @@ def test_scope_discovery_execute_read_citation_and_changed_source_history(pack):
     )
 
 
+@pytest.mark.skipif(
+    os.name != "posix", reason="C locale coercion is a POSIX regression"
+)
+def test_utf8_pack_and_child_roundtrip_with_ascii_default_locale(pack):
+    scope, _, registered, input_path, sources, executable = pack
+    script = r"""
+import codecs, locale, sys
+from pathlib import Path
+from loopx.capabilities.benchmark_toolkit.trajectory_assessment import prepare, read_job
+from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import (
+    PackUnavailable, discover, execute_pack, inspect_pack, read_citation, register_pack,
+)
+assert sys.flags.utf8_mode == 0
+assert codecs.lookup(locale.getencoding()).name == "ascii"
+input_path, sources, runtime = map(Path, sys.argv[1:4])
+scope = dict(runtime_root=runtime, owner_goal_id="research-owner")
+prepared = prepare(input_path, model="gpt-6.1-sol", effort="max", service_tier="fast")
+assert prepared["input"]["objective"] == "Improve native utility · 原始目标"
+registered = register_pack(**scope, input_path=input_path, source_root=sources, title="Case · 原始目标")
+assert registered["pack_id"] == sys.argv[4]
+assert registered["analysis_key"] == prepared["analysis_key"]
+assert discover(**scope)["packs"][0]["title"] == "Case · 原始目标"
+assert inspect_pack(**scope)["job"] is None
+operation = dict(**scope, pack_id=registered["pack_id"], analysis_key=registered["analysis_key"])
+first = execute_pack(**operation, codex=sys.argv[5])
+assert first["status"] == "succeeded" and first["usage"] == dict(input_tokens=10, output_tokens=2)
+view = inspect_pack(**scope)
+assert view["result"]["assessments"][0]["reason"] == "A negative native test still tests benefit · 负向测量。"
+assert view["result"]["metrics"][1]["yes"] == 1
+assert "directory" not in view["job"] and "process_id" not in view["job"]
+assert execute_pack(**operation, codex="must-not-run")["reused"] is True
+directory = next(runtime.glob("benchmark/trajectory-assessments/*/jobs/*"))
+assert read_job(directory)["scope"]["window"] == "one episode · 观察窗口"
+assert "原始目标" in (directory / "prompt.txt").read_text(encoding="utf-8")
+assert "分析完成" in (directory / "events.jsonl").read_text(encoding="utf-8")
+assert (directory / "stderr.txt").read_text(encoding="utf-8") == "合成诊断"
+citation = dict(source_id="source", first_line=2, last_line=2)
+assert read_citation(**operation, **citation)["text"] == "The native measurement did not improve utility · 原始测量。"
+(sources / "result.txt").write_bytes(b"changed")
+assert discover(**scope)["packs"][0]["reason_code"] == "source_changed"
+try:
+    execute_pack(**operation, codex="must-not-run")
+except PackUnavailable as error:
+    assert str(error) == "source_changed"
+else:
+    raise AssertionError("Changed sources must not run a new analysis")
+assert inspect_pack(**scope)["result"] == view["result"]
+assert read_citation(**operation, **citation)["applicability"] == "historical"
+"""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-X",
+            "utf8=0",
+            "-c",
+            script.encode("ascii", "backslashreplace").decode("ascii"),
+            str(input_path),
+            str(sources),
+            str(scope["runtime_root"]),
+            registered["pack_id"],
+            executable,
+        ],
+        env={
+            **os.environ,
+            "LC_ALL": "C",
+            "LANG": "C",
+            "PYTHONUTF8": "0",
+            "PYTHONCOERCECLOCALE": "0",
+            "PYTHONIOENCODING": "ascii",
+        },
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        timeout=60,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
 def test_owner_isolation_unknown_ids_and_keys_are_rejected(pack):
     scope, operation, _, _, _, _ = pack
     other = {**scope, "owner_goal_id": "other-owner"}
@@ -188,7 +277,7 @@ def test_registration_requires_real_hash_literal_excerpt_and_contained_path(
     pack, tmp_path
 ):
     scope, _, _, input_path, sources, _ = pack
-    original = json.loads(input_path.read_text())
+    original = json.loads(input_path.read_text(encoding="utf-8"))
     for change, message in (
         ({"text": "invented"}, "excerpt_mismatch"),
         ({"artifact_sha256": "a" * 64}, "source_changed"),
@@ -196,16 +285,16 @@ def test_registration_requires_real_hash_literal_excerpt_and_contained_path(
     ):
         data = json.loads(json.dumps(original))
         data["sources"][0].update(change)
-        input_path.write_text(json.dumps(data))
+        input_path.write_text(json.dumps(data), encoding="utf-8")
         with pytest.raises(PackUnavailable, match=message):
             register_pack(
                 **scope, input_path=input_path, source_root=sources, title="bad"
             )
     outside = tmp_path / "outside.txt"
-    outside.write_text("outside")
+    outside.write_text("outside", encoding="utf-8")
     (sources / "link.txt").symlink_to(outside)
     original["sources"][0]["artifact"] = "link.txt"
-    input_path.write_text(json.dumps(original))
+    input_path.write_text(json.dumps(original), encoding="utf-8")
     with pytest.raises(PackUnavailable, match="invalid_source_path"):
         register_pack(**scope, input_path=input_path, source_root=sources, title="bad")
     assert len(discover(**scope)["packs"]) == 1
@@ -261,6 +350,6 @@ def upgraded_method(record):
             "benchmark/trajectory-assessments/*/jobs/*/assessment.json"
         )
     )
-    result_file.write_text("{}")
+    result_file.write_text("{}", encoding="utf-8")
     with pytest.raises(PackUnavailable, match="historical_result_unverified"):
         inspect_pack(**scope)

From e3f94526c9bcaf3b4f88a5d8a5f535a84ebfb998 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 08:37:25 +0800
Subject: [PATCH 08/10] test(ci): retire managed runtimes with their disposable
 fixtures

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/conftest.py                            |  19 +++
 tests/managed_effect_runtime_fixture.py      |  58 ++++++++
 tests/test_managed_effect_runtime_fixture.py | 137 +++++++++++++++++++
 3 files changed, 214 insertions(+)
 create mode 100644 tests/managed_effect_runtime_fixture.py
 create mode 100644 tests/test_managed_effect_runtime_fixture.py

diff --git a/tests/conftest.py b/tests/conftest.py
index da0bfb0b83..0cded7d70e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -16,6 +16,11 @@
 
 import pytest  # noqa: E402
 
+from managed_effect_runtime_fixture import (  # noqa: E402
+    capture_disposable_effect_runtimes,
+    stop_disposable_effect_runtimes,
+)
+
 from loopx import paths  # noqa: E402
 from loopx.canary.runner import SMOKE_SUITE_CHOICES  # noqa: E402
 from loopx.semantics.production import NPM_DEV_DEPENDENCIES_MISSING  # noqa: E402
@@ -37,6 +42,20 @@
 _REAL_DEFAULT_ROUTES = (paths.DEFAULT_RUNTIME_ROOT, paths.LEGACY_RUNTIME_ROOT)
 
 
+@pytest.hookimpl(wrapper=True, tryfirst=True)
+def pytest_runtest_teardown(item):
+    temporary = item.funcargs.get("tmp_path")
+    captured = capture_disposable_effect_runtimes(temporary) if isinstance(temporary, Path) else []
+    try:
+        return (yield)
+    finally:
+        # Run after fixture finalizers and monkeypatch restoration, including
+        # failed tests. Ending a fixture must not leave its copied-source Node
+        # processes alive for the runtime's normal five-minute reuse window.
+        if isinstance(temporary, Path):
+            stop_disposable_effect_runtimes(temporary, captured)
+
+
 @pytest.fixture(scope="session", autouse=True)
 def _isolated_worker_effect_runtime(tmp_path_factory):
     # HOME does not namespace the Effect process on POSIX: its locator uses
diff --git a/tests/managed_effect_runtime_fixture.py b/tests/managed_effect_runtime_fixture.py
new file mode 100644
index 0000000000..16bd0bcc46
--- /dev/null
+++ b/tests/managed_effect_runtime_fixture.py
@@ -0,0 +1,58 @@
+"""Retire runtimes in a finished test's disposable namespace, via their owner."""
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+from uuid import uuid4
+
+from loopx.control_plane import effect_runtime
+
+
+def capture_disposable_effect_runtimes(root: Path) -> list[tuple[Path, str]]:
+    if root.is_symlink():
+        return []
+    owned_root = root.resolve()
+    captured = []
+    for locator in root.glob("**/loopx-effect-runtime-*/runtime-*.json"):
+        # Never follow a fixture's borrowed namespace or locator outside the
+        # test directory. Copied packages can legitimately own several revisions.
+        if locator.is_symlink() or not locator.resolve().is_relative_to(owned_root):
+            continue
+        try:
+            captured.append((locator, locator.read_text(encoding="utf-8")))
+        except OSError:
+            continue
+    return captured
+
+
+def stop_disposable_effect_runtimes(
+    root: Path, captured: list[tuple[Path, str]] | None = None,
+) -> None:
+    records = dict.fromkeys([*(captured or []), *capture_disposable_effect_runtimes(root)])
+    if not records:
+        return
+    # pytest may remove passed tmp_path directories during fixture teardown.
+    # Validate the captured bytes with the existing reader after monkeypatches
+    # are restored; the authenticated ping still verifies the exact live owner.
+    with tempfile.TemporaryDirectory(prefix="loopx-test-runtime-cleanup-") as scratch:
+        for locator, raw in records:
+            validation_file = Path(scratch) / locator.name
+            validation_file.write_text(raw, encoding="utf-8")
+            info = effect_runtime._read_info(validation_file, fingerprint=None)
+            if info is None or locator.name != f"runtime-{info['fingerprint'][:16]}.json":
+                continue
+            try:
+                response = effect_runtime._request_with_info(
+                    info, request_id=uuid4().hex, method="runtime.ping", params={}, timeout=1,
+                )
+            except (OSError, RuntimeError):
+                # Corrupt, stale or rejected metadata cannot authorize shutdown.
+                continue
+            ping = response.get("result")
+            if (not isinstance(ping, dict) or ping.get("ready") is not True or ping.get("pid") != info["pid"]
+                    or ping.get("fingerprint") != info["fingerprint"]):
+                continue
+            result = effect_runtime._stop_effect_runtime(
+                locator, info, timeout=5, wait_for_process_exit=True,
+            )
+            assert result["stopped"], f"Disposable runtime did not exit: {locator}"
diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py
new file mode 100644
index 0000000000..ccec08ae9f
--- /dev/null
+++ b/tests/test_managed_effect_runtime_fixture.py
@@ -0,0 +1,137 @@
+"""Disposable pytest namespaces must not retain managed runtime processes."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+from managed_effect_runtime_fixture import stop_disposable_effect_runtimes
+from loopx.control_plane import effect_runtime
+
+
+def _has_exited(pid):
+    if not effect_runtime._pid_is_alive(pid):
+        return True
+    if sys.platform == "linux":
+        # A detached child's zombie has exited but can await the container
+        # init's reap. kill(pid, 0) alone incorrectly calls it a live runtime.
+        try:
+            state = (Path("/proc") / str(pid) / "status").read_text(encoding="utf-8")
+        except FileNotFoundError:
+            return True
+        return "\nState:\tZ" in "\n" + state
+    return False
+
+
+@pytest.mark.parametrize("failing", [False, True])
+@pytest.mark.parametrize("retention", ["all", "none"])
+def test_pytest_retires_source_and_copied_package_runtimes(tmp_path, failing, retention):
+    repository = Path(__file__).resolve().parents[1]
+    records = tmp_path / "runtimes.jsonl"
+    probe = tmp_path / "test_disposable_namespace.py"
+    child_script = """
+import json, sys
+from pathlib import Path
+sys.path.insert(0, sys.argv[1])
+from loopx.control_plane import effect_runtime
+effect_runtime.effect_runtime_result("runtime.ping", {})
+locator = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint())
+print(json.dumps({"path": str(locator), "info": json.loads(locator.read_text(encoding="utf-8"))}))
+"""
+    probe.write_text(textwrap.dedent(f'''
+        import json, os, shutil, subprocess, sys
+        from pathlib import Path
+        import pytest
+
+        @pytest.mark.parametrize("copied", [False, True])
+        def test_owned_runtime(tmp_path, copied):
+            source = Path({str(repository)!r})
+            if copied:
+                receiver = tmp_path / "receiver"
+                shutil.copytree(source / "loopx", receiver / "loopx",
+                    ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
+                source = receiver
+            script = {child_script!r}
+            environment = {{**os.environ, "TMPDIR": str(tmp_path), "TEMP": str(tmp_path),
+                "TMP": str(tmp_path), "PYTHONPATH": str(source)}}
+            child = subprocess.run([sys.executable, "-c", script, str(source)],
+                cwd=tmp_path, env=environment, capture_output=True, text=True, timeout=60)
+            assert child.returncode == 0, child.stdout + child.stderr
+            with Path({str(records)!r}).open("a", encoding="utf-8") as stream:
+                stream.write(child.stdout.strip() + "\\n")
+            assert not {failing!r}, "fixture failure retained"
+    '''), encoding="utf-8")
+    environment = {**os.environ, "PYTHONPATH": os.pathsep.join([
+        str(repository / "tests"), str(repository),
+    ])}
+    observed = []
+    try:
+        result = subprocess.run([
+            sys.executable, "-m", "pytest", "-q", "-p", "conftest",
+            "-o", f"tmp_path_retention_policy={retention}", str(probe),
+        ], cwd=tmp_path, env=environment, capture_output=True, text=True, timeout=120)
+        assert result.returncode == (1 if failing else 0), result.stdout + result.stderr
+        if failing:
+            assert "2 failed" in result.stdout and "fixture failure retained" in result.stdout
+        observed = [json.loads(line) for line in records.read_text(encoding="utf-8").splitlines()]
+        assert len(observed) == 2
+        assert all(_has_exited(row["info"]["pid"]) for row in observed), (
+            "pytest retained a source or copied-package runtime after its test namespace ended"
+        )
+    finally:
+        # Keep the failing baseline bounded too; these exact locators came from
+        # the two child fixtures above, never from a developer's runtime.
+        if records.exists() and not observed:
+            observed = [json.loads(line) for line in records.read_text(encoding="utf-8").splitlines()]
+        for row in observed:
+            effect_runtime._stop_effect_runtime(
+                Path(row["path"]), row["info"], timeout=5, wait_for_process_exit=True,
+            )
+
+
+def test_cleanup_preserves_foreign_namespace_and_rejects_wrong_pid(tmp_path, monkeypatch):
+    foreign = tmp_path / "foreign"
+    owned = tmp_path / "owned"
+    foreign.mkdir()
+    owned.mkdir()
+    with monkeypatch.context() as isolated:
+        isolated.setattr(effect_runtime.tempfile, "tempdir", str(foreign))
+        effect_runtime.effect_runtime_result("runtime.ping", {})
+        locator = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint())
+        info = json.loads(locator.read_text(encoding="utf-8"))
+        try:
+            # A directory link must not turn an unrelated namespace into ours.
+            (owned / locator.parent.name).symlink_to(locator.parent, target_is_directory=True)
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
+            (owned / locator.parent.name).unlink()
+            forged = owned / locator.parent.name / locator.name
+            forged.parent.mkdir()
+            forged.write_text(json.dumps({**info, "pid": os.getpid()}), encoding="utf-8")
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
+            forged.write_text("not JSON", encoding="utf-8")
+            stop_disposable_effect_runtimes(owned)
+        finally:
+            assert effect_runtime.restart_effect_runtime(wait_for_exit=True)["stopped"]
+
+
+def test_cleanup_does_not_hide_an_unfinished_shutdown(tmp_path, monkeypatch):
+    fingerprint = "a" * 64
+    locator = tmp_path / "loopx-effect-runtime-fixture" / f"runtime-{fingerprint[:16]}.json"
+    locator.parent.mkdir()
+    locator.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(effect_runtime, "_read_info", lambda *_a, **_k: {
+        "pid": 123, "fingerprint": fingerprint,
+    })
+    monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *_a, **_k: {
+        "result": {"ready": True, "pid": 123, "fingerprint": fingerprint},
+    })
+    monkeypatch.setattr(effect_runtime, "_stop_effect_runtime", lambda *_a, **_k: {"stopped": False})
+    with pytest.raises(AssertionError, match="Disposable runtime did not exit"):
+        stop_disposable_effect_runtimes(tmp_path)

From 60ee7b36aa761c7515a6aab3f6b35bf6f5ce1e62 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 08:58:56 +0800
Subject: [PATCH 09/10] test(ci): keep malformed locator bytes outside teardown
 authority

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/managed_effect_runtime_fixture.py      | 2 +-
 tests/test_managed_effect_runtime_fixture.py | 3 +++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/tests/managed_effect_runtime_fixture.py b/tests/managed_effect_runtime_fixture.py
index 16bd0bcc46..bdd9b5aeb9 100644
--- a/tests/managed_effect_runtime_fixture.py
+++ b/tests/managed_effect_runtime_fixture.py
@@ -20,7 +20,7 @@ def capture_disposable_effect_runtimes(root: Path) -> list[tuple[Path, str]]:
             continue
         try:
             captured.append((locator, locator.read_text(encoding="utf-8")))
-        except OSError:
+        except (OSError, UnicodeError):
             continue
     return captured
 
diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py
index ccec08ae9f..7ef8c10023 100644
--- a/tests/test_managed_effect_runtime_fixture.py
+++ b/tests/test_managed_effect_runtime_fixture.py
@@ -117,6 +117,9 @@ def test_cleanup_preserves_foreign_namespace_and_rejects_wrong_pid(tmp_path, mon
             assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
             forged.write_text("not JSON", encoding="utf-8")
             stop_disposable_effect_runtimes(owned)
+            forged.write_bytes(b"\xff")
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
         finally:
             assert effect_runtime.restart_effect_runtime(wait_for_exit=True)["stopped"]
 

From 8e23ee3ae20a7218db3663791035a6620aab83dd Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 17 Oct 2026 11:37:25 +0800
Subject: [PATCH 10/10] test: repair refusal and subprocess qualification
 oracles

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_task_lease_cli_native.py             | 29 ++++++++++++++++---
 tests/test_projection_envelope.py             | 10 ++++++-
 tests/test_zcode_host_diagnostics.py          |  9 +++---
 3 files changed, 39 insertions(+), 9 deletions(-)

diff --git a/tests/control_plane/test_task_lease_cli_native.py b/tests/control_plane/test_task_lease_cli_native.py
index abe849eb25..d9be522cb4 100644
--- a/tests/control_plane/test_task_lease_cli_native.py
+++ b/tests/control_plane/test_task_lease_cli_native.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import deepcopy
 from pathlib import Path
 
 from loopx.cli import build_parser
@@ -13,7 +14,7 @@ def _run_acquire(monkeypatch, result: dict) -> dict:
     monkeypatch.setattr(
         task_lease_cli,
         "execute_native_task_lease_acquire",
-        lambda **kwargs: result,
+        lambda **kwargs: deepcopy(result),
     )
     args = build_parser().parse_args(
         [
@@ -77,6 +78,19 @@ def raise_error(**_kwargs):
     return captured[0]
 
 
+def _assert_refusal(payload: dict, expected: dict) -> None:
+    inspection = payload["inspection"]
+    assert set(inspection) == {"cli_args", "reason", "execution_authority_granted"}
+    assert inspection["cli_args"] == [
+        f"--registry={Path('/tmp/registry.json')}", "--format", "json",
+        "task-lease", "inspect", "--goal-id=cli-native-goal",
+        "--todo-id=todo_cli_native",
+    ]
+    assert isinstance(inspection["reason"], str) and inspection["reason"].strip()
+    assert inspection["execution_authority_granted"] is False
+    assert {key: value for key, value in payload.items() if key != "inspection"} == expected
+
+
 def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> None:
     """A typed payload that carries its own schema_version cannot clobber the envelope's."""
 
@@ -88,14 +102,18 @@ def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> No
         "canonical Todo authority is unavailable",
         code="local_authority_todo_list_unavailable",
         payload={
+            "ok": True,
             "schema_version": "loopx_local_coordination_todo_list_result_v0",
+            "action": "release",
+            "error": "payload must not replace the original refusal",
+            "error_code": "payload_error",
             "status": "missing",
             "source_authority": "file_v0",
             "decision_read_from_provider": True,
             "legacy_fallback_used": False,
         },
     )
-    assert _run_acquire_raising(monkeypatch, error) == {
+    _assert_refusal(_run_acquire_raising(monkeypatch, error), {
         "ok": False,
         "schema_version": "task_lease_v0",
         "action": "acquire",
@@ -105,7 +123,7 @@ def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> No
         "source_authority": "file_v0",
         "decision_read_from_provider": True,
         "legacy_fallback_used": False,
-    }
+    })
 
 
 def test_cli_acquire_passes_through_native_success(monkeypatch) -> None:
@@ -177,4 +195,7 @@ def test_cli_acquire_passes_through_native_typed_failure(monkeypatch) -> None:
             },
         },
     }
-    assert _run_acquire(monkeypatch, result) == result
+    expected = deepcopy(result)
+    actual = _run_acquire(monkeypatch, result)
+    assert result == expected
+    _assert_refusal(actual, expected)
diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py
index 62a309f987..6d474a3769 100644
--- a/tests/test_projection_envelope.py
+++ b/tests/test_projection_envelope.py
@@ -584,12 +584,20 @@ def test_receipt_case_collection_does_not_depend_on_the_collecting_process_clock
         "test_global_risks_discloses_unreadable_host_poll_receipts_and_recovers"
     )
     collections = []
-    for clock in ("2026-01-01T00:00:00Z", "2026-01-01T00:00:01Z"):
+    for index, clock in enumerate(("2026-01-01T00:00:00Z", "2026-01-01T00:00:01Z")):
+        # A collector must not observe routes created by its still-running
+        # parent test as writes to the collector's own default namespace.
+        home = tmp_path / f"collector-home-{index}"
+        temporary = tmp_path / f"collector-tmp-{index}"
+        home.mkdir()
+        temporary.mkdir()
         result = subprocess.run(
             [sys.executable, "-m", "pytest", target, "--collect-only", "-q",
              "-o", "addopts=", "-p", "receipt_collection_clock"],
             cwd=Path(__file__).resolve().parents[1],
             env={**os.environ, "LOOPX_RECEIPT_COLLECTION_CLOCK": clock,
+                 "HOME": str(home), "USERPROFILE": str(home),
+                 "TMPDIR": str(temporary), "TEMP": str(temporary), "TMP": str(temporary),
                  "PYTHONPATH": os.pathsep.join((str(tmp_path), os.environ.get("PYTHONPATH", "")))},
             capture_output=True, text=True, timeout=60,
         )
diff --git a/tests/test_zcode_host_diagnostics.py b/tests/test_zcode_host_diagnostics.py
index 71b92f2f80..1ec43cb9fe 100644
--- a/tests/test_zcode_host_diagnostics.py
+++ b/tests/test_zcode_host_diagnostics.py
@@ -338,13 +338,14 @@ def test_probe_deadline_survives_descendant_inheriting_stdout(monkeypatch, capsy
 
 @pytest.mark.parametrize("mode,expected,child_startup_delay", [
     ("timeout_default", "timeout", 0), ("timeout", "timeout", 0),
-    ("output_limit", "output_limit", 0), ("exit_pipe", "timeout", 0),
+    ("output_limit", "output_limit", 0), ("output_limit", "output_limit", .6),
+    ("exit_pipe", "timeout", 0),
     ("exit_closed", "observed", 0), ("exit_closed", "observed", .6),
 ])
 def test_probe_reclaims_persistent_descendants_before_return(tmp_path, monkeypatch, mode, expected, child_startup_delay):
-    # Successful closure uses the shipped deadline so interpreter and child
-    # startup are not constrained by a deadline for deliberately blocked probes.
-    if mode not in {"timeout_default", "exit_closed"}:
+    # Output and successful closure use the shipped deadline so two interpreter
+    # startups are not constrained by the synthetic timeout-case deadline.
+    if mode in {"timeout", "exit_pipe"}:
         monkeypatch.setattr(diagnostics, "PROBE_TIMEOUT_SECONDS", .5)
     # Publishing the ready PID must imply that the first pulse already exists.
     child_script = f"import time; time.sleep({child_startup_delay!r})\nimport pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.with_suffix('.pulse').write_text('0'); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"