fix(context): close evidence and execution policy bypasses

src/perseus/context_contract.py

@@ -435,9 +435,9 @@ def _cc_route(request_class: str, integrations: Any) -> dict[str, Any]:
 def _cc_failure_for_integrations(integrations: Mapping[str, str]) -> str | None:
     if integrations.get("vault") == "timeout" or integrations.get("ledger") == "timeout":
         return "timeout"
-    if integrations.get("vault") == "unavailable":
+    if integrations.get("vault") in {"unavailable", "not_configured"}:
         return "vault_unavailable"
-    if integrations.get("ledger") == "unavailable":
+    if integrations.get("ledger") in {"unavailable", "not_configured"}:
         return "ledger_unavailable"
     return None
 
@@ -692,6 +692,9 @@ def context_rank(
             excluded=[{"candidate_id": item, "reason": "excluded_by_contract"} for item in excluded],
             evidence_required=bool(policy_map.get("evidence_required", False)),
         )
+        if result["status"] == "complete" and result["evidence_projection"]["coverage"]["abstention_required"]:
+            result["status"] = "abstain"
+            result["failure_state"] = "insufficient_evidence"
         return result
     except (TypeError, ValueError) as exc:
         message = str(exc)

src/perseus/context_dag.py

@@ -219,6 +219,7 @@ class CompilationBudget:
     max_depth: int = 4
     max_fanout: int = 6
     max_tokens: int = 4000
+    max_bytes: int | None = None
     deadline_s: float = 30.0
 
     def ledger(self) -> "BudgetLedger":
@@ -233,6 +234,7 @@ def __init__(self, budget: CompilationBudget):
         self.nodes: list[str] = []
         self.depth: dict[str, int] = {}
         self.tokens: dict[str, int] = {}
+        self.bytes: dict[str, int] = {}
         self.children: dict[str, list[str]] = {}
         self.started_at = time.monotonic()
 
@@ -254,9 +256,13 @@ def register_node(self, node: ContextNode, depth: int) -> None:
         self.nodes.append(node.node_id)
         self.depth[node.node_id] = depth
         self.tokens[node.node_id] = dag_tokens(node.content)
+        self.bytes[node.node_id] = len(node.content.encode("utf-8"))
         total = sum(self.tokens.values())
         if total > self.budget.max_tokens:
             raise BudgetExceeded("max_tokens", self.budget.max_tokens, total)
+        total_bytes = sum(self.bytes.values())
+        if self.budget.max_bytes is not None and total_bytes > self.budget.max_bytes:
+            raise BudgetExceeded("max_bytes", self.budget.max_bytes, total_bytes)
 
     def register_edge(self, parent_id: str, child_id: str) -> None:
         self._tick()
@@ -284,12 +290,14 @@ def report(self) -> dict:
             "max_fanout_used": max((len(v) for v in self.children.values()),
                                    default=0),
             "tokens": self.total_tokens,
+            "bytes": sum(self.bytes.values()),
             "wall_clock_s": round(time.monotonic() - self.started_at, 3),
             "limits": {
                 "max_nodes": self.budget.max_nodes,
                 "max_depth": self.budget.max_depth,
                 "max_fanout": self.budget.max_fanout,
                 "max_tokens": self.budget.max_tokens,
+                "max_bytes": self.budget.max_bytes,
                 "deadline_s": self.budget.deadline_s,
             },
             "token_accounting": TOKEN_ACCOUNTING_NOTE,
@@ -780,6 +788,7 @@ def compile_context_dag(*, task_id: str,
                 max_depth=min(int(budget.max_depth), int(profile_budget["max_depth"])),
                 max_fanout=min(int(budget.max_fanout), int(profile_budget["max_fanout"])),
                 max_tokens=min(int(budget.max_tokens), int(profile_budget["max_tokens"])),
+                max_bytes=(int(profile_budget["max_bytes"]) if budget.max_bytes is None else min(int(budget.max_bytes), int(profile_budget["max_bytes"]))),
                 deadline_s=min(float(budget.deadline_s), float(profile_budget["deadline_s"])),
             )
     else:

src/perseus/execution_profiles.py

