diff --git a/core/backends/simulator.py b/core/backends/simulator.py
index a99a4e2..823732d 100644
--- a/core/backends/simulator.py
+++ b/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",
+ }
diff --git a/core/llm/__init__.py b/core/llm/__init__.py
new file mode 100644
index 0000000..f1115dc
--- /dev/null
+++ b/core/llm/__init__.py
@@ -0,0 +1,74 @@
+"""aegis/core/llm/ — the single shared LLM egress (CROSS-1) + model-agnostic adapter (#4).
+
+The ONLY import surface other modules should use. The single-egress test asserts that no
+module outside aegis/core/llm/ imports `anthropic` or POSTs to an LLM completion URL — so
+everyone consumes the egress through these exports.
+"""
+from __future__ import annotations
+
+from .adapter import LLMAdapter
+from .backends import (
+ AnthropicCloudBackend,
+ Capability,
+ HashKind,
+ LLMAttempt,
+ LLMBackend,
+ LLMResult,
+ ModelIdentity,
+ OpenAICompatLocalBackend,
+ Provider,
+ Tier,
+ build_backend,
+)
+from .config import AdapterConfig, BackendSpec, capabilities_for
+from .egress import HookBus, LLMEgress, LLMRequest
+from .errors import (
+ AdapterError,
+ AuthError,
+ ContextExceeded,
+ Permanent,
+ RateLimited,
+ Transient,
+ classify,
+ is_retryable,
+)
+from .identity import IdentitySink, resolve_model_identity, unknown_identity
+
+__all__ = [
+ # egress (CROSS-1)
+ "LLMEgress",
+ "LLMRequest",
+ "HookBus",
+ # adapter façade
+ "LLMAdapter",
+ # backends
+ "LLMBackend",
+ "OpenAICompatLocalBackend",
+ "AnthropicCloudBackend",
+ "build_backend",
+ # data structures
+ "LLMResult",
+ "LLMAttempt",
+ "ModelIdentity",
+ "Provider",
+ "Capability",
+ "HashKind",
+ "Tier",
+ # config
+ "AdapterConfig",
+ "BackendSpec",
+ "capabilities_for",
+ # errors
+ "AdapterError",
+ "AuthError",
+ "RateLimited",
+ "ContextExceeded",
+ "Transient",
+ "Permanent",
+ "classify",
+ "is_retryable",
+ # identity
+ "resolve_model_identity",
+ "unknown_identity",
+ "IdentitySink",
+]
diff --git a/core/llm/adapter.py b/core/llm/adapter.py
new file mode 100644
index 0000000..8633c75
--- /dev/null
+++ b/core/llm/adapter.py
@@ -0,0 +1,64 @@
+"""aegis/core/llm/adapter.py — back-compat façade over the single egress.
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+The fallback-chain logic lives in ONE place — egress.py (LLMEgress). LLMAdapter is a thin
+keyword-args façade preserving the signature the legacy sync callers expect
+(HttpBackend.generate_config, multivendor _llm_query), so there is exactly one chain
+implementation to audit — not two. This is the anti-fragmentation invariant the whole
+proposal set exists to enforce.
+"""
+from __future__ import annotations
+
+from .backends import LLMResult, Tier
+from .egress import HookBus, LLMEgress, LLMRequest
+
+
+class LLMAdapter:
+ """Keyword-args façade. Delegates the chain + retry/cooldown to LLMEgress."""
+
+ def __init__(self, egress: LLMEgress) -> None:
+ self._egress = egress
+
+ @classmethod
+ def from_env(cls, hooks: HookBus | None = None) -> "LLMAdapter":
+ """Build config + backends from env (air-gap checks fire at backend construction)."""
+ return cls(LLMEgress.from_env(hooks=hooks))
+
+ async def complete(
+ self,
+ *,
+ system: str,
+ user: str,
+ max_tokens: int = 500,
+ temperature: float = 0.1,
+ tier: Tier = "cheap",
+ ) -> LLMResult:
+ return await self._egress.complete(
+ LLMRequest(
+ system=system,
+ user=user,
+ max_tokens=max_tokens,
+ temperature=temperature,
+ tier=tier,
+ )
+ )
+
+ def complete_sync(
+ self,
+ *,
+ system: str,
+ user: str,
+ max_tokens: int = 500,
+ temperature: float = 0.1,
+ tier: Tier = "cheap",
+ ) -> LLMResult:
+ return self._egress.complete_sync(
+ LLMRequest(
+ system=system,
+ user=user,
+ max_tokens=max_tokens,
+ temperature=temperature,
+ tier=tier,
+ )
+ )
diff --git a/core/llm/airgap.py b/core/llm/airgap.py
new file mode 100644
index 0000000..4533fde
--- /dev/null
+++ b/core/llm/airgap.py
@@ -0,0 +1,84 @@
+"""aegis/core/llm/airgap.py — air-gap fail-closed invariant (THE WEDGE, part 1).
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+A HARD code invariant that fails BEFORE a request is built — strictly stronger than
+``network_mode: none`` at the container layer:
+ 1. refuses to construct the anthropic-cloud backend at all,
+ 2. refuses any non-loopback base URL,
+ 3. asserts the `anthropic` SDK was never imported into the process.
+
+Import-safe, fully type-hinted, no network — unit-testable on day one. No secrets touched.
+
+NOTE: AEGIS's cloud backend uses raw httpx (no `import anthropic`) precisely so that
+invariant #3 holds even when the cloud path IS exercised in a non-air-gapped build.
+
+Security: `is_loopback` parses the host with `ipaddress` (NOT a string prefix), so a
+spoofed hostname like ``127.evil.com`` or ``127.0.0.1.attacker.com`` — which would pass a
+naive ``startswith("127.")`` yet resolve OFF-perimeter — is correctly rejected. No DNS
+resolution is performed: resolving an attacker-supplied host would itself be egress, which
+is exactly what air-gap mode forbids.
+"""
+from __future__ import annotations
+
+import sys
+from ipaddress import ip_address
+from urllib.parse import urlparse
+
+_CLOUD_PROVIDER = "anthropic-cloud"
+
+# Literal non-IP hostnames allowed in air-gap mode (loopback aliases only).
+_LOOPBACK_NAMES: frozenset[str] = frozenset({"localhost", "ip6-localhost"})
+
+
+def is_loopback(url: str) -> bool:
+ """True only for a genuine loopback address (127.0.0.0/8 or ::1), the literal name
+ 'localhost', or a reserved, non-routable '.local' mDNS host (in-perimeter only).
+
+ Uses strict `ipaddress` parsing rather than a string prefix, so a host that merely
+ *starts with* "127." but is actually a routable name (e.g. ``127.evil.com``) is
+ rejected. An empty/relative host is rejected (never auto-allowed). No DNS lookup is
+ performed — resolving an external host would itself violate the air-gap.
+ """
+ host = (urlparse(url).hostname or "").lower()
+ if not host:
+ return False # reject empty / relative — never auto-allow
+ if host in _LOOPBACK_NAMES:
+ return True
+ try:
+ return ip_address(host).is_loopback # 127.0.0.0/8 and ::1, strictly
+ except ValueError:
+ # Not a literal IP. Only the reserved, non-internet-routable mDNS suffix is
+ # treated as in-perimeter (RFC 6762); everything else is non-loopback.
+ return host.endswith(".local")
+
+
+def assert_airgap_ok(provider: str, base_url: str) -> None:
+ """HARD invariant — raises RuntimeError BEFORE any socket opens.
+
+ Call at backend *construction* time so an air-gap misconfiguration fails closed at
+ startup, not mid-pipeline. A failure here means the pipeline produces NO bundle — the
+ correct fail-closed behavior (AEGIS never seals evidence for a change a model could
+ not have produced).
+ """
+ if provider == _CLOUD_PROVIDER:
+ raise RuntimeError(
+ "AEGIS_AIRGAP=1: cloud backend (anthropic-cloud) is forbidden in air-gap mode"
+ )
+ if not is_loopback(base_url):
+ raise RuntimeError(f"AEGIS_AIRGAP=1: non-loopback egress forbidden: {base_url!r}")
+ if "anthropic" in sys.modules:
+ raise RuntimeError(
+ "AEGIS_AIRGAP=1: the anthropic SDK is loaded in-process — aborting "
+ "(in-process exposure violates the air-gap trust surface)"
+ )
+
+
+def assert_no_cloud_sdk_imported() -> None:
+ """Standalone check for the single-egress test and a startup guard: in air-gap mode
+ the cloud SDK must never be importable into the live process."""
+ if "anthropic" in sys.modules:
+ raise RuntimeError(
+ "air-gap invariant: `anthropic` is present in sys.modules — the cloud SDK "
+ "must stay unimported (AEGIS uses raw httpx for the cloud path)"
+ )
diff --git a/core/llm/backends.py b/core/llm/backends.py
new file mode 100644
index 0000000..12578c9
--- /dev/null
+++ b/core/llm/backends.py
@@ -0,0 +1,291 @@
+"""aegis/core/llm/backends.py — backend Protocol + the two concrete backends.
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+Two backends, no more (NOT a 100+-provider router):
+ - OpenAICompatLocalBackend : POST {base}/v1/chat/completions (vLLM/Ollama/llama.cpp/LM Studio)
+ - AnthropicCloudBackend : POST {base}/v1/messages (RAW httpx — NO SDK import)
+
+Why raw httpx for the cloud path: importing the `anthropic` SDK would put it in
+sys.modules and break the air-gap invariant (airgap.assert_no_cloud_sdk_imported). A raw
+POST keeps the trust surface minimal and the air-gap claim provable.
+
+The local backend's request body is BYTE-COMPATIBLE with what HttpBackend.generate_config()
+posts today (http_backend.py:166-177): {"model","messages":[system,user],"temperature",
+"stream":false} (+ max_tokens, which OpenAI-compat runtimes accept).
+
+All data structures (LLMResult / ModelIdentity / LLMAttempt) are frozen/immutable. The
+backends accept an optional httpx transport so tests drive them with httpx.MockTransport
+(no sockets).
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Literal, Protocol, runtime_checkable
+
+import httpx
+
+from .config import BackendSpec
+from .errors import AdapterError, classify
+
+if TYPE_CHECKING: # avoid importing for runtime; type hints only
+ pass
+
+Provider = Literal["openai-compatible-local", "anthropic-cloud"]
+Capability = Literal["chat", "json", "tools"]
+HashKind = Literal["weights-sha256", "identity-claim"]
+Tier = Literal["cheap", "deep"]
+
+
+@dataclass(frozen=True)
+class ModelIdentity:
+ """Goes verbatim into change.model_identity in the sealed bundle (THE WEDGE).
+
+ LOCAL backend: model_hash is a real SHA-256 of the served weights file
+ (model_hash_kind="weights-sha256"). CLOUD backend: model_hash is None with
+ kind="identity-claim" — NEVER a faked weight hash for a hosted API.
+ """
+
+ provider: Provider
+ model: str
+ model_hash: str | None
+ model_hash_kind: HashKind
+ api_version: str | None
+ capabilities: tuple[Capability, ...] # tuple => immutable + hashable
+ resolved_at_utc: str # ISO-8601
+
+ def to_bundle_dict(self) -> dict[str, object]:
+ """Exact shape sealed into bundle['change']['model_identity'] (schema v1.1)."""
+ return {
+ "provider": self.provider,
+ "model": self.model,
+ "model_hash": self.model_hash,
+ "model_hash_kind": self.model_hash_kind,
+ "api_version": self.api_version,
+ "capabilities": list(self.capabilities),
+ "resolved_at_utc": self.resolved_at_utc,
+ }
+
+
+@dataclass(frozen=True)
+class LLMAttempt:
+ """One try in the fallback chain — recorded for the optional audit trail."""
+
+ provider: Provider
+ model: str
+ ok: bool
+ status: int | None
+ error_kind: str | None # name of the typed error, if any
+ latency_ms: int
+
+
+@dataclass(frozen=True)
+class LLMResult:
+ text: str
+ identity: ModelIdentity
+ attempts: tuple[LLMAttempt, ...]
+ total_latency_ms: int
+
+
+@runtime_checkable
+class LLMBackend(Protocol):
+ """The minimal surface the egress needs from any backend."""
+
+ provider: Provider
+
+ async def complete(
+ self, *, system: str, user: str, max_tokens: int, temperature: float = 0.1
+ ) -> tuple[str, ModelIdentity]:
+ """Single non-streamed completion. Raises a typed AdapterError on failure."""
+ ...
+
+ def identity(self) -> ModelIdentity:
+ """Resolve identity WITHOUT making a request (used for air-gap pre-checks)."""
+ ...
+
+
+def _retry_after_seconds(resp: httpx.Response) -> float | None:
+ """Prefer the HTTP Retry-After header (seconds form)."""
+ raw = resp.headers.get("retry-after")
+ if raw is None:
+ return None
+ try:
+ return float(raw)
+ except ValueError:
+ return None
+
+
+class OpenAICompatLocalBackend:
+ """POST {base}/v1/chat/completions — the universal local contract.
+
+ Construction runs the air-gap pre-check (loopback enforced when AEGIS_AIRGAP=1).
+ """
+
+ provider: Provider = "openai-compatible-local"
+
+ def __init__(
+ self,
+ spec: BackendSpec,
+ timeout_s: float = 60.0,
+ transport: httpx.BaseTransport | None = None,
+ ) -> None:
+ from .airgap import assert_airgap_ok
+
+ if os.environ.get("AEGIS_AIRGAP", "") == "1":
+ assert_airgap_ok(self.provider, spec.base_url)
+ self._spec = spec
+ self._timeout_s = timeout_s
+ self._transport = transport # injected only in tests (httpx.MockTransport)
+
+ def identity(self) -> ModelIdentity:
+ from .identity import resolve_model_identity
+
+ return resolve_model_identity(self._spec)
+
+ async def complete(
+ self, *, system: str, user: str, max_tokens: int, temperature: float = 0.1
+ ) -> tuple[str, ModelIdentity]:
+ body = {
+ "model": self._spec.model,
+ "messages": [
+ {"role": "system", "content": system},
+ {"role": "user", "content": user},
+ ],
+ "temperature": temperature,
+ "stream": False,
+ "max_tokens": max_tokens,
+ }
+ url = f"{self._spec.base_url}/v1/chat/completions"
+ try:
+ async with httpx.AsyncClient(
+ transport=self._transport, timeout=self._timeout_s
+ ) as client:
+ resp = await client.post(url, json=body)
+ except httpx.RequestError as exc: # conn reset / DNS / timeout -> Transient
+ raise classify(None, str(exc), self.provider) from exc
+
+ if resp.status_code >= 400:
+ err = classify(resp.status_code, resp.text, self.provider)
+ from .errors import RateLimited
+
+ if isinstance(err, RateLimited) and err.retry_after_s is None:
+ hdr = _retry_after_seconds(resp)
+ if hdr is not None:
+ err = RateLimited(
+ message=err.message,
+ provider=err.provider,
+ status=err.status,
+ retry_after_s=hdr,
+ )
+ raise err
+
+ try:
+ text = resp.json()["choices"][0]["message"]["content"]
+ except (KeyError, IndexError, ValueError) as exc:
+ from .errors import Permanent
+
+ raise Permanent(
+ message="malformed OpenAI-compat completion response",
+ provider=self.provider,
+ status=resp.status_code,
+ ) from exc
+ return text, self.identity()
+
+
+class AnthropicCloudBackend:
+ """POST {base}/v1/messages via RAW httpx — air-gap forbidden, no SDK import.
+
+ Reads ANTHROPIC_API_KEY at call time; the key is NEVER stored or logged.
+ """
+
+ provider: Provider = "anthropic-cloud"
+
+ def __init__(
+ self,
+ spec: BackendSpec,
+ timeout_s: float = 60.0,
+ transport: httpx.BaseTransport | None = None,
+ ) -> None:
+ from .airgap import assert_airgap_ok
+
+ # In air-gap mode this RAISES — the cloud backend must never construct.
+ if os.environ.get("AEGIS_AIRGAP", "") == "1":
+ assert_airgap_ok(self.provider, spec.base_url)
+ self._spec = spec
+ self._timeout_s = timeout_s
+ self._transport = transport
+
+ def identity(self) -> ModelIdentity:
+ from .identity import resolve_model_identity
+
+ return resolve_model_identity(self._spec)
+
+ async def complete(
+ self, *, system: str, user: str, max_tokens: int, temperature: float = 0.1
+ ) -> tuple[str, ModelIdentity]:
+ from .errors import AuthError, Permanent
+
+ api_key = os.environ.get("ANTHROPIC_API_KEY")
+ if not api_key:
+ raise AuthError(
+ message="ANTHROPIC_API_KEY is not set", provider=self.provider, status=401
+ )
+ headers = {
+ "x-api-key": api_key,
+ "anthropic-version": self._spec.api_version or "2023-06-01",
+ "content-type": "application/json",
+ }
+ body = {
+ "model": self._spec.model,
+ "max_tokens": max_tokens,
+ "temperature": temperature,
+ "system": system,
+ "messages": [{"role": "user", "content": user}],
+ }
+ url = f"{self._spec.base_url}/v1/messages"
+ try:
+ async with httpx.AsyncClient(
+ transport=self._transport, timeout=self._timeout_s
+ ) as client:
+ resp = await client.post(url, json=body, headers=headers)
+ except httpx.RequestError as exc:
+ raise classify(None, str(exc), self.provider) from exc
+
+ if resp.status_code >= 400:
+ err = classify(resp.status_code, resp.text, self.provider)
+ from .errors import RateLimited
+
+ if isinstance(err, RateLimited) and err.retry_after_s is None:
+ hdr = _retry_after_seconds(resp)
+ if hdr is not None:
+ err = RateLimited(
+ message=err.message,
+ provider=err.provider,
+ status=err.status,
+ retry_after_s=hdr,
+ )
+ raise err
+
+ try:
+ text = resp.json()["content"][0]["text"]
+ except (KeyError, IndexError, ValueError) as exc:
+ raise Permanent(
+ message="malformed Anthropic messages response",
+ provider=self.provider,
+ status=resp.status_code,
+ ) from exc
+ return text, self.identity()
+
+
+def build_backend(
+ spec: BackendSpec,
+ timeout_s: float = 60.0,
+ transport: httpx.BaseTransport | None = None,
+) -> LLMBackend:
+ """Factory: construct the right backend for a spec (air-gap checks fire here)."""
+ if spec.provider == "openai-compatible-local":
+ return OpenAICompatLocalBackend(spec, timeout_s=timeout_s, transport=transport)
+ if spec.provider == "anthropic-cloud":
+ return AnthropicCloudBackend(spec, timeout_s=timeout_s, transport=transport)
+ raise AdapterError(message=f"unknown provider {spec.provider!r}", provider=spec.provider)
diff --git a/core/llm/config.py b/core/llm/config.py
new file mode 100644
index 0000000..fc4a989
--- /dev/null
+++ b/core/llm/config.py
@@ -0,0 +1,114 @@
+"""aegis/core/llm/config.py — immutable, env-driven adapter configuration.
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+Cost-aware defaults: the fallback chain is LOCAL-FIRST. Cloud is fallback only, and is
+DROPPED from the chain entirely in air-gap mode (defense in depth — also refused at
+construction by airgap.py).
+
+Env vars read:
+ AEGIS_AIRGAP "1" => air-gap mode (drop cloud, refuse non-loopback)
+ AEGIS_LLM_URL local OpenAI-compat base (default http://localhost:11434)
+ AEGIS_LLM_MODEL local model id (default ai/qwen3:latest)
+ AEGIS_LLM_WEIGHTS path to local weights file for sha256 attestation (optional)
+ MODEL_RUNNER_URL alt local base (Docker Model Runner :12434), optional
+ ANTHROPIC_API_KEY presence enables the cloud fallback (NEVER logged/echoed)
+ ANTHROPIC_MODEL cloud model id (default claude-haiku-4-5-20251001)
+ ANTHROPIC_VERSION cloud api version (default 2023-06-01)
+
+Import-safe, fully type-hinted. `from_env()` is side-effect-free (reads env, builds frozen
+specs). NO secret material is stored on the config — only a derived "cloud available" bool.
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+# Static capability map keyed by model string. Manual refresh only — auto-probing a model
+# would risk egress in air-gap mode. Conservative default ('chat',) for unknown models.
+_CAPABILITIES: dict[str, tuple[str, ...]] = {
+ "ai/qwen3:latest": ("chat", "json"),
+ "gemma4:latest": ("chat",),
+ "claude-haiku-4-5-20251001": ("chat", "json", "tools"),
+}
+
+_DEFAULT_LOCAL_URL = "http://localhost:11434"
+_DEFAULT_LOCAL_MODEL = "ai/qwen3:latest"
+_DEFAULT_CLOUD_MODEL = "claude-haiku-4-5-20251001"
+_DEFAULT_CLOUD_VERSION = "2023-06-01"
+_CLOUD_BASE_URL = "https://api.anthropic.com"
+
+
+@dataclass(frozen=True)
+class BackendSpec:
+ """One ordered link in the fallback chain. Immutable."""
+
+ provider: str # "openai-compatible-local" | "anthropic-cloud"
+ base_url: str
+ model: str
+ api_version: str | None = None
+ weights_path: str | None = None # local only; enables weights-sha256 attestation
+
+
+@dataclass(frozen=True)
+class AdapterConfig:
+ """Immutable adapter configuration. Carries NO secret material."""
+
+ chain: tuple[BackendSpec, ...] # ordered fallback: local first (cost-aware)
+ airgap: bool
+ cloud_available: bool # derived from ANTHROPIC_API_KEY presence; never the key itself
+ max_retries: int = 2
+ base_backoff_s: float = 0.5
+ cooldown_s: float = 60.0 # LiteLLM's failing-provider cooldown idea
+ timeout_s: float = 60.0
+
+ @staticmethod
+ def from_env() -> "AdapterConfig":
+ """Build an immutable config from the environment.
+
+ In air-gap mode the anthropic-cloud spec is DROPPED from the chain here (defense
+ in depth) and ALSO refused at construction by airgap.assert_airgap_ok().
+ """
+ airgap = os.environ.get("AEGIS_AIRGAP", "") == "1"
+
+ local_url = (
+ os.environ.get("AEGIS_LLM_URL")
+ or os.environ.get("MODEL_RUNNER_URL")
+ or _DEFAULT_LOCAL_URL
+ )
+ local_model = os.environ.get("AEGIS_LLM_MODEL", _DEFAULT_LOCAL_MODEL)
+ weights_path = os.environ.get("AEGIS_LLM_WEIGHTS") or None
+
+ local_spec = BackendSpec(
+ provider="openai-compatible-local",
+ base_url=local_url.rstrip("/"),
+ model=local_model,
+ api_version=None,
+ weights_path=weights_path,
+ )
+
+ chain: list[BackendSpec] = [local_spec]
+
+ # presence-only check — the key is NEVER stored on the config object
+ cloud_available = bool(os.environ.get("ANTHROPIC_API_KEY"))
+ if cloud_available and not airgap:
+ chain.append(
+ BackendSpec(
+ provider="anthropic-cloud",
+ base_url=_CLOUD_BASE_URL,
+ model=os.environ.get("ANTHROPIC_MODEL", _DEFAULT_CLOUD_MODEL),
+ api_version=os.environ.get("ANTHROPIC_VERSION", _DEFAULT_CLOUD_VERSION),
+ weights_path=None,
+ )
+ )
+
+ return AdapterConfig(
+ chain=tuple(chain),
+ airgap=airgap,
+ cloud_available=cloud_available,
+ )
+
+
+def capabilities_for(model: str) -> tuple[str, ...]:
+ """Conservative default ('chat',) for any model not in the static map."""
+ return _CAPABILITIES.get(model, ("chat",))
diff --git a/core/llm/egress.py b/core/llm/egress.py
new file mode 100644
index 0000000..9de05e0
--- /dev/null
+++ b/core/llm/egress.py
@@ -0,0 +1,225 @@
+"""aegis/core/llm/egress.py — the single shared LLM egress + hook bus (CROSS-1).
+
+ONE place every LLM call leaves the process. An ordered fallback chain (local-first,
+cost-aware) with retry / exponential backoff / per-provider cooldown — LiteLLM's decision
+*logic* with none of its dependency tree — wrapped in a 4-event hook bus so #2
+(cache_control), #4 (model-identity), and #1 (Reflexion) attach WITHOUT forking the egress.
+
+The core knows nothing about caching, identity schemas, or reflexion — it only emits events.
+This is the load-bearing decoupling: the chain stays small and auditable; features are tenants.
+
+Immutability: LLMRequest / LLMResult / ModelIdentity / LLMAttempt are all frozen. The ONE
+mutable bit is the per-provider cooldown map (process-local, never part of a returned result).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections import defaultdict
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from typing import Literal
+
+from .backends import LLMAttempt, LLMBackend, LLMResult, ModelIdentity, build_backend
+from .config import AdapterConfig
+from .errors import AdapterError, RateLimited, is_retryable
+
+Tier = Literal["cheap", "deep"]
+HookEvent = Literal["before_request", "build_payload", "after_response", "on_result"]
+
+_log = logging.getLogger("aegis.llm.egress")
+
+
+@dataclass(frozen=True)
+class LLMRequest:
+ """Immutable call envelope. trace_id lets #1 correlate generate→verify→repair."""
+
+ system: str
+ user: str
+ max_tokens: int = 500
+ temperature: float = 0.1
+ tier: Tier = "cheap"
+ trace_id: str | None = None
+ # opaque per-call hints (e.g. #2's prefix_kind); the core never interprets these.
+ hints: Mapping[str, object] = field(default_factory=dict)
+
+
+class HookBus:
+ """Deterministic, in-process, synchronous hook dispatch — the ENTIRE extension surface.
+
+ Transform hooks (before_request / build_payload) may return a replacement value;
+ observe hooks (after_response / on_result) return None. A hook that raises is ISOLATED:
+ the egress logs and continues. A broken cache/identity hook must NEVER break an LLM call.
+ """
+
+ def __init__(self) -> None:
+ self._hooks: dict[HookEvent, list[Callable]] = defaultdict(list)
+
+ def register(self, event: HookEvent, fn: Callable) -> None:
+ self._hooks[event].append(fn)
+
+ def transform(self, event: HookEvent, value, *args):
+ for fn in self._hooks[event]:
+ try:
+ out = fn(value, *args)
+ if out is not None:
+ value = out
+ except Exception as exc: # noqa: BLE001 — isolation is the point
+ _log_hook_error(event, fn, exc)
+ return value
+
+ def observe(self, event: HookEvent, *args) -> None:
+ for fn in self._hooks[event]:
+ try:
+ fn(*args)
+ except Exception as exc: # noqa: BLE001
+ _log_hook_error(event, fn, exc)
+
+
+class LLMEgress:
+ """The single egress. Ordered fallback chain + retry/cooldown + hook bus."""
+
+ def __init__(
+ self,
+ config: AdapterConfig,
+ backends: tuple[LLMBackend, ...],
+ hooks: HookBus | None = None,
+ ) -> None:
+ self._cfg = config
+ self._backends = backends # already air-gap-checked at construction
+ self._hooks = hooks or HookBus()
+ self._cooldown_until: dict[str, float] = {} # the ONE mutable bit (process-local)
+
+ @classmethod
+ def from_env(cls, hooks: HookBus | None = None) -> "LLMEgress":
+ cfg = AdapterConfig.from_env() # drops the cloud backend when AEGIS_AIRGAP=1
+ backends = tuple(build_backend(s, timeout_s=cfg.timeout_s) for s in cfg.chain)
+ return cls(cfg, backends, hooks)
+
+ def on(self, event: HookEvent, fn: Callable) -> "LLMEgress":
+ """Register a hook. #2/#4/#1 call this once at wire-up. Fluent for chaining.
+
+ NOTE: ``build_payload`` is registerable today and is fired by #2 once the backends
+ accept a payload hook; PR-1 fires before_request / after_response / on_result.
+ """
+ self._hooks.register(event, fn)
+ return self
+
+ async def complete(self, req: LLMRequest) -> LLMResult:
+ req = self._hooks.transform("before_request", req) # ← #2 / #1 hook point
+ ordered = self._ordered_backends(req.tier)
+ attempts: list[LLMAttempt] = []
+ run_t0 = time.perf_counter()
+
+ for backend in ordered:
+ if self._is_cooled_down(backend.provider):
+ continue
+ outcome = await self._try_backend(backend, req, attempts)
+ if outcome is not None:
+ text, identity = outcome
+ self._hooks.observe("after_response", backend.provider, identity) # ← #4
+ total_ms = int((time.perf_counter() - run_t0) * 1000)
+ result = LLMResult(
+ text=text,
+ identity=identity,
+ attempts=tuple(attempts),
+ total_latency_ms=total_ms,
+ )
+ self._hooks.observe("on_result", result) # ← #1 / #4 hook point
+ return result
+
+ total_ms = int((time.perf_counter() - run_t0) * 1000)
+ raise AdapterError(
+ message=f"all backends failed ({len(attempts)} attempts, {total_ms}ms)",
+ provider="egress",
+ status=None,
+ )
+
+ def complete_sync(self, req: LLMRequest) -> LLMResult:
+ """Sync bridge for legacy callers. Loop-safe: if already inside an event loop
+ (Flask under an async server) offload to a worker thread instead of asyncio.run
+ (which would raise 'event loop already running')."""
+ try:
+ asyncio.get_running_loop()
+ except RuntimeError:
+ return asyncio.run(self.complete(req))
+ import concurrent.futures
+
+ with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+ return pool.submit(lambda: asyncio.run(self.complete(req))).result()
+
+ # ── internals ────────────────────────────────────────────────────────
+ def _ordered_backends(self, tier: Tier) -> tuple[LLMBackend, ...]:
+ """Cost-aware ordering. 'cheap' => chain order (local first). 'deep' => prefer a
+ cloud backend first when present + not air-gapped (in air-gap 'deep' stays local)."""
+ if tier == "deep" and not self._cfg.airgap:
+ cloud = tuple(b for b in self._backends if b.provider == "anthropic-cloud")
+ rest = tuple(b for b in self._backends if b.provider != "anthropic-cloud")
+ return cloud + rest
+ return self._backends
+
+ def _is_cooled_down(self, provider: str) -> bool:
+ until = self._cooldown_until.get(provider)
+ return until is not None and time.monotonic() < until
+
+ def _cool_down(self, provider: str) -> None:
+ self._cooldown_until[provider] = time.monotonic() + self._cfg.cooldown_s
+
+ async def _try_backend(
+ self, backend: LLMBackend, req: LLMRequest, attempts: list[LLMAttempt]
+ ) -> tuple[str, ModelIdentity] | None:
+ """Retry one backend ≤ max_retries. Append an LLMAttempt per try. Returns
+ (text, identity) on success, else None (skip to the next backend in the chain)."""
+ for attempt in range(self._cfg.max_retries + 1):
+ t0 = time.perf_counter()
+ try:
+ text, identity = await backend.complete(
+ system=req.system,
+ user=req.user,
+ max_tokens=req.max_tokens,
+ temperature=req.temperature,
+ )
+ attempts.append(
+ LLMAttempt(
+ provider=backend.provider,
+ model=identity.model,
+ ok=True,
+ status=200,
+ error_kind=None,
+ latency_ms=int((time.perf_counter() - t0) * 1000),
+ )
+ )
+ return text, identity
+ except AdapterError as err:
+ attempts.append(
+ LLMAttempt(
+ provider=backend.provider,
+ model="unknown",
+ ok=False,
+ status=err.status,
+ error_kind=type(err).__name__,
+ latency_ms=int((time.perf_counter() - t0) * 1000),
+ )
+ )
+ if not is_retryable(err) or attempt >= self._cfg.max_retries:
+ if isinstance(err, RateLimited):
+ self._cool_down(backend.provider)
+ return None
+ await asyncio.sleep(self._backoff_s(attempt, err))
+ return None
+
+ def _backoff_s(self, attempt: int, err: AdapterError) -> float:
+ """Exponential backoff; honor Retry-After on 429 when present."""
+ if isinstance(err, RateLimited) and err.retry_after_s is not None:
+ return err.retry_after_s
+ return self._cfg.base_backoff_s * (2**attempt)
+
+
+def _log_hook_error(event: HookEvent, fn: Callable, exc: Exception) -> None:
+ _log.warning(
+ "hook %s on %s failed (isolated, call continues): %s",
+ getattr(fn, "__name__", fn),
+ event,
+ exc,
+ )
diff --git a/core/llm/errors.py b/core/llm/errors.py
new file mode 100644
index 0000000..03956c6
--- /dev/null
+++ b/core/llm/errors.py
@@ -0,0 +1,128 @@
+"""aegis/core/llm/errors.py — typed error taxonomy + retry decision-logic.
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+We port LiteLLM's *ideas* — a small typed exception hierarchy, an is-retryable
+predicate, and Retry-After honoring — WITHOUT its 100+-provider dependency tree.
+Scoped to exactly the two backends AEGIS ships: anthropic-cloud + openai-compat-local.
+
+Import-safe, fully type-hinted, no network. `classify()` is real so the taxonomy is
+unit-testable on day one.
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class AdapterError(Exception):
+ """Base typed error. Shape mirrors LiteLLM's exception (message/provider/status)
+ minus the dependency. Frozen => immutable per coding-style rule."""
+
+ message: str
+ provider: str
+ status: int | None = None
+
+ def __str__(self) -> str: # keep secrets out of logs — only structural fields
+ return f"{type(self).__name__}[{self.provider} status={self.status}]: {self.message}"
+
+
+@dataclass(frozen=True)
+class AuthError(AdapterError):
+ """401 / 403 — bad or missing key. NEVER retried. Skip to next backend."""
+
+
+@dataclass(frozen=True)
+class RateLimited(AdapterError):
+ """429 — back off, honor Retry-After, then cool the provider."""
+
+ retry_after_s: float | None = None
+
+
+@dataclass(frozen=True)
+class ContextExceeded(AdapterError):
+ """400 / 413 with context-window phrasing. NEVER retried (same prompt = same fail)."""
+
+
+@dataclass(frozen=True)
+class Transient(AdapterError):
+ """408 / 5xx / connection reset / timeout. RETRYABLE with backoff."""
+
+
+@dataclass(frozen=True)
+class Permanent(AdapterError):
+ """400 / 404 / 422 (non-context). NEVER retried."""
+
+
+# Phrases that mark a 4xx body as a context-window failure. Deliberately small —
+# scoped to the two runtimes we ship (Anthropic + OpenAI-compat), NOT a cross-provider
+# string zoo.
+_CONTEXT_PHRASES: tuple[str, ...] = (
+ "context window",
+ "maximum context length",
+ "too many tokens",
+ "context_length_exceeded",
+ "prompt is too long",
+)
+
+
+def classify(status: int | None, body: str, provider: str) -> AdapterError:
+ """Map a raw HTTP status + response body into the small internal taxonomy.
+
+ Fail-safe: an unknown 5xx / connection error is treated as retryable Transient.
+ 401/403 -> AuthError (never retry)
+ 429 -> RateLimited (retry, honor Retry-After)
+ 400/413 + context phrasing -> ContextExceeded (never retry)
+ other 4xx -> Permanent (never retry)
+ 408/5xx/None(conn error) -> Transient (retry)
+ """
+ low = (body or "").lower()
+
+ if status in (401, 403):
+ return AuthError(message="authentication failed", provider=provider, status=status)
+
+ if status == 429:
+ return RateLimited(
+ message="rate limited",
+ provider=provider,
+ status=status,
+ retry_after_s=_parse_retry_after(body),
+ )
+
+ if status in (400, 413) and any(p in low for p in _CONTEXT_PHRASES):
+ return ContextExceeded(
+ message="context window exceeded", provider=provider, status=status
+ )
+
+ if status is not None and 400 <= status < 500:
+ return Permanent(
+ message=f"permanent client error {status}", provider=provider, status=status
+ )
+
+ # status is None (connection reset / DNS / timeout) OR a 5xx OR 408 -> retryable.
+ return Transient(
+ message=f"transient error {status if status is not None else 'conn'}",
+ provider=provider,
+ status=status,
+ )
+
+
+def is_retryable(err: AdapterError) -> bool:
+ """5-line decision: retry Transient + RateLimited; never the rest."""
+ return isinstance(err, (Transient, RateLimited))
+
+
+def _parse_retry_after(body: str) -> float | None:
+ """Best-effort Retry-After extraction from a JSON-ish error body.
+
+ NOTE: prefer the HTTP ``Retry-After`` *header* when the backend surfaces it; the
+ backends pass it in via the body fallback only when the header is absent.
+ """
+ m = re.search(r'retry[_-]?after["\s:=]+(\d+(?:\.\d+)?)', (body or "").lower())
+ if m:
+ try:
+ return float(m.group(1))
+ except ValueError:
+ return None
+ return None
diff --git a/core/llm/identity.py b/core/llm/identity.py
new file mode 100644
index 0000000..1fe3450
--- /dev/null
+++ b/core/llm/identity.py
@@ -0,0 +1,109 @@
+"""aegis/core/llm/identity.py — model identity resolution (THE WEDGE, part 2).
+
+Part of PR-1 (CROSS-1 shared LLM egress + #4 model-agnostic adapter).
+
+LOCAL backend => a REAL sha256 of the served weights file ("weights-sha256").
+CLOUD backend => an honest "identity-claim" with model_hash=None (a weight hash for a
+ hosted API is physically impossible; we NEVER fake one).
+
+Multi-GB GGUF/safetensors files are hashed ONCE and cached by (path, mtime, size) so
+generate_config() does not re-hash on every inference.
+
+Import-safe, fully type-hinted, no network.
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+from datetime import datetime, timezone
+
+from .backends import ModelIdentity
+from .config import BackendSpec, capabilities_for
+
+# (path, mtime, size) -> sha256 hex. Auto-invalidates on file change.
+_HASH_CACHE: dict[tuple[str, float, int], str] = {}
+
+_LOCAL_PROVIDER = "openai-compatible-local"
+_CHUNK = 1 << 20 # 1 MiB
+
+
+def _hash_local_weights(path: str) -> str | None:
+ """SHA-256 of the served weights file, cached by (path, mtime, size). Returns None
+ if the path is empty/missing/unreadable — identity then downgrades to an honest
+ 'identity-claim' rather than failing the whole run."""
+ if not path or not os.path.exists(path):
+ return None
+ try:
+ st = os.stat(path)
+ except OSError:
+ return None
+ key = (path, st.st_mtime, st.st_size)
+ cached = _HASH_CACHE.get(key)
+ if cached is not None:
+ return cached
+ h = hashlib.sha256()
+ try:
+ with open(path, "rb") as f:
+ for chunk in iter(lambda: f.read(_CHUNK), b""):
+ h.update(chunk)
+ except OSError:
+ return None
+ digest = h.hexdigest()
+ _HASH_CACHE[key] = digest
+ return digest
+
+
+def resolve_model_identity(spec: BackendSpec) -> ModelIdentity:
+ """Local => real weights-sha256 attestation when the weights file is present.
+ Cloud => identity-claim with model_hash=None (never faked)."""
+ is_local = spec.provider == _LOCAL_PROVIDER
+ weight_hash = _hash_local_weights(spec.weights_path or "") if is_local else None
+ return ModelIdentity(
+ provider=spec.provider, # type: ignore[arg-type]
+ model=spec.model,
+ model_hash=weight_hash,
+ model_hash_kind="weights-sha256" if weight_hash else "identity-claim",
+ api_version=spec.api_version,
+ capabilities=capabilities_for(spec.model), # type: ignore[arg-type]
+ resolved_at_utc=datetime.now(timezone.utc).isoformat(),
+ )
+
+
+def unknown_identity(created_utc: str) -> dict[str, object]:
+ """_UNKNOWN_IDENTITY for config_import runs (no LLM touched the change).
+
+ Keeps every bundle schema-valid (v1.1 requires change.model_identity) and honest —
+ an operator-supplied config never claims a model produced it.
+ """
+ return {
+ "provider": "none",
+ "model": "operator-supplied",
+ "model_hash": None,
+ "model_hash_kind": "identity-claim",
+ "api_version": None,
+ "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
diff --git a/core/orchestrator/pipeline.py b/core/orchestrator/pipeline.py
index 65fe386..8c7c68a 100644
--- a/core/orchestrator/pipeline.py
+++ b/core/orchestrator/pipeline.py
@@ -13,7 +13,8 @@
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 ..risk import authority_record, classify_change, load_max_authorized, unify_severity
from ...evidence.compliance import map_controls
@@ -26,7 +27,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,6 +91,30 @@ 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()
+
+ # authority tier (#5): orthogonal to risk_tier -- a twin-clean LOW-severity change
+ # to a fabric-identity construct (AS/RD/RT) is still BLOCK. Sealed + enforced at G5.
+ authority = authority_record(
+ unify_severity(batfish_errors=bf["errors"],
+ devices_affected=diff["devices_affected"],
+ sessions_dropped=len(diff["sessions_dropped"]),
+ converged=twin["converged"]),
+ classify_change(configs),
+ load_max_authorized(),
+ )
+
return build_bundle(
run_id=run_id, created=created, operator=operator, intent=intent,
source=source, configs=configs, batfish=bf, twin=twin, diff=diff,
@@ -96,6 +122,7 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn",
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, authority=authority,
)
finally:
if twin_id is not None:
diff --git a/core/promote/gate.py b/core/promote/gate.py
index 56cb77f..1ceacb0 100644
--- a/core/promote/gate.py
+++ b/core/promote/gate.py
@@ -12,6 +12,7 @@
from dataclasses import dataclass
from ...evidence.bundler import verify
+from ..risk import Tier, authorize, load_max_authorized
@dataclass(frozen=True)
@@ -43,4 +44,22 @@ def evaluate(bundle: dict, *, approver: str | None, approval_token: str | None,
return GateDecision(False, "live connector blocked — set AEGIS_PROMOTE_ALLOW_LIVE=1 "
"and wire an audited connector")
+ # G5 authority ceiling — NO SELF-ESCALATION. The change's REQUIRED authority is sealed
+ # in the bundle (covered by the G1 integrity check); re-derive the ceiling decision
+ # here so the gate enforces its OWN ceiling at promote time, not a recorded boolean.
+ # A change needing more autonomy than the ceiling allows -- or any hard-BLOCK change
+ # (AS/RD/RT / critical) -- is never promotable.
+ authority = bundle.get("change", {}).get("authority")
+ if not authority:
+ return GateDecision(False, "no authority record — cannot confirm the autonomy bound "
+ "(fail-closed)")
+ try:
+ required = Tier[str(authority.get("required", "")).upper()]
+ except KeyError:
+ return GateDecision(False, f"unrecognized required authority {authority.get('required')!r}")
+ ceiling = load_max_authorized()
+ if not authorize(required, ceiling).allowed: # G5
+ return GateDecision(False, f"no-self-escalation: change requires {required.name} "
+ f"authority, above the {ceiling.name} ceiling — blocked")
+
return GateDecision(True, f"{tier or 'unknown'}-risk {decision} approved for promotion")
diff --git a/core/risk/__init__.py b/core/risk/__init__.py
new file mode 100644
index 0000000..709a0e4
--- /dev/null
+++ b/core/risk/__init__.py
@@ -0,0 +1,29 @@
+"""core/risk/ — deterministic authority model (#5): severity vs authority, hard-force
+overrides for fabric-identity / spine / underlay, and the no-self-escalation ceiling."""
+from __future__ import annotations
+
+from .authority import (
+ AuthorityDecision,
+ ChangeClass,
+ Severity,
+ Tier,
+ authority_record,
+ authorize,
+ classify_change,
+ load_max_authorized,
+ required_authority,
+ unify_severity,
+)
+
+__all__ = [
+ "Severity",
+ "Tier",
+ "ChangeClass",
+ "AuthorityDecision",
+ "classify_change",
+ "unify_severity",
+ "required_authority",
+ "authorize",
+ "load_max_authorized",
+ "authority_record",
+]
diff --git a/core/risk/authority.py b/core/risk/authority.py
new file mode 100644
index 0000000..95f01de
--- /dev/null
+++ b/core/risk/authority.py
@@ -0,0 +1,220 @@
+"""core/risk/authority.py — the deterministic per-risk-tier authority model (#5).
+
+Two ORTHOGONAL axes, both pure and auditable:
+
+ Severity (how bad if it goes wrong) : NONE < LOW < MEDIUM < HIGH < CRITICAL
+ Authority (how much autonomy is allowed): AUTO < HITL < HOTL < BLOCK
+
+The product insight is that these are NOT the same axis. A LOW-severity change to a
+fabric-identity construct (AS number, route-distinguisher / route-target) is still BLOCK,
+and a spine/underlay touch is never fully automated no matter how clean the twin looked.
+`required_authority` derives a base tier from severity, then applies hard-force overrides
+from the change class, and returns the MOST RESTRICTIVE of the two.
+
+`authorize` is the no-self-escalation ceiling: the computed `required` authority must be
+<= a `max_authorized` ceiling, else the change is BLOCKED — an agent can never grant
+itself more autonomy than its ceiling allows. (Binding `max_authorized` to a hardware-PIV
+-signed, compiled-in value is the #5-ceiling / CROSS-3 follow-up; this module ships the
+deterministic invariant that signature will protect — no crypto is claimed here.)
+
+`unify_severity` reconciles AEGIS's two existing severity scorers — guards.risk_tier's
+3-bucket scale and bundler._severity's 5-bucket scale — into one canonical 5-bucket scale.
+The richer engine wins; tests/authority_test.py carries the differential vs both.
+"""
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass
+from enum import IntEnum
+
+from ..backends.base import GeneratedConfig
+
+
+class Severity(IntEnum):
+ NONE = 0
+ LOW = 1
+ MEDIUM = 2
+ HIGH = 3
+ CRITICAL = 4
+
+
+class Tier(IntEnum):
+ """Authority tier — how much autonomy a verified change is allowed. Ordered so a
+ HIGHER value is MORE restrictive (less autonomy)."""
+
+ AUTO = 0 # fully automated, no human
+ HITL = 1 # human-in-the-loop: explicit approval before apply
+ HOTL = 2 # human-on-the-loop: applied under monitoring, human can intervene/abort
+ BLOCK = 3 # never auto-promotable
+
+
+@dataclass(frozen=True)
+class ChangeClass:
+ """What fabric construct the change touches — drives the hard-force overrides.
+
+ AS/RD/RT detection is regex over the candidate config and is reliable across vendors.
+ The spine/underlay flags are a documented HEURISTIC (device-name + underlay-protocol
+ patterns); a production deployment should back them with an authoritative topology
+ role source (containerlab labels / NetBox device role). Until then they bias toward
+ FAIL CLOSED — an IGP/underlay-looking change is treated as touching the underlay."""
+
+ touches_asn: bool = False
+ touches_rd_rt: bool = False
+ touches_spine: bool = False
+ touches_underlay: bool = False
+
+ @property
+ def is_fabric_identity(self) -> bool:
+ """AS number / RD / RT — the identity of the fabric. Never auto-changed."""
+ return self.touches_asn or self.touches_rd_rt
+
+
+# ── change classification (regex, fail-closed) ──────────────────────────────────
+
+_ASN_RE = re.compile(
+ r"(?:autonomous-system|remote-as|peer-as|local-as)\s+\d+|router\s+bgp\s+\d+",
+ re.IGNORECASE,
+)
+_RD_RT_RE = re.compile(
+ r"route-(?:distinguisher|target)|\brd\s+\d+:\d+|\brt\s+\d+:\d+|target:\d+:\d+",
+ re.IGNORECASE,
+)
+_UNDERLAY_CONFIG_RE = re.compile(r"\b(?:underlay|ospf|isis|is-is)\b", re.IGNORECASE)
+_SPINE_NAME_RE = re.compile(r"\bspine\b", re.IGNORECASE)
+_UNDERLAY_NAME_RE = re.compile(r"\b(?:underlay|super-?spine)\b", re.IGNORECASE)
+
+
+def classify_change(configs: list[GeneratedConfig]) -> ChangeClass:
+ """Classify candidate configs into fabric-construct flags (the hard-force triggers)."""
+ text = "\n".join(c.get("config", "") for c in configs)
+ names = " ".join(c.get("device", "") for c in configs)
+ return ChangeClass(
+ touches_asn=bool(_ASN_RE.search(text)),
+ touches_rd_rt=bool(_RD_RT_RE.search(text)),
+ touches_spine=bool(_SPINE_NAME_RE.search(names)),
+ touches_underlay=bool(_UNDERLAY_NAME_RE.search(names) or _UNDERLAY_CONFIG_RE.search(text)),
+ )
+
+
+# ── canonical severity (reconciles guards.risk_tier + bundler._severity) ─────────
+
+def unify_severity(*, batfish_errors: int, devices_affected: int,
+ sessions_dropped: int, converged: bool) -> Severity:
+ """One canonical 5-bucket severity. Mirrors bundler._severity (the richer engine);
+ the differential test shows where guards.risk_tier's 3-bucket scale collapses."""
+ if not converged or batfish_errors > 0:
+ return Severity.CRITICAL
+ if sessions_dropped > 0:
+ return Severity.HIGH
+ if devices_affected >= 4:
+ return Severity.HIGH
+ if devices_affected >= 2:
+ return Severity.MEDIUM
+ if devices_affected >= 1:
+ return Severity.LOW
+ return Severity.NONE
+
+
+# ── severity x change-class -> required authority ───────────────────────────────
+
+_SEVERITY_BASE: dict[Severity, Tier] = {
+ Severity.NONE: Tier.AUTO,
+ Severity.LOW: Tier.AUTO,
+ Severity.MEDIUM: Tier.HITL,
+ Severity.HIGH: Tier.HOTL,
+ Severity.CRITICAL: Tier.BLOCK,
+}
+
+
+def required_authority(severity: Severity, change_class: ChangeClass) -> Tier:
+ """The minimum authority tier a change requires. Base tier from severity, then
+ hard-force overrides from the change class; the result is the MOST RESTRICTIVE.
+
+ Hard-force (regardless of how clean the twin looked):
+ * AS number / RD / RT touch -> BLOCK (fabric identity is never auto-changed)
+ * spine or underlay touch -> >= HOTL (never fully automated)
+ """
+ tier = _SEVERITY_BASE[severity]
+ if change_class.is_fabric_identity:
+ tier = max(tier, Tier.BLOCK)
+ if change_class.touches_spine or change_class.touches_underlay:
+ tier = max(tier, Tier.HOTL)
+ return tier
+
+
+@dataclass(frozen=True)
+class AuthorityDecision:
+ required: Tier
+ max_authorized: Tier
+ allowed: bool
+ effective: Tier
+ reason: str
+
+
+def authorize(required: Tier, max_authorized: Tier) -> AuthorityDecision:
+ """No-self-escalation ceiling: a change may proceed at `required` ONLY if
+ required <= max_authorized; otherwise it is BLOCKED. An agent can never grant itself
+ more autonomy than its ceiling allows.
+
+ `max_authorized` is a plain input here. The #5-ceiling / CROSS-3 follow-up binds it to
+ a hardware-PIV-signed, compiled-in value so the ceiling itself is tamper-proof; this
+ function is the deterministic invariant that signature protects."""
+ allowed = required <= max_authorized
+ effective = required if allowed else Tier.BLOCK
+ if allowed:
+ reason = (f"required {required.name} <= ceiling {max_authorized.name}: "
+ f"allowed at {required.name}")
+ else:
+ reason = (f"NO-SELF-ESCALATION: required {required.name} exceeds ceiling "
+ f"{max_authorized.name} -> BLOCK")
+ return AuthorityDecision(required=required, max_authorized=max_authorized,
+ allowed=allowed, effective=effective, reason=reason)
+
+
+_DEFAULT_MAX_AUTHORIZED = Tier.HOTL
+
+
+def load_max_authorized() -> Tier:
+ """The deployment's autonomy ceiling. Env AEGIS_MAX_AUTHORIZED_TIER (AUTO|HITL|HOTL),
+ default HOTL. BLOCK is rejected as a ceiling (it would authorize fabric-identity /
+ critical changes). An invalid value RAISES (fail-closed, loud) rather than silently
+ defaulting to something permissive.
+
+ NOTE: env-configured today. The CROSS-3 follow-up binds it to a hardware-PIV signature +
+ compiled-in public key so it cannot be raised by editing the environment — the
+ tamper-proofing the no-self-escalation invariant ultimately needs. This ships the
+ deterministic enforcement that signature will protect."""
+ raw = os.environ.get("AEGIS_MAX_AUTHORIZED_TIER")
+ if not raw:
+ return _DEFAULT_MAX_AUTHORIZED
+ try:
+ tier = Tier[raw.strip().upper()]
+ except KeyError:
+ raise ValueError(f"AEGIS_MAX_AUTHORIZED_TIER must be one of AUTO|HITL|HOTL (got {raw!r})")
+ if tier == Tier.BLOCK:
+ raise ValueError("AEGIS_MAX_AUTHORIZED_TIER=BLOCK is invalid — a ceiling cannot "
+ "authorize BLOCK-tier (fabric-identity / critical) changes")
+ return tier
+
+
+def authority_record(severity: Severity, change_class: ChangeClass,
+ max_authorized: Tier) -> dict:
+ """The sealed change.authority object: severity (how bad if it fails) + the required
+ authority (how much autonomy), the ceiling it was checked against, and the
+ no-self-escalation decision. Lowercase strings to match the bundle vocabulary."""
+ required = required_authority(severity, change_class)
+ decision = authorize(required, max_authorized)
+ return {
+ "severity": severity.name.lower(),
+ "required": required.name.lower(),
+ "max_authorized": max_authorized.name.lower(),
+ "allowed": decision.allowed,
+ "effective": decision.effective.name.lower(),
+ "change_class": {
+ "touches_asn": change_class.touches_asn,
+ "touches_rd_rt": change_class.touches_rd_rt,
+ "touches_spine": change_class.touches_spine,
+ "touches_underlay": change_class.touches_underlay,
+ },
+ }
diff --git a/core/seal/__init__.py b/core/seal/__init__.py
new file mode 100644
index 0000000..6b98517
--- /dev/null
+++ b/core/seal/__init__.py
@@ -0,0 +1,32 @@
+"""core/seal/ — the CROSS-3 bounded-autonomy SEAL: a detached, offline-verifiable receipt
+binding the model identity (#4 wedge) + the bounded authority (#5 ceiling) over a specific
+evidence bundle, signed with a pinned key (a YubiKey PIV slot in production)."""
+from __future__ import annotations
+
+from .seal import (
+ SEAL_VERSION,
+ SealError,
+ SealVerdict,
+ build_claims,
+ canonical_claims_bytes,
+ seal_bundle,
+ seal_response,
+ verify_seal,
+)
+from .signing import Ed25519Signer, Ed25519Verifier, Signer, Verifier, key_id
+
+__all__ = [
+ "SEAL_VERSION",
+ "SealError",
+ "SealVerdict",
+ "build_claims",
+ "canonical_claims_bytes",
+ "seal_bundle",
+ "seal_response",
+ "verify_seal",
+ "Signer",
+ "Verifier",
+ "Ed25519Signer",
+ "Ed25519Verifier",
+ "key_id",
+]
diff --git a/core/seal/schema/seal.schema.json b/core/seal/schema/seal.schema.json
new file mode 100644
index 0000000..e5d0f02
--- /dev/null
+++ b/core/seal/schema/seal.schema.json
@@ -0,0 +1,53 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "$id": "https://aegis.local/schema/seal.schema.json",
+ "title": "AEGIS CROSS-3 Bounded-Autonomy Seal (detached receipt)",
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["seal_version", "claims", "signature"],
+ "properties": {
+ "seal_version": { "const": "1.0" },
+ "claims": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["seal_version", "bundle_sha256", "model", "authority", "sealed_at_utc"],
+ "properties": {
+ "seal_version": { "const": "1.0" },
+ "bundle_sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" },
+ "model": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["provider", "model", "model_hash", "model_hash_kind"],
+ "properties": {
+ "provider": { "type": "string" },
+ "model": { "type": "string" },
+ "model_hash": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" },
+ "model_hash_kind": { "enum": ["weights-sha256", "identity-claim"] }
+ }
+ },
+ "authority": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["required", "max_authorized", "allowed", "effective"],
+ "properties": {
+ "required": { "enum": ["auto", "hitl", "hotl", "block"] },
+ "max_authorized": { "enum": ["auto", "hitl", "hotl", "block"] },
+ "allowed": { "type": "boolean" },
+ "effective": { "enum": ["auto", "hitl", "hotl", "block"] }
+ }
+ },
+ "sealed_at_utc": { "type": "string" }
+ }
+ },
+ "signature": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["alg", "key_id", "value"],
+ "properties": {
+ "alg": { "enum": ["ed25519", "piv-ecdsa-p256"] },
+ "key_id": { "type": "string" },
+ "value": { "type": "string" }
+ }
+ }
+ }
+}
diff --git a/core/seal/seal.py b/core/seal/seal.py
new file mode 100644
index 0000000..f7df860
--- /dev/null
+++ b/core/seal/seal.py
@@ -0,0 +1,161 @@
+"""core/seal/seal.py — the CROSS-3 bounded-autonomy SEAL.
+
+A DETACHED receipt that fuses, into one offline-verifiable object, three facts no
+competitor binds together:
+ (#4 wedge) WHICH model produced the change — provider + model + model-hash
+ (#5 ceiling) that it was BOUNDED — required authority <= max_authorized
+ (integrity) over THIS exact evidence bundle — bundle_sha256
+signed with a pinned key (a YubiKey PIV slot in production).
+
+The receipt is DETACHED — it does not mutate the bundle, so the bundle's own integrity
+hash stays stable — and binds to the bundle by its sha256. VERIFY runs OFFLINE against
+only the public key (no secret), and a VALID verdict is a portable cryptographic proof
+that a named+hashed model produced this exact change AND that it was bounded by the ceiling
+— the thing competitors assert but cannot hand an auditor offline.
+"""
+from __future__ import annotations
+
+import base64
+import json
+from dataclasses import dataclass
+
+from ...evidence.bundler import verify as verify_bundle
+from ..risk import Tier
+from .signing import Signer, Verifier
+
+SEAL_VERSION = "1.0"
+
+
+class SealError(Exception):
+ """Refused to produce a seal (the bundle is unbounded or fails integrity)."""
+
+
+def _canonical(obj) -> bytes:
+ return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def canonical_claims_bytes(claims: dict) -> bytes:
+ return _canonical(claims)
+
+
+def build_claims(bundle: dict, sealed_at_utc: str) -> dict:
+ """The exact claim set the seal signs — binds model + authority + the bundle hash."""
+ mi = bundle["change"]["model_identity"]
+ au = bundle["change"]["authority"]
+ return {
+ "seal_version": SEAL_VERSION,
+ "bundle_sha256": bundle["integrity"]["sha256"],
+ "model": {
+ "provider": mi["provider"], "model": mi["model"],
+ "model_hash": mi["model_hash"], "model_hash_kind": mi["model_hash_kind"],
+ },
+ "authority": {
+ "required": au["required"], "max_authorized": au["max_authorized"],
+ "allowed": au["allowed"], "effective": au["effective"],
+ },
+ "sealed_at_utc": sealed_at_utc,
+ }
+
+
+def _is_bounded(authority: dict) -> bool:
+ if not authority.get("allowed") or authority.get("effective") == "block":
+ return False
+ try:
+ return Tier[str(authority["required"]).upper()] <= Tier[str(authority["max_authorized"]).upper()]
+ except KeyError:
+ return False
+
+
+def seal_bundle(bundle: dict, signer: Signer, *, sealed_at_utc: str) -> dict:
+ """Produce a detached seal receipt for a BOUNDED, integrity-valid bundle. Refuses
+ (SealError) to seal a tampered bundle or one whose change exceeds the autonomy ceiling
+ — you never certify a self-escalation."""
+ if not verify_bundle(bundle):
+ raise SealError("refuse to seal: bundle fails its own integrity check")
+ if not _is_bounded(bundle["change"]["authority"]):
+ raise SealError("refuse to seal: change is not within the autonomy ceiling "
+ "(no-self-escalation) — nothing to certify")
+ claims = build_claims(bundle, sealed_at_utc)
+ sig = signer.sign(canonical_claims_bytes(claims))
+ return {
+ "seal_version": SEAL_VERSION,
+ "claims": claims,
+ "signature": {
+ "alg": signer.alg(),
+ "key_id": signer.key_id(),
+ "value": base64.b64encode(sig).decode("ascii"),
+ },
+ }
+
+
+@dataclass(frozen=True)
+class SealVerdict:
+ valid: bool
+ gate: str # the gate that decided (G0..G6) or "OK"
+ reason: str
+
+
+def verify_seal(bundle: dict, seal: dict, verifier: Verifier) -> SealVerdict:
+ """Offline verification — uses ONLY the pinned public key. Runs gates G0..G6; the first
+ failure decides."""
+ # G0 — structure
+ try:
+ claims = seal["claims"]
+ sig = seal["signature"]
+ alg, kid, value = sig["alg"], sig["key_id"], sig["value"]
+ b_sha, model, authority = claims["bundle_sha256"], claims["model"], claims["authority"]
+ except (KeyError, TypeError):
+ return SealVerdict(False, "G0", "malformed seal")
+
+ # G1 — key pinning: only the pinned key + alg are trusted
+ if alg != verifier.alg() or kid != verifier.key_id():
+ return SealVerdict(False, "G1", f"unpinned key/alg ({alg}/{kid})")
+
+ # G2 — signature over the canonical claims
+ try:
+ ok = verifier.verify(canonical_claims_bytes(claims), base64.b64decode(value))
+ except Exception: # noqa: BLE001 — malformed base64, etc.
+ return SealVerdict(False, "G2", "signature decode error")
+ if not ok:
+ return SealVerdict(False, "G2", "signature does not verify against the pinned key")
+
+ # G3 — the seal covers THIS bundle
+ if b_sha != bundle.get("integrity", {}).get("sha256"):
+ return SealVerdict(False, "G3", "seal bundle_sha256 does not match the bundle")
+
+ # G4 — model identity bound + honest (an identity-claim never carries a faked weight hash)
+ mi = bundle.get("change", {}).get("model_identity", {})
+ if (model.get("provider") != mi.get("provider") or model.get("model") != mi.get("model")
+ or model.get("model_hash") != mi.get("model_hash")
+ or model.get("model_hash_kind") != mi.get("model_hash_kind")):
+ return SealVerdict(False, "G4", "sealed model identity does not match the bundle")
+ if model.get("model_hash_kind") == "identity-claim" and model.get("model_hash") is not None:
+ return SealVerdict(False, "G4", "identity-claim must not carry a weight hash")
+
+ # G5 — bounded autonomy: matches the bundle AND is within the ceiling
+ au = bundle.get("change", {}).get("authority", {})
+ if (authority.get("required") != au.get("required")
+ or authority.get("max_authorized") != au.get("max_authorized")
+ or authority.get("allowed") != au.get("allowed")
+ or authority.get("effective") != au.get("effective")):
+ return SealVerdict(False, "G5", "sealed authority does not match the bundle")
+ if not _is_bounded(authority):
+ return SealVerdict(False, "G5", "sealed change exceeds the autonomy ceiling")
+
+ # G6 — the bundle itself integrity-verifies
+ if not verify_bundle(bundle):
+ return SealVerdict(False, "G6", "bundle fails its own integrity check")
+
+ return SealVerdict(True, "OK", "named+hashed model and bounded authority sealed over "
+ "this bundle; signature valid against the pinned key")
+
+
+def seal_response(bundle: dict, signer: Signer, sealed_at_utc: str) -> tuple[dict | None, str | None]:
+ """Live-path convenience: returns (receipt, None) for a bounded, integrity-valid bundle,
+ or (None, reason) when the change cannot be sealed (unbounded / tampered). The caller
+ embeds the receipt under bundle['seal'] (excluded from the bundle hash) so a run emits
+ its proof in one response."""
+ try:
+ return seal_bundle(bundle, signer, sealed_at_utc=sealed_at_utc), None
+ except SealError as exc:
+ return None, str(exc)
diff --git a/core/seal/signing.py b/core/seal/signing.py
new file mode 100644
index 0000000..a356e69
--- /dev/null
+++ b/core/seal/signing.py
@@ -0,0 +1,106 @@
+"""core/seal/signing.py — detached-signature primitives for the CROSS-3 seal.
+
+A `Signer` produces a detached signature over canonical claim bytes; a `Verifier` checks
+one against a PINNED public key. The seal is verifiable OFFLINE with no secret — the
+verifier holds only the public key.
+
+Ed25519Signer/Verifier are the software implementation (real asymmetric crypto via the
+`cryptography` library). The HARDWARE path — a YubiKey PIV slot via PKCS#11 C_Sign, with a
+slot-9a attestation chain pinned to the Yubico PIV root CA — implements the SAME `Signer`
+protocol and is the documented swap-in (it needs the physical token, so it is not built
+here; this module ships the seal mechanics + offline verifier the hardware signer plugs
+into). No hardware signer is stubbed-as-working — that would be a false-green crypto claim.
+"""
+from __future__ import annotations
+
+import hashlib
+from typing import Protocol, runtime_checkable
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+ Ed25519PrivateKey,
+ Ed25519PublicKey,
+)
+
+_ALG = "ed25519"
+
+
+def key_id(public_raw: bytes) -> str:
+ """Short, stable id of a public key — sha256(raw)[:16]. The verifier PINS this; a seal
+ signed by any other key is rejected at gate G1."""
+ return hashlib.sha256(public_raw).hexdigest()[:16]
+
+
+@runtime_checkable
+class Signer(Protocol):
+ def alg(self) -> str: ...
+ def key_id(self) -> str: ...
+ def public_key_bytes(self) -> bytes: ...
+ def sign(self, payload: bytes) -> bytes: ...
+
+
+@runtime_checkable
+class Verifier(Protocol):
+ def alg(self) -> str: ...
+ def key_id(self) -> str: ...
+ def verify(self, payload: bytes, signature: bytes) -> bool: ...
+
+
+class Ed25519Signer:
+ """Software Ed25519 signer (real asymmetric crypto). Production replaces this with a
+ PIV/PKCS#11 signer implementing the same protocol; the seal format is identical."""
+
+ def __init__(self, private_key: Ed25519PrivateKey) -> None:
+ self._sk = private_key
+ self._pub = private_key.public_key().public_bytes(
+ serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+
+ @classmethod
+ def generate(cls) -> "Ed25519Signer":
+ return cls(Ed25519PrivateKey.generate())
+
+ @classmethod
+ def from_private_bytes(cls, raw: bytes) -> "Ed25519Signer":
+ return cls(Ed25519PrivateKey.from_private_bytes(raw))
+
+ def alg(self) -> str:
+ return _ALG
+
+ def key_id(self) -> str:
+ return key_id(self._pub)
+
+ def public_key_bytes(self) -> bytes:
+ return self._pub
+
+ def sign(self, payload: bytes) -> bytes:
+ return self._sk.sign(payload)
+
+ def verifier(self) -> "Ed25519Verifier":
+ """The matching offline verifier (public key only)."""
+ return Ed25519Verifier.from_public_bytes(self._pub)
+
+
+class Ed25519Verifier:
+ """Offline verifier — holds only the PINNED public key (no secret material)."""
+
+ def __init__(self, public_key: Ed25519PublicKey, public_raw: bytes) -> None:
+ self._pk = public_key
+ self._pub = public_raw
+
+ @classmethod
+ def from_public_bytes(cls, raw: bytes) -> "Ed25519Verifier":
+ return cls(Ed25519PublicKey.from_public_bytes(raw), raw)
+
+ def alg(self) -> str:
+ return _ALG
+
+ def key_id(self) -> str:
+ return key_id(self._pub)
+
+ def verify(self, payload: bytes, signature: bytes) -> bool:
+ try:
+ self._pk.verify(signature, payload)
+ return True
+ except InvalidSignature:
+ return False
diff --git a/docs/PIV_HARDWARE_SIGNER_PLAN.md b/docs/PIV_HARDWARE_SIGNER_PLAN.md
new file mode 100644
index 0000000..eb8b7ba
--- /dev/null
+++ b/docs/PIV_HARDWARE_SIGNER_PLAN.md
@@ -0,0 +1,172 @@
+# AEGIS — PIV Hardware-Signer Plan (CROSS-3, final pass)
+
+> **Status:** plan / ready to execute when a YubiKey is in hand.
+> **Scope:** swap the CROSS-3 seal's *software* Ed25519 signer for a *hardware* YubiKey-PIV
+> signer (ECDSA-P256) whose origin is provable via the YubiKey attestation chain, and bind the
+> `max_authorized` ceiling to the same hardware root. The Ed25519 path stays as the dev/CI
+> fallback. No bundle/seal **format** change — the `signature.alg` enum already reserves
+> `piv-ecdsa-p256`.
+
+This is the one piece deliberately *not* built in the seal PRs, because it needs the physical
+token to provision, sign, and verify against. Everything it plugs into already ships and is
+property-tested: the `Signer`/`Verifier` protocol (`core/seal/signing.py`), the detached
+receipt + offline gates **G0–G6** (`core/seal/seal.py`), and the live `/api/seal/*` endpoints.
+
+---
+
+## 0. What already exists (the swap point)
+
+- **`core/seal/signing.py`** — `Signer`/`Verifier` `Protocol`s (`alg() / key_id() /
+ public_key_bytes() / sign()`), plus the software `Ed25519Signer` / `Ed25519Verifier`.
+ **The PIV signer implements the same `Signer` protocol — nothing above it changes.**
+- **`core/seal/seal.py`** — `seal_bundle()` / `verify_seal()` run gates G0–G6; the seal format
+ is `{seal_version, claims, signature:{alg, key_id, value}}`. `value` is base64.
+- **`evidence/schema/.../seal.schema.json`** — `signature.alg ∈ {ed25519, piv-ecdsa-p256}`.
+- **The operator already runs YubiKey PIV/PKCS#11** in the DCN tool (`DCN_SSH_MODE=pkcs11`,
+ `libykcs11.dylib`, a working PIN flow). Reuse that operational knowledge — the plumbing is proven.
+
+---
+
+## 1. Goal
+
+A `PivSigner` backed by a YubiKey PIV slot that:
+1. signs the seal's canonical claim bytes with **ECDSA-P256** via PKCS#11 `C_Sign`,
+2. carries an **attestation chain** (leaf ← slot-`f9` attestation cert ← **Yubico PIV Root CA**)
+ that proves the key is **hardware-resident and non-exportable** — so a software key can't
+ forge a valid seal (this is what makes "hardware-rooted" *structural*, not asserted), and
+3. is selected by config (`AEGIS_SEAL_BACKEND=piv`), with `ed25519` as the default fallback.
+
+Plus: bind **`max_authorized`** (the #5 ceiling) to a PIV-signed policy object so the ceiling
+**cannot be raised by editing the environment** — the explicit `#5-ceiling` follow-up.
+
+---
+
+## 2. Decide up front (hardware + crypto facts)
+
+| Decision | Recommendation | Why |
+|---|---|---|
+| **Slot** | **`9c` (Digital Signature)** for the seal key — *distinct* from the `9a` SSH-auth key | 9c is the PIV signature slot; PIN-per-signature; avoids colliding with the existing SSH key |
+| **Key origin** | **Generate ON-DEVICE** (`ykman piv keys generate`) — never import | Only on-device keys are **attestable**; an imported key has no attestation → defeats the wedge |
+| **Algorithm** | **ECCP256** (NIST P-256 / secp256r1), sign SHA-256 of the canonical claims | PIV-standard ECC; `cryptography` verifies it. (Classic PIV has **no Ed25519** — hence ECDSA here) |
+| **Signature encoding** | `C_Sign`(CKM_ECDSA) returns **raw `r‖s` (64 bytes)** → convert to **DER** for `cryptography` | A real gotcha — `cryptography.verify()` wants DER `(r,s)` via `encode_dss_signature` |
+| **Touch policy** | `never` for unattended sealing; `always` for high-assurance (each seal needs a physical touch) | Tradeoff: `always` blocks automated sealing |
+| **PIN** | `C_Login` before `C_Sign`; source the PIN from a **secrets manager**, never env in prod | PIV locks out after N wrong PINs — never brute-force in tests |
+| **Pinning posture** | **CA + serial allowlist** for regulated fabrics (CA-only survives token swap) | Document the choice; default to CA+serial |
+
+---
+
+## 3. New code (additive — `core/seal/`)
+
+- **`core/seal/piv.py`** — `PivSigner` (PKCS#11 `C_Login` + `C_Sign` on the slot; `alg() ->
+ "piv-ecdsa-p256"`; `key_id()` from the SHA-256 of the attested EC public key; raw→DER sig
+ conversion) and `PivVerifier` (ECDSA-P256 verify against the attested EC public key). Both
+ satisfy the existing `Signer`/`Verifier` protocols verbatim.
+- **`core/seal/attestation.py`** — parse + validate the attestation chain with
+ `cryptography.x509`: leaf ← slot-`f9` intermediate ← **pinned** Yubico PIV Root CA; assert
+ the attested leaf public key **equals** the signing key; optional device-serial allowlist
+ (the serial is an X.509 extension in the attestation cert).
+- **`core/seal/pins.py`** — **compiled-in** constants (NOT file-loaded): the Yubico PIV Root CA
+ fingerprint (SHA-256 of its DER), the pinned seal verification public key + `key_id`, and the
+ serial allowlist. Editing a file must not be able to loosen trust.
+- **`core/seal/seal.py` `verify_seal()`** — when `signature.alg == "piv-ecdsa-p256"`, run the
+ attestation gates **before/with** G1/G2:
+ - **G1a** — attestation chain validates to the pinned Yubico Root CA (+ serial in allowlist);
+ - **G1b** — the attested leaf public key matches `verifier.key_id()` (the pinned key);
+ - **G2** — ECDSA-P256 signature verifies over the canonical claims.
+ The `ed25519` path is unchanged (software/dev/CI).
+- **`core/risk` ceiling** — `load_max_authorized()` verifies a **PIV-signed ceiling policy**
+ `{max_authorized, not_after_utc, key_id, signature}` against the compiled-in pinned key,
+ instead of trusting raw `AEGIS_MAX_AUTHORIZED_TIER`. Env stays as the dev fallback.
+
+---
+
+## 4. Sequencing (the critic's D6–D7 spike). ⚡ = needs the physical token
+
+| Step | Work | HW |
+|---|---|---|
+| **0** | Add the PKCS#11 dep (`PyKCS11`); the EC sign/verify + **raw→DER** helpers; `x509` chain-validation logic — all testable with **software P-256 keys + a self-signed test chain** | — |
+| **1** | Provision the token: generate the seal key on-device in **slot 9c** (ECCP256, PIN policy, touch policy); capture the attestation chain (`ykman piv keys attest 9c` + the `f9` intermediate); export the leaf pubkey | ⚡ |
+| **2** | `PivSigner` against `libykcs11` (`C_Login` w/ PIN, `C_Sign`); first end-to-end **sign → raw→DER → `cryptography` verify** roundtrip | ⚡ |
+| **3** | `attestation.py` chain validation; obtain the **current Yubico PIV Root CA cert from developers.yubico.com**, pin its DER SHA-256, add a checked-in test asserting the pinned bytes match the reference | — |
+| **4** | Capture a **real attestation-chain fixture** once (from the token) → check it into `tests/` so CI validates the chain logic **without** the token | ⚡ (once) |
+| **5** | Extend `verify_seal` with G1a/G1b; **re-run the full G0–G6 + the 0-violation property against the REAL hardware verifier** (the critic's "re-run against the real verify, not the stub") | ⚡ |
+| **6** | Ceiling-policy signing: sign `{max_authorized, not_after_utc, …}` with the PIV key; `load_max_authorized()` verifies it; CI test with a software-signed policy + a one-time hardware one | ⚡ |
+| **7** | `serve.py`: `AEGIS_SEAL_BACKEND=piv` selects `PivSigner`; `/api/seal/pubkey` also publishes the **attestation chain** so a remote verifier validates fully offline | — |
+| **8** | Rotation runbook + CA-only-vs-CA+serial decision documented | — |
+
+---
+
+## 5. Tests — hardware-free vs hardware-gated
+
+**HW-free (run in CI):**
+- ECDSA-P256 sign/verify roundtrip with **software** P-256 keys.
+- **raw `r‖s` → DER** conversion vectors (the easy-to-get-wrong bit).
+- `x509` chain validation against the **checked-in attestation fixture** + negatives: wrong root
+ → fail G1a; a **software-signed leaf must fail G1a** (this is the T6 test); serial not in
+ allowlist → fail.
+- The full **G0–G6** suite with a software-EC verifier (mirrors `seal_test.py`).
+- Ceiling-policy verify with a software key (valid / expired `not_after` / wrong-key / raised-tier).
+
+**HW-gated (run once on the token; `@needs_yubikey`, skipped in CI):**
+- Real `C_Sign` roundtrip; real attestation matches the pinned root; PIN + touch behavior.
+
+---
+
+## 6. Threat-model deltas (what hardware buys — and what it doesn't)
+
+- **Closes T6 (software-key impersonation):** the attestation chain proves the key is
+ hardware-resident + non-exportable; a software signer cannot produce a valid attestation to the
+ pinned Yubico root. *This* makes the "hardware-rooted" claim structural rather than asserted.
+- **Ceiling tamper-proofing:** `max_authorized` becomes PIV-signed → can't be raised by editing
+ env (closes the `#5-ceiling` residual).
+- **Residual (document, do not over-claim):** PIN compromise / coerced operator → mitigated by
+ touch policy + ceiling `not_after` expiry + dual-control (HOTL/BLOCK), **not** eliminated.
+ Supply-chain trust bottoms out at the **pinned Yubico Root CA fingerprint** — which is exactly
+ why Step 3 verifies it against a known-good reference with a checked-in test.
+
+---
+
+## 7. Rotation runbook
+
+- **Seal key rotation:** generate a new on-device key (re-gen `9c` or a new slot), re-capture
+ attestation, update the compiled-in verification pubkey/`key_id`, rebuild + redeploy. Keep a
+ `key_id → pubkey` map so seals from the rotation window still verify (multi-key verifier).
+- **Pinned Root CA fingerprint:** changes only if Yubico rotates the PIV root (rare) — update the
+ compiled-in constant + the checked-in fixture test, rebuild.
+- **Serial allowlist (CA+serial posture):** add/remove device serials, rebuild.
+
+---
+
+## 8. Dependencies / config
+
+- **`PyKCS11`** (or `python-pkcs11`) for the token; **`cryptography`** (already a dep) for EC
+ verify + `x509`; **`ykman`** for provisioning. `libykcs11` module path = the DCN tool's
+ working path.
+- Env: `AEGIS_SEAL_BACKEND=piv|ed25519` · `AEGIS_PIV_MODULE=/path/libykcs11.dylib` ·
+ `AEGIS_PIV_SLOT=9c` · PIN via a secrets manager (not env in prod).
+
+---
+
+## 9. Gotchas (PIV specifics + the DCN tool's experience)
+
+1. **Key must be GENERATED on-device** to be attestable — an imported key has no attestation.
+2. **`C_Sign`(CKM_ECDSA) returns raw `r‖s`, not DER** — convert before `cryptography.verify()`.
+3. macOS `libykcs11.dylib` vs Linux `.so`; the PKCS#11 module must match arch.
+4. **Slot 9c requires the PIN on every signature** (PIV spec) — fine for assurance, but automated
+ sealing needs the PIN available via a secrets manager + an active `C_Login` session.
+5. **Touch policy `always` blocks unattended sealing** — choose per deployment.
+6. PIN lockout after N failures — never brute-force in tests; use a dedicated test token.
+
+---
+
+## 10. Acceptance criteria (done = )
+
+- `AEGIS_SEAL_BACKEND=piv` produces seals whose `signature.alg == "piv-ecdsa-p256"`, carrying an
+ attestation chain that validates to the pinned Yubico Root CA.
+- `verify_seal` accepts a genuine hardware seal and **rejects a software-EC seal at G1a** (the
+ attestation test) — with the **0-violation property re-run against the real verifier**.
+- `load_max_authorized()` rejects a ceiling raised in env when a PIV-signed policy is configured.
+- CI stays green with **no token** (software fallback + checked-in attestation fixture); the
+ hardware path is exercised once, on the token, behind `@needs_yubikey`.
+
+_PIV_HARDWARE_SIGNER_PLAN.md — the final CROSS-3 pass; everything it plugs into already ships._
diff --git a/evidence/bundler.py b/evidence/bundler.py
index 87383e7..3a5af8b 100644
--- a/evidence/bundler.py
+++ b/evidence/bundler.py
@@ -17,6 +17,8 @@ def compute_sha256(bundle: dict) -> str:
"""Hash of the bundle with integrity.sha256 emptied (stable, verifiable)."""
clone = json.loads(json.dumps(bundle))
clone["integrity"]["sha256"] = ""
+ clone.pop("seal", None) # a detached CROSS-3 seal may ride in the response but is NOT
+ # part of the bundle's own integrity hash (it signs that hash)
return hashlib.sha256(_canonical(clone)).hexdigest()
@@ -27,12 +29,14 @@ 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,
+ authority: 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.2",
"run_id": run_id,
"created_utc": created,
"operator": operator,
@@ -50,6 +54,10 @@ 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()),
+ "authority": (authority if authority is not None
+ else _failclosed_authority(severity)),
},
"twin": {
"engine": "containerlab",
@@ -98,3 +106,53 @@ 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,
+ }
+
+
+def _failclosed_authority(severity: str) -> dict:
+ """Fallback when no authority was computed (a direct build_bundle call). Fail CLOSED:
+ an unknown authority is BLOCK, so it can never be promoted. run_preflight always
+ supplies a real record (core.risk.authority_record)."""
+ return {
+ "severity": severity,
+ "required": "block",
+ "max_authorized": "auto",
+ "allowed": False,
+ "effective": "block",
+ "change_class": {"touches_asn": False, "touches_rd_rt": False,
+ "touches_spine": False, "touches_underlay": False},
+ }
diff --git a/evidence/pdf.py b/evidence/pdf.py
index 80c7b52..2b534cd 100644
--- a/evidence/pdf.py
+++ b/evidence/pdf.py
@@ -164,5 +164,21 @@ def render_pdf(bundle: dict) -> bytes:
f"{ig.get('signed', False)}", ss["AegisSub"]))
el.append(Paragraph(f"sha256 {ig['sha256']}", ss["AegisMono"]))
+ # bounded-autonomy seal (CROSS-3) — the offline-verifiable receipt, when present
+ seal = bundle.get("seal")
+ if seal:
+ cl = seal.get("claims", {}); m = cl.get("model", {}); au = cl.get("authority", {})
+ sig = seal.get("signature", {})
+ el.append(Spacer(1, 6))
+ el.append(Paragraph("Bounded-autonomy seal (offline-verifiable receipt)",
+ ss["AegisSub"]))
+ el.append(_kv_table([
+ ["Model", f"{m.get('provider','?')} · {m.get('model','?')} "
+ f"({m.get('model_hash_kind','?')})"],
+ ["Authority", f"required {au.get('required','?')} · ceiling "
+ f"{au.get('max_authorized','?')} · {'within bound' if au.get('allowed') else 'EXCEEDS'}"],
+ ["Signature", f"{sig.get('alg','?')} · key {sig.get('key_id','?')}"],
+ ]))
+
doc.build(el)
return buf.getvalue()
diff --git a/evidence/schema/evidence_bundle.schema.json b/evidence/schema/evidence_bundle.schema.json
index d45c54d..a2dca22 100644
--- a/evidence/schema/evidence_bundle.schema.json
+++ b/evidence/schema/evidence_bundle.schema.json
@@ -7,7 +7,7 @@
"required": ["bundle_version", "run_id", "created_utc", "change", "twin",
"validation", "verdict", "integrity"],
"properties": {
- "bundle_version": { "const": "1.0" },
+ "bundle_version": { "const": "1.2" },
"run_id": { "type": "string", "minLength": 8 },
"created_utc": { "type": "string", "format": "date-time" },
"operator": { "type": "string" },
@@ -15,7 +15,8 @@
"change": {
"type": "object",
"additionalProperties": false,
- "required": ["intent", "source", "generated_configs", "risk_tier", "approval"],
+ "required": ["intent", "source", "generated_configs", "risk_tier", "approval",
+ "model_identity", "authority"],
"properties": {
"intent": { "type": "string" },
"source": { "enum": ["nl_intent", "config_import"] },
@@ -43,6 +44,48 @@
"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 an honest identity-claim (model_hash=null); a missing attestation seals provider 'unknown'/'unattested'; config_import seals 'none'/'operator-supplied'.",
+ "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" }
+ }
+ },
+
+ "authority": {
+ "type": "object",
+ "additionalProperties": false,
+ "description": "The orthogonal authority axis (#5): severity = how bad if it fails; required/effective = how much autonomy is allowed (auto=4.0
flask>=3.0
+cryptography>=42
diff --git a/serve.py b/serve.py
index fac2e1e..ec75a05 100644
--- a/serve.py
+++ b/serve.py
@@ -12,14 +12,17 @@
AEGIS_PORT=9000 python -m aegis.serve
"""
from __future__ import annotations
+import base64
import os
import re
+from datetime import datetime, timezone
from flask import Flask, Response, jsonify, request
from aegis.core.orchestrator.pipeline import run_preflight, PreflightError
from aegis.core.backends.simulator import SimulatorBackend
from aegis.evidence.pdf import render_pdf
+from aegis.core.seal import Ed25519Signer, Ed25519Verifier, seal_response, verify_seal
_UI = os.path.join(os.path.dirname(__file__), "ui", "preflight_screen.html")
@@ -42,6 +45,25 @@
app.config["MAX_CONTENT_LENGTH"] = 2 * 1024 * 1024 # 2 MB
+def _load_signer() -> Ed25519Signer:
+ """Seal signing key. AEGIS_SEAL_KEY = 64 hex chars (an Ed25519 private seed) pins a
+ stable key whose receipts verify across restarts; otherwise an EPHEMERAL demo key is
+ used. Production swaps this for a YubiKey-PIV signer (same Signer protocol)."""
+ raw = (os.environ.get("AEGIS_SEAL_KEY") or "").strip()
+ if raw:
+ try:
+ return Ed25519Signer.from_private_bytes(bytes.fromhex(raw))
+ except Exception as exc: # noqa: BLE001
+ print(f"[aegis.seal] AEGIS_SEAL_KEY invalid ({exc}); using an ephemeral key")
+ print("[aegis.seal] no AEGIS_SEAL_KEY set - EPHEMERAL demo signing key "
+ "(production uses a YubiKey-PIV key; receipts will not persist across restarts)")
+ return Ed25519Signer.generate()
+
+
+_SIGNER = _load_signer()
+_VERIFIER: Ed25519Verifier = _SIGNER.verifier()
+
+
@app.after_request
def _security_headers(resp: Response) -> Response:
resp.headers["X-Content-Type-Options"] = "nosniff"
@@ -94,6 +116,11 @@ def preflight_run():
return jsonify({"error": str(e), "stage": "guard"}), 400
except Exception as e: # noqa: BLE001
return jsonify({"error": f"{type(e).__name__}: {e}", "stage": "pipeline"}), 502
+ # CROSS-3: emit the bounded-autonomy seal in the same response (None if the change is
+ # not within the autonomy ceiling -- see change.authority for why). Detached: the seal
+ # rides under bundle["seal"] but is excluded from the bundle's own integrity hash.
+ seal, _skipped = seal_response(bundle, _SIGNER, datetime.now(timezone.utc).isoformat())
+ bundle["seal"] = seal
return jsonify(bundle)
@@ -114,6 +141,24 @@ def evidence_pdf():
"Content-Disposition": f'attachment; filename="aegis-evidence-{run}.pdf"'})
+@app.get("/api/seal/pubkey")
+def seal_pubkey():
+ """The pinned public key a client uses to verify seals offline (no secret)."""
+ return jsonify({"alg": _VERIFIER.alg(), "key_id": _VERIFIER.key_id(),
+ "public_key_b64": base64.b64encode(_SIGNER.public_key_bytes()).decode("ascii")})
+
+
+@app.post("/api/seal/verify")
+def seal_verify():
+ """Offline verification of a {bundle, seal} pair against THIS server's pinned key."""
+ data = request.get_json(silent=True) or {}
+ bundle, seal = data.get("bundle"), data.get("seal")
+ if not isinstance(bundle, dict) or not isinstance(seal, dict):
+ return jsonify({"error": "body must be {bundle, seal}"}), 400
+ v = verify_seal(bundle, seal, _VERIFIER)
+ return jsonify({"valid": v.valid, "gate": v.gate, "reason": v.reason}), (200 if v.valid else 422)
+
+
def main():
port = int(os.environ.get("AEGIS_PORT", "8088"))
# Bind loopback by default; require an explicit AEGIS_HOST=0.0.0.0 to expose it.
diff --git a/tests/api_test.py b/tests/api_test.py
index a60d1a0..4f60483 100644
--- a/tests/api_test.py
+++ b/tests/api_test.py
@@ -111,12 +111,34 @@ def main() -> int:
check("bundle.egress_none", (bundle.get("integrity") or {}).get("egress") == "none",
str((bundle.get("integrity") or {}).get("egress")))
+ # 13 — CROSS-3 seal emitted in the run response (bounded change -> receipt; else null)
+ au = (bundle.get("change") or {}).get("authority") or {}
+ seal = bundle.get("seal")
+ if au.get("allowed") and au.get("effective") != "block":
+ check("run.seal.present", isinstance(seal, dict) and "signature" in seal,
+ "expected a seal receipt for a bounded change")
+ else:
+ check("run.seal.absent_when_unbounded", seal is None, "unbounded change must not seal")
+
+ # 14 — pinned public key published for offline verification
+ r = c.get("/api/seal/pubkey")
+ pk = r.get_json() if r.status_code == 200 else {}
+ check("seal.pubkey.200", r.status_code == 200, str(r.status_code))
+ check("seal.pubkey.keyid", bool(pk.get("key_id")) and pk.get("alg") == "ed25519", str(pk)[:80])
+
+ # 15 — offline verify endpoint accepts a genuine {bundle, seal} pair
+ if isinstance(seal, dict):
+ r = c.post("/api/seal/verify", json={"bundle": bundle, "seal": seal})
+ vd = r.get_json() if r.status_code == 200 else {}
+ check("seal.verify.valid", r.status_code == 200 and vd.get("valid") is True,
+ f"{r.status_code} {r.get_data(as_text=True)[:120]}")
+
if FAILS:
print("\n=== API ENDPOINT TEST: FAIL ===")
for f in FAILS:
print(" -", f)
return 1
- print(f"\n=== API ENDPOINT TEST: PASS ({CHECKS} checks across 5 routes) ===")
+ print(f"\n=== API ENDPOINT TEST: PASS ({CHECKS} checks across 7 routes) ===")
return 0
diff --git a/tests/authority_test.py b/tests/authority_test.py
new file mode 100644
index 0000000..29790c2
--- /dev/null
+++ b/tests/authority_test.py
@@ -0,0 +1,190 @@
+"""aegis/tests/authority_test.py — #5 deterministic authority gate + no-self-escalation.
+
+House style: pytest-discoverable test_* + a dep-light __main__ runner
+(CI: python3 -m aegis.tests.authority_test). Pure logic, no network. Includes the CROSS-2
+differential of the canonical severity engine vs the two legacy AEGIS scorers
+(guards.risk_tier 3-bucket, bundler._severity 5-bucket).
+"""
+from __future__ import annotations
+
+import itertools
+import sys
+
+from aegis.core.backends.base import GeneratedConfig
+from aegis.core.orchestrator.guards import risk_tier
+from aegis.core.risk import (
+ ChangeClass,
+ Severity,
+ Tier,
+ authorize,
+ classify_change,
+ required_authority,
+ unify_severity,
+)
+from aegis.evidence.bundler import _severity # legacy 5-bucket scorer (differential ref)
+
+
+def _cfg(config: str, device: str = "leaf-1") -> GeneratedConfig:
+ return GeneratedConfig(device=device, vendor="frr", config=config, grounded_commands=[])
+
+
+# ── unify_severity reconciles the two legacy scorers ────────────────────────────
+
+def test_unify_severity_table() -> None:
+ cases = [
+ # (errors, devices, dropped, converged) -> canonical severity
+ (0, 0, 0, True, Severity.NONE),
+ (0, 1, 0, True, Severity.LOW),
+ (0, 3, 0, True, Severity.MEDIUM),
+ (0, 4, 0, True, Severity.HIGH),
+ (0, 1, 1, True, Severity.HIGH), # a dropped session is HIGH
+ (1, 1, 0, True, Severity.CRITICAL), # batfish error
+ (0, 1, 0, False, Severity.CRITICAL), # non-convergence
+ ]
+ for e, d, dr, c, exp in cases:
+ got = unify_severity(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c)
+ assert got == exp, f"{(e, d, dr, c)} -> {got.name}, want {exp.name}"
+
+
+def test_differential_vs_legacy_scorers() -> None:
+ """CROSS-2 differential: canonical vs guards.risk_tier (3-bucket) and bundler._severity
+ (5-bucket). The canonical equals the richer (bundler) engine by construction; guards
+ disagrees in documented ways (it has no NONE/CRITICAL and over-tiers a clean 1-device)."""
+ map3 = {"low": Severity.LOW, "medium": Severity.MEDIUM, "high": Severity.HIGH}
+ map5 = {"none": Severity.NONE, "low": Severity.LOW, "medium": Severity.MEDIUM,
+ "high": Severity.HIGH, "critical": Severity.CRITICAL}
+ disagreements = []
+ for e, d, dr, c in itertools.product((0, 1), (0, 1, 2, 4), (0, 1), (True, False)):
+ canon = unify_severity(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c)
+ bundler = map5[_severity(d, dr, e, c)]
+ guards = map3[risk_tier(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c)]
+ assert canon == bundler, f"canon {canon.name} != bundler {bundler.name} for {(e, d, dr, c)}"
+ if canon != guards:
+ disagreements.append((e, d, dr, c, canon.name, guards.name))
+ assert disagreements, "expected guards-vs-canonical disagreements (the reason to unify)"
+ # the headline disagreement: one clean device -> canonical LOW, guards MEDIUM
+ assert (0, 1, 0, True, "LOW", "MEDIUM") in disagreements
+ # guards collapses non-convergence (CRITICAL) into HIGH
+ assert (0, 1, 0, False, "CRITICAL", "HIGH") in disagreements
+
+
+# ── classify_change: AS/RD/RT (reliable) + spine/underlay (heuristic) ────────────
+
+def test_classify_detects_asn() -> None:
+ assert classify_change([_cfg("router bgp 65001")]).touches_asn
+ assert classify_change([_cfg("set protocols bgp group X peer-as 65002")]).touches_asn
+ assert classify_change([_cfg("neighbor 10.0.0.1 remote-as 65003")]).touches_asn
+ assert not classify_change([_cfg("interface eth0\n ip address 10.0.0.1/31")]).touches_asn
+
+
+def test_classify_detects_rd_rt() -> None:
+ assert classify_change([_cfg("set routing-instances V route-distinguisher 65000:1")]).touches_rd_rt
+ assert classify_change([_cfg("route-target target:65000:100")]).touches_rd_rt
+ assert not classify_change([_cfg("interface eth0")]).touches_rd_rt
+
+
+def test_classify_spine_underlay_heuristic() -> None:
+ assert classify_change([_cfg("x", device="dc1-spine-01")]).touches_spine
+ assert classify_change([_cfg("router ospf 1\n network 10.0.0.0/24 area 0")]).touches_underlay
+ leaf = classify_change([_cfg("# add vlan 10", device="leaf-1")])
+ assert not leaf.touches_spine and not leaf.touches_underlay
+
+
+# ── required_authority: severity base + hard-force overrides ─────────────────────
+
+def test_severity_base_mapping() -> None:
+ none = ChangeClass()
+ assert required_authority(Severity.NONE, none) == Tier.AUTO
+ assert required_authority(Severity.LOW, none) == Tier.AUTO
+ assert required_authority(Severity.MEDIUM, none) == Tier.HITL
+ assert required_authority(Severity.HIGH, none) == Tier.HOTL
+ assert required_authority(Severity.CRITICAL, none) == Tier.BLOCK
+
+
+def test_hardforce_fabric_identity_is_always_block() -> None:
+ assert required_authority(Severity.LOW, ChangeClass(touches_asn=True)) == Tier.BLOCK
+ assert required_authority(Severity.NONE, ChangeClass(touches_rd_rt=True)) == Tier.BLOCK
+
+
+def test_hardforce_spine_underlay_never_auto() -> None:
+ assert required_authority(Severity.LOW, ChangeClass(touches_spine=True)) == Tier.HOTL
+ assert required_authority(Severity.NONE, ChangeClass(touches_underlay=True)) == Tier.HOTL
+ # severity can still escalate above the spine/underlay floor
+ assert required_authority(Severity.CRITICAL, ChangeClass(touches_spine=True)) == Tier.BLOCK
+
+
+def test_hardforce_property_fabric_identity_never_below_block() -> None:
+ for sev in Severity:
+ for asn, rdrt in itertools.product((True, False), repeat=2):
+ t = required_authority(sev, ChangeClass(touches_asn=asn, touches_rd_rt=rdrt))
+ if asn or rdrt:
+ assert t == Tier.BLOCK, f"fabric-identity must BLOCK (sev={sev.name})"
+
+
+def test_spine_underlay_never_auto_property() -> None:
+ for sev in Severity:
+ for sp, ul in itertools.product((True, False), repeat=2):
+ t = required_authority(sev, ChangeClass(touches_spine=sp, touches_underlay=ul))
+ if sp or ul:
+ assert t >= Tier.HOTL, f"spine/underlay must be >= HOTL (sev={sev.name})"
+
+
+# ── no-self-escalation ceiling ──────────────────────────────────────────────────
+
+def test_authorize_allows_within_ceiling() -> None:
+ d = authorize(Tier.HITL, Tier.HOTL)
+ assert d.allowed and d.effective == Tier.HITL
+
+
+def test_authorize_blocks_self_escalation() -> None:
+ d = authorize(Tier.HOTL, Tier.HITL)
+ assert not d.allowed and d.effective == Tier.BLOCK and "NO-SELF-ESCALATION" in d.reason
+
+
+def test_no_self_escalation_zero_violations_property() -> None:
+ """The invariant: effective authority is NEVER above the ceiling. 0 violations over
+ every (required, ceiling) pair."""
+ violations = 0
+ for required in Tier:
+ for ceiling in Tier:
+ d = authorize(required, ceiling)
+ if d.effective != Tier.BLOCK and d.effective > ceiling:
+ violations += 1
+ assert violations == 0, f"{violations} self-escalation violations"
+
+
+def test_end_to_end_low_severity_asn_change_is_blocked() -> None:
+ """A twin-clean, single-device change that edits an AS number must still BLOCK:
+ severity alone (LOW) would say AUTO; the change class forces BLOCK, and an AUTO-ceiling
+ agent cannot push it."""
+ configs = [_cfg("router bgp 65010\n neighbor 10.0.0.1 remote-as 65011", device="leaf-1")]
+ sev = unify_severity(batfish_errors=0, devices_affected=1, sessions_dropped=0, converged=True)
+ assert sev == Severity.LOW
+ cc = classify_change(configs)
+ assert cc.touches_asn
+ req = required_authority(sev, cc)
+ assert req == Tier.BLOCK
+ assert not authorize(req, Tier.AUTO).allowed
+
+
+# ── runner ───────────────────────────────────────────────────────────────────────
+
+def _run_all() -> int:
+ funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+ failures = 0
+ for fn in funcs:
+ try:
+ fn()
+ print(f"PASS {fn.__name__}")
+ except AssertionError as e:
+ failures += 1
+ print(f"FAIL {fn.__name__}: {e}")
+ except Exception as e: # noqa: BLE001
+ failures += 1
+ print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}")
+ print(f"\n{len(funcs) - failures}/{len(funcs)} passed")
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ sys.exit(_run_all())
diff --git a/tests/ceiling_test.py b/tests/ceiling_test.py
new file mode 100644
index 0000000..498a596
--- /dev/null
+++ b/tests/ceiling_test.py
@@ -0,0 +1,186 @@
+"""aegis/tests/ceiling_test.py — #5-ceiling: authority sealed in the bundle + enforced at
+the gate (rule G5, no self-escalation).
+
+House style: pytest-discoverable test_* + a dep-light __main__ runner
+(CI: python3 -m aegis.tests.ceiling_test). jsonschema-based asserts self-skip if absent.
+"""
+from __future__ import annotations
+
+import json
+import os
+import random
+import sys
+from pathlib import Path
+
+from aegis.core.backends.simulator import SimulatorBackend
+from aegis.core.orchestrator.pipeline import run_preflight
+from aegis.core.promote import gate
+from aegis.core.promote.connectors import DryRunConnector
+from aegis.core.promote.promote import PromoteDenied, promote
+from aegis.core.risk import Tier, authorize, load_max_authorized
+from aegis.evidence.bundler import compute_sha256, verify
+
+try:
+ import jsonschema
+ _HAVE = True
+except Exception: # noqa: BLE001
+ _HAVE = False
+
+
+def _schema() -> dict:
+ return json.loads((_root() / "evidence" / "schema" / "evidence_bundle.schema.json").read_text("utf-8"))
+
+
+def _validate(b: dict) -> None:
+ if _HAVE:
+ jsonschema.validate(b, _schema())
+
+
+def _b(intent: str, **kw) -> dict:
+ return run_preflight(intent, backend=SimulatorBackend(), lab="clos-evpn", **kw)
+
+
+def _reseal(b: dict) -> dict:
+ b["integrity"]["sha256"] = ""
+ b["integrity"]["sha256"] = compute_sha256(b)
+ return b
+
+
+# ── the bundle carries + seals the authority record (v1.2) ──────────────────────
+
+def test_bundle_carries_sealed_authority_v12() -> None:
+ b = _b("add vlan 10 to leaf-1")
+ assert b["bundle_version"] == "1.2"
+ a = b["change"]["authority"]
+ assert set(a) == {"severity", "required", "max_authorized", "allowed", "effective", "change_class"}
+ assert a["required"] in ("auto", "hitl", "hotl", "block")
+ assert set(a["change_class"]) == {"touches_asn", "touches_rd_rt", "touches_spine", "touches_underlay"}
+ assert verify(b)
+ _validate(b)
+
+
+def test_authority_is_under_the_seal() -> None:
+ b = _b("add vlan 10 to leaf-1")
+ assert verify(b)
+ b["change"]["authority"]["required"] = "auto" # try to weaken the required tier
+ assert not verify(b), "tampering with the sealed authority must break the integrity hash"
+
+
+def test_asn_change_forces_block_authority() -> None:
+ b = _b("peer bgp with neighbor 10.0.0.1 remote-as 65010")
+ a = b["change"]["authority"]
+ assert a["change_class"]["touches_asn"]
+ assert a["required"] == "block" and a["effective"] == "block" and a["allowed"] is False
+ _validate(b)
+
+
+# ── gate rule G5: no self-escalation ────────────────────────────────────────────
+
+def test_gate_g5_blocks_block_authority_even_when_otherwise_promotable() -> None:
+ """A twin-clean, low-risk AS change is promotable by G1-G4 yet G5 blocks it: the
+ change requires BLOCK authority, above any valid ceiling."""
+ b = _b("peer bgp with neighbor 10.0.0.1 remote-as 65010")
+ assert b["change"]["authority"]["required"] == "block"
+ # make it otherwise fully promotable, keep the (block) authority, re-seal:
+ b["verdict"]["decision"] = "ship_ready"
+ b["change"]["risk_tier"] = "low"
+ _reseal(b)
+ d = gate.evaluate(b, approver=None, approval_token=None, connector_is_live=False, allow_live=False)
+ assert not d.allowed and "self-escalation" in d.reason.lower()
+
+
+def test_gate_g5_allows_within_ceiling() -> None:
+ b = _b("shut interface ethernet-1/1")
+ if b["change"]["authority"]["required"] == "block":
+ return # extremely unlikely for this intent; skip rather than assert a false case
+ b["verdict"]["decision"] = "ship_ready"
+ b["change"]["risk_tier"] = "low"
+ _reseal(b)
+ d = gate.evaluate(b, approver="noc-lead", approval_token="t", connector_is_live=False, allow_live=False)
+ assert d.allowed, d.reason
+
+
+def test_gate_g5_fail_closed_on_missing_authority() -> None:
+ b = _b("add vlan 10 to leaf-1")
+ b["verdict"]["decision"] = "ship_ready"
+ b["change"]["risk_tier"] = "low"
+ del b["change"]["authority"]
+ _reseal(b)
+ d = gate.evaluate(b, approver="noc-lead", approval_token="t", connector_is_live=False, allow_live=False)
+ assert not d.allowed and "authority" in d.reason.lower()
+
+
+# ── 0-violation property: nothing promotes above the ceiling ────────────────────
+
+def test_no_promoted_bundle_exceeds_ceiling_property() -> None:
+ ceiling = load_max_authorized()
+ intents = ["add vlan {n} to leaf-{n}", "peer bgp remote-as 6500{n}",
+ "enable ospf area 0 on edge-{n}", "shut interface ethernet-1/{n}"]
+ rng = random.Random(7)
+ promoted = denied_block = 0
+ for _ in range(300):
+ b = _b(intents[rng.randrange(len(intents))].format(n=rng.randint(1, 9)))
+ try:
+ promote(b, connector=DryRunConnector(), approver="noc-lead",
+ approval_token="t", allow_live=False)
+ promoted += 1
+ req = Tier[b["change"]["authority"]["required"].upper()]
+ assert authorize(req, ceiling).allowed, f"promoted above ceiling: {req.name}"
+ except PromoteDenied:
+ if b["change"]["authority"]["required"] == "block":
+ denied_block += 1
+ assert promoted > 0, "some benign changes must still promote"
+ assert denied_block > 0, "some BLOCK-tier (AS) changes must be denied by the ceiling"
+
+
+# ── ceiling loader: env override + fail-closed ──────────────────────────────────
+
+def test_load_max_authorized_env_and_failclosed() -> None:
+ old = os.environ.get("AEGIS_MAX_AUTHORIZED_TIER")
+ try:
+ assert load_max_authorized() == Tier.HOTL or old is not None # default HOTL
+ os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = "hitl"
+ assert load_max_authorized() == Tier.HITL
+ for bad in ("block", "BLOCK", "bogus", "5"):
+ os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = bad
+ try:
+ load_max_authorized()
+ raise AssertionError(f"ceiling {bad!r} must raise (fail-closed)")
+ except ValueError:
+ pass
+ finally:
+ if old is None:
+ os.environ.pop("AEGIS_MAX_AUTHORIZED_TIER", None)
+ else:
+ os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = old
+
+
+# ── runner ───────────────────────────────────────────────────────────────────────
+
+def _root() -> Path:
+ here = Path(__file__).resolve()
+ for parent in here.parents:
+ if (parent / "evidence" / "schema").exists():
+ return parent
+ return here.parent.parent
+
+
+def _run_all() -> int:
+ funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+ failures = 0
+ for fn in funcs:
+ try:
+ fn()
+ print(f"PASS {fn.__name__}")
+ except AssertionError as e:
+ failures += 1
+ print(f"FAIL {fn.__name__}: {e}")
+ except Exception as e: # noqa: BLE001
+ failures += 1
+ print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}")
+ print(f"\n{len(funcs) - failures}/{len(funcs)} passed")
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ sys.exit(_run_all())
diff --git a/tests/llm_egress_test.py b/tests/llm_egress_test.py
new file mode 100644
index 0000000..fda773c
--- /dev/null
+++ b/tests/llm_egress_test.py
@@ -0,0 +1,466 @@
+"""aegis/tests/llm_egress_test.py — PR-1 single-egress (CROSS-1) + adapter (#4) tests.
+
+Mirrors the repo's 6-suite house style: pytest-discoverable `test_*` functions PLUS a
+dependency-light `__main__` runner (CI runs `python3 -m aegis.tests.llm_egress_test`).
+No network: the wired-path tests drive fake in-process backends and httpx.MockTransport.
+
+PR-1 gate (per the roadmap): unit taxonomy · is_loopback · immutability · the wired
+fallback/retry/cooldown chain · hook-bus isolation · the air-gap fail-closed invariant ·
+no-litellm · the no-foreign-`anthropic`-import single-egress scan.
+
+DEFERRED to a later PR (documented, see test_single_egress_completion_post_DEFERRED):
+the "no /v1/chat/completions outside core/llm" scan — HttpBackend.generate_config still
+owns its POST until the cutover PR; enabling it now would be a false RED.
+
+Run: python3 -m aegis.tests.llm_egress_test
+"""
+from __future__ import annotations
+
+import ast
+import asyncio
+import dataclasses
+import hashlib
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+import httpx
+
+from aegis.core.llm import (
+ AdapterConfig,
+ AdapterError,
+ AuthError,
+ BackendSpec,
+ ContextExceeded,
+ HookBus,
+ LLMEgress,
+ LLMRequest,
+ ModelIdentity,
+ OpenAICompatLocalBackend,
+ Permanent,
+ RateLimited,
+ Transient,
+ classify,
+ is_retryable,
+ resolve_model_identity,
+ unknown_identity,
+)
+from aegis.core.llm.airgap import assert_airgap_ok, is_loopback
+
+
+# ── test doubles ──────────────────────────────────────────────────────────────
+
+def _ident(provider: str = "openai-compatible-local", model: str = "m") -> ModelIdentity:
+ return ModelIdentity(
+ provider=provider, # type: ignore[arg-type]
+ model=model,
+ model_hash=None,
+ model_hash_kind="identity-claim",
+ api_version=None,
+ capabilities=("chat",),
+ resolved_at_utc="2026-05-31T00:00:00+00:00",
+ )
+
+
+class _FakeBackend:
+ """In-process backend driven by a script of ('ok', text) | ('err', AdapterError)."""
+
+ def __init__(self, provider: str, model: str = "m", script=None) -> None:
+ self.provider = provider
+ self._model = model
+ self._script = list(script or [])
+ self.calls = 0
+
+ def identity(self) -> ModelIdentity:
+ return _ident(self.provider, self._model)
+
+ async def complete(self, *, system, user, max_tokens, temperature=0.1):
+ self.calls += 1
+ kind, payload = self._script.pop(0) if self._script else ("ok", user)
+ if kind == "ok":
+ return payload, self.identity()
+ raise payload
+
+
+def _cfg(max_retries: int = 2, base_backoff_s: float = 0.0, cooldown_s: float = 60.0) -> AdapterConfig:
+ return AdapterConfig(
+ chain=(),
+ airgap=False,
+ cloud_available=False,
+ max_retries=max_retries,
+ base_backoff_s=base_backoff_s,
+ cooldown_s=cooldown_s,
+ )
+
+
+def _egress(backends, hooks=None, **cfg_kw) -> LLMEgress:
+ return LLMEgress(_cfg(**cfg_kw), tuple(backends), hooks)
+
+
+def _req(user: str = "hello") -> LLMRequest:
+ return LLMRequest(system="sys", user=user, max_tokens=64)
+
+
+# ── (gate) error taxonomy + is_retryable ───────────────────────────────────────
+
+def test_is_retryable_truth_table() -> None:
+ assert is_retryable(Transient("x", "p"))
+ assert is_retryable(RateLimited("x", "p"))
+ assert not is_retryable(AuthError("x", "p"))
+ assert not is_retryable(ContextExceeded("x", "p"))
+ assert not is_retryable(Permanent("x", "p"))
+
+
+def test_classify_routing() -> None:
+ assert isinstance(classify(401, "", "cloud"), AuthError)
+ assert isinstance(classify(403, "", "cloud"), AuthError)
+ rl = classify(429, "retry_after: 5", "cloud")
+ assert isinstance(rl, RateLimited) and rl.retry_after_s == 5.0
+ assert isinstance(classify(400, "maximum context length", "l"), ContextExceeded)
+ assert isinstance(classify(413, "prompt is too long", "l"), ContextExceeded)
+ assert isinstance(classify(422, "bad", "l"), Permanent)
+ assert isinstance(classify(404, "nope", "l"), Permanent)
+ assert isinstance(classify(503, "", "l"), Transient)
+ assert isinstance(classify(500, "", "l"), Transient)
+ assert isinstance(classify(None, "conn reset", "l"), Transient) # fail-safe
+
+
+def test_adapter_error_str_hides_body() -> None:
+ e = AuthError(message="authentication failed", provider="cloud", status=401)
+ assert "cloud" in str(e) and "401" in str(e)
+
+
+# ── (gate) is_loopback / air-gap fail-closed ───────────────────────────────────
+
+def test_is_loopback() -> None:
+ assert is_loopback("http://127.0.0.1:11434")
+ assert is_loopback("http://localhost:11434")
+ assert is_loopback("http://127.5.5.5:1")
+ assert is_loopback("http://box.local")
+ assert not is_loopback("https://api.anthropic.com")
+ assert not is_loopback("http://10.0.0.5:11434")
+
+
+def test_airgap_refuses_cloud_construction() -> None:
+ raised = False
+ try:
+ assert_airgap_ok("anthropic-cloud", "https://api.anthropic.com")
+ except RuntimeError:
+ raised = True
+ assert raised, "cloud backend must be forbidden in air-gap mode"
+
+
+def test_airgap_refuses_non_loopback() -> None:
+ raised = False
+ try:
+ assert_airgap_ok("openai-compatible-local", "https://evil.example.com")
+ except RuntimeError:
+ raised = True
+ assert raised, "non-loopback egress must be forbidden in air-gap mode"
+
+
+def test_airgap_allows_loopback_local() -> None:
+ assert_airgap_ok("openai-compatible-local", "http://127.0.0.1:11434") # no raise
+
+
+# ── (gate) immutability ────────────────────────────────────────────────────────
+
+def test_frozen_dataclasses_reject_mutation() -> None:
+ req = _req()
+ ident = _ident()
+ for obj, attr, val in ((req, "user", "x"), (ident, "model", "y")):
+ try:
+ setattr(obj, attr, val)
+ raise AssertionError(f"{type(obj).__name__}.{attr} mutated — must be frozen")
+ except dataclasses.FrozenInstanceError:
+ pass
+
+
+# ── model identity: local weights-sha256 vs cloud identity-claim ───────────────
+
+def test_local_identity_carries_real_weights_sha256() -> None:
+ with tempfile.NamedTemporaryFile("wb", suffix=".gguf", delete=False) as f:
+ f.write(b"fake-weights-payload-for-test")
+ wpath = f.name
+ try:
+ expected = hashlib.sha256(b"fake-weights-payload-for-test").hexdigest()
+ spec = BackendSpec(
+ provider="openai-compatible-local",
+ base_url="http://127.0.0.1:11434",
+ model="ai/qwen3:latest",
+ weights_path=wpath,
+ )
+ ident = resolve_model_identity(spec)
+ assert ident.model_hash_kind == "weights-sha256"
+ assert ident.model_hash == expected
+ assert resolve_model_identity(spec).model_hash == expected # cache hit
+ finally:
+ os.unlink(wpath)
+
+
+def test_cloud_identity_is_claim_with_null_hash() -> None:
+ spec = BackendSpec(
+ provider="anthropic-cloud",
+ base_url="https://api.anthropic.com",
+ model="claude-haiku-4-5-20251001",
+ api_version="2023-06-01",
+ )
+ ident = resolve_model_identity(spec)
+ assert ident.model_hash is None
+ assert ident.model_hash_kind == "identity-claim"
+ assert ident.api_version == "2023-06-01"
+
+
+def test_missing_weights_downgrades_to_identity_claim() -> None:
+ spec = BackendSpec(
+ provider="openai-compatible-local",
+ base_url="http://127.0.0.1:11434",
+ model="ai/qwen3:latest",
+ weights_path="/nonexistent/path/model.gguf",
+ )
+ ident = resolve_model_identity(spec)
+ assert ident.model_hash is None
+ assert ident.model_hash_kind == "identity-claim"
+
+
+def test_unknown_identity_shape() -> None:
+ ui = unknown_identity("2026-05-31T10:30:00+00:00")
+ assert set(ui) == {
+ "provider", "model", "model_hash", "model_hash_kind",
+ "api_version", "capabilities", "resolved_at_utc",
+ }
+ assert ui["provider"] == "none" and ui["model_hash"] is None
+
+
+# ── (gate) the WIRED chain: success / fallback / retry / cooldown / fail-closed ─
+
+def test_chain_returns_on_first_success() -> None:
+ b = _FakeBackend("openai-compatible-local", script=[("ok", "ANSWER")])
+ res = asyncio.run(_egress([b]).complete(_req()))
+ assert res.text == "ANSWER"
+ assert len(res.attempts) == 1 and res.attempts[0].ok
+ assert res.identity.provider == "openai-compatible-local"
+ assert b.calls == 1
+
+
+def test_chain_falls_back_on_non_retryable() -> None:
+ b1 = _FakeBackend("openai-compatible-local", script=[("err", AuthError("no key", "local", 401))])
+ b2 = _FakeBackend("anthropic-cloud", script=[("ok", "FROM_CLOUD")])
+ res = asyncio.run(_egress([b1, b2]).complete(_req()))
+ assert res.text == "FROM_CLOUD"
+ assert b1.calls == 1, "non-retryable error must NOT retry the same backend"
+ assert [a.ok for a in res.attempts] == [False, True]
+
+
+def test_chain_retries_transient_then_succeeds() -> None:
+ b = _FakeBackend(
+ "openai-compatible-local",
+ script=[("err", Transient("flaky", "local", 503)), ("ok", "RECOVERED")],
+ )
+ res = asyncio.run(_egress([b], max_retries=2).complete(_req()))
+ assert res.text == "RECOVERED"
+ assert b.calls == 2, "transient error must retry the same backend"
+
+
+def test_ratelimited_cools_provider_and_is_skipped_next_call() -> None:
+ b1 = _FakeBackend(
+ "openai-compatible-local",
+ script=[("err", RateLimited("429", "local", 429)), ("err", RateLimited("429", "local", 429))],
+ )
+ b2 = _FakeBackend("anthropic-cloud", script=[("ok", "A"), ("ok", "B")])
+ eg = _egress([b1, b2], max_retries=1, cooldown_s=60.0)
+ r1 = asyncio.run(eg.complete(_req()))
+ assert r1.text == "A" and b1.calls == 2 # tried twice (attempt 0 + retry), then cooled
+ r2 = asyncio.run(eg.complete(_req()))
+ assert r2.text == "B" and b1.calls == 2, "cooled-down provider must be skipped"
+
+
+def test_chain_exhausted_raises_fail_closed() -> None:
+ b = _FakeBackend("openai-compatible-local", script=[("err", Permanent("nope", "local", 422))])
+ raised = False
+ try:
+ asyncio.run(_egress([b]).complete(_req()))
+ except AdapterError:
+ raised = True
+ assert raised, "no fake result on full failure — must fail closed"
+
+
+def test_deep_tier_prefers_cloud_first() -> None:
+ local = _FakeBackend("openai-compatible-local", script=[("ok", "LOCAL")])
+ cloud = _FakeBackend("anthropic-cloud", script=[("ok", "CLOUD")])
+ res = asyncio.run(_egress([local, cloud]).complete(
+ LLMRequest(system="s", user="u", tier="deep")))
+ assert res.text == "CLOUD" and cloud.calls == 1 and local.calls == 0
+
+
+# ── (gate) hook bus: transform, observe, isolation ─────────────────────────────
+
+def test_before_request_hook_transforms() -> None:
+ echo = _FakeBackend("openai-compatible-local") # echoes req.user
+ hooks = HookBus()
+ hooks.register("before_request", lambda req: dataclasses.replace(req, user="REWRITTEN"))
+ res = asyncio.run(_egress([echo], hooks=hooks).complete(_req("original")))
+ assert res.text == "REWRITTEN"
+
+
+def test_after_response_and_on_result_observed() -> None:
+ b = _FakeBackend("openai-compatible-local", script=[("ok", "X")])
+ seen: dict[str, object] = {}
+ hooks = HookBus()
+ hooks.register("after_response", lambda provider, identity: seen.update(provider=provider, ident=identity))
+ hooks.register("on_result", lambda result: seen.update(result_text=result.text))
+ asyncio.run(_egress([b], hooks=hooks).complete(_req()))
+ assert seen["provider"] == "openai-compatible-local"
+ assert isinstance(seen["ident"], ModelIdentity)
+ assert seen["result_text"] == "X"
+
+
+def test_broken_hook_never_breaks_the_call() -> None:
+ b = _FakeBackend("openai-compatible-local", script=[("ok", "SURVIVED")])
+ hooks = HookBus()
+
+ def _boom(*_a):
+ raise RuntimeError("hook exploded")
+
+ hooks.register("before_request", _boom)
+ hooks.register("after_response", _boom)
+ hooks.register("on_result", _boom)
+ res = asyncio.run(_egress([b], hooks=hooks).complete(_req()))
+ assert res.text == "SURVIVED", "a throwing hook must be isolated"
+
+
+# ── (gate) wired backend over httpx.MockTransport (no sockets) ──────────────────
+
+def _local_backend(handler) -> OpenAICompatLocalBackend:
+ spec = BackendSpec(
+ provider="openai-compatible-local",
+ base_url="http://127.0.0.1:11434",
+ model="ai/qwen3:latest",
+ )
+ return OpenAICompatLocalBackend(spec, timeout_s=5.0, transport=httpx.MockTransport(handler))
+
+
+def test_local_backend_http_happy_path() -> None:
+ def handler(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == "/v1/chat/completions"
+ return httpx.Response(200, json={"choices": [{"message": {"content": "hi-there"}}]})
+
+ text, ident = asyncio.run(_local_backend(handler).complete(system="s", user="u", max_tokens=16))
+ assert text == "hi-there"
+ assert ident.provider == "openai-compatible-local"
+
+
+def test_local_backend_429_classifies_ratelimited_with_header() -> None:
+ def handler(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(429, headers={"Retry-After": "7"}, json={"error": "slow down"})
+
+ try:
+ asyncio.run(_local_backend(handler).complete(system="s", user="u", max_tokens=16))
+ raise AssertionError("expected RateLimited")
+ except RateLimited as err:
+ assert err.retry_after_s == 7.0
+
+
+def test_local_backend_500_is_transient() -> None:
+ def handler(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(500, text="boom")
+
+ try:
+ asyncio.run(_local_backend(handler).complete(system="s", user="u", max_tokens=16))
+ raise AssertionError("expected Transient")
+ except Transient:
+ pass
+
+
+def test_local_backend_connection_error_is_transient() -> None:
+ def handler(request: httpx.Request) -> httpx.Response:
+ raise httpx.ConnectError("refused")
+
+ try:
+ asyncio.run(_local_backend(handler).complete(system="s", user="u", max_tokens=16))
+ raise AssertionError("expected Transient")
+ except Transient:
+ pass
+
+
+# ── single-egress invariant (anthropic import scan) + dep budget ───────────────
+
+def test_single_egress_no_foreign_anthropic_import() -> None:
+ repo = _aegis_root()
+ offenders: list[str] = []
+ for py in repo.rglob("*.py"):
+ rel = py.relative_to(repo).as_posix()
+ if rel.startswith("core/llm/") or "/tests/" in f"/{rel}":
+ continue
+ try:
+ tree = ast.parse(py.read_text(encoding="utf-8"))
+ except (SyntaxError, UnicodeDecodeError):
+ continue
+ for node in ast.walk(tree):
+ if isinstance(node, ast.Import) and any(
+ a.name.split(".")[0] == "anthropic" for a in node.names
+ ):
+ offenders.append(rel)
+ elif isinstance(node, ast.ImportFrom) and (node.module or "").split(".")[0] == "anthropic":
+ offenders.append(rel)
+ assert not offenders, f"non-llm modules import anthropic: {sorted(set(offenders))}"
+
+
+def test_no_litellm_dependency() -> None:
+ req = _aegis_root() / "requirements.txt"
+ text = req.read_text(encoding="utf-8").lower() if req.exists() else ""
+ assert "litellm" not in text, "litellm must never appear in requirements"
+
+
+def test_single_egress_completion_post_DEFERRED() -> None:
+ """DEFERRED to the cutover PR. HttpBackend.generate_config still owns its own POST to
+ /v1/chat/completions; enabling the 'no completion URL outside core/llm' scan now would
+ be a false RED. This placeholder documents the deferral and stays green."""
+ repo = _aegis_root()
+ hb = repo / "core" / "backends" / "http_backend.py"
+ assert hb.exists(), "expected the pre-cutover HttpBackend to still exist in PR-1"
+ print(" NOTE: completion-POST single-egress scan deferred to the http_backend cutover PR")
+
+
+# ── helpers + runner ───────────────────────────────────────────────────────────
+
+def test_is_loopback_rejects_spoofed_prefix() -> None:
+ """Regression for the parser-differential the security review caught: a host that only
+ *starts with* '127.' (or contains 'localhost') but resolves OFF-perimeter must NOT be
+ treated as loopback. Strict ipaddress parsing (no DNS) closes the bypass."""
+ assert not is_loopback("http://127.evil.com")
+ assert not is_loopback("http://127.0.0.1.evil.com")
+ assert not is_loopback("http://localhost.evil.com")
+ assert not is_loopback("http://") # empty / relative host rejected
+ assert is_loopback("http://[::1]:5757") # genuine ipv6 loopback
+ assert is_loopback("http://127.5.5.5") # genuine 127.0.0.0/8
+
+
+def _aegis_root() -> Path:
+ here = Path(__file__).resolve()
+ for parent in here.parents:
+ if (parent / "core" / "llm").exists():
+ return parent
+ return here.parent.parent
+
+
+def _run_all() -> int:
+ funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+ failures = 0
+ for fn in funcs:
+ try:
+ fn()
+ print(f"PASS {fn.__name__}")
+ except AssertionError as e:
+ failures += 1
+ print(f"FAIL {fn.__name__}: {e}")
+ except Exception as e: # noqa: BLE001
+ failures += 1
+ print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}")
+ print(f"\n{len(funcs) - failures}/{len(funcs)} passed")
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ sys.exit(_run_all())
diff --git a/tests/seal_test.py b/tests/seal_test.py
new file mode 100644
index 0000000..ed6bd6d
--- /dev/null
+++ b/tests/seal_test.py
@@ -0,0 +1,257 @@
+"""aegis/tests/seal_test.py — CROSS-3 bounded-autonomy SEAL: detached receipt, offline
+verify, gates G0..G6.
+
+House style: pytest-discoverable test_* + a dep-light __main__ runner
+(CI: python3 -m aegis.tests.seal_test). Uses real Ed25519 (cryptography). The schema test
+self-skips if jsonschema is absent.
+"""
+from __future__ import annotations
+
+import base64
+import json
+import random
+import sys
+from pathlib import Path
+
+from aegis.core.backends.simulator import SimulatorBackend
+from aegis.core.orchestrator.pipeline import run_preflight
+from aegis.core.seal import (
+ Ed25519Signer,
+ Ed25519Verifier,
+ SealError,
+ build_claims,
+ canonical_claims_bytes,
+ seal_bundle,
+ seal_response,
+ verify_seal,
+)
+from aegis.evidence.bundler import compute_sha256, verify
+
+try:
+ import jsonschema
+ _HAVE = True
+except Exception: # noqa: BLE001
+ _HAVE = False
+
+_AT = "2026-06-01T12:00:00+00:00"
+
+
+def _root() -> Path:
+ here = Path(__file__).resolve()
+ for p in here.parents:
+ if (p / "core" / "seal").exists():
+ return p
+ return here.parent.parent
+
+
+def _seal_schema() -> dict:
+ return json.loads((_root() / "core" / "seal" / "schema" / "seal.schema.json").read_text("utf-8"))
+
+
+def _reseal(b: dict) -> dict:
+ b["integrity"]["sha256"] = ""
+ b["integrity"]["sha256"] = compute_sha256(b)
+ return b
+
+
+def _bounded_bundle(required: str = "hitl", max_auth: str = "hotl") -> dict:
+ b = run_preflight("add vlan 10 to leaf-1", backend=SimulatorBackend(), lab="clos-evpn")
+ b["change"]["authority"] = {
+ "severity": "low", "required": required, "max_authorized": max_auth,
+ "allowed": True, "effective": required,
+ "change_class": {"touches_asn": False, "touches_rd_rt": False,
+ "touches_spine": False, "touches_underlay": False},
+ }
+ return _reseal(b)
+
+
+def _unbounded_bundle() -> dict:
+ b = _bounded_bundle()
+ b["change"]["authority"].update(required="block", effective="block", allowed=False)
+ return _reseal(b)
+
+
+def _signed_seal(bundle: dict, signer: Ed25519Signer, claims: dict) -> dict:
+ return {
+ "seal_version": "1.0", "claims": claims,
+ "signature": {"alg": signer.alg(), "key_id": signer.key_id(),
+ "value": base64.b64encode(signer.sign(canonical_claims_bytes(claims))).decode("ascii")},
+ }
+
+
+# ── claims binding ───────────────────────────────────────────────────────────────
+
+def test_build_claims_binds_model_authority_and_hash() -> None:
+ b = _bounded_bundle()
+ c = build_claims(b, _AT)
+ assert c["bundle_sha256"] == b["integrity"]["sha256"]
+ assert c["model"]["provider"] == b["change"]["model_identity"]["provider"]
+ assert c["authority"]["required"] == "hitl"
+
+
+# ── round-trip + offline verify ──────────────────────────────────────────────────
+
+def test_seal_then_verify_is_valid() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ v = verify_seal(b, seal, s.verifier())
+ assert v.valid and v.gate == "OK", v.reason
+
+
+def test_offline_verify_with_public_key_only() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ pub_only = Ed25519Verifier.from_public_bytes(s.public_key_bytes()) # no secret
+ assert verify_seal(b, seal, pub_only).valid
+
+
+# ── G1/G2 — wrong key, forged signature ──────────────────────────────────────────
+
+def test_wrong_key_rejected() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ other = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ v = verify_seal(b, seal, other.verifier()) # not the pinned key
+ assert not v.valid and v.gate in ("G1", "G2")
+
+
+def test_forged_signature_rejected_g2() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ seal["signature"]["value"] = "AAAA" + seal["signature"]["value"][4:] # corrupt the sig
+ v = verify_seal(b, seal, s.verifier())
+ assert not v.valid and v.gate == "G2"
+
+
+# ── G3/G6 — bundle binding + integrity ───────────────────────────────────────────
+
+def test_tampered_bundle_reseal_breaks_g3() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ b["change"]["generated_configs"][0]["config"] += "\n# injected"
+ _reseal(b) # new integrity hash != the seal's claimed bundle_sha256
+ v = verify_seal(b, seal, s.verifier())
+ assert not v.valid and v.gate == "G3"
+
+
+def test_tampered_bundle_no_reseal_breaks_g6() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ b["change"]["generated_configs"][0]["config"] += "\n# injected" # no re-seal
+ v = verify_seal(b, seal, s.verifier())
+ assert not v.valid and v.gate in ("G3", "G6")
+
+
+# ── G4 — model swap (re-signed with the right key to get past G2) ─────────────────
+
+def test_swapped_model_in_claims_rejected_g4() -> None:
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ claims = build_claims(b, _AT)
+ claims["model"]["model"] = "evil/other-model"
+ seal = _signed_seal(b, s, claims)
+ v = verify_seal(b, seal, s.verifier())
+ assert not v.valid and v.gate == "G4"
+
+
+# ── G5 — no self-escalation: never seal / never accept an unbounded change ────────
+
+def test_refuse_to_seal_unbounded() -> None:
+ b = _unbounded_bundle()
+ s = Ed25519Signer.generate()
+ try:
+ seal_bundle(b, s, sealed_at_utc=_AT)
+ raise AssertionError("must refuse to seal an unbounded change")
+ except SealError:
+ pass
+
+
+def test_verify_rejects_unbounded_seal_g5() -> None:
+ b = _unbounded_bundle()
+ s = Ed25519Signer.generate()
+ seal = _signed_seal(b, s, build_claims(b, _AT)) # bypass seal_bundle's refusal
+ v = verify_seal(b, seal, s.verifier())
+ assert not v.valid and v.gate == "G5"
+
+
+# ── 0-violation property: every valid seal certifies a bounded change ─────────────
+
+def test_every_valid_seal_is_bounded_property() -> None:
+ s = Ed25519Signer.generate()
+ ver = s.verifier()
+ rng = random.Random(11)
+ sealed = 0
+ for _ in range(120):
+ b = _bounded_bundle(required=rng.choice(["auto", "hitl", "hotl"]), max_auth="hotl")
+ seal = seal_bundle(b, s, sealed_at_utc=_AT)
+ v = verify_seal(b, seal, ver)
+ assert v.valid, v.reason
+ a = seal["claims"]["authority"]
+ assert a["allowed"] and a["effective"] != "block"
+ sealed += 1
+ assert sealed == 120
+
+
+# ── schema ────────────────────────────────────────────────────────────────────────
+
+def test_seal_matches_schema() -> None:
+ if not _HAVE:
+ print(" SKIP test_seal_matches_schema (jsonschema absent)")
+ return
+ b = _bounded_bundle()
+ s = Ed25519Signer.generate()
+ jsonschema.validate(seal_bundle(b, s, sealed_at_utc=_AT), _seal_schema())
+
+
+# ── live-path wiring: seal_response + embedded-seal hash exclusion ────────────────
+
+def test_seal_response_bounded_returns_receipt() -> None:
+ b = _bounded_bundle(); s = Ed25519Signer.generate()
+ seal, reason = seal_response(b, s, _AT)
+ assert reason is None and seal is not None
+ assert verify_seal(b, seal, s.verifier()).valid
+
+
+def test_seal_response_unbounded_returns_reason() -> None:
+ b = _unbounded_bundle(); s = Ed25519Signer.generate()
+ seal, reason = seal_response(b, s, _AT)
+ assert seal is None and reason and "ceiling" in reason.lower()
+
+
+def test_compute_sha256_excludes_embedded_seal() -> None:
+ """A detached seal embedded under bundle['seal'] (the live-path response shape) must NOT
+ change the bundle's own integrity hash, so verify(bundle) still holds."""
+ b = _bounded_bundle(); s = Ed25519Signer.generate()
+ seal, _ = seal_response(b, s, _AT)
+ b["seal"] = seal
+ assert verify(b), "embedded seal must be excluded from the bundle integrity hash"
+ assert verify_seal(b, b["seal"], s.verifier()).valid
+
+
+# ── runner ───────────────────────────────────────────────────────────────────────
+
+def _run_all() -> int:
+ funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+ failures = 0
+ for fn in funcs:
+ try:
+ fn()
+ print(f"PASS {fn.__name__}")
+ except AssertionError as e:
+ failures += 1
+ print(f"FAIL {fn.__name__}: {e}")
+ except Exception as e: # noqa: BLE001
+ failures += 1
+ print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}")
+ print(f"\n{len(funcs) - failures}/{len(funcs)} passed")
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ sys.exit(_run_all())
diff --git a/tests/stress_test.py b/tests/stress_test.py
index c314a33..bc85faa 100644
--- a/tests/stress_test.py
+++ b/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:
diff --git a/tests/wedge_test.py b/tests/wedge_test.py
new file mode 100644
index 0000000..128228d
--- /dev/null
+++ b/tests/wedge_test.py
@@ -0,0 +1,236 @@
+"""aegis/tests/wedge_test.py — #4 model-identity wedge: seal WHICH model made the change.
+
+House style: pytest-discoverable test_* + a dep-light __main__ runner
+(CI: python3 -m aegis.tests.wedge_test). Tests run against the REAL bundle produced by
+run_preflight + the REAL evidence_bundle schema. jsonschema-based asserts self-skip if
+jsonschema is absent; the seal/verify/round-trip asserts always run.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from aegis.core.backends.simulator import SimulatorBackend
+from aegis.core.llm import (
+ AdapterConfig,
+ BackendSpec,
+ HookBus,
+ IdentitySink,
+ LLMEgress,
+ LLMRequest,
+ ModelIdentity,
+ resolve_model_identity,
+)
+from aegis.core.orchestrator.pipeline import run_preflight
+from aegis.evidence.bundler import verify
+
+try:
+ import jsonschema
+ _HAVE_JSONSCHEMA = True
+except Exception: # noqa: BLE001
+ _HAVE_JSONSCHEMA = False
+
+_INTENT = "add vlan 10 to leaf-1"
+
+
+def _schema() -> dict:
+ return json.loads((_aegis_root() / "evidence" / "schema" / "evidence_bundle.schema.json").read_text("utf-8"))
+
+
+def _validate(bundle: dict) -> None:
+ if _HAVE_JSONSCHEMA:
+ jsonschema.validate(bundle, _schema())
+
+
+def _bundle(**kw) -> dict:
+ return run_preflight(_INTENT, backend=SimulatorBackend(), lab="clos-evpn", **kw)
+
+
+def _local_identity_dict() -> dict:
+ return {
+ "provider": "openai-compatible-local",
+ "model": "ai/qwen3:latest",
+ "model_hash": "a" * 64,
+ "model_hash_kind": "weights-sha256",
+ "api_version": None,
+ "capabilities": ["chat", "json"],
+ "resolved_at_utc": "2026-06-01T10:30:00+00:00",
+ }
+
+
+# ── real pipeline bundles now carry + seal model_identity at v1.1 ────────────────
+
+def test_nl_intent_seals_simulator_identity_v11() -> None:
+ b = _bundle()
+ assert b["bundle_version"] == "1.2"
+ mi = b["change"]["model_identity"]
+ assert mi["provider"] == "simulator" and mi["model"] == "deterministic-sim-v1"
+ assert verify(b)
+ _validate(b)
+
+
+def test_config_import_seals_unknown_identity() -> None:
+ b = run_preflight("", backend=SimulatorBackend(), lab="single", source="config_import",
+ imported_configs=[{"device": "r1", "vendor": "frr", "config": "router bgp 65000"}])
+ mi = b["change"]["model_identity"]
+ assert mi["provider"] == "none" and mi["model"] == "operator-supplied"
+ assert mi["model_hash"] is None and mi["model_hash_kind"] == "identity-claim"
+ assert verify(b)
+ _validate(b)
+
+
+def test_explicit_model_identity_passthrough() -> None:
+ b = _bundle(model_identity=_local_identity_dict())
+ mi = b["change"]["model_identity"]
+ assert mi["provider"] == "openai-compatible-local"
+ assert mi["model_hash"] == "a" * 64 and mi["model_hash_kind"] == "weights-sha256"
+ assert verify(b)
+ _validate(b)
+
+
+def test_model_identity_is_under_the_seal_tamper_detected() -> None:
+ b = _bundle(model_identity=_local_identity_dict())
+ assert verify(b)
+ b["change"]["model_identity"]["model"] = "evil-swapped-model"
+ assert not verify(b), "swapping the attested model must break the integrity seal"
+
+
+def test_cloud_identity_claim_is_never_a_faked_weight_hash() -> None:
+ spec = BackendSpec(provider="anthropic-cloud", base_url="https://api.anthropic.com",
+ model="claude-haiku-4-5-20251001", api_version="2023-06-01")
+ mi = resolve_model_identity(spec).to_bundle_dict()
+ assert mi["model_hash"] is None and mi["model_hash_kind"] == "identity-claim"
+ b = _bundle(model_identity=mi)
+ assert b["change"]["model_identity"]["provider"] == "anthropic-cloud"
+ assert verify(b)
+ _validate(b)
+
+
+# ── schema: v1.1 requires model_identity; bundle_version pinned to 1.1 ───────────
+
+def test_v11_requires_model_identity() -> None:
+ if not _HAVE_JSONSCHEMA:
+ print(" SKIP test_v11_requires_model_identity (jsonschema absent)")
+ return
+ b = _bundle()
+ del b["change"]["model_identity"]
+ try:
+ jsonschema.validate(b, _schema())
+ raise AssertionError("a v1.1 bundle without model_identity must fail validation")
+ except jsonschema.ValidationError:
+ pass
+
+
+def test_schema_pins_bundle_version_to_11() -> None:
+ if not _HAVE_JSONSCHEMA:
+ print(" SKIP test_schema_pins_bundle_version_to_11 (jsonschema absent)")
+ return
+ b = _bundle()
+ b["bundle_version"] = "1.0"
+ try:
+ jsonschema.validate(b, _schema())
+ raise AssertionError("schema must pin bundle_version to 1.1")
+ except jsonschema.ValidationError:
+ pass
+
+
+# ── the WEDGE round-trip: egress after_response -> IdentitySink -> sealed bundle ─
+
+class _FakeGenBackend:
+ """A generating backend whose model the egress attests; the sink hands it to the seal."""
+
+ provider = "openai-compatible-local"
+
+ def identity(self) -> ModelIdentity:
+ return ModelIdentity(
+ provider=self.provider, model="ai/qwen3:latest", model_hash="b" * 64,
+ model_hash_kind="weights-sha256", api_version=None,
+ capabilities=("chat",), resolved_at_utc="2026-06-01T10:30:00+00:00",
+ )
+
+ async def complete(self, *, system, user, max_tokens, temperature=0.1):
+ return "router bgp 65000", self.identity()
+
+
+def test_egress_after_response_sink_into_sealed_bundle() -> None:
+ import asyncio
+
+ sink = IdentitySink()
+ hooks = HookBus()
+ hooks.register("after_response", sink.capture)
+ cfg = AdapterConfig(chain=(), airgap=False, cloud_available=False,
+ max_retries=0, base_backoff_s=0.0)
+ egress = LLMEgress(cfg, (_FakeGenBackend(),), hooks)
+ asyncio.run(egress.complete(LLMRequest(system="s", user=_INTENT, max_tokens=64)))
+
+ assert sink.last is not None and sink.last.model == "ai/qwen3:latest"
+ b = _bundle(model_identity=sink.as_bundle_dict())
+ mi = b["change"]["model_identity"]
+ assert mi["model"] == "ai/qwen3:latest" and mi["model_hash_kind"] == "weights-sha256"
+ assert verify(b)
+ _validate(b)
+
+
+class _NoAttestBackend:
+ """A generating backend (nl_intent) that does NOT attest a model identity --
+ e.g. the live HttpBackend before the generate-path cutover."""
+ def __init__(self) -> None:
+ self._inner = SimulatorBackend()
+ def generate_config(self, intent, lab):
+ return self._inner.generate_config(intent, lab)
+ def batfish_check(self, configs):
+ return self._inner.batfish_check(configs)
+ def spawn_twin(self, lab):
+ return self._inner.spawn_twin(lab)
+ def apply_and_converge(self, twin_id, configs):
+ return self._inner.apply_and_converge(twin_id, configs)
+ def state_diff(self, twin_id):
+ return self._inner.state_diff(twin_id)
+ def teardown_twin(self, twin_id):
+ return self._inner.teardown_twin(twin_id)
+
+
+def test_nl_intent_without_attestation_is_unattested_not_operator() -> None:
+ b = run_preflight(_INTENT, backend=_NoAttestBackend(), lab="clos-evpn")
+ mi = b["change"]["model_identity"]
+ assert mi["provider"] == "unknown" and mi["model"] == "unattested"
+ # MUST NOT be confused with a config_import operator-supplied identity:
+ assert (mi["provider"], mi["model"]) != ("none", "operator-supplied")
+ assert verify(b)
+ _validate(b)
+
+
+def test_empty_sink_yields_none() -> None:
+ assert IdentitySink().as_bundle_dict() is None
+
+
+# ── helpers + runner ────────────────────────────────────────────────────────────
+
+def _aegis_root() -> Path:
+ here = Path(__file__).resolve()
+ for parent in here.parents:
+ if (parent / "evidence" / "schema").exists():
+ return parent
+ return here.parent.parent
+
+
+def _run_all() -> int:
+ funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+ failures = 0
+ for fn in funcs:
+ try:
+ fn()
+ print(f"PASS {fn.__name__}")
+ except AssertionError as e:
+ failures += 1
+ print(f"FAIL {fn.__name__}: {e}")
+ except Exception as e: # noqa: BLE001
+ failures += 1
+ print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}")
+ print(f"\n{len(funcs) - failures}/{len(funcs)} passed")
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ sys.exit(_run_all())