[FIX] Remediation Assess Context-Limit Recovery (#4211)

## Summary

Fix the remediation recipe so context-exhausted `resolve-failures`
assess sessions
verify persisted partial fixes instead of failing the dispatch
immediately. The
pre-merge assess path routes context and rate-limit recovery back to
`test`, while
the merge-gate assess path routes recovery back to `merge_gate_test` so
each loop
keeps its existing bounded counter.

Closes #4206

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/fix-4206-remediation-assess-context-limit/.autoskillit/temp/make-plan/4206-live-plan.md`

🤖 Generated with [Claude Code](https://claude.com/claude-code) via
AutoSkillit

<!-- autoskillit:pipeline-signature
steps=prepare_pr,run_arch_lenses,compose_pr,annotate_pr_diff,review_pr
-->

Author: Trecek

docs/safety/hooks.md

@@ -148,10 +148,11 @@ table to the PR body so reviewers can see per-step token cost.
 
 ### `quota_post_hook.py`
 **Guarded tool:** `run_skill`
-After `run_skill` returns, appends a quota warning to the tool output when
+After `run_skill` returns, replaces the display output with a quota warning when
 the cached binding window marks `should_block=True` (per-window threshold —
 see `quota_guard.py` above). Gives the orchestrator a chance to back off
-voluntarily before the PreToolUse hook starts denying.
+voluntarily before the PreToolUse hook starts denying. The replacement output is
+generated by the hook and does not echo raw `tool_response` content.
 
 ### `review_gate_post_hook.py`
 **Guarded tools:** `run_skill`, `run_python`

src/autoskillit/hooks/quota_post_hook.py

@@ -2,8 +2,8 @@
 """PostToolUse hook: quota warning after run_skill execution.
 
 Fires after run_skill completes and checks whether the cached binding marks
-``should_block=True``. When set, replaces the tool output with a compact
-result summary + quota warning + sleep instruction via updatedMCPToolOutput.
+``should_block=True``. When set, replaces the tool output with a quota warning
+and sleep instruction via updatedMCPToolOutput.
 
 This script is stdlib-only so it can run under any Python interpreter without
 requiring the autoskillit package to be importable.
@@ -37,29 +37,6 @@
 QUOTA_POST_BUDGET_EXCEEDED_TRIGGER: str = "QUOTA BUDGET EXCEEDED"
 
 
-def _extract_run_skill_result(tool_response: str | dict) -> str:
-    """Extract a compact summary from the run_skill double-wrapped JSON response."""
-    try:
-        outer = json.loads(tool_response) if isinstance(tool_response, str) else tool_response
-        if isinstance(outer, dict) and "result" in outer:
-            inner_str = outer["result"]
-            if isinstance(inner_str, str):
-                try:
-                    inner = json.loads(inner_str)
-                    if isinstance(inner, dict):
-                        success = inner.get("success", "unknown")
-                        result_text = inner.get("result", "")
-                        if isinstance(result_text, str) and len(result_text) > 500:
-                            result_text = result_text[:500] + "..."
-                        return f"success: {success}\nresult: {result_text}"
-                except (json.JSONDecodeError, ValueError):
-                    pass
-                return inner_str[:500] if len(inner_str) > 500 else inner_str
-        return str(outer)[:500]
-    except (json.JSONDecodeError, ValueError, TypeError):
-        return str(tool_response)[:500]
-
-
 def main(*, cache_path_override: str | None = None) -> None:
     try:
         raw = sys.stdin.read()
@@ -71,8 +48,6 @@ def main(*, cache_path_override: str | None = None) -> None:
         sys.exit(0)
 
     tool_name = event.get("tool_name", "")
-    tool_response = event.get("tool_response") or ""
-
     settings = resolve_quota_settings(cache_path_override=cache_path_override)
     if settings.disabled:
         sys.exit(0)  # quota guard disabled for this session
@@ -140,8 +115,6 @@ def main(*, cache_path_override: str | None = None) -> None:
     else:
         n = settings.buffer_seconds
 
-    result_summary = _extract_run_skill_result(tool_response)
-
     session_deadline_str = os.environ.get("AUTOSKILLIT_SESSION_DEADLINE")
     budget_exceeded = False
     remaining_budget = float("inf")
@@ -157,7 +130,6 @@ def main(*, cache_path_override: str | None = None) -> None:
     if budget_exceeded:
         resets_at_display = resets_at_str or "unknown"
         warning_text = (
-            f"{result_summary}\n\n"
             f"{QUOTA_POST_BUDGET_EXCEEDED_TRIGGER}\n"
             f"Post-execution utilization: {utilization:.0f}% on window '{window_name}' "
             f"(threshold: {effective_threshold:.0f}%)\n"
@@ -173,7 +145,6 @@ def main(*, cache_path_override: str | None = None) -> None:
         )
     else:
         warning_text = (
-            f"{result_summary}\n\n"
             f"{QUOTA_POST_WARNING_TRIGGER}\n"
             f"Post-execution utilization: {utilization:.0f}% on window '{window_name}' "
             f"(threshold: {effective_threshold:.0f}%)\n"

src/autoskillit/recipes/diagrams/remediation.md

@@ -1,4 +1,4 @@
-<!-- autoskillit-recipe-hash: sha256:b46326a0d43bc9c41fa2615979dfbe9df18d296cd2e171be01348976a34d4b1a -->
+<!-- autoskillit-recipe-hash: sha256:c8205806500c54e98e1dd94e2f09f106c0b1603fcc60e2958ab70788c59a39c9 -->
 <!-- autoskillit-diagram-format: v7 -->
 ## remediation
 Investigate, rectify, implement, and merge a bug fix with CI and PR gates.
@@ -12,10 +12,13 @@ investigate → rectify → review_approach (optional)
 |
 dry_walkthrough → implement ↔ [retry_worktree on context limit]
 |
-test → assess
+test ↔ assess
+     [assess context/rate limit → test]
 |
 +-- audit_impl → reset_test_fix_counter → reset_merge_test_fix_counter
     → reset_ref_push_counter → pre_remediation_merge → remediate (optional)
+    → merge_gate_test ↔ merge_gate_assess
+       [merge_gate_assess context/rate limit → merge_gate_test]
 |
 make_plan → commit_guard → merge → push
 |

src/autoskillit/recipes/remediation.json

@@ -777,8 +777,8 @@
           "route": "release_issue_failure"
         }
       ],
-      "on_context_limit": "release_issue_failure",
-      "on_rate_limit": "release_issue_failure",
+      "on_context_limit": "test",
+      "on_rate_limit": "test",
       "on_failure": "release_issue_failure",
       "on_exhausted": "release_issue_failure",
       "optional": true,
@@ -855,8 +855,8 @@
           "route": "release_issue_failure"
         }
       ],
-      "on_context_limit": "release_issue_failure",
-      "on_rate_limit": "release_issue_failure",
+      "on_context_limit": "merge_gate_test",
+      "on_rate_limit": "merge_gate_test",
       "on_failure": "release_issue_failure",
       "on_exhausted": "release_issue_failure",
       "optional": true,

src/autoskillit/recipes/remediation.yaml

@@ -716,8 +716,8 @@ steps:
     - when: ${{ result.verdict }} == 'no_test_infrastructure'
       route: release_issue_failure
     - route: release_issue_failure
-    on_context_limit: release_issue_failure
-    on_rate_limit: release_issue_failure
+    on_context_limit: test
+    on_rate_limit: test
     on_failure: release_issue_failure
     on_exhausted: release_issue_failure
     optional: true
@@ -772,8 +772,8 @@ steps:
     - when: ${{ result.verdict }} == 'no_test_infrastructure'
       route: release_issue_failure
     - route: release_issue_failure
-    on_context_limit: release_issue_failure
-    on_rate_limit: release_issue_failure
+    on_context_limit: merge_gate_test
+    on_rate_limit: merge_gate_test
     on_failure: release_issue_failure
     on_exhausted: release_issue_failure
     optional: true

tests/hooks/test_quota_post_check.py

@@ -184,17 +184,18 @@ def test_qpc6_warning_contains_sleep_instruction(tmp_path):
     assert "time.sleep" in output
 
 
-# T7: Warning preserves run_skill result summary
-def test_qpc7_warning_preserves_result_summary(tmp_path):
-    """updatedMCPToolOutput includes the run_skill result (success/fail, key fields)."""
+# T7: Warning output is hook-generated only
+def test_qpc7_warning_does_not_forward_tool_response(tmp_path):
+    """updatedMCPToolOutput must not echo arbitrary raw tool_response content."""
     cache = tmp_path / "quota_cache.json"
     _write_cache(cache, utilization=90.0)
     event = _build_event(success=True, result_text="plan written")
     out, _ = _run_hook(event=event, cache_path=cache)
     data = json.loads(out)
     output = data["hookSpecificOutput"]["updatedMCPToolOutput"]
-    assert "success: True" in output
-    assert "plan written" in output
+    assert "QUOTA WARNING" in output
+    assert "success: True" not in output
+    assert "plan written" not in output
 
 
 # T8: JSONL event logging for post-check warning

tests/recipe/test_bundled_recipes_general.py

@@ -85,8 +85,10 @@ def test_pre_merge_test_fix_cycle_has_guard(recipe_name: str) -> None:
 
 
 @pytest.mark.parametrize("recipe_name", _PRE_MERGE_CYCLE_RECIPES)
-def test_pre_merge_fix_on_context_limit_routes_to_failure(recipe_name: str) -> None:
-    """fix/assess on_context_limit must route to release_issue_failure (safe default)."""
+def test_pre_merge_fix_on_context_limit_uses_expected_recovery(
+    recipe_name: str,
+) -> None:
+    """fix/assess context-limit routing must preserve resumable partial work."""
     recipe = load_recipe(_resolve_recipe_path(recipe_name))
     test_step = recipe.steps["test"]
     guard_name = test_step.on_failure
@@ -98,9 +100,10 @@ def test_pre_merge_fix_on_context_limit_routes_to_failure(recipe_name: str) -> N
     )
     assert fix_step_name is not None
     fix_step = recipe.steps[fix_step_name]
-    assert fix_step.on_context_limit == "release_issue_failure", (
+    expected_route = "test" if recipe_name == "remediation" else "release_issue_failure"
+    assert fix_step.on_context_limit == expected_route, (
         f"{recipe_name}: {fix_step_name}.on_context_limit must route to "
-        f"'release_issue_failure', got: {fix_step.on_context_limit!r}"
+        f"{expected_route!r}, got: {fix_step.on_context_limit!r}"
     )
 
 

tests/recipe/test_bundled_recipes_pipeline_structure.py

@@ -1005,13 +1005,25 @@ def test_remediation_no_add_dir_dead_param(self, recipe) -> None:
             f"{[(f.step_name, f.message) for f in add_dir_dead]}"
         )
 
-    def test_remediation_assess_step_on_context_limit_routes_to_failure(self, recipe) -> None:
-        """assess step must route on_context_limit to release_issue_failure (safe default)."""
+    def test_remediation_assess_step_on_context_limit_routes_to_test(self, recipe) -> None:
+        """assess context exhaustion must verify persisted partial fixes."""
         assess = recipe.steps["assess"]
-        assert assess.on_context_limit == "release_issue_failure", (
-            f"remediation.yaml assess step must declare on_context_limit: release_issue_failure, "
+        assert assess.on_context_limit == "test", (
+            f"remediation.yaml assess step must declare on_context_limit: test, "
             f"got: {assess.on_context_limit!r}"
         )
+        assert assess.on_rate_limit == assess.on_context_limit
+
+    def test_remediation_merge_gate_assess_context_limit_routes_to_merge_gate_test(
+        self, recipe
+    ) -> None:
+        """merge-gate assess context exhaustion must stay on the merge-gate loop."""
+        assess = recipe.steps["merge_gate_assess"]
+        assert assess.on_context_limit == "merge_gate_test", (
+            "remediation.yaml merge_gate_assess step must declare "
+            f"on_context_limit: merge_gate_test, got: {assess.on_context_limit!r}"
+        )
+        assert assess.on_rate_limit == assess.on_context_limit
 
 
 class TestDisplayConditionContract: