test(release): preserve context recovery oracles and timed-wait guidance

CONTRIBUTING.md

@@ -189,6 +189,7 @@ touch, mention it in the PR instead of fixing it there.
 | `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 |
 | `release-artifacts.yml` | `loopx/`, packaging, and lockfile paths | no | release identity and a release build from this source |
+| `runtime-snapshots.yml` | runtime artifact, installer, and updater paths | no | qualified, revision-addressed runtime archives and the main channel pointer |
 | `ark-turn.yml` | Turn driver and collaboration paths | no | optional Ark Turn package, stdio MCP, and DSH parity |
 | `memt-turn.yml` | optional MemT package, Turn driver, lock, and packaging paths | no | explicit optional dependency installation, complete adapter workflow, stdio MCP, and driver binding |
 | `dsh-plugin.yml` | DSH plugin and distribution workflow paths | no | Linux/Windows package install and uninstall, typed host contracts, and Linux real LoopX/web runtime admission |

examples/blocker-push-runtime-smoke.py

@@ -220,7 +220,7 @@ def main() -> int:
         # current compact sentence instead of the retired per-shell phrasing.
         assert "reuse the value on retries" in compact_prompt, prompt
         assert "obey projected replan/stop, never wake counts" in compact_prompt, prompt
-        assert "re-enter guard." in compact_prompt, prompt
+        assert "re-enter as instructed, do/verify authorized work." 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

examples/control_plane/heartbeat_prompt_fixtures.py

@@ -39,10 +39,9 @@ def normalized(text: str) -> str:
 
 
 def prompt_budget_text(text: str) -> str:
-    return text.replace(GOAL_ID, "<GOAL_ID>").replace(
-        str(ACTIVE_STATE),
-        "<ACTIVE_GOAL_STATE_PATH>",
-    )
+    # Normalize the complete path before its Goal substring, matching the
+    # independently declared authored-guidance measurement contract.
+    return text.replace(str(ACTIVE_STATE), "<ACTIVE_STATE>").replace(GOAL_ID, "<GOAL_ID>")
 
 
 def assert_prompt_budget(label: str, text: str) -> None:

examples/install-local-smoke.py

@@ -57,7 +57,7 @@ def normal_turns_use_cli_interaction_contract(task_body: str) -> bool:
 def read_installed_workflow_skill(
     skill: Path, *, references: set[str]
 ) -> dict[str, str]:
-    """Read the installed entrypoint and its complete, self-contained procedures."""
+    """Read installed procedures and verify links stay in the declared bundle."""
     skill_root = skill.parent.resolve()
     documents = {"SKILL.md": skill.read_text(encoding="utf-8")}
     linked_references = {
@@ -81,8 +81,12 @@ def read_installed_workflow_skill(
             if "://" in destination or destination.startswith("#"):
                 continue
             resource = (document.parent / destination.split("#", 1)[0]).resolve()
-            assert resource.is_relative_to(skill_root), (document, destination)
+            bundle_root = skill_root.parent
+            assert resource.is_relative_to(bundle_root), (document, destination)
+            bundled_path = resource.relative_to(bundle_root)
+            assert bundled_path.parts[0] in {"loopx", *PACKAGED_HOST_SKILL_IDS}, (document, destination)
             assert resource.is_file(), (document, destination)
+            assert resource.read_bytes() == (REPO_ROOT / "skills" / bundled_path).read_bytes(), resource
     return documents
 
 
@@ -917,7 +921,7 @@ 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 "obey projected replan/stop, never wake counts" in payload["task_body"], payload
-        assert "re-enter guard." in payload["task_body"], payload
+        assert "re-enter as instructed, do/verify authorized work." 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

loopx/control_plane/testing/cli_output_budget.py

@@ -312,13 +312,15 @@ class CliOutputCommandClassification:
         ),
         markdown_anchor="# Heartbeat Automation Prompt",
         max_chars={
-            # The complete first-guard selection guidance and supported
-            # 320-char scope emit 4,532 JSON chars at immutable b47bbc6c9.
-            # Retain the old 4,500 failure; allow 68 chars of JSON headroom.
-            # Body presentation stays 4,200; Markdown/line guards are unchanged.
-            "small": {"json": 4_600, "markdown": 6_200},
-            "crowded": {"json": 4_600, "markdown": 6_200},
-            "multi_agent": {"json": 4_600, "markdown": 6_200},
+            # Same fixed-path, 320-char scope workload at 9cae3b289 emitted
+            # 4,532 JSON / 6,648 Markdown chars; timed-wait guidance adds 321
+            # chars, retaining the original Turn, due time and fresh admission.
+            # All three scenarios now emit 4,853 / 6,969 chars and 63 Markdown
+            # lines. Preserve historical misses; retain complete instructions,
+            # the current 4,500 body allowance and unchanged line/growth guards.
+            "small": {"json": 5_000, "markdown": 7_100},
+            "crowded": {"json": 5_000, "markdown": 7_100},
+            "multi_agent": {"json": 5_000, "markdown": 7_100},
         },
         max_lines={
             "small": {"json": 18, "markdown": 72},

tests/control_plane/test_cli_output_budget.py

@@ -1712,7 +1712,7 @@ def test_thin_json_budget_retains_full_supported_scope(tmp_path: Path) -> None:
         )
         payload = measurement["payload"]
         assert scope in payload["task_body"]
-        assert payload["interface_budget"]["max_chars"] == 4200
+        assert payload["interface_budget"]["max_chars"] == 4500
         assert payload["interface_budget"]["within_budget"] is True
         assert "refresh-state --goal-outcome-update-json" in payload["task_body"]
         assert "Local outcomes never replace root continuation, budget or stop criteria." in payload["task_body"]

tests/control_plane/test_codex_context_boundary.py

@@ -177,7 +177,7 @@ def test_native_reader_requires_complete_hook_output_even_when_model_saw_page(tm
         name: [{"hooks": [handler]}] for name in ("SessionStart", "PostToolUse")}}))
     publish = workspace / "publish.py"
     publish.write_text(
-        "import json, pathlib, time\n"
+        "import json, pathlib, time, subprocess, shlex\n"
         "from loopx.control_plane.turn_driver.codex_context import project_codex_context\n"
         "from loopx.cli_commands.quota_capture import native_context_capture_view\n"
         f"capture=pathlib.Path({str(capture)!r})\n"
@@ -186,6 +186,10 @@ def test_native_reader_requires_complete_hook_output_even_when_model_saw_page(tm
         f"full=project_codex_context(body,directory={str(root)!r},capture_directory=capture,"
         "goal_id='fixture',agent_id='agent',surface='work_context',capture_id='fixture-capture')\n"
         f"view=native_context_capture_view(full,capture,context_directory={str(root)!r})\n"
+        # Read the returned reference through its real command. The test still
+        # needs an actual page in model input to isolate complete-hook coverage.
+        "view=json.loads(subprocess.run(shlex.split(view['command']),check=True,"
+        "capture_output=True,text=True).stdout)\n"
         "text=json.dumps(view,separators=(',',':'))\n"
         f"pathlib.Path({str(view_file)!r}).write_text(text)\n"
         "print(text,flush=True)\n"

tests/control_plane/test_codex_context_capture.py

@@ -9,6 +9,7 @@
 import json
 import os
 from pathlib import Path
+import re
 import shlex
 import subprocess
 import sys
@@ -371,11 +372,30 @@ def response(request, number, message):
                 assert "LoopX model-context baseline invalidated." not in text
                 # The recovery entry must really be present after compaction;
                 # the fixture may not supply a hidden saved capture path.
-                prefix = "Before dependent action, recover missing or uncertain necessary facts"
-                permanent = text[text.index(prefix):]
-                # Read only the advertised command sentence. Subsequent
-                # recovery guidance is not part of the shell invocation.
-                cmd = permanent.split("with: ", 1)[1].split(". ", 1)[0]
+                commands = set()
+                for advertised in re.findall(r"with: ([^\n]+?)(?:\. |$)", text):
+                    try:
+                        argv = shlex.split(advertised)
+                        quota = argv.index("quota")
+                        if argv[quota:quota + 2] != ["quota", "context-read"]:
+                            continue
+                        # The worker advertises its own trial-local directory
+                        # in the actual guard; it need not use this fixture's.
+                        if (argv[argv.index(context_option) + 1] != command[command.index(context_option) + 1]
+                                or argv[argv.index("--goal-id") + 1] != GOAL_ID
+                                or argv[argv.index("--agent-id") + 1] != AGENT_ID):
+                            continue
+                    except (ValueError, IndexError):
+                        continue
+                    # Voluntary full traversal serves this byte oracle. Index
+                    # and selected reads do not themselves establish a baseline.
+                    for option, width in (("--select", 2), ("--list", 1)):
+                        while option in argv:
+                            index = argv.index(option)
+                            del argv[index:index + width]
+                    commands.add(shlex.join(argv))
+                assert len(commands) == 1, "One scoped permanent reader must be visible in model input"
+                cmd = commands.pop()
                 stages["phase"] = "lookup_read"
                 assert delivered(request, "cold") == "" and delivered(request, "warm") == ""
                 return tool(cmd, number)

tests/control_plane/test_frontier_deadline.py

@@ -136,9 +136,9 @@ def test_quota_planning_keeps_gate_deadline_before_visibility_limits(monkeypatch
     original = json.loads(json.dumps(gates))
     crossings = []
     invoke = quota_selection.effect_runtime_result
-    def observed(method, params):
+    def observed(method, params, **kwargs):
         crossings.append(method)
-        return invoke(method, params)
+        return invoke(method, params, **kwargs)
     monkeypatch.setattr(quota_selection, "effect_runtime_result", observed)
     summary = quota_summary.summarize_user_todos_for_quota(
         {"schema_version": "todo_summary_v0", "items": gates, "total_count": 26,

tests/control_plane/test_goal_state_context_transport.py

@@ -86,6 +86,11 @@ def guard():
     page, raw = json.loads(output.getvalue()), bytearray()
     assert code == 0 and page["ok"], (page.get("reason"), page.get("error"), page.get("error_code"))
     digest = page["sha256"]
+    if page.get("schema_version") != "context_capture_page_v1":
+        page = read_context_capture(build_parser().parse_args([
+            "--format", "json", "quota", "context-read", "--capture-file", page["path"],
+            "--capture-sha256", digest, "--offset", "0", "--page-bytes", "3072"]))
+        assert page["ok"], page
     while True:
         assert page["offset"] == len(raw) and page["sha256"] == digest
         raw.extend(page["content"].encode())

tests/control_plane/test_host_prompt_behavior.py

@@ -54,7 +54,10 @@ def test_probe_uses_current_production_prompts_and_hidden_independent_oracle():
             assert "heartbeat_recommendation.agent_must_attempt" in body
             assert "projected monitor/continuation" in body
             assert "no invented successor" in body
-            assert "no wait spend" in body.lower()
+            assert "no wait spend" in body.lower() or "no spend for waiting alone." in body.lower()
+            assert "resume_at:<RFC3339>" in body
+            assert "yield to host scheduler" in body
+            assert "Fresh admission on resume." in body
             assert "--codex-app" in body
             assert "LOOPX_TURN=<current_time_iso>" in body
             assert case["id"] not in json.dumps(messages)

tests/control_plane/test_quota_monitor_clock.py

@@ -70,9 +70,9 @@ def clock():
         clocks.append(True)
         return NOW
 
-    def record(method, params):
+    def record(method, params, **kwargs):
         calls.append(method)
-        return original(method, params)
+        return original(method, params, **kwargs)
 
     monkeypatch.setattr(quota_selection, "now_utc", clock, raising=False)
     monkeypatch.setattr(quota_selection, "effect_runtime_result", record)

tests/control_plane/test_quota_summary_source_precedence.py

@@ -39,9 +39,9 @@ def test_only_preferred_available_source_is_projected(
     calls = []
     original = quota_selection.effect_runtime_result
 
-    def record(method, params):
+    def record(method, params, **kwargs):
         calls.append(method)
-        return original(method, params)
+        return original(method, params, **kwargs)
 
     monkeypatch.setattr(quota_selection, "effect_runtime_result", record)
     summary = quota_summary.select_quota_todo_summary(