From f638fb4a1ce6dca9fee17d060ab393b88c544f03 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:13:37 +0800 Subject: [PATCH 1/2] feat(finance): add a provider-neutral confirmation gate for venue executors The Core typed-action store owns the confirmation receipt, the single execution claim and the canonical envelope, but a venue executor still has to prove that the effect it is about to perform is the effect the operator approved. Without a typed gate every adapter re-implements that comparison ad hoc, and the easy mistake is to treat a matching digest - which exists before the human clicks - as proof of confirmation. verify_bound_confirmation() checks a canonical proposal against a frozen plan: the confirmation receipt must exist, come from the expected provider and the authorized principal, and carry the message id of this operation's own card; the record and lifecycle must still be live; a one-shot claim must exist; the confirmed payload terms must equal the frozen plan terms; and the operation must not be expired. Refusals are typed codes instead of exceptions so a caller can report them honestly instead of retrying or working around them. Coverage uses synthetic fixtures only. The module performs no IO, reads no credentials, talks to no venue and grants no execution authority. Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../confirmation_gate.py | 179 ++++++++++++++++++ .../test_finance_confirmation_gate.py | 153 +++++++++++++++ 2 files changed, 332 insertions(+) create mode 100644 packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py create mode 100644 tests/extensions/test_finance_confirmation_gate.py diff --git a/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py new file mode 100644 index 0000000000..9c38e6135f --- /dev/null +++ b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py @@ -0,0 +1,179 @@ +"""Provider-neutral admission gate for one human-confirmed financial operation. + +The Core typed-action store owns the confirmation receipt, the single execution +claim and the canonical operation envelope. A venue executor must additionally +prove that the effect it is about to perform is the effect the operator +approved. + +This module performs that check against records that are already loaded from +the canonical store. It never mutates state, never reads credentials and never +talks to a venue, so it can run before any effect-bearing call. A refusal is a +typed result, not an exception: callers must report it honestly instead of +retrying or working around it. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime, timezone +import hashlib +import json +from typing import Any + + +GATE_SCHEMA_VERSION = "finance_confirmation_gate_v0" + +#: Record statuses that still represent a live, human-confirmed operation. +#: ``applying`` is the ordinary state while the claimed effect is in flight. +LIVE_RECORD_STATUSES = frozenset({"applied", "applying"}) + +#: Canonical lifecycle states an executor may act on. +LIVE_LIFECYCLE_STATES = frozenset({"claimed", "consumed", "outcome_observed"}) + +#: Refusal codes are stable public tokens; they never carry private material. +CONFIRMATION_MISSING = "confirmation_missing" +CONFIRMATION_PROVIDER_MISMATCH = "confirmation_provider_mismatch" +CONFIRMATION_PRINCIPAL_MISMATCH = "confirmation_principal_mismatch" +CONFIRMATION_CARD_MISMATCH = "confirmation_card_mismatch" +OPERATION_NOT_LIVE = "operation_not_live" +CLAIM_MISSING = "claim_missing" +PAYLOAD_TERMS_MISSING = "payload_terms_missing" +TERMS_MISMATCH = "terms_mismatch" +EXPIRED = "expired" + + +def _mapping(value: object, *, field: str) -> dict[str, Any]: + if isinstance(value, str): + try: + value = json.loads(value) + except ValueError as exc: + raise ValueError(f"{field} must be a JSON object") from exc + if not isinstance(value, Mapping): + raise ValueError(f"{field} must be an object") + return dict(value) + + +def _canonical(value: object) -> str: + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + +def _digest(value: object) -> str | None: + if value is None: + return None + return hashlib.sha256(_canonical(value).encode()).hexdigest() + + +def _timestamp(value: object) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def verify_bound_confirmation( + proposal: Mapping[str, Any], + plan: Mapping[str, Any], + *, + expected_provider: str = "lark", + now: datetime | None = None, +) -> dict[str, Any]: + """Check one canonical operation against a frozen execution plan. + + ``proposal`` is the record as stored by the canonical action store + (``normalized_parameters`` plus ``operation``). ``plan`` is the frozen plan + the executor would run: ``terms_key`` (default ``bound_order``), ``terms``, + optional ``authorized_principal`` and ``expires_at``, and + ``require_claim`` (default ``True``). + """ + + parameters = _mapping(proposal.get("normalized_parameters"), field="normalized_parameters") + operation = _mapping(proposal.get("operation"), field="operation") + payload = parameters.get("payload") + payload = dict(payload) if isinstance(payload, Mapping) else {} + + confirmation = operation.get("confirmation") + confirmation = dict(confirmation) if isinstance(confirmation, Mapping) else None + claim = operation.get("claim") + claim = dict(claim) if isinstance(claim, Mapping) else None + delivery = operation.get("delivery") + delivery = dict(delivery) if isinstance(delivery, Mapping) else {} + + terms_key = str(plan.get("terms_key") or "bound_order") + expected_terms = plan.get("terms") + authorized_principal = plan.get("authorized_principal") + require_claim = plan.get("require_claim", True) is not False + clock = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + + refusals: list[str] = [] + card_message_ids = { + value + for value in (delivery.get("message_id"), operation.get("card_message_id")) + if isinstance(value, str) and value + } + + if confirmation is None: + refusals.append(CONFIRMATION_MISSING) + else: + if str(confirmation.get("provider") or "") != expected_provider: + refusals.append(CONFIRMATION_PROVIDER_MISMATCH) + if authorized_principal and confirmation.get("principal") != authorized_principal: + refusals.append(CONFIRMATION_PRINCIPAL_MISMATCH) + confirmed_card = confirmation.get("message_id") + # A confirmation is only valid for the card that carried this exact + # operation; a receipt from any other message proves nothing here. + if not isinstance(confirmed_card, str) or not confirmed_card or ( + card_message_ids and confirmed_card not in card_message_ids + ): + refusals.append(CONFIRMATION_CARD_MISMATCH) + + record_status = str(proposal.get("status") or "") + lifecycle_state = str(operation.get("lifecycle_state") or "") + if record_status not in LIVE_RECORD_STATUSES or lifecycle_state not in LIVE_LIFECYCLE_STATES: + refusals.append(OPERATION_NOT_LIVE) + + if require_claim and claim is None: + refusals.append(CLAIM_MISSING) + + payload_terms = payload.get(terms_key) + if not isinstance(payload_terms, Mapping) or expected_terms is None: + refusals.append(PAYLOAD_TERMS_MISSING) + elif _canonical(payload_terms) != _canonical(expected_terms): + refusals.append(TERMS_MISMATCH) + + expiry = _timestamp(plan.get("expires_at")) or _timestamp(payload.get("expires_at")) + if expiry is not None and clock > expiry: + refusals.append(EXPIRED) + + return { + "schema_version": GATE_SCHEMA_VERSION, + "allowed": not refusals, + "refusals": refusals, + "evidence": { + "operation_id": operation.get("operation_id") or proposal.get("proposal_id"), + "record_status": record_status, + "lifecycle_state": lifecycle_state, + "confirmation_digest": operation.get("confirmation_digest"), + "confirmation_provider": (confirmation or {}).get("provider"), + "confirmation_principal": (confirmation or {}).get("principal"), + "confirmation_card_message_id": (confirmation or {}).get("message_id"), + "operation_card_message_id": delivery.get("message_id"), + "claim_id": (claim or {}).get("claim_id"), + "terms_key": terms_key, + "terms_digest": _digest(payload_terms) if isinstance(payload_terms, Mapping) else None, + "expected_terms_digest": _digest(expected_terms), + "expires_at": expiry.isoformat() if expiry else None, + "observed_at": clock.isoformat(), + }, + } diff --git a/tests/extensions/test_finance_confirmation_gate.py b/tests/extensions/test_finance_confirmation_gate.py new file mode 100644 index 0000000000..be2c90bfc4 --- /dev/null +++ b/tests/extensions/test_finance_confirmation_gate.py @@ -0,0 +1,153 @@ +"""Synthetic coverage for the provider-neutral confirmation gate.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import copy +import json +import sys +from pathlib import Path + + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "packages" / "loopx-finance-execution" / "src")) + +from loopx_finance_execution.confirmation_gate import ( # noqa: E402 + CLAIM_MISSING, + CONFIRMATION_CARD_MISMATCH, + CONFIRMATION_MISSING, + CONFIRMATION_PRINCIPAL_MISMATCH, + CONFIRMATION_PROVIDER_MISMATCH, + EXPIRED, + OPERATION_NOT_LIVE, + PAYLOAD_TERMS_MISSING, + TERMS_MISMATCH, + verify_bound_confirmation, +) + + +NOW = datetime(2026, 10, 17, 4, 0, tzinfo=timezone.utc) +CARD = "om_synthetic_card_1" +PRINCIPAL = "lark:ou_synthetic_operator" +TERMS = { + "enabled": True, + "type": "market", + "side": "long", + "amount_usdc": "20", + "limit_price": None, + "reduce_only": False, + "take_profit_stop_loss": None, +} + + +def _proposal() -> dict: + return { + "proposal_id": "proposal-synthetic-1", + "status": "applying", + "normalized_parameters": json.dumps( + { + "payload": { + "schema_version": "finance_bound_order_v1", + "bound_order": copy.deepcopy(TERMS), + "expires_at": "2026-10-17T05:00:00Z", + } + } + ), + "operation": { + "operation_id": "proposal-synthetic-1", + "lifecycle_state": "claimed", + "confirmation_digest": "a" * 64, + "confirmation": { + "provider": "lark", + "event_id": "synthetic-event-1", + "principal": PRINCIPAL, + "message_id": CARD, + }, + "claim": {"claim_id": "claim-synthetic-1"}, + "delivery": {"message_id": CARD}, + }, + } + + +def _plan() -> dict: + return { + "terms_key": "bound_order", + "terms": copy.deepcopy(TERMS), + "authorized_principal": PRINCIPAL, + } + + +def test_confirmed_operation_is_admitted() -> None: + result = verify_bound_confirmation(_proposal(), _plan(), now=NOW) + assert result["allowed"] is True + assert result["refusals"] == [] + assert result["evidence"]["claim_id"] == "claim-synthetic-1" + assert result["evidence"]["confirmation_card_message_id"] == CARD + + +def test_missing_confirmation_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("confirmation") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert CONFIRMATION_MISSING in result["refusals"] + + +def test_confirmation_from_another_channel_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["provider"] = "web" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_PROVIDER_MISMATCH in result["refusals"] + + +def test_confirmation_from_another_principal_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["principal"] = "lark:ou_someone_else" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_PRINCIPAL_MISMATCH in result["refusals"] + + +def test_confirmation_from_another_card_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["message_id"] = "om_other_card" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_CARD_MISMATCH in result["refusals"] + + +def test_cancelled_operation_is_not_live() -> None: + proposal = _proposal() + proposal["status"] = "cancelled" + proposal["operation"]["lifecycle_state"] = "prepared" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert OPERATION_NOT_LIVE in result["refusals"] + + +def test_missing_claim_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("claim") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + +def test_missing_payload_terms_refuses() -> None: + proposal = _proposal() + parameters = json.loads(proposal["normalized_parameters"]) + parameters["payload"].pop("bound_order") + proposal["normalized_parameters"] = json.dumps(parameters) + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert PAYLOAD_TERMS_MISSING in result["refusals"] + + +def test_terms_drift_refuses() -> None: + plan = _plan() + plan["terms"]["amount_usdc"] = "500" + result = verify_bound_confirmation(_proposal(), plan, now=NOW) + assert TERMS_MISMATCH in result["refusals"] + assert result["evidence"]["terms_digest"] != result["evidence"]["expected_terms_digest"] + + +def test_expired_operation_refuses() -> None: + result = verify_bound_confirmation( + _proposal(), _plan(), now=datetime(2026, 10, 17, 6, 0, tzinfo=timezone.utc) + ) + assert EXPIRED in result["refusals"] + assert result["allowed"] is False From 80f2a75add23583eb19507da0cc65482a8fd981f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 20:28:57 +0800 Subject: [PATCH 2/2] fix(finance): fail closed on card, claim and every deadline Review follow-up on the confirmation gate: - Require the authoritative original-card identity. A missing or self- contradictory delivery/card binding is refused instead of being treated as 'no card to compare'. - Require a canonical claim id unconditionally and drop the require_claim bypass, so an omitted or empty claim can never be admitted as a confirmed execution precheck. - Validate the canonical envelope deadline as required, and enforce the plan and payload deadlines independently when present. An invalid deadline is refused and the boundary is inclusive (now >= expiry), so a later payload deadline can no longer extend an expired canonical envelope. Coverage adds missing/conflicting origin card, omitted/empty/foreign claim, missing/invalid canonical deadline, the equality boundary and conflicting deadline cases. Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../confirmation_gate.py | 95 +++++++++++---- .../test_finance_confirmation_gate.py | 112 ++++++++++++++++-- 2 files changed, 172 insertions(+), 35 deletions(-) diff --git a/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py index 9c38e6135f..49c68c8af0 100644 --- a/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py +++ b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py @@ -10,6 +10,10 @@ talks to a venue, so it can run before any effect-bearing call. A refusal is a typed result, not an exception: callers must report it honestly instead of retrying or working around it. + +Every check fails closed. A record that does not carry the authoritative +original-card identity, a canonical claim id or a valid canonical deadline is +refused rather than admitted by default. """ from __future__ import annotations @@ -30,15 +34,18 @@ #: Canonical lifecycle states an executor may act on. LIVE_LIFECYCLE_STATES = frozenset({"claimed", "consumed", "outcome_observed"}) -#: Refusal codes are stable public tokens; they never carry private material. CONFIRMATION_MISSING = "confirmation_missing" CONFIRMATION_PROVIDER_MISMATCH = "confirmation_provider_mismatch" CONFIRMATION_PRINCIPAL_MISMATCH = "confirmation_principal_mismatch" CONFIRMATION_CARD_MISMATCH = "confirmation_card_mismatch" +CARD_BINDING_MISSING = "card_binding_missing" +CARD_BINDING_CONFLICT = "card_binding_conflict" OPERATION_NOT_LIVE = "operation_not_live" CLAIM_MISSING = "claim_missing" PAYLOAD_TERMS_MISSING = "payload_terms_missing" TERMS_MISMATCH = "terms_mismatch" +DEADLINE_MISSING = "deadline_missing" +DEADLINE_INVALID = "deadline_invalid" EXPIRED = "expired" @@ -82,6 +89,23 @@ def _timestamp(value: object) -> datetime | None: return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) +def _origin_card_id(operation: Mapping[str, Any]) -> tuple[str | None, list[str]]: + """Return the authoritative original-card id and every observed conflict.""" + + delivery = operation.get("delivery") + delivery = dict(delivery) if isinstance(delivery, Mapping) else {} + observed: list[str] = [] + for candidate in (delivery.get("message_id"), operation.get("card_message_id")): + if isinstance(candidate, str) and candidate.strip(): + observed.append(candidate.strip()) + distinct = sorted(set(observed)) + if not distinct: + return None, [] + if len(distinct) > 1: + return None, distinct + return distinct[0], [] + + def verify_bound_confirmation( proposal: Mapping[str, Any], plan: Mapping[str, Any], @@ -94,8 +118,11 @@ def verify_bound_confirmation( ``proposal`` is the record as stored by the canonical action store (``normalized_parameters`` plus ``operation``). ``plan`` is the frozen plan the executor would run: ``terms_key`` (default ``bound_order``), ``terms``, - optional ``authorized_principal`` and ``expires_at``, and - ``require_claim`` (default ``True``). + optional ``authorized_principal`` and an optional ``expires_at``. + + The canonical envelope deadline is required; the payload and plan deadlines + are optional but, when present, must be valid and are enforced + independently of each other. """ parameters = _mapping(proposal.get("normalized_parameters"), field="normalized_parameters") @@ -106,22 +133,15 @@ def verify_bound_confirmation( confirmation = operation.get("confirmation") confirmation = dict(confirmation) if isinstance(confirmation, Mapping) else None claim = operation.get("claim") - claim = dict(claim) if isinstance(claim, Mapping) else None - delivery = operation.get("delivery") - delivery = dict(delivery) if isinstance(delivery, Mapping) else {} + claim = dict(claim) if isinstance(claim, Mapping) else {} terms_key = str(plan.get("terms_key") or "bound_order") expected_terms = plan.get("terms") authorized_principal = plan.get("authorized_principal") - require_claim = plan.get("require_claim", True) is not False clock = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) refusals: list[str] = [] - card_message_ids = { - value - for value in (delivery.get("message_id"), operation.get("card_message_id")) - if isinstance(value, str) and value - } + origin_card_id, card_conflicts = _origin_card_id(operation) if confirmation is None: refusals.append(CONFIRMATION_MISSING) @@ -130,11 +150,18 @@ def verify_bound_confirmation( refusals.append(CONFIRMATION_PROVIDER_MISMATCH) if authorized_principal and confirmation.get("principal") != authorized_principal: refusals.append(CONFIRMATION_PRINCIPAL_MISMATCH) + + if origin_card_id is None: + refusals.append(CARD_BINDING_CONFLICT if card_conflicts else CARD_BINDING_MISSING) + if confirmation is not None: confirmed_card = confirmation.get("message_id") - # A confirmation is only valid for the card that carried this exact - # operation; a receipt from any other message proves nothing here. - if not isinstance(confirmed_card, str) or not confirmed_card or ( - card_message_ids and confirmed_card not in card_message_ids + # Only a receipt from this operation's own card proves the operator + # confirmed this exact card; anything else proves nothing here. + if ( + not isinstance(confirmed_card, str) + or not confirmed_card.strip() + or origin_card_id is None + or confirmed_card.strip() != origin_card_id ): refusals.append(CONFIRMATION_CARD_MISMATCH) @@ -143,7 +170,10 @@ def verify_bound_confirmation( if record_status not in LIVE_RECORD_STATUSES or lifecycle_state not in LIVE_LIFECYCLE_STATES: refusals.append(OPERATION_NOT_LIVE) - if require_claim and claim is None: + # Core remains the only first-consumption owner; the gate only requires + # that a canonical claim id exists. There is no claim bypass. + claim_id = claim.get("claim_id") + if not isinstance(claim_id, str) or not claim_id.strip(): refusals.append(CLAIM_MISSING) payload_terms = payload.get(terms_key) @@ -152,9 +182,24 @@ def verify_bound_confirmation( elif _canonical(payload_terms) != _canonical(expected_terms): refusals.append(TERMS_MISMATCH) - expiry = _timestamp(plan.get("expires_at")) or _timestamp(payload.get("expires_at")) - if expiry is not None and clock > expiry: - refusals.append(EXPIRED) + deadlines: list[tuple[str, datetime]] = [] + envelope_deadline = _timestamp(operation.get("expires_at")) + if envelope_deadline is None: + refusals.append(DEADLINE_MISSING) + else: + deadlines.append(("canonical", envelope_deadline)) + for label, value in (("plan", plan.get("expires_at")), ("payload", payload.get("expires_at"))): + if value is None: + continue + parsed = _timestamp(value) + if parsed is None: + refusals.append(DEADLINE_INVALID) + continue + deadlines.append((label, parsed)) + for _label, deadline in deadlines: + if clock >= deadline: + refusals.append(EXPIRED) + break return { "schema_version": GATE_SCHEMA_VERSION, @@ -168,12 +213,16 @@ def verify_bound_confirmation( "confirmation_provider": (confirmation or {}).get("provider"), "confirmation_principal": (confirmation or {}).get("principal"), "confirmation_card_message_id": (confirmation or {}).get("message_id"), - "operation_card_message_id": delivery.get("message_id"), - "claim_id": (claim or {}).get("claim_id"), + "origin_card_message_id": origin_card_id, + "card_conflict_ids": card_conflicts, + "claim_id": claim_id if isinstance(claim_id, str) else None, "terms_key": terms_key, "terms_digest": _digest(payload_terms) if isinstance(payload_terms, Mapping) else None, "expected_terms_digest": _digest(expected_terms), - "expires_at": expiry.isoformat() if expiry else None, + "canonical_expires_at": ( + envelope_deadline.isoformat() if envelope_deadline else None + ), + "enforced_deadlines": {label: moment.isoformat() for label, moment in deadlines}, "observed_at": clock.isoformat(), }, } diff --git a/tests/extensions/test_finance_confirmation_gate.py b/tests/extensions/test_finance_confirmation_gate.py index be2c90bfc4..a749883f9d 100644 --- a/tests/extensions/test_finance_confirmation_gate.py +++ b/tests/extensions/test_finance_confirmation_gate.py @@ -5,18 +5,23 @@ from datetime import datetime, timezone import copy import json -import sys from pathlib import Path +import sys -sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "packages" / "loopx-finance-execution" / "src")) +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "packages" / "loopx-finance-execution" / "src")) from loopx_finance_execution.confirmation_gate import ( # noqa: E402 + CARD_BINDING_CONFLICT, + CARD_BINDING_MISSING, CLAIM_MISSING, CONFIRMATION_CARD_MISMATCH, CONFIRMATION_MISSING, CONFIRMATION_PRINCIPAL_MISMATCH, CONFIRMATION_PROVIDER_MISMATCH, + DEADLINE_INVALID, + DEADLINE_MISSING, EXPIRED, OPERATION_NOT_LIVE, PAYLOAD_TERMS_MISSING, @@ -28,6 +33,7 @@ NOW = datetime(2026, 10, 17, 4, 0, tzinfo=timezone.utc) CARD = "om_synthetic_card_1" PRINCIPAL = "lark:ou_synthetic_operator" +CANONICAL_EXPIRY = "2026-10-17T05:00:00Z" TERMS = { "enabled": True, "type": "market", @@ -48,7 +54,7 @@ def _proposal() -> dict: "payload": { "schema_version": "finance_bound_order_v1", "bound_order": copy.deepcopy(TERMS), - "expires_at": "2026-10-17T05:00:00Z", + "expires_at": CANONICAL_EXPIRY, } } ), @@ -56,6 +62,7 @@ def _proposal() -> dict: "operation_id": "proposal-synthetic-1", "lifecycle_state": "claimed", "confirmation_digest": "a" * 64, + "expires_at": CANONICAL_EXPIRY, "confirmation": { "provider": "lark", "event_id": "synthetic-event-1", @@ -76,12 +83,23 @@ def _plan() -> dict: } +def _payload_of(proposal: dict) -> dict: + return json.loads(proposal["normalized_parameters"])["payload"] + + +def _store_payload(proposal: dict, payload: dict) -> None: + parameters = json.loads(proposal["normalized_parameters"]) + parameters["payload"] = payload + proposal["normalized_parameters"] = json.dumps(parameters) + + def test_confirmed_operation_is_admitted() -> None: result = verify_bound_confirmation(_proposal(), _plan(), now=NOW) assert result["allowed"] is True assert result["refusals"] == [] assert result["evidence"]["claim_id"] == "claim-synthetic-1" - assert result["evidence"]["confirmation_card_message_id"] == CARD + assert result["evidence"]["origin_card_message_id"] == CARD + assert set(result["evidence"]["enforced_deadlines"]) == {"canonical", "payload"} def test_missing_confirmation_refuses() -> None: @@ -113,6 +131,23 @@ def test_confirmation_from_another_card_refuses() -> None: assert CONFIRMATION_CARD_MISMATCH in result["refusals"] +def test_missing_origin_card_identity_fails_closed() -> None: + proposal = _proposal() + proposal["operation"].pop("delivery") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert CARD_BINDING_MISSING in result["refusals"] + assert CONFIRMATION_CARD_MISMATCH in result["refusals"] + + +def test_conflicting_origin_card_identity_fails_closed() -> None: + proposal = _proposal() + proposal["operation"]["card_message_id"] = "om_second_card" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CARD_BINDING_CONFLICT in result["refusals"] + assert set(result["evidence"]["card_conflict_ids"]) == {CARD, "om_second_card"} + + def test_cancelled_operation_is_not_live() -> None: proposal = _proposal() proposal["status"] = "cancelled" @@ -121,18 +156,32 @@ def test_cancelled_operation_is_not_live() -> None: assert OPERATION_NOT_LIVE in result["refusals"] -def test_missing_claim_refuses() -> None: +def test_omitted_claim_refuses() -> None: proposal = _proposal() proposal["operation"].pop("claim") result = verify_bound_confirmation(proposal, _plan(), now=NOW) assert CLAIM_MISSING in result["refusals"] +def test_empty_claim_refuses() -> None: + proposal = _proposal() + proposal["operation"]["claim"] = {} + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + +def test_foreign_claim_shape_refuses() -> None: + proposal = _proposal() + proposal["operation"]["claim"] = {"claimed_at": "2026-10-17T03:59:00Z"} + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + def test_missing_payload_terms_refuses() -> None: proposal = _proposal() - parameters = json.loads(proposal["normalized_parameters"]) - parameters["payload"].pop("bound_order") - proposal["normalized_parameters"] = json.dumps(parameters) + payload = _payload_of(proposal) + payload.pop("bound_order") + _store_payload(proposal, payload) result = verify_bound_confirmation(proposal, _plan(), now=NOW) assert PAYLOAD_TERMS_MISSING in result["refusals"] @@ -145,9 +194,48 @@ def test_terms_drift_refuses() -> None: assert result["evidence"]["terms_digest"] != result["evidence"]["expected_terms_digest"] -def test_expired_operation_refuses() -> None: - result = verify_bound_confirmation( - _proposal(), _plan(), now=datetime(2026, 10, 17, 6, 0, tzinfo=timezone.utc) - ) +def test_missing_canonical_deadline_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("expires_at") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert DEADLINE_MISSING in result["refusals"] + + +def test_invalid_deadline_refuses() -> None: + proposal = _proposal() + proposal["operation"]["expires_at"] = "not-a-timestamp" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert DEADLINE_MISSING in result["refusals"] + plan = _plan() + plan["expires_at"] = "31/12/2026" + result = verify_bound_confirmation(_proposal(), plan, now=NOW) + assert DEADLINE_INVALID in result["refusals"] + + +def test_deadline_boundary_is_inclusive() -> None: + exact = datetime(2026, 10, 17, 5, 0, tzinfo=timezone.utc) + assert EXPIRED in verify_bound_confirmation(_proposal(), _plan(), now=exact)["refusals"] + before = datetime(2026, 10, 17, 4, 59, 59, tzinfo=timezone.utc) + assert verify_bound_confirmation(_proposal(), _plan(), now=before)["allowed"] is True + + +def test_conflicting_deadlines_all_enforced() -> None: + """A later payload deadline must not extend an expired canonical envelope.""" + + proposal = _proposal() + payload = _payload_of(proposal) + payload["expires_at"] = "2026-10-17T09:00:00Z" + _store_payload(proposal, payload) + later = datetime(2026, 10, 17, 6, 0, tzinfo=timezone.utc) + result = verify_bound_confirmation(proposal, _plan(), now=later) assert EXPIRED in result["refusals"] assert result["allowed"] is False + + +def test_plan_deadline_is_independent() -> None: + plan = _plan() + plan["expires_at"] = "2026-10-17T04:30:00Z" + later = datetime(2026, 10, 17, 4, 45, tzinfo=timezone.utc) + result = verify_bound_confirmation(_proposal(), plan, now=later) + assert EXPIRED in result["refusals"]