fix: MEDIUM tier cont. — reconstruction ambiguity, governor fail-open, sim-scaffold marker, parity coverage gate

src/pi_extension_governor/governor.py

@@ -107,10 +107,12 @@ def process_bundle(
 
         # Scan source code for shadow/hidden parameters using the standalone PiSchemaGhost micro-agent
         from pi_micro_agents.pi_schema_ghost import detect_shadow_parameters
-        from pi_micro_agents.pi_schema_ghost import is_strict_mode as is_ghost_strict_mode
 
         ghost_risk, ghost_violations = detect_shadow_parameters(entrypoint_source)
-        if ghost_risk >= 71.0 and is_ghost_strict_mode():
+        # Reject high risk unconditionally (like detect_prompt_injection). Gating
+        # this on a strict-mode toggle made it an env-reachable per-detector kill
+        # switch that silently admitted high-risk extensions.
+        if ghost_risk >= 71.0:
             return ExtensionAdmissionResult(
                 manifest_id=manifest.extension_id,
                 admitted=False,
@@ -126,10 +128,9 @@ def process_bundle(
 
         # Scan source code for invisible guardrail evasions using the standalone PiCoTShadow micro-agent
         from pi_micro_agents.pi_cot_shadow import detect_invisible_guardrails
-        from pi_micro_agents.pi_cot_shadow import is_strict_mode as is_cot_strict_mode
 
         cot_risk, cot_violations = detect_invisible_guardrails(entrypoint_source)
-        if cot_risk >= 71.0 and is_cot_strict_mode():
+        if cot_risk >= 71.0:
             return ExtensionAdmissionResult(
                 manifest_id=manifest.extension_id,
                 admitted=False,
@@ -145,10 +146,9 @@ def process_bundle(
 
         # Scan source code for illegal surplus sub-key leakage using the standalone PiTokenSurplusOrchestrator micro-agent
         from pi_micro_agents.pi_surplus_orchestrator import detect_surplus_violations
-        from pi_micro_agents.pi_surplus_orchestrator import is_strict_mode as is_surplus_strict_mode
 
         surplus_risk, surplus_violations = detect_surplus_violations(entrypoint_source)
-        if surplus_risk >= 71.0 and is_surplus_strict_mode():
+        if surplus_risk >= 71.0:
             return ExtensionAdmissionResult(
                 manifest_id=manifest.extension_id,
                 admitted=False,
@@ -164,10 +164,9 @@ def process_bundle(
 
         # Scan source code for spend/cost anomalies using the standalone SpendAnomalyHunter micro-agent
         from pi_micro_agents.pi_spend_hunter import detect_spend_anomalies
-        from pi_micro_agents.pi_spend_hunter import is_strict_mode as is_spend_strict_mode
 
         spend_risk, spend_violations = detect_spend_anomalies(entrypoint_source)
-        if spend_risk >= 71.0 and is_spend_strict_mode():
+        if spend_risk >= 71.0:
             return ExtensionAdmissionResult(
                 manifest_id=manifest.extension_id,
                 admitted=False,

src/pi_interoperability_layer/platform/execution_fabric.py

@@ -1,9 +1,15 @@
 """Shard-Coordinated Deterministic Execution Fabric.
 
-Compiler-style distributed execution with global barriers.
-No swarm semantics. No autonomous behavior.
-Deterministic partitioning, phase-locked orchestration,
-ephemeral worker leasing, and replay recovery.
+⚠️  SIMULATION / REFERENCE SCAFFOLD — NOT a live execution path.
+
+This models the *shape* of a compiler-style distributed fabric (deterministic
+partitioning, phase-locked orchestration, worker leasing, replay recovery), but
+``execute_phase`` does NOT distribute or run anything: it resolves every step to
+a hash of the input via ``_simulate_execution`` (see the "# simulated" markers).
+It has no production caller — only its own unit/integration tests import it. Do
+not treat it as evidence that real distributed/barrier execution exists. Wire it
+to a real dispatcher, or move it under an examples/ namespace, before relying on
+it as a platform capability.
 """
 
 from __future__ import annotations

src/pi_micro_agents/orchestrator/consensus.py

@@ -161,15 +161,26 @@ def _find_output_model(agent_class, result_keys):
     if mod is None:
         return None
     want = set(result_keys)
-    for v in vars(mod).values():
+    matches = [
+        v
+        for v in vars(mod).values()
         if (
             isinstance(v, type)
             and issubclass(v, BaseModel)
             and v is not BaseModel
             and set(v.model_fields.keys()) == want
-        ):
-            return v
-    return None
+        )
+    ]
+    if len(matches) > 1:
+        # Ambiguous: ≥2 models share this exact field set, so reconstructing by
+        # field-set match would arbitrarily pick one (vars() iteration order) and
+        # could build the WRONG type. Refuse — the caller (_try_rust_agent) catches
+        # this and falls back to the Python agent rather than risk a wrong result.
+        raise ValueError(
+            f"ambiguous Rust output reconstruction in {getattr(mod, '__name__', '?')}: "
+            f"{[m.__name__ for m in matches]} all match field set {sorted(want)}"
+        )
+    return matches[0] if matches else None
 
 
 def _try_rust_agent(agent_name, agent_class, perturbed):

tests/conformance/test_rust_parity_coverage.py

@@ -0,0 +1,44 @@
+"""Every Rust-registered agent must have a parity spec (coverage gate).
+
+The Rust core is sold as "byte-for-byte equivalent to Python". The existing
+rust/parity check only verifies specs ⊆ registry (every spec maps to a real
+agent). This adds the missing direction — registry ⊆ specs — so a Rust agent can
+never be added without a parity spec, which would otherwise silently run
+unverified-against-Python. Pure-Python (parses sources), so it runs in the main
+CI without building the cdylib.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parents[2]
+_REGISTRY = _REPO / "rust" / "crates" / "pi-agents" / "src" / "registry.rs"
+_SPEC_DIR = _REPO / "rust" / "parity" / "specs"
+
+
+def _registered_agents() -> set:
+    text = _REGISTRY.read_text(encoding="utf-8")
+    return set(re.findall(r'm\.insert\(\s*"([^"]+)"', text))
+
+
+def _spec_rust_names() -> set:
+    names = set()
+    for fp in _SPEC_DIR.glob("*.py"):
+        if fp.name.startswith("_"):
+            continue
+        m = re.search(r'RUST_NAME\s*=\s*"([^"]+)"', fp.read_text(encoding="utf-8"))
+        if m:
+            names.add(m.group(1))
+    return names
+
+
+def test_every_registered_rust_agent_has_a_parity_spec():
+    registered = _registered_agents()
+    specs = _spec_rust_names()
+    # Guard against a vacuous pass (e.g. a parse regression returning empty sets).
+    assert len(registered) >= 200, f"registry parse looks wrong: only {len(registered)} agents"
+    assert len(specs) >= 200, f"spec parse looks wrong: only {len(specs)} specs"
+    missing = registered - specs
+    assert not missing, f"Rust agents registered but with NO parity spec (add one): {sorted(missing)}"

tests/unit/pi-extension-governor/test_governor_failopen.py

@@ -0,0 +1,63 @@
+"""Governor must reject high-risk extensions regardless of per-detector strict mode.
+
+Finding: four of the five source scanners only rejected a >=71 risk when their
+strict-mode toggle was on (`if risk >= 71.0 and is_*_strict_mode()`), so an
+operator-controlled env var / config could silently downgrade a high-risk
+admission to advisory — a per-detector kill switch. Only detect_prompt_injection
+rejected unconditionally. The four must match it.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from pi_extension_governor.governor import ExtensionGovernor
+from pi_extension_governor.manifest import CapabilityClass, ExtensionBundle, ExtensionManifest, TrustZone
+from pi_extension_governor.policy import ExtensionGovernancePolicy
+from pi_extension_governor.provenance import ExtensionProvenanceLedger
+from pi_extension_governor.trust_zones import TrustZoneEnforcer
+
+
+def _governor(td):
+    return ExtensionGovernor(
+        ExtensionGovernancePolicy(),
+        ExtensionProvenanceLedger(ledger_dir=Path(td) / "ledger"),
+        TrustZoneEnforcer(),
+    )
+
+
+def _bundle():
+    manifest = ExtensionManifest(
+        extension_id="x",
+        package_name="x",
+        package_version="1.0.0",
+        package_hash="h",
+        capability_class=CapabilityClass.OPENAPI_TOOLING,
+        trust_zone=TrustZone.GOVERNED_EXTENSION,
+    )
+    return ExtensionBundle(bundle_id="b", manifest=manifest, payload_hash="ph")
+
+
+def test_high_shadow_risk_rejected_even_with_strict_mode_off(monkeypatch):
+    # Force the shadow-parameter scanner to report high risk, and turn its
+    # strict-mode toggle OFF. The bundle must still be rejected.
+    monkeypatch.setattr(
+        "pi_micro_agents.pi_schema_ghost.detect_shadow_parameters", lambda src: (95.0, ["shadow_param"])
+    )
+    monkeypatch.setattr("pi_micro_agents.pi_schema_ghost.is_strict_mode", lambda: False)
+
+    with tempfile.TemporaryDirectory() as td:
+        result = _governor(td).process_bundle(_bundle(), "OUTPUT = {}", {})
+    assert result.admitted is False
+    assert "shadow" in result.reason.lower()
+
+
+def test_high_spend_risk_rejected_even_with_strict_mode_off(monkeypatch):
+    monkeypatch.setattr("pi_micro_agents.pi_spend_hunter.detect_spend_anomalies", lambda src: (88.0, ["spend_anomaly"]))
+    monkeypatch.setattr("pi_micro_agents.pi_spend_hunter.is_strict_mode", lambda: False)
+
+    with tempfile.TemporaryDirectory() as td:
+        result = _governor(td).process_bundle(_bundle(), "OUTPUT = {}", {})
+    assert result.admitted is False
+    assert "spend" in result.reason.lower()

tests/unit/pi-micro-agents/test_find_output_model.py

@@ -0,0 +1,54 @@
+"""_find_output_model must not silently pick a model on an ambiguous field-set match.
+
+Finding: the Rust-output reconstruction selects the agent module's pydantic model
+whose field set equals the Rust JSON keys, returning the FIRST match in vars()
+iteration order. If a module defines two models with identical field sets, the
+choice is arbitrary (iteration-order dependent) and could reconstruct the wrong
+type. It must instead refuse (raise) on ambiguity so the caller falls back to the
+Python agent rather than risking a wrong reconstruction.
+"""
+
+from __future__ import annotations
+
+import types
+
+import pytest
+from pydantic import BaseModel
+
+from pi_micro_agents.orchestrator import consensus
+
+
+def test_find_output_model_single_match(monkeypatch):
+    class OutA(BaseModel):
+        x: int
+        y: int
+
+    mod = types.ModuleType("fake_single_mod")
+    mod.OutA = OutA
+    monkeypatch.setitem(__import__("sys").modules, "fake_single_mod", mod)
+
+    class A:
+        __module__ = "fake_single_mod"
+
+    assert consensus._find_output_model(A, {"x", "y"}) is OutA
+
+
+def test_find_output_model_raises_on_ambiguous_field_set(monkeypatch):
+    class OutA(BaseModel):
+        x: int
+        y: int
+
+    class OutB(BaseModel):
+        x: int
+        y: int
+
+    mod = types.ModuleType("fake_ambiguous_mod")
+    mod.OutA = OutA
+    mod.OutB = OutB
+    monkeypatch.setitem(__import__("sys").modules, "fake_ambiguous_mod", mod)
+
+    class A:
+        __module__ = "fake_ambiguous_mod"
+
+    with pytest.raises(Exception):
+        consensus._find_output_model(A, {"x", "y"})