@@ -33,6 +33,7 @@
     "available_memory_mb", "available_compute_units", "resource_metrics",
 })
 _EP_NETWORK_MODES = frozenset({"offline", "local", "approved_network"})
+_EP_NETWORK_RANK = {"offline": 0, "local": 1, "approved_network": 2}
 _EP_DEGRADATION_POLICIES = frozenset({"fail_closed", "partial", "omit_low_priority"})
 _EP_RETRIEVAL_STATES = frozenset({"complete", "partial", "degraded", "unavailable", "timeout"})
 _EP_MODE_DEFAULTS = {
@@ -105,12 +106,9 @@ def _ep_id(value: Any, field: str, *, default: str = "") -> str:
 
 
 def _ep_limit(value: Any, field: str, *, maximum: int = 10_000_000) -> int:
-    if isinstance(value, bool):
+    if isinstance(value, bool) or not isinstance(value, int):
         raise ExecutionProfileError(f"{field} must be a positive integer")
-    try:
-        number = int(value)
-    except (TypeError, ValueError):
-        raise ExecutionProfileError(f"{field} must be a positive integer") from None
+    number = value
     if number < 1 or number > maximum:
         raise ExecutionProfileError(f"{field} must be between 1 and {maximum}")
     return number
@@ -157,7 +155,7 @@ class ExecutionProfile:
     @classmethod
     def from_mapping(cls, value: Mapping[str, Any] | "ExecutionProfile" | None) -> "ExecutionProfile":
         if isinstance(value, cls):
-            return value
+            value = value.to_dict()
         if value is None:
             value = {"mode": "standard-local", "profile_id": "default-local"}
         if not isinstance(value, Mapping):
@@ -294,13 +292,13 @@ def execution_profile_compilation_budget(resolved: Mapping[str, Any]) -> dict[st
     if not isinstance(resolved, Mapping) or not isinstance(resolved.get("effective"), Mapping):
         raise ExecutionProfileError("resolved execution profile is malformed")
     effective = resolved["effective"]
-    max_tokens = min(int(effective["max_context_tokens"]), max(1, int(effective["max_context_bytes"]) // 4))
     latency = effective.get("latency_target_ms")
     return {
         "max_nodes": max(1, int(effective["max_items"])),
         "max_depth": max(1, int(effective["max_depth"])),
         "max_fanout": max(1, int(effective["max_items"])),
-        "max_tokens": max_tokens,
+        "max_tokens": int(effective["max_context_tokens"]),
+        "max_bytes": int(effective["max_context_bytes"]),
         "deadline_s": max(0.001, float(latency) / 1000.0) if latency is not None else 30.0,
     }
 
@@ -317,17 +315,22 @@ def _ep_resolve_execution_profile_impl(
     req = _ep_requirements(requirements)
     if retrieval_status not in _EP_RETRIEVAL_STATES:
         raise ExecutionProfileError("retrieval_status is unsupported")
-    if req.get("network_mode") and base.network_mode == "offline" and req["network_mode"] != "offline":
-        raise ExecutionProfileError("offline profile cannot satisfy a network requirement")
-    if req.get("require_offline") and base.network_mode != "offline":
+    requested_network = req.get("network_mode")
+    if requested_network and _EP_NETWORK_RANK[requested_network] > _EP_NETWORK_RANK[base.network_mode]:
+        raise ExecutionProfileError(f"profile network policy {base.network_mode} cannot satisfy requested {requested_network} requirement")
+    if req.get("require_offline") and _EP_NETWORK_RANK[base.network_mode] < _EP_NETWORK_RANK["offline"]:
         raise ExecutionProfileError("profile does not satisfy required offline mode")
     missing = sorted(set(req.get("required_capabilities", ())) - set(base.runtime_capabilities))
     if missing:
         raise ExecutionProfileError(f"required capabilities are unsupported: {', '.join(missing)}")
     safe_resources, resource_state = _ep_resources(resources)
     effective = base.to_dict()
+    if req.get("require_offline") or requested_network is not None:
+        effective["network_mode"] = "offline" if req.get("require_offline") else requested_network
     for field in ("max_context_tokens", "max_context_bytes", "max_items", "max_depth", "latency_target_ms"):
         if field in req and req[field] is not None:
+            if field == "latency_target_ms" and effective[field] is None:
+                raise ExecutionProfileError("latency target cannot be resolved without a profile latency bound")
             effective[field] = min(int(effective[field]), int(req[field]))
     if effective["max_context_tokens"] < 1 or effective["max_context_bytes"] < 1 or effective["max_items"] < 1 or effective["max_depth"] < 1:
         raise ExecutionProfileError("requirements leave no usable context budget")

src/perseus/runtime_adapter.py

@@ -22,6 +22,7 @@
 _RA_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/#@+\-]{0,159}$")
 _RA_DIGEST_RE = re.compile(r"^(?:sha256:)?[0-9a-fA-F]{64}$")
 _RA_EXECUTION_MODES = frozenset({"offline", "local", "approved_network"})
+_RA_NETWORK_RANK = {"offline": 0, "local": 1, "approved_network": 2}
 _RA_RESULT_STATUSES = frozenset({"success", "partial", "unavailable", "timeout", "cancelled", "malformed"})
 _RA_CAPABILITY_FIELDS = frozenset({
     "schema_version", "backend_id", "backend_version", "model_id", "model_version",
@@ -87,12 +88,9 @@ def _ra_digest(value: Any, field: str) -> str:
 
 
 def _ra_limit(value: Any, field: str, *, maximum: int = 10_000_000) -> int:
-    if isinstance(value, bool):
+    if isinstance(value, bool) or not isinstance(value, int):
         raise RuntimeAdapterError(f"{field} must be a positive integer")
-    try:
-        number = int(value)
-    except (TypeError, ValueError):
-        raise RuntimeAdapterError(f"{field} must be a positive integer") from None
+    number = value
     if number < 1 or number > maximum:
         raise RuntimeAdapterError(f"{field} must be between 1 and {maximum}")
     return number
@@ -140,24 +138,27 @@ class RuntimeCapabilities:
     @classmethod
     def from_mapping(cls, value: Mapping[str, Any] | "RuntimeCapabilities") -> "RuntimeCapabilities":
         if isinstance(value, cls):
-            return value
+            value = value.to_dict()
         if not isinstance(value, Mapping):
             raise RuntimeAdapterError("runtime capabilities must be an object")
         _ra_forbidden_keys(value, "capabilities")
+        missing = _RA_CAPABILITY_FIELDS - set(value)
+        if missing:
+            raise RuntimeAdapterError(f"capabilities missing required fields: {sorted(map(str, missing))}")
         unknown = set(value) - _RA_CAPABILITY_FIELDS
         if unknown:
             raise RuntimeAdapterError(f"unsupported capability fields: {sorted(map(str, unknown))}")
-        if value.get("schema_version", _RA_CAPABILITIES_SCHEMA) != _RA_CAPABILITIES_SCHEMA:
+        if value["schema_version"] != _RA_CAPABILITIES_SCHEMA:
             raise RuntimeAdapterError("unsupported runtime capabilities schema version")
-        modes = _ra_string_list(value.get("execution_modes", ()), "execution_modes")
+        modes = _ra_string_list(value["execution_modes"], "execution_modes")
         if not modes or not set(modes).issubset(_RA_EXECUTION_MODES):
             raise RuntimeAdapterError("execution_modes must contain offline, local, or approved_network")
-        metrics = _ra_string_list(value.get("resource_metrics", ()), "resource_metrics")
+        metrics = _ra_string_list(value["resource_metrics"], "resource_metrics")
         for field in ("backend_id", "backend_version", "model_id", "tokenizer_id", "auth_mode", "provider_ref"):
-            _ra_id(value.get(field, ""), field)
-        model_version = _ra_id(value.get("model_version", "unknown"), "model_version")
-        hardware_class = _ra_id(value.get("hardware_class", "unknown"), "hardware_class")
-        if not isinstance(value.get("streaming", False), bool) or not isinstance(value.get("tools", False), bool):
+            _ra_id(value[field], field)
+        model_version = _ra_id(value["model_version"], "model_version")
+        hardware_class = _ra_id(value["hardware_class"], "hardware_class")
+        if not isinstance(value["streaming"], bool) or not isinstance(value["tools"], bool):
             raise RuntimeAdapterError("streaming and tools must be booleans")
         return cls(
             schema_version=_RA_CAPABILITIES_SCHEMA,
@@ -168,8 +169,8 @@ def from_mapping(cls, value: Mapping[str, Any] | "RuntimeCapabilities") -> "Runt
             tokenizer_id=_ra_id(value["tokenizer_id"], "tokenizer_id"),
             context_capacity_tokens=_ra_limit(value["context_capacity_tokens"], "context_capacity_tokens"),
             execution_modes=modes,
-            streaming=value.get("streaming", False),
-            tools=value.get("tools", False),
+            streaming=value["streaming"],
+            tools=value["tools"],
             hardware_class=hardware_class,
             resource_metrics=metrics,
             auth_mode=_ra_id(value["auth_mode"], "auth_mode"),
@@ -213,14 +214,17 @@ class AdapterRequest:
     @classmethod
     def from_mapping(cls, value: Mapping[str, Any] | "AdapterRequest") -> "AdapterRequest":
         if isinstance(value, cls):
-            return value
+            value = value.to_dict()
         if not isinstance(value, Mapping):
             raise RuntimeAdapterError("adapter request must be an object")
         _ra_forbidden_keys(value, "request")
+        missing = _RA_REQUEST_FIELDS - set(value)
+        if missing:
+            raise RuntimeAdapterError(f"request missing required fields: {sorted(map(str, missing))}")
         unknown = set(value) - _RA_REQUEST_FIELDS
         if unknown:
             raise RuntimeAdapterError(f"unsupported request fields: {sorted(map(str, unknown))}")
-        if value.get("schema_version", _RA_REQUEST_SCHEMA) != _RA_REQUEST_SCHEMA:
+        if value["schema_version"] != _RA_REQUEST_SCHEMA:
             raise RuntimeAdapterError("unsupported runtime request schema version")
         profile = value.get("execution_profile")
         if not isinstance(profile, Mapping):
@@ -244,10 +248,15 @@ def from_mapping(cls, value: Mapping[str, Any] | "AdapterRequest") -> "AdapterRe
             normalized_required["resource_metrics"] = list(_ra_string_list(requested["resource_metrics"], "resource_metrics"))
         if "min_context_tokens" in requested:
             normalized_required["min_context_tokens"] = _ra_limit(requested["min_context_tokens"], "min_context_tokens")
-        mode = _ra_text(value.get("execution_mode", "local"), "execution_mode", max_length=32)
+        mode = _ra_text(value["execution_mode"], "execution_mode", max_length=32)
         if mode not in _RA_EXECUTION_MODES:
             raise RuntimeAdapterError("execution_mode is unsupported")
-        profile_digest = _ra_digest(value.get("execution_profile_digest", profile["profile_digest"]), "execution_profile_digest")
+        effective_profile = profile.get("effective")
+        if not isinstance(effective_profile, Mapping) or effective_profile.get("network_mode") not in _RA_NETWORK_RANK:
+            raise RuntimeAdapterError("execution_profile effective network policy is missing")
+        if _RA_NETWORK_RANK[mode] > _RA_NETWORK_RANK[effective_profile["network_mode"]]:
+            raise RuntimeAdapterError("execution_mode exceeds execution_profile network policy")
+        profile_digest = _ra_digest(value["execution_profile_digest"], "execution_profile_digest")
         if profile_digest != str(profile["profile_digest"]).lower().removeprefix("sha256:"):
             raise RuntimeAdapterError("execution_profile_digest does not match execution_profile")
         return cls(
@@ -295,40 +304,43 @@ class AdapterResult:
     @classmethod
     def from_mapping(cls, value: Mapping[str, Any] | "AdapterResult") -> "AdapterResult":
         if isinstance(value, cls):
-            return value
+            value = value.to_dict()
         if not isinstance(value, Mapping):
             raise RuntimeAdapterError("adapter result must be an object")
         _ra_forbidden_keys(value, "result")
         unknown = set(value) - _RA_RESULT_FIELDS
         if unknown:
             raise RuntimeAdapterError(f"unsupported result fields: {sorted(map(str, unknown))}")
-        if value.get("schema_version", _RA_RESULT_SCHEMA) != _RA_RESULT_SCHEMA:
+        missing = _RA_RESULT_FIELDS - set(value)
+        if missing:
+            raise RuntimeAdapterError(f"result missing required fields: {sorted(map(str, missing))}")
+        if value["schema_version"] != _RA_RESULT_SCHEMA:
             raise RuntimeAdapterError("unsupported runtime result schema version")
-        status = _ra_text(value.get("status", ""), "status", max_length=32)
+        status = _ra_text(value["status"], "status", max_length=32)
         if status not in _RA_RESULT_STATUSES:
             raise RuntimeAdapterError("unsupported runtime result status")
-        output = value.get("output")
+        output = value["output"]
         if output is not None:
             output = _ra_text(output, "output", max_length=1_000_000, allow_empty=True)
-        usage_raw = value.get("usage", {})
+        usage_raw = value["usage"]
         if not isinstance(usage_raw, Mapping) or set(usage_raw) - _RA_USAGE_FIELDS:
             raise RuntimeAdapterError("usage contains unsupported fields")
         usage: dict[str, int] = {}
         for key, raw in usage_raw.items():
             if isinstance(raw, bool) or not isinstance(raw, int) or raw < 0:
                 raise RuntimeAdapterError(f"usage.{key} must be a non-negative integer")
             usage[key] = raw
-        runtime_raw = value.get("runtime", {})
+        runtime_raw = value["runtime"]
         if not isinstance(runtime_raw, Mapping) or set(runtime_raw) - _RA_RUNTIME_FIELDS:
             raise RuntimeAdapterError("runtime provenance contains unsupported fields")
         runtime = {str(key): _ra_id(raw, f"runtime.{key}") for key, raw in runtime_raw.items()}
-        error_code = value.get("error_code")
+        error_code = value["error_code"]
         if error_code is not None:
             error_code = _ra_id(error_code, "error_code")
-        error_message = value.get("error_message")
+        error_message = value["error_message"]
         if error_message is not None:
             error_message = _ra_text(error_message, "error_message", max_length=256)
-        fallback = value.get("external_fallback_allowed", False)
+        fallback = value["external_fallback_allowed"]
         if fallback is not False:
             raise RuntimeAdapterError("external fallback is permanently disabled by the core contract")
         if status in {"success", "partial"} and output is None:

tests/test_context_evidence.py

@@ -81,15 +81,12 @@ def test_score_does_not_become_a_truth_gate_and_exclusions_are_bounded():
     assert projection["excluded"] == [{"candidate_id": "missing", "reason": "scope mismatch"}]
 
 
-def test_raw_material_is_never_emitted_even_when_used_for_digest():
-    projection = perseus.project_context_evidence(
-        [_entry(content="password=top-secret", body="private body")],
-        evidence_required=True,
-    )
-    serialized = json.dumps(projection, sort_keys=True).lower()
-    assert "top-secret" not in serialized
-    assert "private body" not in serialized
-    assert projection["selected"][0]["evidence_digest"]
+def test_raw_material_digest_mismatch_is_rejected_before_projection():
+    with pytest.raises(perseus.ContextEvidenceError, match="evidence"):
+        perseus.project_context_evidence(
+            [_entry(content="password=top-secret", body="private body")],
+            evidence_required=True,
+        )
 
 
 def test_item_without_source_reference_is_excluded_even_with_a_digest():

tests/test_contract_schemas.py

@@ -47,6 +47,7 @@ def test_new_contract_schemas_are_valid_and_accept_reference_envelopes():
         "schema_version": "perseus-runtime-request/v1",
         "request_id": "schema-request",
         "execution_profile": profile,
+        "execution_profile_digest": profile["profile_digest"],
         "context_digest": "a" * 64,
         "evidence_digest": "b" * 64,
         "input_digest": "c" * 64,

tests/test_runtime_adapter.py

@@ -52,6 +52,7 @@ def _request(**extra):
         "schema_version": "perseus-runtime-request/v1",
         "request_id": "request-1",
         "execution_profile": _profile(),
+        "execution_profile_digest": _profile()["profile_digest"],
         "context_digest": "a" * 64,
         "evidence_digest": "b" * 64,
         "input_digest": "c" * 64,