fix(quota): preserve authority recovery in status and plan

loopx/cli_commands/quota_failure_report.py

@@ -87,6 +87,13 @@ def quota_failure_payload(
         if isinstance(error, EffectRuntimeHostPermissionError)
         else "fix quota/status collection before spending automatic compute"
     )
+    # Observation and execution callers consume the same authority rejection.
+    # Preserve its typed recovery instead of losing it at the early read return.
+    authority_failure = (
+        {"error_code": error.code, "reason": str(error), **error.payload}
+        if isinstance(error, (LegacyCoordinationWriterFenced, LocalCoordinationAuthorityUnavailable))
+        else {}
+    )
     if command not in QUOTA_EVENT_KINDS:
         return {
             "ok": False,
@@ -114,6 +121,7 @@ def quota_failure_payload(
             ],
             **verbose_debug,
             **lock_timeout_fields,
+            **authority_failure,
         }
 
     # A managed-runtime failure already carries a bounded, public-safe message
@@ -223,13 +231,7 @@ def quota_failure_payload(
         if error.agent_id is not None:
             payload["agent_id"] = error.agent_id
     elif isinstance(error, (LegacyCoordinationWriterFenced, LocalCoordinationAuthorityUnavailable)):
-        payload.update(
-            {
-                "error_code": error.code,
-                "reason": str(error),
-                **error.payload,
-            }
-        )
+        payload.update(authority_failure)
     if lock_timeout_fields:
         payload["recommended_action"] = "inspect the lock holder before retrying"
     if command == "monitor-poll":

tests/control_plane/test_quota_observation_scope.py

@@ -47,6 +47,7 @@ def goals(tmp_path, request):
     return registry, runtime, project
 
 
+@pytest.mark.parametrize("goals", ["file", "sqlite"], indirect=True)
 @pytest.mark.parametrize("mode", ["status", "plan"])
 def test_selected_goal_excludes_unrelated_broken_goal(goals, mode):
     registry, runtime, project = goals
@@ -67,6 +68,27 @@ def test_selected_goal_excludes_unrelated_broken_goal(goals, mode):
     assert {row["goal_id"] for group in global_payload["groups"].values() for row in group} == {"selected", "unrelated"}
 
 
+@pytest.mark.parametrize("goals", ["legacy"], indirect=True)
+@pytest.mark.parametrize("mode", ["status", "plan"])
+def test_unmigrated_selected_goal_retains_authority_recovery(goals, mode):
+    registry, runtime, project = goals
+    before = (project / "state.md").read_bytes()
+    code, payload = run_json_cli_result("quota", mode, "--goal-id", "selected",
+                                        registry_path=registry, runtime_root=runtime)
+    assert code == 1 and payload["ok"] is False, payload
+    assert payload["error_code"] == "canonical_authority_required"
+    assert payload["goal_id"] == "selected"
+    assert payload["recovery"] == {"action": "migrate_goal_authority", "backup_required": True}
+    assert payload["write_admitted"] is False
+    assert payload["decision_read_from_provider"] is False
+    assert payload["legacy_fallback_used"] is False
+    assert not payload["groups"]
+    assert "back up and migrate" in render_quota_markdown(payload)
+    assert (project / "state.md").read_bytes() == before
+    assert not (runtime / "authority").exists()
+    assert not list((runtime / "goals" / "selected" / "runs").glob("*.json*"))
+
+
 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",

tests/test_cli_argument_diagnostics.py

@@ -349,7 +349,9 @@ def test_deprecated_scheduler_detail_alias_is_hidden_from_help(
     assert "--include-scheduler-detail" not in help_text
 
 
+@pytest.mark.parametrize("mode", ["status", "plan"])
 def test_quota_collection_failure_payload_is_path_free_by_default(
+    mode: str,
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
@@ -360,7 +362,7 @@ def fail_collect_status(**_kwargs: object) -> dict[str, object]:
 
     monkeypatch.setattr(quota_command, "collect_status", fail_collect_status)
 
-    exit_code = main(["--format", "json", "quota", "status"])
+    exit_code = main(["--format", "json", "quota", mode])
 
     assert exit_code == 1
     payload = json.loads(capsys.readouterr().out)