From 0caa350456f7d85891c0038d059ca350d83d1208 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 27 Oct 2026 15:37:39 +0800
Subject: [PATCH] fix(quota): preserve authority recovery in observation
 failures

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota_failure_report.py    | 16 ++++++++------
 .../test_quota_observation_scope.py           | 22 +++++++++++++++++++
 tests/test_cli_argument_diagnostics.py        |  4 +++-
 3 files changed, 34 insertions(+), 8 deletions(-)

diff --git a/loopx/cli_commands/quota_failure_report.py b/loopx/cli_commands/quota_failure_report.py
index 69df8bb43e..5e04b82dc1 100644
--- a/loopx/cli_commands/quota_failure_report.py
+++ b/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":
diff --git a/tests/control_plane/test_quota_observation_scope.py b/tests/control_plane/test_quota_observation_scope.py
index d28123de3d..8602f35858 100644
--- a/tests/control_plane/test_quota_observation_scope.py
+++ b/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",
diff --git a/tests/test_cli_argument_diagnostics.py b/tests/test_cli_argument_diagnostics.py
index 3c7e312ad8..4171dea543 100644
--- a/tests/test_cli_argument_diagnostics.py
+++ b/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)