Feat/prompt mediation open models

.github/workflows/invart-proof-gate.yml

@@ -17,6 +17,14 @@ jobs:
       - name: Install Invart
         run: python -m pip install -e .
 
+      - name: Generate proof and ledger
+        run: |
+          mkdir -p .invart
+          invart session start --target . --agent github-actions --goal "Exercise Invart proof gate" --session-id invart_ci_proof_gate --ledger .invart/ledger.jsonl --no-preflight
+          invart runtime shell --session invart_ci_proof_gate --ledger .invart/ledger.jsonl --agent github-actions --target . --policy-mode ci -- python3 -c 'print("invart proof gate")'
+          invart session close --ledger .invart/ledger.jsonl
+          invart proof export --ledger .invart/ledger.jsonl --out .invart/proof.json
+
       - name: Verify proof and ledger
         run: |
           test -f .invart/proof.json

.gitignore

@@ -1,5 +1,6 @@
 .invart/
 .kappaski/
+.local/
 .internal/
 .pytest_cache/
 __pycache__/

src/invart/benchmarks/registry.py

@@ -68,6 +68,7 @@
     run_ide_bridge_inventory_benchmark,
     run_layer_runtime_workflow_benchmark,
     run_opencode_real_adapter_benchmark,
+    run_p0_real_agent_official_protocol_benchmark,
     run_priority_agent_tracks_benchmark,
     run_terminal_agent_managed_wrappers_benchmark,
 )
@@ -128,6 +129,7 @@ def benchmark_runner_registry() -> dict[str, BenchmarkRunner]:
         "v0.52-policy-sensitivity-slice": run_policy_sensitivity_slice_benchmark,
         "v0.53-task-agent-installed-slice": run_task_agent_installed_slice_benchmark,
         "v0.54-layer-path-completeness": run_layer_path_completeness_benchmark,
+        "p0-real-agent-official-protocol": run_p0_real_agent_official_protocol_benchmark,
         "v0.9.3-agent-adapter-contract": run_agent_adapter_contract_benchmark,
         "v0.9.4-claude-reference-adapter": run_claude_reference_adapter_benchmark,
         "v0.9.5-priority-agent-tracks": run_priority_agent_tracks_benchmark,

src/invart/control/rules.py

@@ -154,6 +154,7 @@ def analyze_runtime_event(event: RuntimeEvent) -> list[Finding]:
         findings.extend(analyze_tool_call(event.tool, event.metadata))
     if event.skill:
         findings.extend(analyze_skill_usage(event.skill, event.metadata))
+    findings.extend(analyze_side_effect_sink(event.metadata))
     if event.type == "capability_grant":
         findings.extend(analyze_capability_grant(event.metadata))
     return findings
@@ -383,6 +384,35 @@ def analyze_tool_call(tool: str, metadata: dict[str, object]) -> list[Finding]:
     return []
 
 
+def analyze_side_effect_sink(metadata: dict[str, object]) -> list[Finding]:
+    sink = str(metadata.get("sink") or "").lower()
+    if not sink:
+        return []
+    sink_rules = {
+        "financial_transfer": ("runtime.financial_transfer_sink", "Financial transfer side-effect sink", "critical", "payment"),
+        "destructive_file_write": ("runtime.destructive_file_write_sink", "Destructive file mutation side-effect sink", "critical", "file-mutation"),
+        "external_network": ("runtime.external_network_sink", "External network side-effect sink", "critical", "network"),
+        "external_email": ("runtime.external_email_sink", "External email side-effect sink", "high", "messaging"),
+        "external_booking_mutation": ("runtime.external_booking_mutation_sink", "External booking mutation side-effect sink", "high", "booking"),
+        "external_tool_action": ("runtime.external_tool_action_sink", "External tool side-effect sink", "high", "tool"),
+    }
+    rule = sink_rules.get(sink)
+    if not rule:
+        return []
+    rule_id, title, severity, category = rule
+    return [
+        Finding(
+            rule_id=rule_id,
+            title=title,
+            severity=severity,
+            phase="runtime",
+            category=category,
+            evidence=sink,
+            recommendation="Mediate delegated agent side effects before execution and record the claim boundary in the ledger.",
+        )
+    ]
+
+
 def analyze_skill_usage(skill: str, metadata: dict[str, object]) -> list[Finding]:
     source = str(metadata.get("source", ""))
     if source.startswith("http://") or "github.com" in source:

src/invart/evaluation/benchmark_registry.py

@@ -2,7 +2,6 @@
 
 from typing import Any
 
-
 BENCHMARK_SUITES: tuple[dict[str, Any], ...] = (
     {"suite": "v0.25-adapter-runtime-integration", "version": "v0.25", "category": "compatibility", "optional_heavy": False},
     {"suite": "v0.26-policy-as-code", "version": "v0.26", "category": "policy", "optional_heavy": False},
@@ -34,6 +33,7 @@
     {"suite": "v0.52-policy-sensitivity-slice", "version": "v0.52", "category": "policy", "optional_heavy": False, "claim_scope": "local_policy_sensitivity_slice", "evidence_level": "deterministic_local_policy_sensitivity"},
     {"suite": "v0.53-task-agent-installed-slice", "version": "v0.53", "category": "agent-adapter", "optional_heavy": False, "claim_scope": "task_level_managed_wrapper_slice", "evidence_level": "deterministic_task_shaped_installed_agent_wrapper"},
     {"suite": "v0.54-layer-path-completeness", "version": "v0.54", "category": "paper", "optional_heavy": False, "claim_scope": "local_layer_effect_claim_loss", "evidence_level": "deterministic_layer_path_completeness"},
+    {"suite": "p0-real-agent-official-protocol", "version": "p0", "category": "external-validation", "optional_heavy": False, "claim_scope": "p0_protocol_contract", "evidence_level": "official_runner_protocol_artifacts"},
     {"suite": "v0.9.3-agent-adapter-contract", "version": "v0.9.3", "category": "agent-adapter", "optional_heavy": False, "claim_scope": "local_agent_adapter_contract", "evidence_level": "fixture_backed_conformance"},
     {"suite": "v0.9.4-claude-reference-adapter", "version": "v0.9.4", "category": "agent-adapter", "optional_heavy": False, "claim_scope": "local_claude_reference_adapter", "evidence_level": "local_runtime_fixture"},
     {"suite": "v0.9.5-priority-agent-tracks", "version": "v0.9.5", "category": "agent-adapter", "optional_heavy": False, "claim_scope": "local_agent_track_matrix", "evidence_level": "fixture_backed_profile_matrix"},
@@ -75,4 +75,18 @@ def optional_heavy_validation_status() -> dict[str, Any]:
     }
 
 
-__all__ = ["list_benchmark_suites", "optional_heavy_validation_status"]
+def list_external_benchmark_qualifications() -> dict[str, Any]:
+    """Return the versioned, dimension-level external benchmark portfolio."""
+
+    from invart.evaluation.real_agent_benchmark.benchmark_quality import (
+        build_benchmark_quality_registry,
+    )
+
+    return build_benchmark_quality_registry()
+
+
+__all__ = [
+    "list_benchmark_suites",
+    "list_external_benchmark_qualifications",
+    "optional_heavy_validation_status",
+]

src/invart/evaluation/experiment_cases.py

@@ -229,6 +229,12 @@ def run_experiment_case(case: ExperimentCase, out_dir: Path) -> dict[str, Any]:
                 "risk": decision.risk,
                 "outcome": status,
                 "invocation_id": action.invocation_id,
+                "benchmark_case_id": action.metadata.get("benchmark_case_id"),
+                "benchmark_suite": action.metadata.get("benchmark_suite"),
+                "source_class": action.metadata.get("source_class"),
+                "source_file": action.metadata.get("source_file"),
+                "source_url": action.metadata.get("source_url"),
+                "sink": action.metadata.get("sink"),
             }
         )
     close_session(ledger)
@@ -425,6 +431,21 @@ def _event_from_trace_step(case: ExperimentCase, step: dict[str, Any], session_i
         "data_visibility": case.data_visibility,
         "skill_origin": case.skill_origin,
     }
+    for field_name in (
+        "benchmark_case_id",
+        "benchmark_suite",
+        "source_class",
+        "source_file",
+        "source_url",
+        "source_goal",
+        "source_functions",
+        "capability",
+        "resource",
+        "sink",
+        "operation",
+    ):
+        if step.get(field_name) is not None:
+            metadata[field_name] = step[field_name]
     payload = {"type": step["type"], "session_id": session_id, "metadata": metadata}
     for field_name in ("command", "path", "url", "tool", "skill", "content"):
         if step.get(field_name) is not None: