diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 1975e0bab3..4216452681 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -184,6 +184,7 @@ touch, mention it in the PR instead of fixing it there.
 | --- | --- | --- | --- |
 | `python-tests.yml` | every PR | yes (`merge-gate`) | lint, mypy, sharded pytest, TypeScript core and coverage, minimum Node.js, Windows PowerShell, dashboard presentation |
 | `dco.yml` | every PR | yes (`Sign-off`) | `Signed-off-by` on contribution commits; verified GitHub-generated two-parent merges are exempt |
+| `ci-maintenance.yml` | closed or updated PR (`pull_request_target`, trusted base code only) | no | cancels obsolete Python qualification runs after rechecking the PR and source branch |
 | `dependency-review.yml` | every PR | no | dependency changes introduced by the PR |
 | `postgresql-integration.yml` | control-plane or npm lockfile paths | no | PostgreSQL authority store and service on a temporary instance |
 | `package-smoke.yml` | extension package paths | no | extension packages install, entrypoints, and example schemas |
diff --git a/docs/product/release-readiness.md b/docs/product/release-readiness.md
index e3fc2afe7d..2c4ae6d07a 100644
--- a/docs/product/release-readiness.md
+++ b/docs/product/release-readiness.md
@@ -864,6 +864,24 @@ path, and canary route rather than as a user-facing release baseline.
   SQLite qualification remain unqualified. No paired-version benchmark uplift
   or isolated Astra causal effect is claimed.
 
+- `v1.4.1` on 2026-10-15: workspace visibility, semantic replan and recovery
+  hardening at frozen commit `4f27801c4e26fd51246d352eada72c98f8e7e657`.
+  The [published release](https://github.com/loopx-project/loopx/releases/tag/v1.4.1)
+  records nineteen bilingual operating entries and concrete community credits.
+  GitHub/PyPI packages, four desktop installers, the signed macOS update feed,
+  `desktop-stable`, source `stable`, the website download path and the personal
+  illustrated guide were independently read back. An actual PyPI
+  1.4.0-to-1.4.1 upgrade preserved existing work; package hashes and source bytes
+  matched the immutable product tag. Reviewed publication-fixture recovery did
+  not move that tag. At the final source, 515 public smokes and 2,213 affected
+  Python checks passed, with five explicit native-environment skips; this was
+  not a fresh full-tree Python sweep. Doubao Evolving/high completed the fixed
+  22-scenario, two-repeat portfolio and six contrasts with zero failures or
+  skips. Mac artifacts retain ad-hoc signing without Apple notarization;
+  Windows artifacts were verified without native client execution. No paired
+  benchmark, sustained scientific continuation or isolated model causal uplift
+  is claimed.
+
 When a new public release is promoted, add it here only after the matching tag,
 release note, stable ref, update path, and focused release canary agree.
 
diff --git a/docs/reference/contracts/interface-budget-contract.md b/docs/reference/contracts/interface-budget-contract.md
index 3c8772e2b7..d29073c8a4 100644
--- a/docs/reference/contracts/interface-budget-contract.md
+++ b/docs/reference/contracts/interface-budget-contract.md
@@ -80,6 +80,22 @@ Do not abbreviate admission, continuation, scope, safety or settlement instructi
 solely to fit a historical character count, or copy dynamic quota decisions into
 the static prompt.
 
+The same two-scope Codex App brief fixture measures 4,284 normalized body
+characters on v1.4.1 and 4,447 with the shared stop/readback rule: 163 additional
+characters preserve the exact heartbeat target, one attempt, verified stopping,
+unconfirmed-result handling and prohibition on work/spend/retry fallback. The
+brief presentation allowance becomes 4,600, leaving 153 characters of headroom.
+The two original scope paragraphs and current-Turn settlement instructions stay
+intact. This is a regression-budget adjustment, not a model context limit or an
+execution allowance; all other mode limits, field counts, quota and authority
+remain unchanged. The original 4,300-character failure is retained in validation.
+
+同一 Codex App brief 双范围 fixture 在 v1.4.1 为 4,284 个归一化正文字符，加入共享停止/
+读回规则后为 4,447，增加 163；保留准确停止对象、单次尝试、停止验证、未确认结果处理
+以及禁止 work/spend/retry fallback。brief 展示预算调至 4,600，保留 153 字符余量，
+不删除两段原范围与当前 Turn 结算规则。其余模式、字段数、执行配额和权限不变；
+原 4,300 上限失败保留，不能将新展示预算称为模型上下文或执行额度提升。
+
 The earlier readable-peer qualification raised the thin body ceiling from 2,500
 to 3,000; those historical measurements follow.
 On the same real CLI fixture with a host scope and three runtime capabilities,
diff --git a/examples/blocker-push-runtime-smoke.py b/examples/blocker-push-runtime-smoke.py
index f847f6af4a..a4372be45a 100644
--- a/examples/blocker-push-runtime-smoke.py
+++ b/examples/blocker-push-runtime-smoke.py
@@ -217,7 +217,8 @@ def main() -> int:
         # The bootstrap rule is shared from heartbeat.rules after #4201; assert the
         # current compact sentence instead of the retired per-shell phrasing.
         assert "reuse the value on retries" in compact_prompt, prompt
-        assert "guard; 2 stalls->replan" in compact_prompt, prompt
+        assert "re-enter guard." in compact_prompt, prompt
+        assert "2 stalls->replan" not in compact_prompt, prompt
         assert "Exact monitor settlement=no refresh/spend" in compact_prompt, prompt
         assert "Admitted work: settlement_plan even if artifacts unchanged" in compact_prompt, prompt
         # The shipped writeback sentence moved to the mixed-language form in the
diff --git a/examples/control_plane/control-plane-integrated-canary-smoke.py b/examples/control_plane/control-plane-integrated-canary-smoke.py
index 63049aa45b..a784e4f554 100644
--- a/examples/control_plane/control-plane-integrated-canary-smoke.py
+++ b/examples/control_plane/control-plane-integrated-canary-smoke.py
@@ -31,7 +31,7 @@
 CANARY_TODO_ID = "todo_integrated_canary"
 CANARY_TODO_TITLE = "Design bounded status/quota/review-packet/Todo/read-path canary"
 SUCCESSOR_TODO_TITLE = "Continue integrated Markdown successor routing canary"
-MARKDOWN_CONTINUATION_TITLE = "Continue same-agent markdown continuation through status and quota."
+MARKDOWN_CONTINUATION_TITLE = "[P1] Continue same-agent markdown continuation through status and quota."
 MONITOR_TODO_ID = "todo_integrated_due_monitor"
 MONITOR_TARGET_KEY = "integrated-due-monitor-watch"
 VALIDATED_PROGRESS_CLASSIFICATION = "integrated_canary_validated_progress"
diff --git a/examples/control_plane/quota-no-git-status-smoke.py b/examples/control_plane/quota-no-git-status-smoke.py
index 6cf51ef55d..6ff81a01a5 100644
--- a/examples/control_plane/quota-no-git-status-smoke.py
+++ b/examples/control_plane/quota-no-git-status-smoke.py
@@ -118,7 +118,7 @@ def main() -> int:
         )
         assert quota.get("status") != "quota_collection_failed", quota
         assert quota["should_run"] is True, quota
-        assert quota["recommended_action"] == "Solve the no-git quota fixture.", quota
+        assert quota["recommended_action"] == "[P1] Solve the no-git quota fixture.", quota
     return 0
 
 
diff --git a/examples/control_plane/runtime-handoff-status-read-path-smoke.py b/examples/control_plane/runtime-handoff-status-read-path-smoke.py
index 7ef879820c..8f57342ea3 100644
--- a/examples/control_plane/runtime-handoff-status-read-path-smoke.py
+++ b/examples/control_plane/runtime-handoff-status-read-path-smoke.py
@@ -206,9 +206,11 @@ def assert_status_read_path(
         runtime_root=runtime_root,
     )
     item = status_item(payload)
-    assert item["status"] == POST_HANDOFF_CLASSIFICATION, item
+    assert item["status"] == "active_state_agent_todo", item
+    assert item["source"] == "active_state", item
+    assert item.get("stale_latest_run_warning") is None, item
     assert item["waiting_on"] == "codex", item
-    assert item["recommended_action"] == POST_HANDOFF_ACTION, item
+    assert item["recommended_action"] == f"[P1] {TODO_TITLE}", item
     assert item["project_asset"]["active_state_next_action"] == GOAL_NEXT_ACTION, item
     assert item["project_asset"]["latest_validation"]["classification"] == POST_HANDOFF_CLASSIFICATION, item
     assert item["project_asset"]["latest_validation"]["summary"] == POST_HANDOFF_ACTION, item
diff --git a/examples/control_plane/todo-cli-smoke.py b/examples/control_plane/todo-cli-smoke.py
index 693129e18f..822b5038ab 100644
--- a/examples/control_plane/todo-cli-smoke.py
+++ b/examples/control_plane/todo-cli-smoke.py
@@ -22,13 +22,13 @@
 
 
 GOAL_ID = "todo-cli-goal"
-USER_TODO = "Review the owner decision checklist before approving delivery."
-SCOPED_USER_GATE = "Choose the side-agent publishing channel before posting externally."
-AGENT_TODO = "Summarize the read-only evidence after the user checklist is done."
+USER_TODO = "[P1] Review the owner decision checklist before approving delivery."
+SCOPED_USER_GATE = "[P1] Choose the side-agent publishing channel before posting externally."
+AGENT_TODO = "[P1] Summarize the read-only evidence after the user checklist is done."
 AGENT_TODO_NOTE = "Start from the public-safe evidence summary."
 UPDATED_AGENT_TODO_NOTE = "Preserve the validated summary as the handoff context."
-UPDATED_AGENT_TODO = "Publish the compact evidence summary after validation passes."
-MONITOR_TODO = "Monitor the release-note draft PR until the next scheduled check."
+UPDATED_AGENT_TODO = "[P1] Publish the compact evidence summary after validation passes."
+MONITOR_TODO = "[P1] Monitor the release-note draft PR until the next scheduled check."
 MONITOR_DUE_AT = "2026-01-02T00:00:00+00:00"
 MONITOR_NEXT_DUE_AT = "2026-01-03T00:00:00+00:00"
 MONITOR_EXPIRES_AT = "2026-01-04T00:00:00+00:00"
@@ -299,7 +299,7 @@ def main() -> int:
         assert "has no coordination.registered_agents" in legacy_claim["error"], legacy_claim
         assert "Register this peer identity first" in legacy_claim["error"], legacy_claim
         assert "--registered-agent codex-main-control" in legacy_claim["error"], legacy_claim
-        wrapped_agent_todo = "- [ ] Summarize the read-only evidence after the user\n  checklist is done."
+        wrapped_agent_todo = "- [ ] [P1] Summarize the read-only evidence after the user\n  checklist is done."
         state_file.write_text(
             state_file.read_text(encoding="utf-8").replace(f"- [ ] {AGENT_TODO}", wrapped_agent_todo),
             encoding="utf-8",
@@ -351,8 +351,8 @@ def main() -> int:
             }
         ], metadata_payload
         after_metadata = state_file.read_text(encoding="utf-8")
-        assert after_metadata.count("- [ ] Summarize the read-only evidence after the user") == 1, after_metadata
-        agent_block_start = after_metadata.index("- [ ] Summarize the read-only evidence after the user")
+        assert after_metadata.count("- [ ] [P1] Summarize the read-only evidence after the user") == 1, after_metadata
+        agent_block_start = after_metadata.index("- [ ] [P1] Summarize the read-only evidence after the user")
         agent_metadata_start = after_metadata.index("<!-- loopx:todo", agent_block_start)
         assert after_metadata.index("checklist is done.", agent_block_start) < agent_metadata_start, after_metadata
         assert "status=open task_class=advancement_task action_kind=run_eval" in after_metadata
diff --git a/examples/control_plane/todo-concurrent-write-lock-smoke.py b/examples/control_plane/todo-concurrent-write-lock-smoke.py
index ee88eb10eb..dd63516df6 100644
--- a/examples/control_plane/todo-concurrent-write-lock-smoke.py
+++ b/examples/control_plane/todo-concurrent-write-lock-smoke.py
@@ -20,9 +20,9 @@
 
 
 GOAL_ID = "todo-concurrent-write-lock-goal"
-CHILD_ADD_TODO = "Child add should preserve the parent todo written under lock."
+CHILD_ADD_TODO = "[P1] Child add should preserve the parent todo written under lock."
 PARENT_ADD_TODO = "Parent todo written while child add waits on the lock."
-UPDATE_TARGET_TODO = "Update target todo should keep unrelated concurrent additions."
+UPDATE_TARGET_TODO = "[P1] Update target todo should keep unrelated concurrent additions."
 PARENT_UPDATE_TODO = "Parent todo written while child update waits on the lock."
 
 
diff --git a/examples/control_plane/todo-lifecycle-cli-smoke.py b/examples/control_plane/todo-lifecycle-cli-smoke.py
index 023e169551..7903069a89 100644
--- a/examples/control_plane/todo-lifecycle-cli-smoke.py
+++ b/examples/control_plane/todo-lifecycle-cli-smoke.py
@@ -286,11 +286,11 @@ def assert_nonblocking_review_continuation() -> None:
             "--evidence",
             "feature PR is validated and ready for review",
             "--next-user-todo",
-            "Review the validated experiment feature PR.",
+            "[P1] Review the validated experiment feature PR.",
             "--next-user-task-class",
             "user_action",
             "--next-agent-todo",
-            "Continue the next independent experiment feature.",
+            "[P1] Continue the next independent experiment feature.",
             "--next-claimed-by",
             "codex-main-control",
             "--next-action-kind",
@@ -301,7 +301,7 @@ def assert_nonblocking_review_continuation() -> None:
         review = next(
             item
             for item in fields["user_todos"]["items"]
-            if item["text"] == "Review the validated experiment feature PR."
+            if item["text"] == "[P1] Review the validated experiment feature PR."
         )
         assert review["task_class"] == "user_action", review
         assert review["bound_agent"] == "codex-main-control", review
@@ -311,7 +311,7 @@ def assert_nonblocking_review_continuation() -> None:
         continuation = next(
             item
             for item in fields["agent_todos"]["items"]
-            if item["text"] == "Continue the next independent experiment feature."
+            if item["text"] == "[P1] Continue the next independent experiment feature."
         )
         assert continuation["status"] == "open", continuation
         assert continuation["claimed_by"] == "codex-main-control", continuation
diff --git a/examples/install-local-smoke.py b/examples/install-local-smoke.py
index 34699eb2ef..4c811e1d89 100644
--- a/examples/install-local-smoke.py
+++ b/examples/install-local-smoke.py
@@ -833,7 +833,8 @@ def main() -> int:
         )
         assert "```sh\nLOOPX_TURN=<current_time_iso>\n" in payload["task_body"], payload
         assert "not a command-prefix assignment" in payload["task_body"], payload
-        assert "guard; 2 stalls->replan" in payload["task_body"], payload
+        assert "re-enter guard." in payload["task_body"], payload
+        assert "2 stalls->replan" not in payload["task_body"], payload
         assert "Exact monitor settlement=no refresh/spend" in payload["task_body"], payload
         assert "Admitted work: settlement_plan even if artifacts unchanged" in payload["task_body"], payload
         assert "no-change=surface_only/no spend" not in payload["task_body"], payload
diff --git a/examples/project/global-manager-command-cli-smoke.py b/examples/project/global-manager-command-cli-smoke.py
index 38e07d7ec8..4703554556 100644
--- a/examples/project/global-manager-command-cli-smoke.py
+++ b/examples/project/global-manager-command-cli-smoke.py
@@ -320,9 +320,9 @@ def assert_risks_payload(payload: dict[str, object]) -> None:
     ], payload
 
     stale_rows = payload["groups"]["stale_runs"]
-    assert stale_rows, payload
-    assert stale_rows[0]["kind"] == "stale_latest_run_projection", stale_rows
-    assert stale_rows[0]["goal_id"] == GOAL_ID, stale_rows
+    # Current Todo reads reconcile this ordinary stale progress projection.
+    # Historical failures and unavailable sources retain their independent risks.
+    assert stale_rows == [], payload
     assert any(
         warning["reason_code"] == "global_registry_unavailable"
         for warning in payload["source_warnings"]
diff --git a/examples/project/project-agent-adoption-smoke.py b/examples/project/project-agent-adoption-smoke.py
index c5316418e2..bccc2230c1 100644
--- a/examples/project/project-agent-adoption-smoke.py
+++ b/examples/project/project-agent-adoption-smoke.py
@@ -21,7 +21,7 @@
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 GOAL_ID = "adoption-main-control"
-USER_TODO = "Review the P0 owner-risk checklist before approving delivery."
+USER_TODO = "[P1] Review the P0 owner-risk checklist before approving delivery."
 AGENT_TODO = "Run the read-only map dry-run after the operator approval is recorded."
 APPROVED_COMMAND = f"loopx read-only-map --goal-id {GOAL_ID} --dry-run"
 QUOTA_EXECUTION_CONTEXT = (
diff --git a/loopx/control_plane/collaboration/operation_wake.py b/loopx/control_plane/collaboration/operation_wake.py
index cd04e03b7b..af5cb2bd57 100644
--- a/loopx/control_plane/collaboration/operation_wake.py
+++ b/loopx/control_plane/collaboration/operation_wake.py
@@ -140,7 +140,7 @@ def dispatch_confirmed_operation_wake(
         if parameters["goal_id"] != configuration["goal_id"]:
             return {**base, "state": "out_of_scope"}
         config_path = _configuration_path(Path(configuration["project"]), configuration["execution_config"])
-        raw_config = json.loads(config_path.read_text())
+        raw_config = json.loads(config_path.read_text(encoding="utf-8"))
         if raw_config.get("schema_version") != "loopx_local_delegation_v0":
             raise ValueError("unsupported operation launch configuration")
         bindings = raw_config.get("bindings")
diff --git a/loopx/control_plane/heartbeat/budget.py b/loopx/control_plane/heartbeat/budget.py
index 3e9c91ebbd..3e10c860d3 100644
--- a/loopx/control_plane/heartbeat/budget.py
+++ b/loopx/control_plane/heartbeat/budget.py
@@ -12,7 +12,7 @@
 INTERFACE_BUDGET_CHARS = {
     "full": 12_000,
     "compact": 6_700,
-    "brief": 4_300,
+    "brief": 4_600,
     "thin": 3_200,
     "visible_goal": 4_000,
 }
diff --git a/loopx/control_plane/turn_driver/dsh_operation_host.py b/loopx/control_plane/turn_driver/dsh_operation_host.py
index 5f3c3b0d03..f269cf2540 100644
--- a/loopx/control_plane/turn_driver/dsh_operation_host.py
+++ b/loopx/control_plane/turn_driver/dsh_operation_host.py
@@ -58,7 +58,7 @@ def _sdk_runtime_pin(runtime_bin: Path | None = None) -> dict[str, Any]:
     if runtime_bin is not None:
         entry = runtime_bin.resolve(strict=True)
         package = entry.parent.parent
-        manifest = json.loads((package / "package.json").read_text())
+        manifest = json.loads((package / "package.json").read_text(encoding="utf-8"))
         modules = package.parent.parent
         if (
             sdk.version != "0.2.1a3"
diff --git a/loopx/presentation/codex_attached_api.py b/loopx/presentation/codex_attached_api.py
index 0fffbf536c..223036f810 100644
--- a/loopx/presentation/codex_attached_api.py
+++ b/loopx/presentation/codex_attached_api.py
@@ -153,7 +153,7 @@ def _codex_attached_read(self) -> None:
                         "No delivery receipt is recorded for this input.", status=404
                     )
                     return
-                receipt = json.loads(receipt_path.read_text())
+                receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
                 if receipt.get("binding_ref") != binding["binding_ref"]:
                     raise ValueError("Delivery belongs to an older binding")
                 http._send_json(receipt_readback(receipt))
@@ -300,7 +300,7 @@ def _codex_attached_send(self) -> None:
             fingerprint = input_fingerprint(body["binding_ref"], body["text"])
             with exclusive_file_lock(path):
                 if path.exists():
-                    receipt = json.loads(path.read_text())
+                    receipt = json.loads(path.read_text(encoding="utf-8"))
                     if receipt["fingerprint"] != fingerprint:
                         raise ValueError("Operation id belongs to a different input")
                     http._send_json(receipt_readback(receipt))
diff --git a/tests/architecture/test_control_plane_import_boundaries.py b/tests/architecture/test_control_plane_import_boundaries.py
index 4734e5f2bb..4b51d689a9 100644
--- a/tests/architecture/test_control_plane_import_boundaries.py
+++ b/tests/architecture/test_control_plane_import_boundaries.py
@@ -74,7 +74,7 @@ def _module_name(path: Path) -> str:
 def _resolved_imports(path: Path) -> set[str]:
     module_name = _module_name(path)
     package_name = module_name if path.name == "__init__.py" else module_name.rpartition(".")[0]
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     imports: set[str] = set()
     for node in ast.walk(tree):
         if isinstance(node, ast.Import):
@@ -92,7 +92,7 @@ def _resolved_from_imports(path: Path) -> dict[str, set[str]]:
     if path.is_relative_to(PACKAGE_ROOT):
         module_name = _module_name(path)
         package_name = module_name if path.name == "__init__.py" else module_name.rpartition(".")[0]
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     imports: dict[str, set[str]] = {}
     for node in ast.walk(tree):
         if not isinstance(node, ast.ImportFrom):
@@ -107,7 +107,7 @@ def _resolved_from_imports(path: Path) -> dict[str, set[str]]:
 
 
 def _top_level_imported_names(path: Path) -> set[str]:
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     return {
         alias.asname or alias.name
         for node in tree.body
@@ -117,7 +117,7 @@ def _top_level_imported_names(path: Path) -> set[str]:
 
 
 def _top_level_function_names(path: Path) -> set[str]:
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     return {
         node.name
         for node in tree.body
@@ -128,7 +128,7 @@ def _top_level_function_names(path: Path) -> set[str]:
 def _top_level_import_bindings(path: Path) -> dict[str, str]:
     module_name = _module_name(path)
     package_name = module_name.rpartition(".")[0]
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     bindings: dict[str, str] = {}
     for node in tree.body:
         if isinstance(node, ast.Import):
@@ -144,7 +144,7 @@ def _top_level_import_bindings(path: Path) -> dict[str, str]:
 
 
 def _public_import_only_bindings(path: Path) -> dict[str, str]:
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = ast.parse(path.read_bytes(), filename=str(path))
     loaded_names = {
         node.id
         for node in ast.walk(tree)
@@ -161,7 +161,7 @@ def _public_contract_evidence() -> dict[tuple[str, str], set[str]]:
     evidence: dict[tuple[str, str], set[str]] = {}
     for root in PUBLIC_CONTRACT_ROOTS:
         for path in root.rglob("*.py"):
-            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+            tree = ast.parse(path.read_bytes(), filename=str(path))
             # Aliases such as ``from loopx import status as status_module`` are a
             # real consumer of a facade export, reached by attribute access
             # rather than a from-import. Without this map the audit would report
diff --git a/tests/architecture/test_source_session_registry_denial.py b/tests/architecture/test_source_session_registry_denial.py
index 8fe8a05098..adaa637b9f 100644
--- a/tests/architecture/test_source_session_registry_denial.py
+++ b/tests/architecture/test_source_session_registry_denial.py
@@ -8,6 +8,7 @@
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 DIRECT_LOADER_ALLOWLIST = {
+    "loopx/runtime.py",
     "loopx/authority.py",
     "loopx/control_plane/coordination/new_goal_storage_transport.py",
     "loopx/chat_server.py",
@@ -49,6 +50,7 @@
 # Storage-location readers grant no runtime action. Exceptions name the exact
 # functions; another direct loader in either module must still fail.
 METADATA_READER_FUNCTIONS = {
+    "loopx/runtime.py": {"_archive_source_goal_refs"},
     "loopx/capabilities/native_chat/project_context.py": {"coordination_runtime_root"},
     "loopx/control_plane/goals/source_session_recreation.py": {"_canonical_runtime_root"},
 }
@@ -135,7 +137,7 @@ def test_direct_project_registry_loaders_have_source_session_denial() -> None:
         "loopx/presentation/request_control_api.py": {"_request_control_source"},
         "loopx/chat_server.py": {"serve_chat"},
         "loopx/presentation/goal_direction_api.py": {"_direction_route"},
-        "loopx/chat_status_api.py": {"_status"},
+        "loopx/chat_status_api.py": {"_discovery_registry"},
         "loopx/control_plane/goals/direction_runtime.py": {"_source"},
         "loopx/control_plane/coordination/new_goal_storage_transport.py": {"initialize_goal_storage_target"},
         "loopx/presentation/goal_amendment_api.py": {"_amendment_inspect"},
@@ -156,8 +158,8 @@ def test_direct_project_registry_loaders_have_source_session_denial() -> None:
     assert "source_session_goal_lifetime" in attached_owner
 
 
-def test_metadata_exception_rejects_another_direct_loader() -> None:
-    relative = "loopx/control_plane/goals/source_session_recreation.py"
+@pytest.mark.parametrize("relative", tuple(METADATA_READER_FUNCTIONS))
+def test_metadata_exception_rejects_another_direct_loader(relative: str) -> None:
     source = (REPO_ROOT / relative).read_text(encoding="utf-8")
     tree = ast.parse(source + "\n\ndef execute_goal(registry_path):\n    return load_project_registry(registry_path)\n")
     with pytest.raises(AssertionError):
diff --git a/tests/conftest.py b/tests/conftest.py
index 5c6ff5d628..da0bfb0b83 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -37,6 +37,22 @@
 _REAL_DEFAULT_ROUTES = (paths.DEFAULT_RUNTIME_ROOT, paths.LEGACY_RUNTIME_ROOT)
 
 
+@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
+    # getuid() and the process temp directory. A restart in one xdist worker
+    # must not retire another worker's real runtime.
+    temporary = tmp_path_factory.mktemp("effect-runtime")
+    with pytest.MonkeyPatch.context() as patch:
+        for variable in ("TMPDIR", "TEMP", "TMP"):
+            patch.setenv(variable, str(temporary))
+        patch.setattr(tempfile, "tempdir", str(temporary))
+        yield
+        from loopx.control_plane.effect_runtime import restart_effect_runtime
+
+        restart_effect_runtime(wait_for_exit=True)
+
+
 @pytest.fixture
 def isolated_effect_runtime(tmp_path, monkeypatch):
     temporary = tmp_path / "temporary"
diff --git a/tests/control_plane/test_advancement_frontier_classification.py b/tests/control_plane/test_advancement_frontier_classification.py
index 98f0053eb2..5f0657538c 100644
--- a/tests/control_plane/test_advancement_frontier_classification.py
+++ b/tests/control_plane/test_advancement_frontier_classification.py
@@ -135,7 +135,7 @@ def test_real_quota_frontier_reads_canonical_claims(tmp_path, monkeypatch, provi
         "current_agent_claimed_advancement_count": 1, "unclaimed_advancement_count": 1,
         "other_agent_claimed_advancement_count": 1,
     }
-    assert packet["selected_todo"]["text"] == "Own direction"
+    assert packet["selected_todo"]["text"] == "[P1] Own direction"
     assert state.exists()
 
 
diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py
index 5fc6cd690a..489877cfe4 100644
--- a/tests/control_plane/test_canonical_todo_writer_isolation.py
+++ b/tests/control_plane/test_canonical_todo_writer_isolation.py
@@ -171,7 +171,7 @@ def history():
         assert transactions[-1]["receipts"][0]["write_class"] == "todo_add"
         assert transactions[-1]["receipts"][0]["seq"] == 2
         assert {todo["text"] for todo in recovered["head"]["todos"]} == {
-            "Already delivered source work", "Pending source work"}
+            "[P1] Already delivered source work", "[P1] Pending source work"}
         assert json.loads((directory / "drain-cursor.json").read_text())["last_seq"] == 2
         assert sorted(f.name for f in directory.iterdir()) == ["drain-cursor.json"]
         assert w.drain()["outcome"] == "nothing_pending"
diff --git a/tests/control_plane/test_canonical_update_dispatch.py b/tests/control_plane/test_canonical_update_dispatch.py
index 050f56b62f..2a2c778d95 100644
--- a/tests/control_plane/test_canonical_update_dispatch.py
+++ b/tests/control_plane/test_canonical_update_dispatch.py
@@ -197,7 +197,7 @@ def test_empty_promoted_edits_reject_without_the_source_writer(update_workspace,
     first = json.loads(result.stdout)
     current = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
     assert current["todos"][0]["note"] == "Keep the original note"
-    assert current["todos"][0]["text"] == "Updated authoritative record"
+    assert current["todos"][0]["text"] == "[P1] Updated authoritative record"
 
     later = cli_update("--note", "Newer note", "--update-operation-id", "later-update")
     assert later.returncode == 0, later.stdout + later.stderr
diff --git a/tests/control_plane/test_chat_reviewed_create.py b/tests/control_plane/test_chat_reviewed_create.py
index 51cb1e12c5..bc09978a3c 100644
--- a/tests/control_plane/test_chat_reviewed_create.py
+++ b/tests/control_plane/test_chat_reviewed_create.py
@@ -37,7 +37,7 @@ def interleaved(**kwargs):
     assert conflict.value.receipt["write_attempted"] is (timing == "after-final-preview")
     assert ("current_preview_id" in conflict.value.receipt) is (timing == "before-confirmation")
     snapshot = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
-    assert [todo["text"] for todo in snapshot["todos"]] == ["Independent accepted request"]
+    assert [todo["text"] for todo in snapshot["todos"]] == ["[P1] Independent accepted request"]
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
@@ -51,4 +51,4 @@ def test_current_review_creates_once_and_duplicate_remains_noop(tmp_path, monkey
             text="Reviewed request", preview_id=preview["preview_id"])
         assert result["applied"] is True
     snapshot = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
-    assert [todo["text"] for todo in snapshot["todos"]] == ["Reviewed request"]
+    assert [todo["text"] for todo in snapshot["todos"]] == ["[P1] Reviewed request"]
diff --git a/tests/control_plane/test_cold_source_import_cli.py b/tests/control_plane/test_cold_source_import_cli.py
index 982dc2b3d9..690d244b9a 100644
--- a/tests/control_plane/test_cold_source_import_cli.py
+++ b/tests/control_plane/test_cold_source_import_cli.py
@@ -389,7 +389,8 @@ def listing():
     assert records["todo_current"]["text"] == body
     assert records["todo_current"]["note"] == "Work retained after migration"
     assert records["todo_archived"]["evidence"] == "original"
-    assert records[created["todo_id"]]["text"] == "New work after migration"
+    later = records[created["todo_id"]]
+    assert (later["text"], later["priority"]) == ("[P1] New work after migration", "P1")
     assert len(complete_after["leases"]) == 1
     assert complete_after["leases"][0]["todo_id"] == "todo_current"
     assert complete_after["leases"][0]["owner"] == "agent-a"
diff --git a/tests/control_plane/test_goal_attention_integration.py b/tests/control_plane/test_goal_attention_integration.py
index 5c1f4bbeca..729d923287 100644
--- a/tests/control_plane/test_goal_attention_integration.py
+++ b/tests/control_plane/test_goal_attention_integration.py
@@ -104,7 +104,7 @@ def test_status_automatically_rebuilds_stale_run_routing(tmp_path, monkeypatch,
         else:
             assert item["source"] == "active_state"
             assert item["status"] == "active_state_agent_todo"
-            assert item["recommended_action"] == "Verify the current deliverable"
+            assert item["recommended_action"] == "[P1] Verify the current deliverable"
             assert item.get("stale_latest_run_warning") is None
         historical = status["run_history"]["goals"][0]["latest_status_run"]
         assert historical["generated_at"] == record["generated_at"]
diff --git a/tests/control_plane/test_heartbeat_prompt_support.py b/tests/control_plane/test_heartbeat_prompt_support.py
index 11e448f830..d5fcffb45c 100644
--- a/tests/control_plane/test_heartbeat_prompt_support.py
+++ b/tests/control_plane/test_heartbeat_prompt_support.py
@@ -259,7 +259,7 @@ def test_brief_diagnostics_keep_only_the_selected_prompt_command() -> None:
 
 @pytest.mark.parametrize(
     ("mode", "base_budget"),
-    [("full", 12000), ("compact", 6700), ("brief", 4300), ("thin", 3200)],
+    [("full", 12000), ("compact", 6700), ("brief", 4600), ("thin", 3200)],
 )
 def test_reward_memory_prompt_headroom_is_fixed_and_feature_scoped(
     mode: str, base_budget: int
diff --git a/tests/control_plane/test_host_bootstrap_lifecycle.py b/tests/control_plane/test_host_bootstrap_lifecycle.py
index 10fcfb5a08..a789db11ef 100644
--- a/tests/control_plane/test_host_bootstrap_lifecycle.py
+++ b/tests/control_plane/test_host_bootstrap_lifecycle.py
@@ -188,7 +188,7 @@ def test_app_brief_with_registry_profile_keeps_budget_and_current_settlement(reg
     body = packet["task_body"]
     assert all(scope.rstrip(".!?") in body for scope in scopes)
     assert packet["agent_scope_source"] == "agent_profile_v1"
-    assert packet["interface_budget"]["max_chars"] == 4300
+    assert packet["interface_budget"]["max_chars"] == 4600
     assert packet["interface_budget"]["within_budget"], packet["interface_budget"]
     assert packet["cli_preflight"] in body
     assert "--codex-app" in body
diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py
index d40b850ecc..b47debecfd 100644
--- a/tests/control_plane/test_local_coordination_authority.py
+++ b/tests/control_plane/test_local_coordination_authority.py
@@ -2398,7 +2398,7 @@ def test_real_canonical_provider_preserves_complete_complex_todo_semantics(
     created_item = next(
         item for item in after_create["todos"] if item["todo_id"] == created["todo_id"]
     )
-    assert created_item["text"] == "Create directly against promoted provider"
+    assert created_item["text"] == "[P1] Create directly against promoted provider"
     assert created_item["claimed_by"] == "agent-a"
     assert after_create["authority_read"]["todo_read_model"]["todo_count"] == 4
     assert claimed_item["note"] == "Correction context"
diff --git a/tests/control_plane/test_native_user_completion_update.py b/tests/control_plane/test_native_user_completion_update.py
index f4c9ca5143..f39710ed55 100644
--- a/tests/control_plane/test_native_user_completion_update.py
+++ b/tests/control_plane/test_native_user_completion_update.py
@@ -53,7 +53,7 @@ def test_public_user_update_done_uses_canonical_transaction(tmp_path: Path, prov
     assert result["source_authority"] == f"{provider}_v0"
     assert result["legacy_fallback_used"] is False
     todo = list_goal_todos(registry_path=registry, goal_id="goal-a")["todos"][0]
-    assert (todo["status"], todo["text"], todo["note"]) == ("done", "Observed outcome", "Reviewed")
+    assert (todo["status"], todo["text"], todo["note"]) == ("done", "[P1] Observed outcome", "Reviewed")
     assert state.is_file()
 
 
diff --git a/tests/control_plane/test_new_goal_storage_defaults.py b/tests/control_plane/test_new_goal_storage_defaults.py
index 744a47ce4a..71b2e22fc6 100644
--- a/tests/control_plane/test_new_goal_storage_defaults.py
+++ b/tests/control_plane/test_new_goal_storage_defaults.py
@@ -703,7 +703,7 @@ def storage_effect(method, payload):
             source = read_canonical_todos_if_promoted(runtime_root=project.parent / "runtime", goal_id="recovery", include_leases=True)
             assert source["handoff_mode"] == mode
             assert len(source["todos"]) == 1
-            assert source["todos"][0]["text"] == "Verify recovery"
+            assert (source["todos"][0]["text"], source["todos"][0]["priority"]) == ("[P1] Verify recovery", "P1")
             assert source["source_authority"] == f"{provider}_v0"
             assert selections[-1]["authority_initialized"] is True
         if provider == "sqlite":
diff --git a/tests/control_plane/test_promoted_archive_retention.py b/tests/control_plane/test_promoted_archive_retention.py
index f35b74b6d0..95e04a7612 100644
--- a/tests/control_plane/test_promoted_archive_retention.py
+++ b/tests/control_plane/test_promoted_archive_retention.py
@@ -87,7 +87,7 @@ def test_promoted_archive_scope_and_later_write_recovery(tmp_path, monkeypatch,
     recovered = ws.cli("coordination-shadow", "recover-promotion", "--reviewed-plan", str(reviewed), "--execute")
     assert recovered["promotion"]["status"] == "replayed"
     current = ws.cli("todo", "list", "--todo-id", later["todo_id"])
-    assert current["todo"]["text"] == "Later canonical work"
+    assert current["todo"]["text"] == "[P1] Later canonical work"
     assert current["authority_read"]["legacy_fallback_used"] is False
     assert not ws.state.exists()
 
diff --git a/tests/control_plane/test_runtime_shadow_writer_capture.py b/tests/control_plane/test_runtime_shadow_writer_capture.py
index 58a3fc5879..3413b9a65d 100644
--- a/tests/control_plane/test_runtime_shadow_writer_capture.py
+++ b/tests/control_plane/test_runtime_shadow_writer_capture.py
@@ -107,7 +107,7 @@ def test_runtime_shadow_todo_writer_captures_full_records_and_reuses_one_store(
     )
     head = view["head"]
     assert head["schema_version"] == "loopx_coordination_runtime_shadow_projection_v0"
-    assert head["todos"][0]["text"] == "Retain complete canonical Todo fields."
+    assert head["todos"][0]["text"] == "[P1] Retain complete canonical Todo fields."
     assert head["todos"][0]["note"] == "transaction-bound"
     assert head["todo_read_model"]["todo_count"] == 1
     assert view["cursor"] == "3"
diff --git a/tests/control_plane/test_shadow_registry_source.py b/tests/control_plane/test_shadow_registry_source.py
index f35162f080..9c4767aed5 100644
--- a/tests/control_plane/test_shadow_registry_source.py
+++ b/tests/control_plane/test_shadow_registry_source.py
@@ -206,7 +206,7 @@ def test_terminal_lease_survives_public_cutover_and_original_recovery(tmp_path:
     assert recovered["promotion"]["status"] == "replayed", recovered
     assert ws.cli("task-lease", "inspect", "--todo-id", todo_id)["lease"] == expected
     retained = ws.cli("todo", "list", "--todo-id", later["todo_id"])
-    assert retained["todo"]["text"] == "Later canonical work must survive recovery"
+    assert (retained["todo"]["text"], retained["todo"]["priority"]) == ("[P1] Later canonical work must survive recovery", "P1")
     assert lease.read_bytes() == original
     assert not ws.state.exists()
 
diff --git a/tests/control_plane/test_steward_goal_notice_integration.py b/tests/control_plane/test_steward_goal_notice_integration.py
index 30f32a7f59..836b42cedb 100644
--- a/tests/control_plane/test_steward_goal_notice_integration.py
+++ b/tests/control_plane/test_steward_goal_notice_integration.py
@@ -90,7 +90,7 @@ def read():
     todo("update", "--todo-id", next(row["todo_id"] for row in read() if row["role"] == "user"), "--role", "user", "--status", "blocked",
          "--reason", "Publication needs an owner decision", "--update-operation-id", "gate-blocked", "--agent-id", "agent-a")
     source_before = read()
-    records = {row["text"]: row for row in list_goal_todos(
+    records = {row["title"]: row for row in list_goal_todos(
         registry_path=registry, runtime_root_arg=str(runtime), goal_id="goal-a")["todos"]}
     selected = records["Review publication evidence"]
     fallback = records["Continue independent verification"]
diff --git a/tests/control_plane/test_todo_handoff_retirement_contract.py b/tests/control_plane/test_todo_handoff_retirement_contract.py
index 300610530a..86dad5bd92 100644
--- a/tests/control_plane/test_todo_handoff_retirement_contract.py
+++ b/tests/control_plane/test_todo_handoff_retirement_contract.py
@@ -126,7 +126,7 @@ def cli(*args):
         "--update-expected-provider-revision", first["authority_read"]["provider_revision"])
     readback = cli("todo", "list", "--goal-id", "goal-a", "--todo-id", original)
     row = readback["todo"]
-    assert row["text"] == "Original complete requirement"
+    assert (row["text"], row["priority"]) == ("[P1] Original complete requirement", "P1")
     assert row["status"] == "open" and row["successor_todo_ids"] == [successor]
     # Exact canonical reads return the source record. Derived handoff context
     # belongs to status consumers and must never become a second stored source.
diff --git a/tests/control_plane/test_todo_list_read_transport.py b/tests/control_plane/test_todo_list_read_transport.py
index c76061c803..a3033c5d95 100644
--- a/tests/control_plane/test_todo_list_read_transport.py
+++ b/tests/control_plane/test_todo_list_read_transport.py
@@ -112,7 +112,11 @@ def test_read_filters_errors_and_fresh_provider(read_cli):
         assert added.returncode == 0, added.stdout + added.stderr
     fresh = run()
     assert fresh["todo_count"] == 4
-    assert any(item["text"].startswith("Fresh ") for item in fresh["todos"])
+    if provider == "legacy":
+        assert any(item["text"] == "Fresh legacy requirement" for item in fresh["todos"])
+    else:
+        current = next(item for item in fresh["todos"] if item["todo_id"] == json.loads(added.stdout)["todo_id"])
+        assert (current["text"], current["priority"]) == ("[P1] Fresh canonical requirement", "P1")
     if provider != "legacy":
         assert fresh["authority_read"]["provider_revision"] != full["authority_read"]["provider_revision"]
         assert fresh["authority_read"]["decision_read_from_provider"]
diff --git a/tests/test_collaboration_mcp.py b/tests/test_collaboration_mcp.py
index 34826ad591..65088c8d8e 100644
--- a/tests/test_collaboration_mcp.py
+++ b/tests/test_collaboration_mcp.py
@@ -188,7 +188,7 @@ async def call(name, args):
             assert resumed["recorded_decision"] == "adopt" and resumed["answer_owed"]
             assert resumed["step"] == "review_request_work"
             assert [(t["todo_id"], t["title"], t["status"]) for t in resumed["linked_todos"]] == [
-                (own, "Current accepted work", "open")
+                (own, "[P1] Current accepted work", "open")
             ]
             await call("return_result", {"request_id": rid, "text": conclusion})
             assert (await call("read_context", {}))["items"] == []
diff --git a/tests/test_independent_delegation_validation.py b/tests/test_independent_delegation_validation.py
index d0c5bf0b21..b2dcd8b706 100644
--- a/tests/test_independent_delegation_validation.py
+++ b/tests/test_independent_delegation_validation.py
@@ -121,7 +121,7 @@ def checked_with_task_change(**kwargs):
             runner._validate(binding)
     current = demo.canonical_tasks(root)[todo_id]
     field = {"note": "note", "text": "text", "claim": "claimed_by"}[change]
-    assert current[field] == {"note": "Current progress observation", "text": "Validate a different requested result",
+    assert current[field] == {"note": "Current progress observation", "text": "[P1] Validate a different requested result",
                               "claim": "reviewer"}[change]
     assert not current["done"]
     assert not (root / "analyst/initial/host-invocations").exists()
diff --git a/tests/test_manager_context_tracking.py b/tests/test_manager_context_tracking.py
index d0f72c5ca9..60eebf1b39 100644
--- a/tests/test_manager_context_tracking.py
+++ b/tests/test_manager_context_tracking.py
@@ -319,14 +319,14 @@ def add(owner, goal="goal-a"):
         "--agent-id", "agent-a", "--text", "Verified current work")
     row = cli("manager-inbox", "status", *inbox)["rows"][0]
     assert [(t["todo_id"], t["title"], t["status"]) for t in row["linked_todos"]] == [
-        (tid, "Verified current work", "open")
+        (tid, "[P1] Verified current work", "open")
     ]
     assert "Verified current work" not in path.read_text()
     resumed = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"][0]
     follow = resumed["receiver_followthrough"]
     assert follow["recorded_decision"] == "adopt" and follow["step"] == "review_request_work"
     assert [(t["todo_id"], t["title"], t["status"]) for t in follow["linked_todos"]] == [
-        (tid, "Verified current work", "open")
+        (tid, "[P1] Verified current work", "open")
     ]
     assert follow["answer_owed"] is True
     # Evidence-only linking must survive the next ordinary CLI receiver read.