refactor(quota): construct settled replay without action cleanup

loopx/cli_commands/quota.py

@@ -209,8 +209,8 @@ def _apply_requested_quota_action_selection_preflight(
     requested_todo_id: str | None,
     receipt_bound_todo_id: str | None,
     receipt_bound_replan_obligation_id: str | None,
-    receipt_pending_action_todo_id: str | None,
-    receipt_identity_upgraded: bool,
+    receipt_pending_action_todo_id: str | None = None,
+    receipt_identity_upgraded: bool = False,
 ) -> bool:
     if not requested_todo_id:
         return False

loopx/control_plane/quota/_supporting_projections.py

@@ -0,0 +1,136 @@
+"""Supporting diagnostics shared by active and settled quota packets."""
+from __future__ import annotations
+
+from typing import Any
+
+from ..agents.agent_scope import _action_scope_tokens_from_text
+from ..runtime.decision_freshness import decision_freshness_warning as _decision_freshness_warning
+from ..runtime.promotion_readiness import promotion_readiness_warning as _promotion_readiness_warning
+from ..todos.user_gate import build_gate_prompt as _build_gate_prompt
+from .recent_runs import goal_latest_runs as _goal_latest_runs
+
+
+def _recent_reward_lessons(status_payload: dict[str, Any], *, goal_id: str) -> list[dict[str, Any]]:
+    lessons: list[dict[str, Any]] = []
+    for run in _goal_latest_runs(status_payload, goal_id=goal_id):
+        reward = run.get("human_reward") if isinstance(run.get("human_reward"), dict) else {}
+        lesson = reward.get("lesson") if isinstance(reward.get("lesson"), dict) else {}
+        if not lesson:
+            continue
+        lessons.append(
+            {
+                "generated_at": run.get("generated_at"),
+                "decision": reward.get("decision"),
+                "reward": reward.get("reward"),
+                "kind": lesson.get("kind"),
+                "summary": lesson.get("summary"),
+                "avoid": lesson.get("avoid") if isinstance(lesson.get("avoid"), list) else [],
+                "prefer": lesson.get("prefer") if isinstance(lesson.get("prefer"), list) else [],
+            }
+        )
+    return lessons
+
+
+def _reward_lesson_projection_warning(
+    status_payload: dict[str, Any],
+    *,
+    goal_id: str,
+    recommended_action: str | None,
+) -> dict[str, Any] | None:
+    action = str(recommended_action or "").strip()
+    if not action:
+        return None
+    action_lower = action.lower()
+    action_tokens = _action_scope_tokens_from_text(action)
+    matches: list[dict[str, Any]] = []
+    for lesson in _recent_reward_lessons(status_payload, goal_id=goal_id):
+        for avoid in lesson.get("avoid") or []:
+            avoid_text = str(avoid or "").strip()
+            if not avoid_text:
+                continue
+            avoid_tokens = _action_scope_tokens_from_text(avoid_text)
+            exact_match = avoid_text.lower() in action_lower
+            if not exact_match and not avoid_tokens:
+                continue
+            token_overlap = sorted(action_tokens & avoid_tokens)
+            if not exact_match and len(token_overlap) < min(2, len(avoid_tokens)):
+                continue
+            matches.append(
+                {
+                    "generated_at": lesson.get("generated_at"),
+                    "decision": lesson.get("decision"),
+                    "kind": lesson.get("kind"),
+                    "summary": lesson.get("summary"),
+                    "avoid": avoid_text,
+                    "token_overlap": token_overlap[:5],
+                }
+            )
+    if not matches:
+        return None
+    return {
+        "schema_version": "reward_lesson_projection_warning_v0",
+        "source": "run_history.human_reward.lesson",
+        "goal_id": goal_id,
+        "message": (
+            "recommended_action overlaps a recent human_reward lesson avoid rule; "
+            "rebase the route or update the affected todo/next action before continuing"
+        ),
+        "recommended_action": action,
+        "match_count": len(matches),
+        "matches": matches[:3],
+    }
+
+
+def _attach_truthy_fields(payload: dict[str, Any], **fields: Any) -> None:
+    payload.update({key: value for key, value in fields.items() if value})
+
+
+def _dict_field(payload: dict[str, Any], key: str) -> dict[str, Any] | None:
+    return payload.get(key) if isinstance(payload.get(key), dict) else None
+
+
+def _attach_quota_supporting_projections(
+    payload: dict[str, Any],
+    *,
+    status_payload: dict[str, Any],
+    item: dict[str, Any],
+    project_asset: dict[str, Any],
+    goal_id: str,
+    selected_recommended_action: Any,
+    state: str,
+    user_todo_summary: dict[str, Any] | None,
+    should_run: bool,
+    state_action_projection_warning: dict[str, Any] | None,
+    next_action_warning: dict[str, Any] | None,
+    replan_obligation: dict[str, Any] | None,
+    notify_gate: bool = True,
+) -> None:
+    _attach_truthy_fields(
+        payload,
+        stale_latest_run_warning=_dict_field(item, "stale_latest_run_warning"),
+        state_action_projection_warning=state_action_projection_warning,
+        next_action_projection_warning=next_action_warning,
+        backlog_hygiene_warning=_dict_field(item, "backlog_hygiene_warning"),
+        completed_todo_archive_warning=_dict_field(item, "completed_todo_archive_warning"),
+        autonomous_replan_obligation=replan_obligation,
+        dreaming_proposal=_dict_field(item, "dreaming_proposal"),
+        dreaming_lane_badge=_dict_field(item, "dreaming_lane_badge"),
+        interface_budget_cadence=_dict_field(project_asset, "interface_budget_cadence"),
+        decision_freshness_warning=_decision_freshness_warning(status_payload, goal_id=goal_id),
+        promotion_readiness_warning=_promotion_readiness_warning(status_payload),
+        reward_lesson_projection_warning=_reward_lesson_projection_warning(
+            status_payload,
+            goal_id=goal_id,
+            recommended_action=selected_recommended_action,
+        ),
+    )
+    if state == "operator_gate" and (
+        gate_prompt := _build_gate_prompt(item, user_todo_summary=user_todo_summary)
+    ):
+        payload["gate_prompt"] = gate_prompt
+        if notify_gate:
+            payload["notify_user_on_gate"] = True
+    _attach_truthy_fields(
+        payload, next_handoff_condition=item.get("next_handoff_condition"),
+        agent_command=item.get("agent_command") if should_run else None,
+    )

loopx/control_plane/quota/live_decision.py

@@ -13,10 +13,10 @@
     InteractionProjectionHookRegistration,
     dispatch_interaction_projection_hooks,
 )
+from .effect_program import ReceiptBoundReplayPhase
 from .settlement import (
     read_heartbeat_settlement,
 )
-from .effect_program import ReceiptBoundReplayPhase
 from ..work_items.interaction_contract import (
     build_interaction_contract,
     build_protocol_action_packet,

loopx/control_plane/quota/settlement_precedence.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 from .effective_action import EffectiveAction
 
-from typing import Any, Protocol
+from typing import Any
 
-from ..effect_program import ReceiptBoundReplayPhase
 
 
 HEARTBEAT_SETTLED_REPLAY_REASON = (
@@ -40,66 +39,6 @@
 )
 
 
-class SettledReplayRoute(Protocol):
-    normal_delivery_allowed: bool
-    recovery_allowed: bool
-    self_repair_allowed: bool
-    capability_repair_allowed: bool
-    workspace_repair_allowed: bool
-    should_run: bool
-    effective_action: str
-    reason: str
-    replan_decision_allowed: bool
-    receipt_bound_replan_decision: bool
-    heartbeat_recommendation: dict[str, Any]
-    external_evidence_observation: dict[str, Any] | None
-    external_evidence_observation_recent: dict[str, Any] | None
-    selected_recommended_action: Any
-    agent_lane_next_action: dict[str, Any] | None
-    agent_scope_frontier: dict[str, Any] | None
-    agent_lane_frontier_hint: dict[str, Any] | None
-    state_action_projection_warning: dict[str, Any] | None
-    next_action_warning: dict[str, Any] | None
-    goal_route_hint: dict[str, Any] | None
-
-
-def apply_settled_replay_route_precedence(
-    route: SettledReplayRoute,
-    *,
-    replay_phase: ReceiptBoundReplayPhase | None,
-) -> None:
-    """Prevent a settled heartbeat identity from selecting work in the same turn."""
-
-    if replay_phase is not ReceiptBoundReplayPhase.SETTLED:
-        return
-    route.normal_delivery_allowed = False
-    route.recovery_allowed = False
-    route.self_repair_allowed = False
-    route.capability_repair_allowed = False
-    route.workspace_repair_allowed = False
-    route.should_run = False
-    route.effective_action = EffectiveAction.HEARTBEAT_SETTLED_SKIP.value
-    route.reason = HEARTBEAT_SETTLED_REPLAY_REASON
-    route.replan_decision_allowed = False
-    route.receipt_bound_replan_decision = False
-    route.heartbeat_recommendation = {
-        **route.heartbeat_recommendation,
-        "recommended_mode": route.effective_action,
-        "notify": "DONT_NOTIFY",
-        "reason": route.reason,
-        "spend_policy": "no quota spend for an already-settled heartbeat turn",
-    }
-    route.external_evidence_observation = None
-    route.external_evidence_observation_recent = None
-    route.selected_recommended_action = route.reason
-    route.agent_lane_next_action = None
-    route.agent_scope_frontier = None
-    route.agent_lane_frontier_hint = None
-    route.state_action_projection_warning = None
-    route.next_action_warning = None
-    route.goal_route_hint = None
-
-
 def clear_quota_action_projections(
     payload: dict[str, Any],
     *,
@@ -109,72 +48,44 @@ def clear_quota_action_projections(
         payload.pop(key, None)
 
 
-def apply_settled_replay_payload_precedence(
-    payload: dict[str, Any],
-    *,
-    replay_phase: ReceiptBoundReplayPhase | None,
-) -> None:
-    """Keep late supporting projections from reopening a settled heartbeat turn."""
-
-    if replay_phase is not ReceiptBoundReplayPhase.SETTLED:
-        return
+def settled_replay_fields() -> dict[str, Any]:
+    """Construct the authority fields of a verified, already-settled Turn."""
     reason = HEARTBEAT_SETTLED_REPLAY_REASON
-    payload.update(
-        {
-            "decision": "skip",
-            "should_run": False,
-            "normal_delivery_allowed": False,
-            "recovery_delivery_allowed": False,
-            "self_repair_allowed": False,
-            "capability_repair_allowed": False,
-            "workspace_repair_allowed": False,
-            "effective_action": EffectiveAction.HEARTBEAT_SETTLED_SKIP.value,
-            "actionable_by_codex": False,
-            "reason": reason,
-            "requires_user_action": False,
-            "recommended_action": (
-                "Finish this heartbeat without another action; use a fresh turn "
-                "identity for successor selection."
-            ),
-            "heartbeat_recommendation": {
-                "recommended_mode": "heartbeat_settled_skip",
-                "notify": "DONT_NOTIFY",
-                "reason": reason,
-                "spend_policy": "no quota spend for an already-settled heartbeat turn",
-                "agent_must_attempt": False,
-            },
-            "execution_obligation": {
-                "must_attempt_work": False,
-                "kind": "heartbeat_settled_skip",
-                "delivery_allowed": False,
-                "notify_is_execution_gate": False,
-                "reason": reason,
-                "spend_policy": "no quota spend for an already-settled heartbeat turn",
-            },
-        }
-    )
-    frontier = payload.get("goal_frontier_projection")
-    if isinstance(frontier, dict):
-        payload["goal_frontier_projection"] = {
-            key: value
-            for key, value in frontier.items()
-            if key
-            not in {
-                "autonomous_replan_decision",
-                "replan_ack_feedback",
-                "replan_obligation",
-                "vision_continuation_audit",
-                "vision_wait_state",
-            }
-        }
-    clear_quota_action_projections(
-        payload,
-        additional_keys=(
-            "boundary_projection_gap",
-            "operator_question",
-            "selected_todo",
-            "state_projection_gap",
-            "task_orchestration_contract",
-            "todo_write_hint",
+    return {
+        "decision": "skip",
+        "should_run": False,
+        "normal_delivery_allowed": False,
+        "recovery_delivery_allowed": False,
+        "self_repair_allowed": False,
+        "capability_repair_allowed": False,
+        "workspace_repair_allowed": False,
+        # A settled Turn grants no safe bypass: the heartbeat task body reads
+        # safe_bypass_allowed as permission to run a bounded step and spend, so
+        # a fresh Turn must recompute any fallback instead of inheriting one.
+        "safe_bypass_allowed": False,
+        "safe_bypass_kind": None,
+        "safe_bypass_policy": None,
+        "effective_action": EffectiveAction.HEARTBEAT_SETTLED_SKIP.value,
+        "actionable_by_codex": False,
+        "reason": reason,
+        "requires_user_action": False,
+        "recommended_action": (
+            "Finish this heartbeat without another action; use a fresh turn "
+            "identity for successor selection."
         ),
-    )
+        "heartbeat_recommendation": {
+            "recommended_mode": "heartbeat_settled_skip",
+            "notify": "DONT_NOTIFY",
+            "reason": reason,
+            "spend_policy": "no quota spend for an already-settled heartbeat turn",
+            "agent_must_attempt": False,
+        },
+        "execution_obligation": {
+            "must_attempt_work": False,
+            "kind": "heartbeat_settled_skip",
+            "delivery_allowed": False,
+            "notify_is_execution_gate": False,
+            "reason": reason,
+            "spend_policy": "no quota spend for an already-settled heartbeat turn",
+        },
+    }