Skip inactive Anti-Entropy validation in turn context

loopx/capabilities/anti_entropy/README.md

@@ -80,10 +80,9 @@ task is an explicit choice. When enabled, one closed loop per task is:
      that is a task-scoped typed enable, not a Goal-wide switch. A Goal-level
      `available_capabilities` declaration is availability only and does not
      activate unrelated or later-added Todos in the same Goal;
-   - the turn-context hook itself is composed unconditionally in the shared
-     `attach_turn_context` owner and is called by both `turn run-once
-     --execute` and the heartbeat `quota should-run` path; it decides
-     off/applicable per task and injects nothing when the task has not opted in;
+   - both `turn run-once --execute` and heartbeat `quota should-run` use
+     `attach_turn_context`. The Anti-Entropy provider declares a readback slot
+     when the task enables it, and the shared owner dispatches those registrations;
    - the post-change workspace is the Goal repository from the registry
      (`load_goal_from_registry`), never the process working directory.
 

loopx/capabilities/anti_entropy/catalog_entry.py

@@ -94,7 +94,7 @@
     "docs": [_README],
     "boundaries": [
         "The capability is default-off and task-scoped: a turn receives the retirement context only when the task explicitly declares a retirement basis or enables the capability; disabled turns get zero injection.",
-        "The turn-context hook is composed unconditionally into the shared attach_turn_context owner (turn run-once --execute and heartbeat quota should-run paths) and injects declared targets, the inspect descriptor, and delete-first guidance only for a task that opted in; it grants no execution authority.",
+        "The shared attach_turn_context owner dispatches the hook for enabled tasks in turn run-once --execute and heartbeat quota should-run. Applicable tasks receive declared targets, the inspect descriptor and delete-first guidance.",
         "Task intent is an explicit retirement_basis / anti_entropy_basis namespace in the selected task's own text/note/description/instruction carriers, or a task-scoped typed enable (Todo required_capabilities / direct flag); generic target_paths or target_symbols in metadata/params are not intent, and a Goal-level available-capabilities declaration is availability, not activation.",
         "LoopX has no capability-owned host tool registry: the inspect descriptor is projected as turn-context data, and the CLI (or the Python entry it wraps) remains the callable route.",
         "The CLI (verify/inspect) remains the shared verification entry for humans and agents; final delivery acceptance is owned by the existing change-quality prepare/record/verify owner.",

loopx/capabilities/anti_entropy/turn_context.py

@@ -556,11 +556,12 @@ def anti_entropy_context_registration(
     and the Goal registry. Disabled turns declare no readback slot, so no empty
     guidance field is written into the host payload.
     """
-    workspace = _resolve_workspace(
-        payload, registry_path=registry_path, goal_id=goal_id
-    )
     basis = _resolve_retirement_basis(payload, decision)
     enabled = _is_anti_entropy_enabled(payload, decision, basis)
+    workspace = (
+        _resolve_workspace(payload, registry_path=registry_path, goal_id=goal_id)
+        if enabled else None
+    )
 
     def produce() -> Mapping[str, Any]:
         return anti_entropy_context_result(

loopx/capabilities/turn_context.py

@@ -27,27 +27,26 @@ def attach_turn_context(
     decision: Mapping[str, Any],
     turn_instance_id: str,
 ) -> None:
-    dispatch_turn_context_hooks(
+    memory = memory_context_registration(
         payload,
-        (
-            memory_context_registration(
-                payload,
-                registry_path=registry_path,
-                goal_id=goal_id,
-                agent_id=agent_id,
-                decision=decision,
-                turn_instance_id=turn_instance_id,
-            ),
-            anti_entropy_context_registration(
-                payload,
-                registry_path=registry_path,
-                goal_id=goal_id,
-                agent_id=agent_id,
-                decision=decision,
-                turn_instance_id=turn_instance_id,
-            ),
-        ),
+        registry_path=registry_path,
+        goal_id=goal_id,
+        agent_id=agent_id,
+        decision=decision,
+        turn_instance_id=turn_instance_id,
+    )
+    anti_entropy = anti_entropy_context_registration(
+        payload,
+        registry_path=registry_path,
+        goal_id=goal_id,
+        agent_id=agent_id,
+        decision=decision,
+        turn_instance_id=turn_instance_id,
     )
+    # The provider declares a slot for every enabled task, including when it
+    # needs to report that no retirement targets apply.
+    registrations = (memory, anti_entropy) if anti_entropy.readback_slots else (memory,)
+    dispatch_turn_context_hooks(payload, registrations)
 
 
 def configured_post_settlement(

tests/capabilities/test_anti_entropy_feedback.py

@@ -452,16 +452,33 @@ def test_turn_context_hook_lifecycle_all_states(tmp_path: Path) -> None:
     assert any("anti-entropy" in inst.lower() for inst in projected.get("capability_instructions", []))
 
 
-def test_attach_turn_context_integration(tmp_path: Path) -> None:
+@pytest.mark.parametrize(("intent", "status"), [
+    ({}, None),
+    ({"anti_entropy_enabled": True}, "inapplicable"),
+    ({"retirement_basis": {"target_symbols": ["old_api"], "rationale": "replaced"}}, "active"),
+    ({"retirement_basis": {"target_symbols": ["old_api"]}, "anti_entropy_progress_fact": True}, "recovering"),
+], ids=["disabled", "enabled", "active", "recovery"])
+def test_attach_turn_context_integration(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, intent: dict[str, object], status: str | None,
+) -> None:
     """Verify loopx.capabilities.turn_context.attach_turn_context wires anti-entropy with memory."""
     from loopx.capabilities.turn_context import attach_turn_context
+    from loopx.control_plane import capability_hooks
+
+    invoke = capability_hooks.effect_runtime_result
+    methods = []
+
+    def observe(method, params, **kwargs):
+        if (params.get("registration") or {}).get("hook_id") == "anti-entropy.turn-context":
+            methods.append(method)
+        return invoke(method, params, **kwargs)
 
-    basis_data = {"target_symbols": ["old_api"], "rationale": "replaced"}
+    monkeypatch.setattr(capability_hooks, "effect_runtime_result", observe)
     payload = {
         "goal_id": "g-1",
         "agent_id": "a-1",
-        "retirement_basis": basis_data,
         "workspace": str(tmp_path),
+        **intent,
     }
 
     attach_turn_context(
@@ -473,9 +490,17 @@ def test_attach_turn_context_integration(tmp_path: Path) -> None:
         turn_instance_id="turn-42",
     )
 
-    assert "anti_entropy_guidance" in payload
-    assert payload["anti_entropy_guidance"]["status"] == "active"
-    assert payload["anti_entropy_guidance"]["target_symbols"] == ["old_api"]
+    if status is None:
+        assert "anti_entropy_guidance" not in payload
+        assert methods == []
+    else:
+        assert methods == ["capability_hook.turn_context.validate_registration", "capability_hook.turn_context.validate"]
+        guidance = payload["anti_entropy_guidance"]
+        if status == "inapplicable":
+            assert guidance == {"enabled": True, "applicable": False}
+        else:
+            assert guidance["status"] == status
+            assert guidance["target_symbols"] == ["old_api"]
 
 
 def test_feedback_cycle_pure_path_delivery_ready(tmp_path: Path) -> None:
@@ -860,7 +885,7 @@ def fake_attach(payload: dict, **kwargs: object) -> None:
         captured["payload"] = payload
         captured["kwargs"] = kwargs
 
-    monkeypatch.setattr(quota_reward_memory, "attach_turn_context", fake_attach)
+    monkeypatch.setattr("loopx.capabilities.turn_context.attach_turn_context", fake_attach)
     payload = {
         "ok": True,
         "should_run": True,
@@ -897,7 +922,7 @@ def fake_attach(payload: dict, **kwargs: object) -> None:
         nonlocal called
         called = True
 
-    monkeypatch.setattr(quota_reward_memory, "attach_turn_context", fake_attach)
+    monkeypatch.setattr("loopx.capabilities.turn_context.attach_turn_context", fake_attach)
     quota_reward_memory.attach_reward_memory_recall_after_should_run(
         {"ok": True, "should_run": False, "heartbeat_receipt": {"status": "committed"}},
         quota_command="should-run",

tests/control_plane/test_quota_memory_context.py

@@ -21,6 +21,7 @@
 )
 from tests.capabilities.test_agent_turn_recall import RecallProvider, active_record
 from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
 from tests.control_plane.reward_memory_host_fixture import enable_live_memory
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, _initialize_fixture_authority, _write_fixture,
@@ -32,7 +33,7 @@ def run_cli(args: list[str]) -> dict:
     with redirect_stdout(output):
         code = main(args)
     assert code == 0, output.getvalue()[:1200]
-    return json.loads(output.getvalue())
+    return read_cli_policy_payload(output.getvalue(), returncode=code)
 
 
 def quota_fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str):