Add explicit policy-only decision boundary

Author: mdheller

src/sourceos_syncd/policy.py

@@ -3,6 +3,11 @@
 The real PolicyFabric service will eventually own policy evaluation. This module
 provides a local, deterministic contract-compatible evaluator so State Integrity
 Reports can already carry policy counts and explanation codes.
+
+Boundary invariant: this module emits policy decisions only. It does not perform
+runtime effects, mutate agent grants, repair state, write ledgers, or replicate
+payloads. Downstream systems must consume explicit decision refs before taking
+any action.
 """
 
 from __future__ import annotations
@@ -15,6 +20,7 @@
 
 POLICY_DECISION_SCHEMA = "sourceos.policy-decision/v1alpha1"
 POLICY_ENGINE = "policy-fabric-local-stub"
+DECISION_SCOPE = "policy-only"
 
 ACTIONS = {
     "index",
@@ -44,6 +50,31 @@ class PolicyRequest:
     context: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(frozen=True)
+class DecisionBoundary:
+    """Hard boundary carried by local policy decisions."""
+
+    decision_scope: str = DECISION_SCOPE
+    runtime_effect_performed: bool = False
+    authority_mutation_performed: bool = False
+    state_repair_performed: bool = False
+    ledger_write_performed: bool = False
+    downstream_refs: tuple[str, ...] = (
+        "SourceOS-Linux/sourceos-spec#113",
+        "SourceOS-Linux/sourceos-syncd#30",
+    )
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "decision_scope": self.decision_scope,
+            "runtime_effect_performed": self.runtime_effect_performed,
+            "authority_mutation_performed": self.authority_mutation_performed,
+            "state_repair_performed": self.state_repair_performed,
+            "ledger_write_performed": self.ledger_write_performed,
+            "downstream_refs": list(self.downstream_refs),
+        }
+
+
 @dataclass(frozen=True)
 class PolicyDecision:
     decision_id: str
@@ -57,6 +88,7 @@ class PolicyDecision:
     engine: str = POLICY_ENGINE
     schema: str = POLICY_DECISION_SCHEMA
     generated_at: str = field(default_factory=utc_now)
+    boundary: DecisionBoundary = field(default_factory=DecisionBoundary)
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -71,9 +103,28 @@ def as_dict(self) -> dict[str, Any]:
             "subject": self.subject,
             "object_id": self.object_id,
             "data_class": self.data_class,
+            "decision_boundary": self.boundary.as_dict(),
         }
 
 
+def validate_decision_boundary(decision: dict[str, Any]) -> None:
+    """Reject collapsed policy→runtime/authority/state records."""
+
+    boundary = decision.get("decision_boundary")
+    if not isinstance(boundary, dict):
+        raise ValueError("policy decision missing decision_boundary")
+    if boundary.get("decision_scope") != DECISION_SCOPE:
+        raise ValueError("policy decision scope must be policy-only")
+    for key in (
+        "runtime_effect_performed",
+        "authority_mutation_performed",
+        "state_repair_performed",
+        "ledger_write_performed",
+    ):
+        if boundary.get(key) is not False:
+            raise ValueError(f"policy decision must not perform {key}")
+
+
 def _decision_id(request: PolicyRequest, status: str, reason: str) -> str:
     payload = {
         "action": request.action,
@@ -131,13 +182,16 @@ def evaluate_report_policy(lanes: list[dict[str, Any]], subject: str = "sourceos
         lane_name = str(lane.get("name", "normal"))
         for action in ("index", "retain", "replicate", "agent_access"):
             decision = evaluate_policy(PolicyRequest(action=action, lane=lane_name, subject=subject))
-            decisions.append(decision.as_dict())
+            payload = decision.as_dict()
+            validate_decision_boundary(payload)
+            decisions.append(payload)
     return decisions
 
 
 def decision_counts(decisions: list[dict[str, Any]]) -> dict[str, int]:
     counts = {status: 0 for status in sorted(STATUSES)}
     for decision in decisions:
+        validate_decision_boundary(decision)
         status = str(decision.get("status", "deferred"))
         if status not in counts:
             status = "deferred"
@@ -146,6 +200,8 @@ def decision_counts(decisions: list[dict[str, Any]]) -> dict[str, int]:
 
 
 def policy_summary(decisions: list[dict[str, Any]], sample_limit: int = 12) -> dict[str, Any]:
+    for decision in decisions:
+        validate_decision_boundary(decision)
     return {
         "engine": POLICY_ENGINE,
         "counts": decision_counts(decisions),