fix(quota): scope read-only plans to the selected Goal

docs/quota-allocation.md

@@ -9,6 +9,30 @@ This replaces the current ad hoc pattern where priority is encoded only by
 changing automation periods. A timer can wake the executor, but the product
 policy should live in LoopX.
 
+## Read-only observation scope
+
+`loopx quota status --goal-id example` and `loopx quota plan --goal-id example`
+collect only the selected Goal's status and history. Previously these commands
+silently ignored `--goal-id` and collected the whole registry. The selector now
+also partitions the optional status projection cache; it is not an output-only
+filter. Omitting it retains the global view and ordering.
+
+Scoped JSON includes `goal_filter`. Its counts, groups and `next_automatic_turn`
+describe that selected Goal only, not its rank among all Goals. An unknown or
+unregistered Goal returns `status=goal_not_found` and a nonzero exit code.
+Goal selection reuses the existing status health contract: unrelated Goal-local
+errors are outside the query, while applicable global errors remain visible.
+Observation neither opens a Turn nor grants execution/spend authority; use
+`quota should-run` with the current lane's identity for execution admission.
+`status_projection_envelope` retains the TS-owned source freshness and coverage
+of that observation, including explicit cache use; the Markdown view shows it.
+
+`quota status/plan --goal-id` 现在按指定 Goal 收集状态与历史，缓存也按同一范围隔离；
+之前会静默忽略选择器并读取全部 Goal。不传仍是全局视图。定向结果中的计数、分组和
+`next_automatic_turn` 只针对该 Goal，不代表全局排名；不存在或未注册的 Goal 明确
+报错。沿用既有 status 健康合同，保留相关全局错误。这是只读观测，不创建 Turn、
+不授予执行／spend 权限，不能代替带当前身份的 `quota should-run`。
+
 ## Product Scope
 
 In v0.1, quota means **compute quota only**.

loopx/cli_commands/quota_context.py

@@ -301,7 +301,9 @@ def prepare_quota_command_context(
         require_monitor_poll_source_available(
             runtime_root=runtime_root, goal_id=args.goal_id,
         )
-    status_goal_id = args.goal_id if command not in {"status", "plan"} else None
+    # Observation commands use the same scoped collector/cache as execution
+    # commands. Only an omitted selector requests the whole registry.
+    status_goal_id = args.goal_id
     projection_cache_ttl_seconds = int(
         getattr(args, "projection_cache_ttl_seconds", 120)
     )

loopx/cli_commands/quota_registration.py

@@ -40,7 +40,7 @@ def register_quota_command(
     )
     quota_parser.add_argument(
         "--goal-id",
-        help="Goal id to check. Required for one-goal quota commands, including should-run, scheduler ACK/failure, spend, and void.",
+        help="Goal id to check. Optional for status/plan (omit for all goals); required for should-run, scheduler ACK/failure, spend, and void. A scoped status/plan ranks only the selected goal, not the global next turn.",
     )
     quota_parser.add_argument(
         "--agent-id",

loopx/presentation/renderers/quota_markdown.py

@@ -4,6 +4,7 @@
 
 from ...control_plane import control_plane_policy_summary
 from ...control_plane.quota.states import QUOTA_STATE_ORDER
+from ...control_plane.projection_envelope_facts import render_projection_envelope_markdown
 from ...control_plane.runtime.decision_freshness import (
     DECISION_FRESHNESS_WARNING_ITEM_LIMIT,
 )
@@ -192,6 +193,14 @@ def render_quota_markdown(payload: dict[str, Any]) -> str:
         f"- goals: `{payload.get('goal_count')}`",
         f"- runs: `{payload.get('run_count')}`",
     ]
+    if payload.get("goal_filter"):
+        lines.append(f"- goal_filter: `{markdown_scalar(payload['goal_filter'])}` (counts and ranking cover this Goal only)")
+    if payload.get("reason"):
+        lines.append(f"- reason: {markdown_scalar(payload['reason'])}")
+    if payload.get("recommended_action"):
+        lines.append(f"- action: {markdown_scalar(payload['recommended_action'])}")
+    if payload.get("status_projection_envelope"):
+        lines.extend(render_projection_envelope_markdown(payload["status_projection_envelope"]))
     summary = as_dict(payload.get("summary"))
     states = as_dict(summary.get("states"))
     state_text = ", ".join(f"{state}={states.get(state, 0)}" for state in QUOTA_STATE_ORDER)
@@ -204,7 +213,8 @@ def render_quota_markdown(payload: dict[str, Any]) -> str:
     lines.append(f"- states: {state_text}")
 
     next_turn = as_dict(payload.get("next_automatic_turn"))
-    lines.extend(["", "## Next Automatic Turn"])
+    next_title = "Next Automatic Turn Within Selected Goal" if payload.get("goal_filter") else "Next Automatic Turn"
+    lines.extend(["", f"## {next_title}"])
     if next_turn:
         quota = as_dict(next_turn.get("quota"))
         lines.append(

loopx/quota.py

@@ -817,7 +817,7 @@ def build_quota_plan(
     if groups.get("unknown"):
         summary["states"]["unknown"] = len(groups["unknown"])
 
-    return {
+    result = {
         "ok": status_payload.get("ok"),
         "mode": mode,
         "registry": status_payload.get("registry"),
@@ -829,6 +829,34 @@ def build_quota_plan(
         "groups": groups,
         "health_items": health_items,
     }
+    return _describe_quota_observation(result, status_payload, mode=mode)
+
+
+def _describe_quota_observation(
+    result: dict[str, Any], status_payload: dict[str, Any], *, mode: str
+) -> dict[str, Any]:
+    """Carry read-only selection and source proof without changing execution plans."""
+    if mode not in {"status", "plan"}:
+        return result
+    goal_filter = status_payload.get("goal_filter")
+    if goal_filter:
+        result["goal_filter"] = goal_filter
+        if not any(
+            item["goal_id"] == goal_filter
+            for group in result["groups"].values()
+            for item in group
+        ):
+            result.update(
+                ok=False,
+                status="goal_not_found",
+                reason="goal is not present in the registered quota plan",
+                recommended_action="run `loopx registry` and connect or sync the selected goal",
+            )
+    if isinstance(status_payload.get("projection_envelope"), dict):
+        # Freshness and coverage remain owned by the typed status envelope,
+        # including when this observation was served from an explicit cache.
+        result["status_projection_envelope"] = status_payload["projection_envelope"]
+    return result
 
 
 def _build_quota_plan_for_goal(

tests/control_plane/test_quota_observation_scope.py

@@ -0,0 +1,147 @@
+"""Read-only Goal selection must constrain collection, not just displayed rows."""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+
+import pytest
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+
+from loopx.cli import build_parser
+from loopx.cli_commands.quota_context import prepare_quota_command_context
+from loopx.control_plane.effect_runtime import restart_effect_runtime
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.projection_envelope_facts import seal_projection_envelope, source_fact
+from loopx.control_plane.runtime.time import now_utc_iso
+from loopx.control_plane.testing.canary_harness import run_json_cli_result, write_fixture_registry
+from loopx.presentation.renderers.quota_markdown import render_quota_markdown
+from loopx.quota import build_quota_plan
+
+
+@pytest.fixture(autouse=True)
+def isolated_runtime(tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    yield
+    assert restart_effect_runtime()["status"] in {"stopped", "not_running"}
+
+
+@pytest.fixture(params=["legacy", "file", "sqlite"])
+def goals(tmp_path, request):
+    project = tmp_path / "project"
+    project.mkdir()
+    state = project / "state.md"
+    state.write_text("---\nstatus: active\nwaiting_on: codex\n---\n# Example\n\n## Agent Todo\n")
+    registry, runtime = tmp_path / "registry.json", tmp_path / "runtime"
+    write_fixture_registry(project=project, runtime_root=runtime, registry_path=registry,
+                           goal_id="selected", domain="engineering", adapter_kind="generic_project_goal_v0",
+                           state_file=str(state), registered_agents=["worker"])
+    doc = json.loads(registry.read_text())
+    other = deepcopy(doc["goals"][0])
+    other.update(id="unrelated", state_file=str(project / "missing.md"))
+    doc["goals"].append(other)
+    registry.write_text(json.dumps(doc))
+    if request.param != "legacy":
+        projection = build_todo_runtime_shadow_projection(goal_id="selected", todos=[], handoff_mode="soft_claim")
+        initialize_canonical_authority(runtime, "selected", projection, state_path=state, provider=request.param)
+    return registry, runtime, project
+
+
+@pytest.mark.parametrize("mode", ["status", "plan"])
+def test_selected_goal_excludes_unrelated_broken_goal(goals, mode):
+    registry, runtime, project = goals
+    code, payload = run_json_cli_result("quota", mode, "--goal-id", "selected", "--scan-root", str(project),
+                                        registry_path=registry, runtime_root=runtime)
+    assert code == 0, payload
+    assert payload["goal_filter"] == "selected"
+    assert {row["goal_id"] for group in payload["groups"].values() for row in group} == {"selected"}
+    assert payload["summary"]["registered_goals"] == 1
+    assert not any(row.get("goal_id") == "unrelated" for row in payload["health_items"])
+    assert not list((runtime / "goals" / "selected" / "runs").glob("*.json*"))
+    assert payload["status_projection_envelope"]["coverage"]["scope"] == "goal"
+    assert "Within Selected Goal" in render_quota_markdown(payload)
+
+    _, global_payload = run_json_cli_result("quota", mode, "--scan-root", str(project),
+                                            registry_path=registry, runtime_root=runtime)
+    assert "goal_filter" not in global_payload
+    assert {row["goal_id"] for group in global_payload["groups"].values() for row in group} == {"selected", "unrelated"}
+
+
+def test_unknown_goal_does_not_become_a_successful_empty_or_global_plan(goals):
+    registry, runtime, project = goals
+    code, payload = run_json_cli_result("quota", "status", "--goal-id", "missing", "--scan-root", str(project),
+                                        registry_path=registry, runtime_root=runtime)
+    assert code == 1
+    assert payload["status"] == "goal_not_found"
+    assert payload["goal_filter"] == "missing"
+    assert not any(payload["groups"].values())
+    assert "connect or sync" in render_quota_markdown(payload)
+
+
+@pytest.mark.parametrize("mode", ["status", "plan"])
+@pytest.mark.parametrize("goal_id", [None, "selected"])
+def test_scope_reaches_collector_and_cache_identity(tmp_path, mode, goal_id):
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(tmp_path / "runtime"), "goals": []}))
+    argv = ["quota", mode, "--write-projection-cache"]
+    if goal_id:
+        argv.extend(["--goal-id", goal_id])
+    args = build_parser().parse_args(argv)
+    calls = []
+
+    def collect(**kwargs):
+        calls.append(kwargs["goal_id"])
+        now = now_utc_iso()
+        envelope = seal_projection_envelope(projection="status", observed_at=now,
+            sources=[source_fact("registry", last_read_at=now)],
+            coverage={"scope": "goal" if kwargs["goal_id"] else "registry", "expected_count": 1,
+                      "included_count": 1, "omitted": [], "shown_count": 1, "available_count": 1})
+        return {"ok": True, "goal_filter": kwargs["goal_id"], "projection_envelope": envelope}
+
+    context = prepare_quota_command_context(args, registry_path=registry, runtime_root_arg=None,
+        status_collector=collect, operator_inbox_urgency_projector_factory=lambda **kwargs: lambda **kw: {})
+    assert calls == [goal_id]
+    assert context.status_goal_id == goal_id
+    args.write_projection_cache = False
+    args.use_projection_cache = True
+    cached = prepare_quota_command_context(args, registry_path=registry, runtime_root_arg=None,
+        status_collector=collect, operator_inbox_urgency_projector_factory=lambda **kwargs: lambda **kw: {})
+    assert calls == [goal_id]
+    assert cached.status_payload["goal_filter"] == goal_id
+    cached_plan = build_quota_plan(cached.status_payload)
+    assert cached_plan["status_projection_envelope"]["served_from_cache"] is True
+    assert cached_plan["status_projection_envelope"]["observed_at"] == context.status_payload["projection_envelope"]["observed_at"]
+    # Switching scope must miss; the previous global/Goal cache cannot answer it.
+    args.goal_id = None if goal_id else "selected"
+    other = prepare_quota_command_context(args, registry_path=registry, runtime_root_arg=None,
+        status_collector=collect, operator_inbox_urgency_projector_factory=lambda **kwargs: lambda **kw: {})
+    assert calls == [goal_id, args.goal_id]
+    assert other.cache_metadata["hit"] is False
+
+
+def test_scoped_plan_keeps_global_health_and_upstream_observation_proof():
+    now = now_utc_iso()
+    envelope = seal_projection_envelope(projection="status", observed_at=now,
+        sources=[source_fact("registry", last_read_at=now)],
+        coverage={"scope": "goal", "expected_count": 1, "included_count": 1,
+                  "omitted": [], "shown_count": 1, "available_count": 1})
+    health = {"goal_id": "global", "source": "contract", "severity": "error",
+              "recommended_action": "Repair the shared registry boundary"}
+    status = {"ok": False, "goal_filter": "selected", "projection_envelope": envelope,
+              "run_history": {"goals": [{"id": "selected", "registry_member": True,
+                                          "quota": {"state": "eligible", "compute": 1}}]},
+              "attention_queue": {"items": [health]}}
+    result = build_quota_plan(status)
+    assert result["ok"] is False
+    assert result["health_items"] == [health]
+    assert result["status_projection_envelope"] == envelope
+    assert "Repair the shared registry boundary" in render_quota_markdown(result)
+
+
+def test_observation_metadata_does_not_change_internal_execution_plan():
+    result = build_quota_plan({"ok": True, "goal_filter": "missing",
+                               "projection_envelope": {"served_from_cache": True}}, mode="should-run")
+    assert result["ok"] is True
+    assert "goal_filter" not in result
+    assert "status" not in result
+    assert "status_projection_envelope" not in result