refactor(quota): share inbox admission construction

loopx/chat_server.py

@@ -1055,18 +1055,21 @@ def _action_preview(self) -> None:
             status=201,
         )
 
+    def _action_not_found(self) -> None:
+        self._send_error(
+            "typed Chat action proposal was not found",
+            status=404,
+            error_code="action_not_found",
+        )
+
     def _action_snapshot(self, proposal_id: str) -> None:
         try:
             proposal = self.server.action_service.load(proposal_id)
         except ValueError as exc:
             self._send_error(str(exc), status=400, error_code="invalid_proposal_id")
             return
         if proposal is None:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         self._send_json(
             {
@@ -1107,11 +1110,7 @@ def _action_cancel(self, proposal_id: str) -> None:
                 raise ValueError("action cancel request must be empty")
             proposal = self.server.action_service.cancel(proposal_id)
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1144,11 +1143,7 @@ def _action_transition(self, proposal_id: str, transition: str) -> None:
             else:
                 raise ValueError("unsupported action transition")
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1190,11 +1185,7 @@ def _action_apply(self, proposal_id: str) -> None:
             )
             return
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")

loopx/control_plane/quota/decision_summary.py

@@ -328,29 +328,24 @@ def resolve_quota_run_decision(
     if automation_prompt_upgrade_required and not terminal_no_followup:
         should_run = False
         effective_action = EffectiveAction.AUTOMATION_PROMPT_UPGRADE_REQUIRED.value
-    elif inbox_reply_due:
+    elif inbox_priority_due:
         should_run = True
         normal_delivery_allowed = True
         recovery_delivery_allowed = False
         self_repair_allowed = False
         capability_repair_allowed = False
         workspace_repair_allowed = False
-        effective_action = EffectiveAction.LARK_INBOX_REPLY_DUE.value
-        reason = (
-            "a direct Lark question, bot mention, or verified reply to the bot "
-            "is pending reply"
-        )
-    elif inbox_material_review_due:
-        should_run = True
-        normal_delivery_allowed = True
-        recovery_delivery_allowed = False
-        self_repair_allowed = False
-        capability_repair_allowed = False
-        workspace_repair_allowed = False
-        effective_action = EffectiveAction.OPERATOR_INBOX_MATERIAL_REVIEW_DUE.value
-        reason = (
-            "captured unaddressed operator-inbox material is pending bounded review"
-        )
+        if inbox_reply_due:
+            effective_action = EffectiveAction.LARK_INBOX_REPLY_DUE.value
+            reason = (
+                "a direct Lark question, bot mention, or verified reply to the bot "
+                "is pending reply"
+            )
+        else:
+            effective_action = EffectiveAction.OPERATOR_INBOX_MATERIAL_REVIEW_DUE.value
+            reason = (
+                "captured unaddressed operator-inbox material is pending bounded review"
+            )
 
     effective_action, reason = _task_orchestration_effective_action(
         task_orchestration_contract,

loopx/extensions/lark/goal_topic_connections.py

@@ -92,6 +92,7 @@
     IngressMode,
     ReplyMode,
     _routing_value,
+    _connection_routing_modes,
     decide_lark_topic_route_event,
 )
 from .presentation.kanban import (
@@ -1035,29 +1036,7 @@ def list_lark_connections(
             )
             connector_status: dict[str, Any] | None = None
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 raw_connector = binding.get("connector")
                 if raw_connector is not None:
                     if not isinstance(raw_connector, Mapping):
@@ -1216,29 +1195,7 @@ def decide_lark_topic_event(
                 else {}
             )
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 connector = binding.get("connector")
                 if connector is not None:
                     if not isinstance(connector, Mapping):

loopx/extensions/lark/goal_topic_routing.py

@@ -47,6 +47,37 @@ def _routing_value(
         raise ValueError(f"{field} must be one of: {allowed}") from exc
 
 
+def _connection_routing_modes(
+    routing: Mapping[str, Any],
+) -> tuple[str, str, str]:
+    """Normalize persisted modes for both connection readback and event routing."""
+
+    capture_scope = _routing_value(
+        CaptureScope,
+        routing.get("capture_scope")
+        or (
+            "configured_chat_all"
+            if routing.get("incoming_mode") == "all"
+            else "addressed_only"
+        ),
+        default=CaptureScope.ADDRESSED_ONLY.value,
+        field="capture_scope",
+    )
+    ingress_mode = _routing_value(
+        IngressMode,
+        routing.get("ingress_mode"),
+        default=IngressMode.DIRECT_SESSION.value,
+        field="ingress_mode",
+    )
+    reply_mode = _routing_value(
+        ReplyMode,
+        routing.get("reply_mode"),
+        default=ReplyMode.TOPIC_REPLY.value,
+        field="reply_mode",
+    )
+    return capture_scope, ingress_mode, reply_mode
+
+
 def _normalize_mention_name(name: str) -> str:
     cleaned = str(name or "").strip()
     if cleaned.startswith("@"):

tests/control_plane/test_quota_run_decision.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+import pytest
+
 from loopx.control_plane.goals.goal_frontier import (
     GOAL_TERMINAL_SOURCE_COMPLETENESS_SCHEMA_VERSION,
     GOAL_TERMINAL_STATE_SCHEMA_VERSION,
@@ -108,8 +110,8 @@ def test_replan_precedes_normal_delivery() -> None:
     assert decision.effective_action == "autonomous_replan_required"
 
 
-def test_terminal_closure_precedes_automation_upgrade() -> None:
-    projection = {
+def _terminal_projection() -> dict[str, Any]:
+    return {
         "terminal_state": {
             "schema_version": GOAL_TERMINAL_STATE_SCHEMA_VERSION,
             "kind": "no_followup",
@@ -146,10 +148,13 @@ def test_terminal_closure_precedes_automation_upgrade() -> None:
         "autonomy_blockers": [],
         "replan_required": False,
     }
+
+
+def test_terminal_closure_precedes_automation_upgrade() -> None:
     decision = _resolve(
         automation_prompt_upgrade={"reason": "refresh prompt identity"},
         automation_prompt_upgrade_required=True,
-        goal_frontier_projection=projection,
+        goal_frontier_projection=_terminal_projection(),
     )
 
     assert decision.should_run is False
@@ -168,3 +173,69 @@ def test_task_orchestration_refines_normal_run_action() -> None:
     assert decision.should_run is True
     assert decision.normal_delivery_allowed is True
     assert decision.effective_action == "coordinate_task_bundle"
+
+
+@pytest.mark.parametrize(
+    "reply,material,action,reason",
+    [
+        (
+            True,
+            False,
+            "lark_inbox_reply_due",
+            "a direct Lark question, bot mention, or verified reply to the bot is pending reply",
+        ),
+        (
+            False,
+            True,
+            "operator_inbox_material_review_due",
+            "captured unaddressed operator-inbox material is pending bounded review",
+        ),
+        (
+            True,
+            True,
+            "lark_inbox_reply_due",
+            "a direct Lark question, bot mention, or verified reply to the bot is pending reply",
+        ),
+    ],
+)
+@pytest.mark.parametrize("terminal", [False, True])
+@pytest.mark.parametrize("upgrade", [False, True])
+def test_inbox_admission_preserves_guard_and_source_priority(
+    reply: bool,
+    material: bool,
+    action: str,
+    reason: str,
+    terminal: bool,
+    upgrade: bool,
+) -> None:
+    decision = _resolve(
+        normal_delivery_allowed=False,
+        recovery_delivery_allowed=True,
+        self_repair_allowed=True,
+        capability_gate={"action": "repair_bridge"},
+        workspace_guard={"reason": "workspace repair"},
+        inbox_reply_due=reply,
+        inbox_material_review_due=material,
+        goal_frontier_projection=_terminal_projection() if terminal else None,
+        automation_prompt_upgrade_required=upgrade,
+        automation_prompt_upgrade={"reason": "refresh prompt identity"},
+        replan_obligation={"required": True, "agent_id": "agent-a"},
+        task_orchestration_contract={"execution_state": "ready", "mode": "adaptive"},
+    )
+
+    # A nonterminal prompt upgrade wins; terminal inbox work retains the
+    # existing exception. Neither inbox source becomes coordinator work.
+    blocked = upgrade and not terminal
+    assert decision.should_run is (not blocked)
+    assert decision.normal_delivery_allowed is (not blocked)
+    assert decision.effective_action == (
+        "automation_prompt_upgrade_required" if blocked else action
+    )
+    assert decision.reason == ("refresh prompt identity" if blocked else reason)
+    assert decision.recovery_delivery_allowed is False
+    assert decision.self_repair_allowed is False
+    assert decision.capability_repair_allowed is False
+    assert decision.workspace_repair_allowed is False
+    assert decision.replan_decision_allowed is False
+    assert decision.state == "eligible"
+    assert decision.quota == {"state": "eligible"}

tests/extensions/test_lark_goal_topic_connections.py

@@ -2122,6 +2122,75 @@ def _prep_goal_channel_target(root: Path) -> Path:
     return target_path
 
 
+@pytest.mark.parametrize(
+    ("routing", "expected"),
+    [
+        ({}, ("addressed_only", "direct_session", "topic_reply")),
+        (
+            {"incoming_mode": "all"},
+            ("configured_chat_all", "direct_session", "topic_reply"),
+        ),
+        (
+            {
+                "incoming_mode": "all",
+                "capture_scope": " ADDRESSED_ONLY ",
+                "ingress_mode": " SESSION_QUEUE ",
+                "reply_mode": " TOPIC_REPLY ",
+            },
+            ("addressed_only", "session_queue", "topic_reply"),
+        ),
+        ({"capture_scope": "invalid"}, None),
+        ({"ingress_mode": "async-inbox"}, None),
+        ({"reply_mode": "invalid"}, None),
+    ],
+)
+def test_connection_readback_and_event_route_share_persisted_mode_rules(
+    tmp_path: Path,
+    routing: dict[str, str],
+    expected: tuple[str, str, str] | None,
+) -> None:
+    target_path = _prep_goal_channel_target(tmp_path)
+    binding_path = tmp_path / "binding.json"
+    payload = _legacy_v0_binding_payload("om_topic_alpha", "agent-alpha")
+    payload["bindings"]["goal-alpha"]["routing"] = routing
+    write_goal_channel_binding(binding_path, payload)
+    before = binding_path.read_bytes()
+    rows = list_lark_connections(
+        registry=_registry(tmp_path),
+        target_path=target_path,
+        binding_paths={"goal-alpha": binding_path},
+        runner=_runner({}),
+    )
+    decision = decide_lark_topic_event(
+        target_payload=read_goal_channel_targets(target_path),
+        binding_payloads={"goal-alpha": read_goal_channel_binding(binding_path)},
+        event={
+            "chat_id": CHAT_ID,
+            "root_id": "om_topic_alpha",
+            "message_id": "om_incoming",
+            "content": "@mew bot hello",
+        },
+    )
+    assert len(rows) == 1
+    if expected is None:
+        assert rows[0]["reply_ready"] is False
+        assert rows[0]["health_error_code"] == "invalid_routing_state"
+        assert decision == {
+            "matched": False,
+            "reason": "invalid_routing_state",
+            "route": None,
+        }
+    else:
+        assert rows[0]["reply_ready"] is True
+        assert decision["matched"] is True
+        for key, value in zip(
+            ("capture_scope", "ingress_mode", "reply_mode"), expected
+        ):
+            assert rows[0][key] == value
+            assert decision["route"][key] == value
+    assert binding_path.read_bytes() == before
+
+
 def test_reconnect_after_upgrade_reuses_legacy_topic_root_without_resend(
     tmp_path: Path,
 ) -> None: