test(quota): contract prose never triggers replan; advise stale runtimes

loopx/doctor.py

@@ -21,6 +21,7 @@
 from .python_install_owner import PythonInstallOwner, python_distribution_upgrade_command, resolve_python_install_owner
 from .capabilities.project_skill_delivery import discover_project_scoped_skill_ids
 from .registry_writability import probe_registry_write_path
+from .release_advisories import build_release_advisories_section
 from .release_manifest import load_release_manifest, release_version_tag
 from .skill_install_readback import (
     ARK_MANAGED_AGENT_REQUIRED_SKILL_IDS,
@@ -823,6 +824,7 @@ def collect_doctor(
             latest_promotion_readiness_event(DEFAULT_RUNTIME_ROOT)
         ),
     }
+    release_advisories = build_release_advisories_section(__version__)
     install_freshness = build_install_freshness(
         command_path=command_path,
         release_root=release_root,
@@ -1106,6 +1108,7 @@ def collect_doctor(
         "typescript_control_plane": typescript_control_plane,
         "install_freshness": install_freshness,
         "upgrade_hint": install_freshness,
+        "release_advisories": release_advisories,
         "skill": {
             "path": str(skill_path),
             "exists": bool(project_skill.get("exists")),
@@ -1286,6 +1289,26 @@ def render_doctor_markdown(payload: dict[str, Any]) -> str:
                 "```",
             ]
         )
+    advisories_section = (
+        payload.get("release_advisories")
+        if isinstance(payload.get("release_advisories"), dict)
+        else {}
+    )
+    for advisory in advisories_section.get("advisories") or []:
+        if not isinstance(advisory, dict):
+            continue
+        lines.extend(
+            [
+                "",
+                "## Release Advisory",
+                f"- id: `{advisory.get('id')}`",
+                f"- current_version: `{advisories_section.get('current_version')}`",
+                f"- last_affected_release: `{advisory.get('last_affected_release')}`",
+                f"- fixed_in_release: `{advisory.get('fixed_in_release')}`",
+                f"- summary: {advisory.get('summary')}",
+                f"- guidance: {advisory.get('guidance')}",
+            ]
+        )
     if typescript_control_plane:
         lines.extend(
             [

loopx/release_advisories.py

@@ -0,0 +1,70 @@
+"""Version-aware release advisories for known runtime defects.
+
+An advisory names the last release that shipped a known defect and the first
+release that fixed it. `loopx doctor` surfaces an advisory when the running
+runtime is at or below the last affected release, so an operator sees the
+bounded failure mode and the upgrade path. Advisories describe runtime
+behavior only; they never interpret the user's project text.
+"""
+
+from __future__ import annotations
+
+import re
+
+RELEASE_ADVISORIES_SCHEMA_VERSION = "loopx_release_advisories_v0"
+
+_VERSION_TOKEN = re.compile(r"\d+")
+
+
+def _version_tuple(version: str) -> tuple[int, ...]:
+    return tuple(int(part) for part in _VERSION_TOKEN.findall(version)) or (0,)
+
+
+_RELEASE_ADVISORIES = (
+    {
+        "id": "legacy_replan_prose_stall_matcher",
+        "last_affected_release": "0.4.5",
+        "fixed_in_release": "0.4.6",
+        "summary": (
+            "Releases through 0.4.5 inferred autonomous-replan stalls from "
+            "free-form run summaries with a substring matcher, so successful "
+            "records containing words such as 'installed' or 'installation' "
+            "could produce a no_progress_streak trigger and an erroneous "
+            "autonomous_replan_required decision."
+        ),
+    },
+)
+
+
+def release_advisories_for(current_version: str) -> tuple[dict[str, str], ...]:
+    """Return the advisories that affect the running release."""
+
+    running = _version_tuple(current_version)
+    advisories: list[dict[str, str]] = []
+    for advisory in _RELEASE_ADVISORIES:
+        if running <= _version_tuple(advisory["last_affected_release"]):
+            advisories.append(
+                {
+                    "id": advisory["id"],
+                    "current_version": current_version,
+                    "last_affected_release": advisory["last_affected_release"],
+                    "fixed_in_release": advisory["fixed_in_release"],
+                    "summary": advisory["summary"],
+                    "guidance": (
+                        "Upgrade to "
+                        f"{advisory['fixed_in_release']} or newer: run "
+                        "`loopx update check` and then `loopx update apply`."
+                    ),
+                }
+            )
+    return tuple(advisories)
+
+
+def build_release_advisories_section(current_version: str) -> dict[str, object]:
+    affected = release_advisories_for(current_version)
+    return {
+        "schema_version": RELEASE_ADVISORIES_SCHEMA_VERSION,
+        "current_version": current_version,
+        "requires_upgrade": bool(affected),
+        "advisories": list(affected),
+    }

tests/control_plane/test_quota_should_run_prose_contract.py

@@ -0,0 +1,223 @@
+"""Public quota should-run contract: prose never creates a stall signal.
+
+Issue #4336: releases through 0.4.5 inferred autonomous-replan stalls from
+free-form run summaries with a substring matcher, so the word `stalled`
+inside `installed` turned successful progress records into a
+`no_progress_streak` trigger. Main sources replan decisions from typed
+progress observations only; these tests pin that contract through the real
+public CLI rather than private helper state.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+GOAL_ID = "prose-stall-contract"
+AGENT_ID = "codex-prose-contract"
+TURN_ID = "turn-prose-contract-1"
+
+
+def _write_fixture(root: Path) -> tuple[Path, Path, Path]:
+    project = root / "project"
+    runtime = root / "runtime"
+    state_file = f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state_path = project / state_file
+    registry_path = project / ".loopx" / "registry.json"
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+    state_path.write_text(
+        "---\n"
+        "status: active-read-only\n"
+        "owner_mode: goal\n"
+        'objective: "Install the CLI and settle the delivery."\n'
+        "updated_at: 2026-01-01T00:00:00+00:00\n"
+        "---\n\n"
+        "# Prose Stall Contract Fixture\n\n"
+        "## Objective\n\n"
+        "Install the CLI and settle the delivery.\n\n"
+        "## Next Action\n\n"
+        "- Validate and settle the selected delivery.\n\n"
+        "## Agent Todo\n\n"
+        "- [ ] [P1] Validate and settle the selected delivery.\n"
+        "  <!-- loopx:todo todo_id=todo_prose_contract status=open "
+        "task_class=advancement_task action_kind=validate -->\n",
+        encoding="utf-8",
+    )
+    registry_path.parent.mkdir(parents=True, exist_ok=True)
+    registry_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "0.1",
+                "updated_at": "2026-01-01T00:00:00+00:00",
+                "common_runtime_root": str(runtime),
+                "goals": [
+                    {
+                        "id": GOAL_ID,
+                        "domain": "prose-stall-contract",
+                        "status": "active-read-only",
+                        "repo": str(project),
+                        "state_file": state_file,
+                        "adapter": {
+                            "kind": "read_only_project_map_v0",
+                            "status": "connected-read-only",
+                        },
+                        "coordination": {
+                            "registered_agents": [AGENT_ID],
+                            "agent_model": "peer_v1",
+                        },
+                        "authority_sources": [],
+                        "quota": {
+                            "compute": 1.0,
+                            "window_hours": 24,
+                            "allowed_slots": 2,
+                        },
+                    }
+                ],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    return project, runtime, registry_path
+
+
+def _write_runs(runtime: Path, runs: list[dict[str, Any]]) -> None:
+    runs_dir = runtime / "goals" / GOAL_ID / "runs"
+    runs_dir.mkdir(parents=True, exist_ok=True)
+    indexed: list[dict[str, Any]] = []
+    for minute, run in enumerate(runs, start=1):
+        run = {**run, "generated_at": f"2026-01-01T00:0{minute}:00+00:00"}
+        json_path = runs_dir / f"run-{minute}.json"
+        markdown_path = runs_dir / f"run-{minute}.md"
+        json_path.write_text(json.dumps(run) + "\n", encoding="utf-8")
+        markdown_path.write_text("# Run fixture\n", encoding="utf-8")
+        indexed.append(
+            {
+                **run,
+                "json_path": str(json_path),
+                "markdown_path": str(markdown_path),
+            }
+        )
+    (runs_dir / "index.jsonl").write_text(
+        "".join(json.dumps(run) + "\n" for run in indexed),
+        encoding="utf-8",
+    )
+
+
+def _run_should_run(
+    registry_path: Path,
+    runtime: Path,
+    project: Path,
+) -> tuple[int, dict[str, Any]]:
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "loopx.cli",
+            "--registry",
+            str(registry_path),
+            "--runtime-root",
+            str(runtime),
+            "--format",
+            "json",
+            "quota",
+            "should-run",
+            "--codex-app",
+            "--goal-id",
+            GOAL_ID,
+            "--agent-id",
+            AGENT_ID,
+            "--turn-instance-id",
+            TURN_ID,
+            "--scan-path",
+            str(project),
+        ],
+        cwd=REPO_ROOT,
+        check=False,
+        capture_output=True,
+        text=True,
+        env={**os.environ, "PYTHONPATH": str(REPO_ROOT)},
+    )
+    return result.returncode, json.loads(result.stdout)
+
+
+_PROSE_SUMMARIES = (
+    "CLI installed successfully and the tool is ready for use.",
+    "package uninstalled; installation completed for the CLI tool.",
+)
+
+_TYPED_OBSERVATION = {
+    "schema_version": "typed_progress_observation_v0",
+    "result_class": "unchanged",
+    "work_item_id": "todo_prose_contract",
+    "surface_id": "surface-existing",
+    "hypothesis_id": "hypothesis-existing",
+    "probe_kind": "probe-existing",
+    "evidence_ids": ["evidence-existing"],
+}
+
+
+def _run_base(**extra: Any) -> dict[str, Any]:
+    return {
+        "goal_id": GOAL_ID,
+        "agent_id": AGENT_ID,
+        "classification": "bounded_fixture_probe",
+        "delivery_batch_scale": "single_surface",
+        "delivery_outcome": "surface_only",
+        **extra,
+    }
+
+
+def test_installed_wording_never_creates_a_no_progress_trigger(
+    tmp_path: Path,
+) -> None:
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    _write_runs(
+        runtime,
+        [_run_base(summary=summary) for summary in _PROSE_SUMMARIES],
+    )
+
+    exit_code, payload = _run_should_run(registry_path, runtime, project)
+
+    assert exit_code == 0, payload
+    assert payload["decision"] != "autonomous_replan_required", payload
+    assert payload["decision"] == "run", payload
+    serialized = json.dumps(payload)
+    assert "no_progress_streak" not in serialized
+    assert "typed_progress_repeat" not in serialized
+
+
+def test_typed_no_progress_observation_still_requires_replan(
+    tmp_path: Path,
+) -> None:
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    _write_runs(
+        runtime,
+        [
+            _run_base(
+                summary=summary,
+                progress_observation=dict(_TYPED_OBSERVATION),
+            )
+            for summary in _PROSE_SUMMARIES
+        ],
+    )
+
+    exit_code, payload = _run_should_run(registry_path, runtime, project)
+
+    assert exit_code == 0, payload
+    assert payload["decision"] == "autonomous_replan_required", payload
+    triggers = (
+        payload.get("replan_obligation", {}).get("triggers")
+        or payload.get("autonomous_replan_obligation", {}).get("triggers")
+        or []
+    )
+    assert any(
+        trigger.get("kind") == "typed_progress_repeat" for trigger in triggers
+    ), payload

tests/test_release_advisories.py

@@ -0,0 +1,47 @@
+"""Release advisories must surface affected legacy runtimes in diagnosis."""
+
+from __future__ import annotations
+
+from loopx.doctor import render_doctor_markdown
+from loopx.release_advisories import (
+    build_release_advisories_section,
+    release_advisories_for,
+)
+
+
+def test_affected_releases_receive_actionable_upgrade_guidance() -> None:
+    for version in ("0.4.3", "0.4.5"):
+        advisories = release_advisories_for(version)
+        assert len(advisories) == 1, (version, advisories)
+        advisory = advisories[0]
+        assert advisory["id"] == "legacy_replan_prose_stall_matcher"
+        assert advisory["last_affected_release"] == "0.4.5"
+        assert advisory["fixed_in_release"] == "0.4.6"
+        assert "installed" in advisory["summary"]
+        assert "0.4.6" in advisory["guidance"]
+        assert "loopx update check" in advisory["guidance"]
+        assert "loopx update apply" in advisory["guidance"]
+
+
+def test_fixed_releases_receive_no_advisory() -> None:
+    for version in ("0.4.6", "1.0.3"):
+        assert release_advisories_for(version) == ()
+
+
+def test_doctor_section_reports_stale_runtime_for_affected_install() -> None:
+    section = build_release_advisories_section("0.4.3")
+    assert section["requires_upgrade"] is True
+    rendered = render_doctor_markdown({"release_advisories": section})
+    assert "## Release Advisory" in rendered
+    assert "legacy_replan_prose_stall_matcher" in rendered
+    assert "last_affected_release: `0.4.5`" in rendered
+    assert "fixed_in_release: `0.4.6`" in rendered
+    assert "loopx update apply" in rendered
+
+
+def test_doctor_section_omits_advisory_for_current_runtime() -> None:
+    section = build_release_advisories_section("1.0.3")
+    assert section["requires_upgrade"] is False
+    assert section["advisories"] == []
+    rendered = render_doctor_markdown({"release_advisories": section})
+    assert "## Release Advisory" not in rendered