feat(evidence): seal model identity into the bundle — #4-wedge (PR-2)

core/backends/simulator.py

@@ -144,3 +144,18 @@ def state_diff(self, twin_id: str) -> DiffResult:
     def teardown_twin(self, twin_id: str) -> None:
         # simulator: nothing to free. real backend: clab destroy + ns cleanup.
         return None
+
+    # --- model identity (the #4 wedge) ------------------------------------
+    def model_identity(self) -> dict:
+        """Deterministic attestation of the simulated generator. Fixed fields keep the
+        sealed bundle reproducible (stress INV-3). The real local/cloud backends attest a
+        weights-sha256 / identity-claim at the generate-path cutover."""
+        return {
+            "provider": "simulator",
+            "model": "deterministic-sim-v1",
+            "model_hash": None,
+            "model_hash_kind": "identity-claim",
+            "api_version": None,
+            "capabilities": [],
+            "resolved_at_utc": "1970-01-01T00:00:00+00:00",
+        }

core/llm/__init__.py

@@ -32,7 +32,7 @@
     classify,
     is_retryable,
 )
-from .identity import resolve_model_identity, unknown_identity
+from .identity import IdentitySink, resolve_model_identity, unknown_identity
 
 __all__ = [
     # egress (CROSS-1)
@@ -70,4 +70,5 @@
     # identity
     "resolve_model_identity",
     "unknown_identity",
+    "IdentitySink",
 ]

core/llm/identity.py

@@ -84,3 +84,26 @@ def unknown_identity(created_utc: str) -> dict[str, object]:
         "capabilities": [],
         "resolved_at_utc": created_utc,
     }
+
+
+class IdentitySink:
+    """Captures the ModelIdentity from the egress 'after_response' hook so the pipeline can
+    seal WHICH model produced a change into the evidence bundle (#4 wedge).
+
+    Wire-up:  egress.on("after_response", sink.capture)
+    Read:     sink.last  /  sink.as_bundle_dict()  -> pass to run_preflight(model_identity=...)
+    Process-local, last-write-wins (the egress makes one successful call per generate)."""
+
+    def __init__(self) -> None:
+        self._last: ModelIdentity | None = None
+
+    @property
+    def last(self) -> ModelIdentity | None:
+        return self._last
+
+    def capture(self, provider: str, identity: ModelIdentity) -> None:
+        """after_response(provider, identity) observer — never raises into the egress."""
+        self._last = identity
+
+    def as_bundle_dict(self) -> dict[str, object] | None:
+        return self._last.to_bundle_dict() if self._last is not None else None

core/orchestrator/pipeline.py

@@ -13,7 +13,7 @@
 
 from ..backends.base import Backend, GeneratedConfig
 from . import guards, rollback
-from ...evidence.bundler import build_bundle
+from ...evidence.bundler import build_bundle, unattested_identity
 from ...evidence.compliance import map_controls
 
 
@@ -26,7 +26,8 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn",
                   operator: str = "unknown",
                   source: str = "nl_intent",
                   approver: str | None = None,
-                  imported_configs: list[dict] | None = None) -> dict:
+                  imported_configs: list[dict] | None = None,
+                  model_identity: dict | None = None) -> dict:
     frameworks = frameworks or ["pci_dss_v4"]
     run_id = uuid.uuid4().hex
     created = datetime.now(timezone.utc).isoformat()
@@ -89,13 +90,27 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn",
         decision, reason = _verdict(bf, twin, tier, needs_approval,
                                     approval_granted)
 
+        # seal WHICH model produced the change (#4 wedge). config_import is operator-
+        # supplied (no model) -> None -> build_bundle seals the honest _UNKNOWN_IDENTITY.
+        # nl_intent ALWAYS went through a model (generate_config), so the backend MUST
+        # attest which; if it cannot, seal an explicit UNATTESTED marker -- NEVER the
+        # operator-supplied identity (that would falsely deny model authorship).
+        if model_identity is not None:
+            mi = model_identity
+        elif source == "config_import":
+            mi = None
+        else:
+            attested = backend.model_identity() if hasattr(backend, "model_identity") else None
+            mi = attested if attested is not None else unattested_identity()
+
         return build_bundle(
             run_id=run_id, created=created, operator=operator, intent=intent,
             source=source, configs=configs, batfish=bf, twin=twin, diff=diff,
             compliance=compliance, tier=tier, needs_approval=needs_approval,
             approver=approver if approval_granted else None,
             approval_utc=created if approval_granted else None,
             rollback_plan=plan, decision=decision, reason=reason,
+            model_identity=mi,
         )
     finally:
         if twin_id is not None:

evidence/bundler.py

@@ -27,12 +27,13 @@ def verify(bundle: dict) -> bool:
 def build_bundle(*, run_id, created, operator, intent, source, configs, batfish,
                  twin, diff, compliance, tier, needs_approval, approver,
                  approval_utc, rollback_plan, decision, reason,
+                 model_identity: dict | None = None,
                  signed: bool = False) -> dict:
     severity = _severity(diff["devices_affected"],
                          len(diff["sessions_dropped"]),
                          batfish["errors"], twin["converged"])
     bundle = {
-        "bundle_version": "1.0",
+        "bundle_version": "1.1",
         "run_id": run_id,
         "created_utc": created,
         "operator": operator,
@@ -50,6 +51,8 @@ def build_bundle(*, run_id, created, operator, intent, source, configs, batfish,
                 "granted_by": approver,
                 "granted_utc": approval_utc,
             },
+            "model_identity": (model_identity if model_identity is not None
+                               else _unknown_identity()),
         },
         "twin": {
             "engine": "containerlab",
@@ -98,3 +101,38 @@ def _severity(devices, dropped, errors, converged) -> str:
     if devices >= 1:
         return "low"
     return "none"
+
+
+# Fixed timestamp for synthetic identities (UNKNOWN / simulator) so stress determinism
+# (INV-3: same intent -> identical bundle) holds. Real backends carry their own resolved_at.
+_SYNTHETIC_RESOLVED_AT = "1970-01-01T00:00:00+00:00"
+
+
+def _unknown_identity() -> dict:
+    """The honest _UNKNOWN_IDENTITY for a change no model produced (config_import /
+    operator-supplied) — keeps the v1.1 bundle schema-valid WITHOUT claiming authorship."""
+    return {
+        "provider": "none",
+        "model": "operator-supplied",
+        "model_hash": None,
+        "model_hash_kind": "identity-claim",
+        "api_version": None,
+        "capabilities": [],
+        "resolved_at_utc": _SYNTHETIC_RESOLVED_AT,
+    }
+
+
+def unattested_identity() -> dict:
+    """A model DID produce this change (nl_intent) but the backend could not attest WHICH.
+    Deliberately DISTINCT from _unknown_identity (operator-supplied) so a missing
+    attestation can NEVER be read as 'no model was involved' (that would falsely deny AI
+    authorship). provider 'unknown' + model 'unattested' make the gap auditable."""
+    return {
+        "provider": "unknown",
+        "model": "unattested",
+        "model_hash": None,
+        "model_hash_kind": "identity-claim",
+        "api_version": None,
+        "capabilities": [],
+        "resolved_at_utc": _SYNTHETIC_RESOLVED_AT,
+    }

evidence/schema/evidence_bundle.schema.json

@@ -7,15 +7,16 @@
   "required": ["bundle_version", "run_id", "created_utc", "change", "twin",
                "validation", "verdict", "integrity"],
   "properties": {
-    "bundle_version": { "const": "1.0" },
+    "bundle_version": { "const": "1.1" },
     "run_id":         { "type": "string", "minLength": 8 },
     "created_utc":    { "type": "string", "format": "date-time" },
     "operator":       { "type": "string" },
 
     "change": {
       "type": "object",
       "additionalProperties": false,
-      "required": ["intent", "source", "generated_configs", "risk_tier", "approval"],
+      "required": ["intent", "source", "generated_configs", "risk_tier", "approval",
+                   "model_identity"],
       "properties": {
         "intent": { "type": "string" },
         "source": { "enum": ["nl_intent", "config_import"] },
@@ -43,6 +44,23 @@
             "granted_by":  { "type": ["string", "null"] },
             "granted_utc": { "type": ["string", "null"], "format": "date-time" }
           }
+        },
+
+        "model_identity": {
+          "type": "object",
+          "additionalProperties": false,
+          "description": "The proposing model, sealed into the bundle SHA-256 (the #4 wedge). LOCAL backends carry a real weights-sha256; CLOUD backends carry an honest identity-claim with model_hash=null; config_import/no-model runs carry _UNKNOWN_IDENTITY (provider 'none'); the simulator attests provider 'simulator'.",
+          "required": ["provider", "model", "model_hash", "model_hash_kind",
+                       "api_version", "capabilities", "resolved_at_utc"],
+          "properties": {
+            "provider":        { "enum": ["openai-compatible-local", "anthropic-cloud", "simulator", "none", "unknown"] },
+            "model":           { "type": "string" },
+            "model_hash":      { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" },
+            "model_hash_kind": { "enum": ["weights-sha256", "identity-claim"] },
+            "api_version":     { "type": ["string", "null"] },
+            "capabilities":    { "type": "array", "items": { "enum": ["chat", "json", "tools"] } },
+            "resolved_at_utc": { "type": "string" }
+          }
         }
       }
     },
@@ -112,7 +130,7 @@
             "additionalProperties": false,
             "required": ["framework", "control", "status", "evidence"],
             "properties": {
-              "framework": { "enum": ["pci_dss_v4", "soc2", "nist_800-53"] },
+              "framework": { "enum": ["pci_dss_v4", "soc2", "nist_800-53", "eu_ai_act"] },
               "control":   { "type": "string" },
               "status":    { "enum": ["pass", "fail", "n/a"] },
               "evidence":  { "type": "string" }

tests/stress_test.py

@@ -77,6 +77,8 @@ def state_diff(self, twin_id):
     def teardown_twin(self, twin_id):
         self.torn += 1
         return self.inner.teardown_twin(twin_id)
+    def model_identity(self):
+        return self.inner.model_identity()
 
 
 def _random_intent(rng: random.Random) -> str: