fix(control-plane): fail closed on malformed runtime recovery and settlement payloads (#3525)

* fix(control-plane): fail closed on malformed runtime recovery and settlement payloads

Harden the TS/Python boundary so malformed settlement metadata and failed-Turn recovery state are rejected instead of being silently treated as absent. Add regressions for corrupted settlement receipts, malformed failure payloads, and corrupted failed-turn recovery journals.

Signed-off-by: duang777 <duangjl007@gmail.com>

* fix(control-plane): continue past corrupted event-log candidates

Signed-off-by: duang777 <duangjl007@gmail.com>

* refactor(settlement): centralize runtime payload decoding

Signed-off-by: huangruiteng <huangrt01@163.com>

---------

Signed-off-by: duang777 <duangjl007@gmail.com>
Signed-off-by: huangruiteng <huangrt01@163.com>
Co-authored-by: huangruiteng <huangrt01@163.com>

Author: Duang777

loopx/control_plane/effect_program.py

@@ -313,37 +313,107 @@ def _runtime_result(result: SettlementResult[Any]) -> dict[str, Any]:
     }
 
 
-def _runtime_result_metadata(
+def _required_settlement_string(
+    payload: Mapping[str, Any],
+    key: str,
+    *,
+    shape: str,
+) -> str:
+    value = payload.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise RuntimeError(f"TypeScript settlement result {shape} shape mismatch")
+    return value
+
+
+_SettlementEnum = TypeVar("_SettlementEnum", bound=StrEnum)
+
+
+def _required_settlement_enum(
+    enum_type: type[_SettlementEnum],
+    payload: Mapping[str, Any],
+    key: str,
+    *,
+    shape: str,
+) -> _SettlementEnum:
+    try:
+        return enum_type(_required_settlement_string(payload, key, shape=shape))
+    except ValueError:
+        raise RuntimeError(
+            f"TypeScript settlement result {shape} shape mismatch"
+        ) from None
+
+
+def decode_settlement_receipt_payload(payload: Any) -> SettlementReceipt:
+    if not isinstance(payload, Mapping):
+        raise RuntimeError("TypeScript settlement result receipts shape mismatch")
+    source_ref = payload.get("source_ref")
+    if source_ref is not None and (
+        not isinstance(source_ref, str) or not source_ref.strip()
+    ):
+        raise RuntimeError("TypeScript settlement result receipts shape mismatch")
+    return SettlementReceipt(
+        step_kind=_required_settlement_enum(
+            SettlementStepKind,
+            payload,
+            "step_kind",
+            shape="receipts",
+        ),
+        status=_required_settlement_string(payload, "status", shape="receipts"),
+        effect_id=_required_settlement_string(
+            payload,
+            "effect_id",
+            shape="receipts",
+        ),
+        source_ref=source_ref,
+    )
+
+
+def decode_settlement_result_payload(
     payload: Any,
-) -> tuple[tuple[SettlementReceipt, ...], SettlementFailure | None]:
+) -> tuple[Any, tuple[SettlementReceipt, ...], SettlementFailure | None]:
+    """Decode the shared TS settlement envelope at the Python boundary."""
+
     if not isinstance(payload, Mapping):
         raise RuntimeError("TypeScript settlement result shape mismatch")
+    if "value" not in payload or "failure" not in payload:
+        raise RuntimeError("TypeScript settlement result shape mismatch")
     receipts_value = payload.get("receipts")
-    receipts = (
-        tuple(
-            SettlementReceipt(
-                step_kind=SettlementStepKind(str(receipt["step_kind"])),
-                status=str(receipt["status"]),
-                effect_id=str(receipt["effect_id"]),
-                source_ref=str(receipt.get("source_ref") or "") or None,
-            )
-            for receipt in receipts_value
-            if isinstance(receipt, Mapping)
-        )
-        if isinstance(receipts_value, list)
-        else ()
+    if not isinstance(receipts_value, list):
+        raise RuntimeError("TypeScript settlement result receipts shape mismatch")
+    receipts = tuple(
+        decode_settlement_receipt_payload(receipt) for receipt in receipts_value
     )
     failure_value = payload.get("failure")
     failure = None
-    if isinstance(failure_value, Mapping):
+    if failure_value is not None:
+        if not isinstance(failure_value, Mapping):
+            raise RuntimeError("TypeScript settlement result failure shape mismatch")
+        if payload.get("value") is not None:
+            raise RuntimeError("TypeScript settlement result shape mismatch")
         details = failure_value.get("details")
+        if details is not None and not isinstance(details, Mapping):
+            raise RuntimeError("TypeScript settlement result failure shape mismatch")
         failure = SettlementFailure(
-            kind=SettlementFailureKind(str(failure_value["kind"])),
-            step_kind=SettlementStepKind(str(failure_value["step_kind"])),
-            reason=str(failure_value["reason"]),
-            details=dict(details) if isinstance(details, Mapping) else None,
+            kind=_required_settlement_enum(
+                SettlementFailureKind,
+                failure_value,
+                "kind",
+                shape="failure",
+            ),
+            step_kind=_required_settlement_enum(
+                SettlementStepKind,
+                failure_value,
+                "step_kind",
+                shape="failure",
+            ),
+            reason=_required_settlement_string(
+                failure_value,
+                "reason",
+                shape="failure",
+            ),
+            details=dict(details) if details is not None else None,
         )
-    return receipts, failure
+    return payload.get("value"), receipts, failure
 
 
 T = TypeVar("T")
@@ -412,7 +482,7 @@ def bind(self, step: Callable[[T], SettlementResult[U]]) -> SettlementResult[U]:
                 "next": _runtime_result(next_result),
             },
         )
-        receipts, failure = _runtime_result_metadata(reduced)
+        _, receipts, failure = decode_settlement_result_payload(reduced)
         return SettlementResult(
             value=next_result.value,
             receipts=receipts,

loopx/control_plane/goals/active_state_event_projection.py

@@ -56,6 +56,7 @@ def active_state_event_projection_fields(
     event_log_basename: str = DEFAULT_STATE_EVENT_LOG_BASENAME,
 ) -> dict[str, Any]:
     goal_id = str(goal.get("id") or "").strip()
+    first_warning: dict[str, Any] | None = None
     for event_log_path in state_event_log_candidates(
         goal,
         state_path=state_path,
@@ -79,15 +80,17 @@ def active_state_event_projection_fields(
                 item_limit=item_limit,
             )
         except (OSError, StateEventError) as exc:
-            return {
-                "state_event_projection_warning": {
-                    "schema_version": STATE_EVENT_READ_WARNING_SCHEMA_VERSION,
-                    "source": "event_log",
-                    "event_log": event_log_path.name,
-                    "fallback": "markdown_active_state",
-                    "reason": type(exc).__name__,
+            if first_warning is None:
+                first_warning = {
+                    "state_event_projection_warning": {
+                        "schema_version": STATE_EVENT_READ_WARNING_SCHEMA_VERSION,
+                        "source": "event_log",
+                        "event_log": event_log_path.name,
+                        "fallback": "markdown_active_state",
+                        "reason": type(exc).__name__,
+                    }
                 }
-            }
+            continue
         if fields:
             fields["state_event_projection"] = {
                 "schema_version": STATE_EVENT_PROJECTION_SCHEMA_VERSION,
@@ -100,4 +103,4 @@ def active_state_event_projection_fields(
                 "projection_version": projection.get("projection_version"),
             }
             return fields
-    return {}
+    return first_warning or {}

loopx/control_plane/settlement_driver.py

@@ -12,12 +12,12 @@
 from typing import Any
 
 from .effect_program import (
-    SettlementFailure,
-    SettlementFailureKind,
     SettlementIdentity,
     SettlementReceipt,
     SettlementResult,
     SettlementStepKind,
+    decode_settlement_receipt_payload,
+    decode_settlement_result_payload,
 )
 from .effect_runtime import effect_runtime_result
 
@@ -26,44 +26,13 @@ def _identity_payload(identity: SettlementIdentity) -> dict[str, Any]:
     return identity.as_dict()
 
 
-def _receipt_from_payload(payload: Mapping[str, Any]) -> SettlementReceipt:
-    return SettlementReceipt(
-        step_kind=SettlementStepKind(str(payload["step_kind"])),
-        status=str(payload["status"]),
-        effect_id=str(payload["effect_id"]),
-        source_ref=str(payload.get("source_ref") or "") or None,
-    )
-
-
 def decode_settlement_result(
     payload: Any,
     *,
     value_decoder: Callable[[Any], Any] | None = None,
     projection_payload: Mapping[str, Any] | None = None,
 ) -> SettlementResult[Any]:
-    if not isinstance(payload, Mapping):
-        raise RuntimeError("TypeScript settlement result shape mismatch")
-    receipts_value = payload.get("receipts")
-    receipts = (
-        tuple(
-            _receipt_from_payload(receipt)
-            for receipt in receipts_value
-            if isinstance(receipt, Mapping)
-        )
-        if isinstance(receipts_value, list)
-        else ()
-    )
-    failure_value = payload.get("failure")
-    failure = None
-    if isinstance(failure_value, Mapping):
-        details = failure_value.get("details")
-        failure = SettlementFailure(
-            kind=SettlementFailureKind(str(failure_value["kind"])),
-            step_kind=SettlementStepKind(str(failure_value["step_kind"])),
-            reason=str(failure_value["reason"]),
-            details=dict(details) if isinstance(details, Mapping) else None,
-        )
-    value = payload.get("value")
+    value, receipts, failure = decode_settlement_result_payload(payload)
     if value_decoder is not None and value is not None:
         value = value_decoder(value)
     return SettlementResult(
@@ -113,4 +82,4 @@ def settlement_receipt(
     )
     if not isinstance(payload, Mapping):
         raise RuntimeError("TypeScript settlement receipt shape mismatch")
-    return _receipt_from_payload(payload)
+    return decode_settlement_receipt_payload(payload)

loopx/control_plane/todos/event_writeback.py

@@ -108,7 +108,7 @@ def _canonical_event_projection_source(
                 continue
             projection = build_state_projection(events, goal_id=goal_id or None)
         except (OSError, StateEventError):
-            return None
+            continue
         if all(
             projection.get(key) == projection_authority.get(key)
             for key in authority_keys

loopx/control_plane/turn_driver/session_recovery.py

@@ -36,10 +36,12 @@ def reconcile_failed_turn_retry_request(
     *,
     session_binding_resolver: SessionBindingResolver | None,
 ) -> dict[str, Any]:
-    recovery_value = journal.get("host_recovery")
-    recovery = dict(recovery_value) if isinstance(recovery_value, Mapping) else {}
-    if not recovery:
+    if "host_recovery" not in journal:
         return dict(request)
+    recovery_value = journal.get("host_recovery")
+    if not isinstance(recovery_value, Mapping):
+        raise ValueError("failed-Turn host recovery shape mismatch")
+    recovery = dict(recovery_value)
     if recovery.get("schema_version") != HOST_RECOVERY_SCHEMA_VERSION:
         raise ValueError("failed-Turn host recovery has an unsupported schema")
     if recovery.get("kind") not in HOST_RECOVERY_KINDS:

tests/control_plane/test_quota_settlement.py

@@ -3,6 +3,7 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import patch
 
 import pytest
 
@@ -284,6 +285,28 @@ def spend(_value: int) -> SettlementResult[str]:
     assert result.receipts == failed.receipts
 
 
+def test_settlement_result_bind_rejects_corrupted_receipts() -> None:
+    def writeback(value: int) -> SettlementResult[int]:
+        return SettlementResult.pure(value + 1)
+
+    with patch.object(core_effect_program, "effect_runtime_result") as runtime_result:
+        runtime_result.side_effect = [
+            {"execute": True},
+            {
+                "value": {
+                    "completed_phases": [],
+                    "writeback": None,
+                    "quota_spend": None,
+                },
+                "receipts": ["corrupted-receipt"],
+                "failure": None,
+            },
+        ]
+
+        with pytest.raises(RuntimeError, match="receipts shape mismatch"):
+            SettlementResult.pure(2).bind(writeback)
+
+
 def test_codex_app_plan_projects_one_identity_across_settlement_steps() -> None:
     plan = build_codex_app_settlement_plan(
         goal_id=GOAL_ID,

tests/control_plane/test_settlement_driver.py

@@ -7,6 +7,7 @@
     SettlementStepKind,
 )
 from loopx.control_plane.settlement_driver import (
+    decode_settlement_result,
     effect_ids_match,
     settlement_receipt,
 )
@@ -209,3 +210,69 @@ def test_turn_settlement_fails_closed_when_prepared_outcome_is_unknown(
     assert result.failure.kind.value == "effect_outcome_unknown"
     assert result.failure.step_kind is SettlementStepKind.DURABLE_WRITEBACK
     assert calls["writeback"] == 0
+
+
+def test_decode_settlement_result_rejects_corrupted_failure_payload() -> None:
+    with pytest.raises(RuntimeError, match="failure shape mismatch"):
+        decode_settlement_result(
+            {
+                "value": {
+                    "completed_phases": [],
+                    "writeback": None,
+                    "quota_spend": None,
+                },
+                "receipts": [
+                    {
+                        "step_kind": "validation",
+                        "status": "committed",
+                        "effect_id": "effect-1",
+                    },
+                ],
+                "failure": "corrupted-failure",
+            }
+        )
+
+
+@pytest.mark.parametrize(
+    ("payload", "message"),
+    [
+        (
+            {"value": {}, "receipts": [], "failure": {"kind": "cancelled"}},
+            "result shape mismatch",
+        ),
+        (
+            {
+                "value": None,
+                "receipts": [],
+                "failure": {
+                    "kind": "cancelled",
+                    "step_kind": "durable_writeback",
+                    "reason": "cancelled",
+                    "details": "corrupted-details",
+                },
+            },
+            "failure shape mismatch",
+        ),
+        (
+            {
+                "value": {},
+                "receipts": [
+                    {
+                        "step_kind": "validation",
+                        "status": 1,
+                        "effect_id": "effect-1",
+                    }
+                ],
+                "failure": None,
+            },
+            "receipts shape mismatch",
+        ),
+        ({"value": {}, "receipts": []}, "result shape mismatch"),
+    ],
+)
+def test_decode_settlement_result_rejects_invalid_union_states(
+    payload: object,
+    message: str,
+) -> None:
+    with pytest.raises(RuntimeError, match=message):
+        decode_settlement_result(payload)