feat(execution_kernel): B13-b T1–T3 pure guards + unit tests

src/quant_platform_kit/execution_kernel/__init__.py

@@ -0,0 +1,32 @@
+"""Pure execution-kernel guards (B13-b T1–T3).
+
+These helpers interpret an already-built snapshot and return allow/deny.
+They never touch broker clients, GCS/Firestore claims, or platform locks.
+Platform adapters remain responsible for durable identity storage.
+"""
+
+from .guards import (
+    deny_blind_retry_after_uncertain_transport,
+    deny_live_submit_without_durable_identity,
+    deny_new_cycle_when_submission_unknown,
+)
+from .types import (
+    ExecutionGuardDecision,
+    ExecutionMode,
+    LiveSubmitSnapshot,
+    SubmissionCertainty,
+    SubmissionRetrySnapshot,
+    SubmissionUnknownSnapshot,
+)
+
+__all__ = [
+    "ExecutionGuardDecision",
+    "ExecutionMode",
+    "LiveSubmitSnapshot",
+    "SubmissionCertainty",
+    "SubmissionRetrySnapshot",
+    "SubmissionUnknownSnapshot",
+    "deny_blind_retry_after_uncertain_transport",
+    "deny_live_submit_without_durable_identity",
+    "deny_new_cycle_when_submission_unknown",
+]

src/quant_platform_kit/execution_kernel/guards.py

@@ -0,0 +1,76 @@
+"""B13-b T1–T3 pure guards. Table-driven; no storage side effects."""
+
+from __future__ import annotations
+
+from .types import (
+    ExecutionGuardDecision,
+    ExecutionMode,
+    LiveSubmitSnapshot,
+    SubmissionCertainty,
+    SubmissionRetrySnapshot,
+    SubmissionUnknownSnapshot,
+)
+
+REASON_MISSING_DURABLE_IDENTITY = "missing_durable_identity"
+REASON_SUBMISSION_UNCERTAIN = "submission_uncertain"
+REASON_RECONCILE_REQUIRED = "reconcile_required"
+REASON_OK = "ok"
+
+
+def deny_live_submit_without_durable_identity(
+    snapshot: LiveSubmitSnapshot,
+) -> ExecutionGuardDecision:
+    """T1 / G1: live ∧ ¬bypass ∧ ¬identity → deny.
+
+    Dry-run / paper / configured bypass do not require durable identity here;
+    adapters still must keep broker writes closed on those paths.
+    """
+    requires_identity = (
+        snapshot.mode is ExecutionMode.LIVE and not snapshot.dry_run_bypass
+    )
+    if requires_identity and not snapshot.identity_held:
+        return ExecutionGuardDecision(
+            decision="deny",
+            reason_code=REASON_MISSING_DURABLE_IDENTITY,
+            notes="live submit requires adapter-held durable identity",
+        )
+    return ExecutionGuardDecision(decision="allow", reason_code=REASON_OK)
+
+
+def deny_new_cycle_when_submission_unknown(
+    snapshot: SubmissionUnknownSnapshot,
+) -> ExecutionGuardDecision:
+    """T2 / G2: certainty=unknown ∧ new-cycle submit → deny."""
+    if (
+        snapshot.requesting_new_cycle_submit
+        and snapshot.submission_certainty is SubmissionCertainty.UNKNOWN
+    ):
+        return ExecutionGuardDecision(
+            decision="deny",
+            reason_code=REASON_SUBMISSION_UNCERTAIN,
+            certainty=SubmissionCertainty.UNKNOWN,
+            notes="unknown submission blocks a new cycle submit",
+        )
+    return ExecutionGuardDecision(
+        decision="allow",
+        reason_code=REASON_OK,
+        certainty=snapshot.submission_certainty,
+    )
+
+
+def deny_blind_retry_after_uncertain_transport(
+    snapshot: SubmissionRetrySnapshot,
+) -> ExecutionGuardDecision:
+    """T3 / G4: transport uncertain ∧ unreconciled ∧ blind retry → deny."""
+    if (
+        snapshot.requesting_blind_retry
+        and snapshot.transport_uncertain
+        and not snapshot.reconciled
+    ):
+        return ExecutionGuardDecision(
+            decision="deny",
+            reason_code=REASON_RECONCILE_REQUIRED,
+            certainty=SubmissionCertainty.UNKNOWN,
+            notes="uncertain transport requires reconcile before retry",
+        )
+    return ExecutionGuardDecision(decision="allow", reason_code=REASON_OK)

src/quant_platform_kit/execution_kernel/types.py

@@ -0,0 +1,68 @@
+"""Shared vocabulary for B13-b shadow guards (no I/O)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Literal
+
+
+class ExecutionMode(str, Enum):
+    """Runtime mode presented to the guard (adapter-normalized)."""
+
+    LIVE = "live"
+    DRY_RUN = "dry_run"
+    PAPER = "paper"
+
+
+class SubmissionCertainty(str, Enum):
+    """Shared submission certainty (B13-a §2). Does not replace broker strings."""
+
+    KNOWN_FAILED = "known_failed"
+    UNKNOWN = "unknown"
+    FILLED_ACCOUNTING_PENDING = "filled_accounting_pending"
+    TERMINAL = "terminal"
+    NOT_APPLICABLE = "not_applicable"
+
+
+DecisionLiteral = Literal["allow", "deny"]
+
+
+@dataclass(frozen=True, slots=True)
+class ExecutionGuardDecision:
+    """Pure guard outcome. Not a broker ACK, fill proof, or digest health signal."""
+
+    decision: DecisionLiteral
+    reason_code: str
+    certainty: SubmissionCertainty | None = None
+    notes: str = ""
+
+    @property
+    def allowed(self) -> bool:
+        return self.decision == "allow"
+
+
+@dataclass(frozen=True, slots=True)
+class LiveSubmitSnapshot:
+    """Inputs for T1 — deny live submit without durable identity."""
+
+    mode: ExecutionMode
+    dry_run_bypass: bool
+    identity_held: bool
+
+
+@dataclass(frozen=True, slots=True)
+class SubmissionUnknownSnapshot:
+    """Inputs for T2 — deny new-cycle submit when submission is unknown."""
+
+    submission_certainty: SubmissionCertainty
+    requesting_new_cycle_submit: bool
+
+
+@dataclass(frozen=True, slots=True)
+class SubmissionRetrySnapshot:
+    """Inputs for T3 — deny blind retry after uncertain transport."""
+
+    transport_uncertain: bool
+    reconciled: bool
+    requesting_blind_retry: bool

tests/test_execution_kernel_guards.py

@@ -0,0 +1,178 @@
+"""Table-driven unit tests for B13-b T1–T3 (no I/O, no broker)."""
+
+from __future__ import annotations
+
+import unittest
+
+from quant_platform_kit.execution_kernel import (
+    ExecutionMode,
+    LiveSubmitSnapshot,
+    SubmissionCertainty,
+    SubmissionRetrySnapshot,
+    SubmissionUnknownSnapshot,
+    deny_blind_retry_after_uncertain_transport,
+    deny_live_submit_without_durable_identity,
+    deny_new_cycle_when_submission_unknown,
+)
+from quant_platform_kit.execution_kernel.guards import (
+    REASON_MISSING_DURABLE_IDENTITY,
+    REASON_OK,
+    REASON_RECONCILE_REQUIRED,
+    REASON_SUBMISSION_UNCERTAIN,
+)
+
+
+class T1DurableIdentityTests(unittest.TestCase):
+    def test_live_without_identity_denies(self) -> None:
+        decision = deny_live_submit_without_durable_identity(
+            LiveSubmitSnapshot(
+                mode=ExecutionMode.LIVE,
+                dry_run_bypass=False,
+                identity_held=False,
+            )
+        )
+        self.assertFalse(decision.allowed)
+        self.assertEqual(decision.reason_code, REASON_MISSING_DURABLE_IDENTITY)
+
+    def test_live_with_identity_allows(self) -> None:
+        decision = deny_live_submit_without_durable_identity(
+            LiveSubmitSnapshot(
+                mode=ExecutionMode.LIVE,
+                dry_run_bypass=False,
+                identity_held=True,
+            )
+        )
+        self.assertTrue(decision.allowed)
+        self.assertEqual(decision.reason_code, REASON_OK)
+
+    def test_dry_run_without_identity_allows_guard(self) -> None:
+        decision = deny_live_submit_without_durable_identity(
+            LiveSubmitSnapshot(
+                mode=ExecutionMode.DRY_RUN,
+                dry_run_bypass=False,
+                identity_held=False,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+    def test_live_bypass_without_identity_allows_guard(self) -> None:
+        decision = deny_live_submit_without_durable_identity(
+            LiveSubmitSnapshot(
+                mode=ExecutionMode.LIVE,
+                dry_run_bypass=True,
+                identity_held=False,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+    def test_paper_without_identity_allows_guard(self) -> None:
+        decision = deny_live_submit_without_durable_identity(
+            LiveSubmitSnapshot(
+                mode=ExecutionMode.PAPER,
+                dry_run_bypass=False,
+                identity_held=False,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+
+class T2SubmissionUnknownTests(unittest.TestCase):
+    def test_unknown_new_cycle_denies(self) -> None:
+        decision = deny_new_cycle_when_submission_unknown(
+            SubmissionUnknownSnapshot(
+                submission_certainty=SubmissionCertainty.UNKNOWN,
+                requesting_new_cycle_submit=True,
+            )
+        )
+        self.assertFalse(decision.allowed)
+        self.assertEqual(decision.reason_code, REASON_SUBMISSION_UNCERTAIN)
+        self.assertEqual(decision.certainty, SubmissionCertainty.UNKNOWN)
+
+    def test_unknown_without_new_cycle_allows(self) -> None:
+        decision = deny_new_cycle_when_submission_unknown(
+            SubmissionUnknownSnapshot(
+                submission_certainty=SubmissionCertainty.UNKNOWN,
+                requesting_new_cycle_submit=False,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+    def test_terminal_new_cycle_allows(self) -> None:
+        decision = deny_new_cycle_when_submission_unknown(
+            SubmissionUnknownSnapshot(
+                submission_certainty=SubmissionCertainty.TERMINAL,
+                requesting_new_cycle_submit=True,
+            )
+        )
+        self.assertTrue(decision.allowed)
+        self.assertEqual(decision.certainty, SubmissionCertainty.TERMINAL)
+
+    def test_not_applicable_new_cycle_allows(self) -> None:
+        decision = deny_new_cycle_when_submission_unknown(
+            SubmissionUnknownSnapshot(
+                submission_certainty=SubmissionCertainty.NOT_APPLICABLE,
+                requesting_new_cycle_submit=True,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+
+class T3BlindRetryTests(unittest.TestCase):
+    def test_uncertain_unreconciled_blind_retry_denies(self) -> None:
+        decision = deny_blind_retry_after_uncertain_transport(
+            SubmissionRetrySnapshot(
+                transport_uncertain=True,
+                reconciled=False,
+                requesting_blind_retry=True,
+            )
+        )
+        self.assertFalse(decision.allowed)
+        self.assertEqual(decision.reason_code, REASON_RECONCILE_REQUIRED)
+
+    def test_uncertain_but_reconciled_allows(self) -> None:
+        decision = deny_blind_retry_after_uncertain_transport(
+            SubmissionRetrySnapshot(
+                transport_uncertain=True,
+                reconciled=True,
+                requesting_blind_retry=True,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+    def test_uncertain_without_retry_allows(self) -> None:
+        decision = deny_blind_retry_after_uncertain_transport(
+            SubmissionRetrySnapshot(
+                transport_uncertain=True,
+                reconciled=False,
+                requesting_blind_retry=False,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+    def test_certain_transport_blind_retry_allows_guard(self) -> None:
+        decision = deny_blind_retry_after_uncertain_transport(
+            SubmissionRetrySnapshot(
+                transport_uncertain=False,
+                reconciled=False,
+                requesting_blind_retry=True,
+            )
+        )
+        self.assertTrue(decision.allowed)
+
+
+class CertaintyVocabTests(unittest.TestCase):
+    def test_certainty_enum_values_stable(self) -> None:
+        self.assertEqual(
+            {item.value for item in SubmissionCertainty},
+            {
+                "known_failed",
+                "unknown",
+                "filled_accounting_pending",
+                "terminal",
+                "not_applicable",
+            },
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()