diff --git a/integrations/agentrust-confidential-workflow-harness/README.md b/integrations/agentrust-confidential-workflow-harness/README.md index a352a39f..121a15c3 100644 --- a/integrations/agentrust-confidential-workflow-harness/README.md +++ b/integrations/agentrust-confidential-workflow-harness/README.md @@ -1,5 +1,46 @@ # Confidential Workflow Acceptance Harness +The [canonical acceptance procedure](https://github.com/agentrust-io/integrations/issues/199) +defines the paired controls and evidence boundaries. The utility-specific +reproduction steps and local trust boundary are documented below. + +## Offline synthetic evidence reappraisal + +`harness.reappraisal` retains an immutable copy of a deterministic run result +and compares it with explicit current verifier inputs: scenario, policy/trust +digests, a named dependency inventory, evaluation time and freshness budget. +Changed inputs or expired evidence return `reassessment_required` with reasons; +the historical result is retained. A changed dependency calls for reassessment, +not automatic refusal of the new workload. Refused, unknown and incomplete +historical results cannot become applicable without a fresh passing run. + +Run with `HarnessRunner().run(scenario, observed_at=...)` to retain the full +scenario digest and original observation time in memory. These fields are +private local reuse metadata and are always excluded from public `to_dict()` +output. Both the result and each observation use explicit public-field lists; +new internal dataclass fields are excluded by default, and emitted mutable +values are copied. The public result schema rejects the private metadata. `EvidenceSnapshot.capture` +retains them separately inside its local `result_json`; the snapshot, its +content digests and dataclass representations are private and must not be +published. Capture rejects mismatched scenarios and missing or future +observation times. Later capture never resets evidence age. Legacy run objects +without both fields require a fresh run before retention for reappraisal. + +Use `EvidenceSnapshot.capture(scenario, result, context, captured_at=...)`, then +`appraise(snapshot, scenario, current_context, evaluated_at=..., +max_age_seconds=...)`. Times are UTC Unix integer seconds. The optional run timestamp is caller-trusted +fixture input, not authenticated time. Evidence +expires exactly at the verifier-selected age limit. This is evidence freshness, +not the credential-expiry verification exercised by protocol-specific adapters. +The existing unittest discovery command runs the paired reappraisal controls. + +This utility accepts caller-trusted local synthetic fixtures only. It does not +authenticate a snapshot, verify completeness of caller-supplied dependencies, +provide persistent revocation, execute an adapter, or authorize a new action. +Its hashes identify bytes; replacing both content and hashes is outside this +local trust boundary. `applicable` is limited to the selected synthetic claim +and explicit inputs, not protocol acceptance or permission to execute. + A deterministic software harness for issue #199. It keeps authorization, delivery, receipt, admission, execution, response verification and release outcomes separate and preserves unavailable evidence instead of inferring success from dispatch. @@ -22,6 +63,7 @@ attestation, or completion of the full software milestone. ## Reproduce ```bash +python -m pip install --require-hashes -r ../../requirements/validate.txt python -m unittest discover -s tests -v ``` diff --git a/integrations/agentrust-confidential-workflow-harness/harness/model.py b/integrations/agentrust-confidential-workflow-harness/harness/model.py index 20492855..3a4249cb 100644 --- a/integrations/agentrust-confidential-workflow-harness/harness/model.py +++ b/integrations/agentrust-confidential-workflow-harness/harness/model.py @@ -1,5 +1,8 @@ from __future__ import annotations from dataclasses import dataclass, field, asdict +from copy import deepcopy +import hashlib +import json from enum import Enum from typing import Any @@ -50,6 +53,11 @@ class Scenario: waiting_on_release: str | None = None inputs: dict[str, Any] = field(default_factory=dict) + def digest(self) -> str: + raw = json.dumps(asdict(self), sort_keys=True, separators=(",", ":"), + allow_nan=False).encode("utf-8") + return "sha256:" + hashlib.sha256(raw).hexdigest() + @dataclass class RunResult: scenario_id: str @@ -63,15 +71,32 @@ class RunResult: limitations: list[str] package_versions: dict[str, str] = field(default_factory=dict) + scenario_digest: str | None = None + observed_at: int | None = None + def to_dict(self) -> dict[str, Any]: - d = asdict(self) - d["observations"] = [ - { - **asdict(o), - "boundary": o.boundary.value, - "outcome": o.outcome.value, - "evidence_class": o.evidence_class.value, - } - for o in self.observations - ] - return d + # Explicit public projection: new internal fields are private by default. + # Values in declared public fields still require caller-side classification. + return deepcopy({ + "scenario_id": self.scenario_id, + "transaction_id": self.transaction_id, + "claim": self.claim, + "status": self.status, + "observations": [ + { + "boundary": o.boundary.value, + "outcome": o.outcome.value, + "source": o.source, + "evidence_class": o.evidence_class.value, + "lineage": o.lineage, + "caused_by": o.caused_by, + "detail": o.detail, + } + for o in self.observations + ], + "boundary_outcomes": self.boundary_outcomes, + "mutation_proof": self.mutation_proof, + "blocked": self.blocked, + "limitations": self.limitations, + "package_versions": self.package_versions, + }) diff --git a/integrations/agentrust-confidential-workflow-harness/harness/reappraisal.py b/integrations/agentrust-confidential-workflow-harness/harness/reappraisal.py new file mode 100644 index 00000000..64b85992 --- /dev/null +++ b/integrations/agentrust-confidential-workflow-harness/harness/reappraisal.py @@ -0,0 +1,150 @@ +"""Offline applicability checks for retained deterministic harness results. + +Inputs are local, caller-trusted fixtures, not authenticated protocol evidence. +An applicable result is not permission to execute, a freshness attestation, or a +replacement for independent acceptance. No adapter or live transport is called. +""" +from dataclasses import asdict, dataclass +import hashlib +import json +import re + +from .model import EvidenceClass, RunResult, Scenario +from .verdict import status, summarize + + +def _digest(value: object) -> str: + raw = json.dumps(value, sort_keys=True, separators=(",", ":"), + allow_nan=False).encode("utf-8") + return "sha256:" + hashlib.sha256(raw).hexdigest() + + +def _check_digest(value: str) -> None: + if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value): + raise ValueError("expected a lowercase sha256 digest") + + +def _check_time(value: int) -> None: + if type(value) is not int or value < 0: + raise ValueError("time must be a non-negative integer UTC Unix second") + + +@dataclass(frozen=True) +class EvaluationContext: + """Verifier-selected fixture inputs, supplied separately from the producer.""" + + policy_digest: str + trust_digest: str + dependencies: tuple[tuple[str, str], ...] + + def __post_init__(self): + _check_digest(self.policy_digest) + _check_digest(self.trust_digest) + if not isinstance(self.dependencies, tuple) or not self.dependencies: + raise ValueError("dependencies must be a non-empty immutable tuple") + names = [] + for entry in self.dependencies: + if not isinstance(entry, tuple) or len(entry) != 2: + raise ValueError("dependency entries must be (name, digest) tuples") + name, digest = entry + if not isinstance(name, str) or not name.strip(): + raise ValueError("dependency names must be non-empty") + _check_digest(digest) + names.append(name) + if len(set(names)) != len(names): + raise ValueError("duplicate dependency name") + + +@dataclass(frozen=True) +class EvidenceSnapshot: + """Immutable local record; the digest is identity, not authentication.""" + + scenario_digest: str + result_json: str + result_digest: str + context: EvaluationContext + captured_at: int + + @classmethod + def capture(cls, scenario: Scenario, result: RunResult, + context: EvaluationContext, *, captured_at: int): + _check_time(captured_at) + if not isinstance(context, EvaluationContext): + raise ValueError("explicit evaluation context required") + if (scenario.scenario_id, scenario.transaction_id, scenario.claim) != ( + result.scenario_id, result.transaction_id, result.claim): + raise ValueError("result does not match scenario identity") + if result.scenario_digest != scenario.digest(): + raise ValueError("result does not match full scenario digest") + _check_time(result.observed_at) + if result.observed_at > captured_at: + raise ValueError("capture precedes observation") + if scenario.mutation is not None or result.mutation_proof != { + "mutation": None, "gate_weakened": False}: + raise ValueError("weakened-gate evidence cannot be retained for reuse") + if not result.observations or any( + o.evidence_class != EvidenceClass.SYNTHETIC + for o in result.observations): + raise ValueError("only explicit synthetic observations are supported") + boundaries = summarize(result.observations) + blocked = [scenario.waiting_on_release] if scenario.waiting_on_release else [] + if (result.boundary_outcomes != boundaries or result.blocked != blocked + or result.status != status(boundaries, blocked)): + raise ValueError("result conflicts with retained observations") + # Local retention is deliberately separate from public serialization. + data = result.to_dict() + data.update(scenario_digest=result.scenario_digest, observed_at=result.observed_at) + encoded = json.dumps(data, sort_keys=True, separators=(",", ":"), allow_nan=False) + return cls(result.scenario_digest, encoded, _digest(data), context, captured_at) + + +@dataclass(frozen=True) +class Reappraisal: + historical_status: str + applicability: str + reasons: tuple[str, ...] + evidence_class: str = "synthetic" + + +def appraise(snapshot: EvidenceSnapshot, scenario: Scenario, + current: EvaluationContext, *, evaluated_at: int, + max_age_seconds: int) -> Reappraisal: + """Compare a retained fixture against current verifier inputs. + + The caller selects a freshness budget on EVERY appraisal. Exactly at the + deadline the evidence is expired. This is evidence-age checking, not a + credential-expiry implementation. A changed context requires a fresh run; + it does not imply that the new workload is incompatible or unauthorized. + """ + _check_time(evaluated_at) + _check_time(snapshot.captured_at) + if type(max_age_seconds) is not int or max_age_seconds <= 0: + raise ValueError("a positive verifier-selected freshness budget is required") + if not isinstance(current, EvaluationContext): + raise ValueError("explicit current evaluation context required") + data = json.loads(snapshot.result_json) + if _digest(data) != snapshot.result_digest: + raise ValueError("snapshot content identity mismatch") + _check_time(data["observed_at"]) + if data["observed_at"] > snapshot.captured_at: + raise ValueError("capture precedes observation") + if data["scenario_digest"] != snapshot.scenario_digest: + raise ValueError("snapshot scenario identity mismatch") + reasons = [] + if snapshot.scenario_digest != _digest(asdict(scenario)): + reasons.append("scenario_changed") + if snapshot.context.policy_digest != current.policy_digest: + reasons.append("policy_changed") + if snapshot.context.trust_digest != current.trust_digest: + reasons.append("trust_changed") + if dict(snapshot.context.dependencies) != dict(current.dependencies): + reasons.append("dependencies_changed") + if evaluated_at < snapshot.captured_at: + reasons.append("evaluation_precedes_capture") + elif evaluated_at - data["observed_at"] >= max_age_seconds: + reasons.append("evidence_expired") + if data["status"] != "passed": + reasons.append("historical_result_not_passed") + return Reappraisal(data["status"], + "reassessment_required" if reasons else "applicable", + tuple(reasons)) diff --git a/integrations/agentrust-confidential-workflow-harness/harness/runner.py b/integrations/agentrust-confidential-workflow-harness/harness/runner.py index 33bb24ef..1ec3dc2b 100644 --- a/integrations/agentrust-confidential-workflow-harness/harness/runner.py +++ b/integrations/agentrust-confidential-workflow-harness/harness/runner.py @@ -1,4 +1,5 @@ from __future__ import annotations +import time from .adapters.deterministic import DeterministicWorkflowAdapter from .model import RunResult, Scenario from .verdict import status, summarize @@ -13,13 +14,23 @@ class HarnessRunner: def __init__(self, adapter=None): self.adapter = adapter or DeterministicWorkflowAdapter() - def run(self, scenario: Scenario) -> RunResult: + def run(self, scenario: Scenario, *, observed_at: int | None = None) -> RunResult: + # The timestamp is fixed at run entry, never at later snapshot capture. + if observed_at is None: + observed_at = int(time.time()) + if type(observed_at) is not int or observed_at < 0: + raise ValueError("observation time must be a non-negative integer UTC Unix second") + scenario_digest = scenario.digest() if scenario.mutation is not None and scenario.mutation not in SUPPORTED_MUTATIONS: raise ValueError(f"unsupported mutation: {scenario.mutation!r}") observations = list(self.adapter.run(scenario).observations) + if scenario.digest() != scenario_digest: + raise ValueError("scenario changed during run") boundaries = summarize(observations) blocked = [scenario.waiting_on_release] if scenario.waiting_on_release else [] return RunResult( + scenario_digest=scenario_digest, + observed_at=observed_at, scenario_id=scenario.scenario_id, transaction_id=scenario.transaction_id, claim=scenario.claim, diff --git a/integrations/agentrust-confidential-workflow-harness/tests/test_public_privacy.py b/integrations/agentrust-confidential-workflow-harness/tests/test_public_privacy.py new file mode 100644 index 00000000..ed8ed847 --- /dev/null +++ b/integrations/agentrust-confidential-workflow-harness/tests/test_public_privacy.py @@ -0,0 +1,114 @@ +"""Public projection and private retention regression controls.""" +from dataclasses import field, fields, make_dataclass, replace +import json +import unittest + +from harness import HarnessRunner +from harness.evidence import validate_no_sensitive_values +from harness.model import Observation, RunResult +from harness.reappraisal import EvidenceSnapshot, EvaluationContext, appraise +from test_harness import scenario + + +class PublicPrivacyTests(unittest.TestCase): + def test_candidate_matching_has_no_public_scenario_digest(self): + candidates = ('ALPHA', 'BRAVO', 'CHARLIE') + scenarios = [scenario(inputs={**scenario().inputs, 'protected_code': value}) + for value in candidates] + result = HarnessRunner().run(scenarios[0], observed_at=100) + validate_no_sensitive_values(result, candidates) + packet = result.to_dict() + self.assertNotIn('scenario_digest', packet) + self.assertEqual([value for value, selected in zip(candidates, scenarios) + if selected.digest() == packet.get('scenario_digest')], []) + self.assertEqual(result.scenario_digest, scenarios[0].digest()) + + def setUp(self): + digest = 'sha256:' + 'a' * 64 + self.context = EvaluationContext(digest, digest, (('checker', digest),)) + + def test_private_metadata_absent_for_every_terminal_status(self): + cases = [scenario(), + scenario(inputs={**scenario().inputs, 'revoked_before_use': True}), + scenario(inputs={**scenario().inputs, 'timeout_after_dispatch': True}), + scenario(waiting_on_release='missing API')] + statuses = set() + for selected in cases: + result = HarnessRunner().run(selected, observed_at=100) + before = (result.scenario_digest, result.observed_at) + public = result.to_dict() + statuses.add(public['status']) + self.assertNotIn('scenario_digest', public) + self.assertNotIn('observed_at', public) + self.assertNotIn(selected.digest(), json.dumps(public)) + self.assertEqual((result.scenario_digest, result.observed_at), before) + self.assertEqual(statuses, {'passed', 'refused', 'unknown', 'incomplete'}) + + def test_unobserved_private_input_does_not_change_public_result(self): + public = [] + for value in ('synthetic-private-A', 'synthetic-private-B'): + selected = scenario(inputs={**scenario().inputs, 'private_fixture': value}) + result = HarnessRunner().run(selected, observed_at=100) + public.append(result.to_dict()) + self.assertNotIn(result.scenario_digest, json.dumps(public[-1])) + self.assertEqual(public[0], public[1]) + + def test_capture_retains_private_metadata_and_expiry(self): + selected = scenario() + result = HarnessRunner().run(selected, observed_at=100) + result.to_dict() + snapshot = EvidenceSnapshot.capture(selected, result, self.context, captured_at=105) + private = json.loads(snapshot.result_json) + self.assertEqual(private['scenario_digest'], selected.digest()) + self.assertEqual(private['observed_at'], 100) + self.assertEqual(appraise(snapshot, selected, self.context, + evaluated_at=109, max_age_seconds=10).applicability, + 'applicable') + self.assertEqual(appraise(snapshot, selected, self.context, + evaluated_at=110, max_age_seconds=10).reasons, + ('evidence_expired',)) + + def test_public_serialization_does_not_weaken_pairing(self): + selected = scenario() + result = HarnessRunner().run(selected, observed_at=100) + result.to_dict() + changed = replace(selected, inputs={**selected.inputs, 'workload': 'other'}) + with self.assertRaises(ValueError): + EvidenceSnapshot.capture(changed, result, self.context, captured_at=100) + legacy = replace(result, scenario_digest=None, observed_at=None) + self.assertEqual(result.to_dict(), legacy.to_dict()) + with self.assertRaises(ValueError): + EvidenceSnapshot.capture(selected, legacy, self.context, captured_at=100) + + def test_future_result_field_is_not_public(self): + result = HarnessRunner().run(scenario(), observed_at=100) + extended_type = make_dataclass( + "ExtendedResult", [("private_note", str, field(default="synthetic-private"))], + bases=(RunResult,)) + extended = extended_type(**{f.name: getattr(result, f.name) for f in fields(RunResult)}) + self.assertEqual(extended.to_dict(), result.to_dict()) + self.assertNotIn("synthetic-private", json.dumps(extended.to_dict())) + + def test_future_observation_field_is_not_public(self): + result = HarnessRunner().run(scenario(), observed_at=100) + observation = result.observations[0] + extended_type = make_dataclass( + "ExtendedObservation", [("private_note", str, field(default="synthetic-private"))], + bases=(Observation,), frozen=True) + extended = extended_type(**{f.name: getattr(observation, f.name) + for f in fields(Observation)}) + changed = replace(result, observations=[extended, *result.observations[1:]]) + self.assertEqual(changed.to_dict(), result.to_dict()) + self.assertNotIn("synthetic-private", json.dumps(changed.to_dict())) + + def test_public_projection_copies_mutable_fields(self): + result = HarnessRunner().run(scenario(), observed_at=100) + baseline = result.to_dict() + public = result.to_dict() + public["mutation_proof"]["gate_weakened"] = True + public["boundary_outcomes"].clear() + public["blocked"].append("synthetic-extra") + public["limitations"].append("synthetic-extra") + public["package_versions"].clear() + public["observations"][0]["source"] = "synthetic-other" + self.assertEqual(result.to_dict(), baseline) diff --git a/integrations/agentrust-confidential-workflow-harness/tests/test_reappraisal.py b/integrations/agentrust-confidential-workflow-harness/tests/test_reappraisal.py new file mode 100644 index 00000000..9dd6a7ba --- /dev/null +++ b/integrations/agentrust-confidential-workflow-harness/tests/test_reappraisal.py @@ -0,0 +1,190 @@ +from dataclasses import asdict, replace +import json +import unittest +from unittest.mock import patch + +from harness import HarnessRunner +from harness.model import EvidenceClass +from harness.reappraisal import EvaluationContext, EvidenceSnapshot, appraise +from test_harness import scenario + + +A = "sha256:" + "a" * 64 +B = "sha256:" + "b" * 64 +C = "sha256:" + "c" * 64 + + +class ReappraisalTests(unittest.TestCase): + def setUp(self): + self.scenario = scenario() + self.context = EvaluationContext(A, B, (("lockfile", C), ("checker", A))) + self.result = HarnessRunner().run(self.scenario, observed_at=100) + self.snapshot = EvidenceSnapshot.capture( + self.scenario, self.result, self.context, captured_at=100) + + def evaluate(self, **kwargs): + args = dict(snapshot=self.snapshot, scenario=self.scenario, + current=self.context, evaluated_at=109, max_age_seconds=10) + args.update(kwargs) + return appraise(**args) + + def test_unchanged_valid_control(self): + self.assertEqual(self.evaluate().applicability, "applicable") + reordered = replace(self.context, dependencies=tuple(reversed(self.context.dependencies))) + self.assertEqual(self.evaluate(current=reordered).applicability, "applicable") + + def test_expiry_boundary_and_current_budget(self): + for instant in (110, 111): + with self.subTest(instant=instant): + r = self.evaluate(evaluated_at=instant) + self.assertEqual(r.reasons, ("evidence_expired",)) + self.assertEqual(r.applicability, "reassessment_required") + self.assertEqual(r.historical_status, "passed") + self.assertEqual(self.evaluate(max_age_seconds=9).reasons, ("evidence_expired",)) + self.assertEqual(self.evaluate(evaluated_at=99).reasons, + ("evaluation_precedes_capture",)) + + def test_policy_trust_and_dependency_changes_with_valid_twins(self): + cases = [ + (replace(self.context, policy_digest=C), "policy_changed"), + (replace(self.context, trust_digest=C), "trust_changed"), + (replace(self.context, dependencies=(("lockfile", A), ("checker", A))), + "dependencies_changed"), + (replace(self.context, dependencies=(("lockfile", C),)), "dependencies_changed"), + (replace(self.context, dependencies=(*self.context.dependencies, ("runtime", B))), + "dependencies_changed"), + ] + for changed, reason in cases: + with self.subTest(reason=reason, changed=changed): + self.assertEqual(self.evaluate().applicability, "applicable") + r = self.evaluate(current=changed) + self.assertEqual(r.reasons, (reason,)) + self.assertEqual(r.applicability, "reassessment_required") + self.assertEqual(r.historical_status, "passed") + + def test_changed_scenario_cannot_reuse_old_evidence(self): + for changed in (replace(self.scenario, transaction_id="other"), + replace(self.scenario, claim="different claim"), + replace(self.scenario, inputs={**self.scenario.inputs, "workload": "X"})): + with self.subTest(changed=changed): + self.assertEqual(self.evaluate(scenario=changed).reasons, ("scenario_changed",)) + + def test_capture_rejects_same_identity_changed_scenario(self): + changed = replace(self.scenario, inputs={**self.scenario.inputs, "workload": "X"}) + self.assertNotEqual(HarnessRunner().run(changed, observed_at=100).status, "passed") + with self.assertRaisesRegex(ValueError, "full scenario"): + EvidenceSnapshot.capture(changed, self.result, self.context, captured_at=100) + self.scenario.inputs["workload"] = "X" + with self.assertRaisesRegex(ValueError, "full scenario"): + EvidenceSnapshot.capture(self.scenario, self.result, self.context, captured_at=100) + + def test_delayed_capture_does_not_refresh_old_observation(self): + delayed = EvidenceSnapshot.capture( + self.scenario, self.result, self.context, captured_at=1000) + self.assertEqual(self.evaluate(snapshot=delayed, evaluated_at=1001).reasons, + ("evidence_expired",)) + fresh_result = HarnessRunner().run(self.scenario, observed_at=1000) + fresh = EvidenceSnapshot.capture( + self.scenario, fresh_result, self.context, captured_at=1000) + self.assertEqual(self.evaluate(snapshot=fresh, evaluated_at=1001).applicability, + "applicable") + + def test_missing_invalid_or_future_observation_is_rejected(self): + for value in (None, True, -1, 1.5, 101): + with self.subTest(value=value), self.assertRaises(ValueError): + EvidenceSnapshot.capture(self.scenario, replace(self.result, observed_at=value), + self.context, captured_at=100) + with self.assertRaisesRegex(ValueError, "full scenario"): + EvidenceSnapshot.capture(self.scenario, replace(self.result, scenario_digest=None), + self.context, captured_at=100) + + def test_compound_changes_do_not_mask_each_other(self): + current = EvaluationContext(C, A, (("other", B),)) + self.scenario.inputs["workload"] = "changed" + result = self.evaluate(current=current, evaluated_at=110) + self.assertEqual(result.reasons, ("scenario_changed", "policy_changed", "trust_changed", + "dependencies_changed", "evidence_expired")) + self.assertEqual(result.applicability, "reassessment_required") + self.assertEqual(result.historical_status, "passed") + + def test_fresh_run_after_change_preserves_old_snapshot(self): + before = asdict(self.snapshot) + changed = replace(self.context, dependencies=(("lockfile", A),)) + self.assertEqual(self.evaluate(current=changed).applicability, "reassessment_required") + fresh = EvidenceSnapshot.capture( + self.scenario, HarnessRunner().run(self.scenario, observed_at=110), changed, captured_at=110) + self.assertEqual(self.evaluate(snapshot=fresh, current=changed, + evaluated_at=110).applicability, "applicable") + self.assertEqual(asdict(self.snapshot), before) + + def test_unavailable_refused_and_blocked_results_never_upgrade(self): + for overrides in ({"inputs": {**self.scenario.inputs, "timeout_after_dispatch": True}}, + {"inputs": {**self.scenario.inputs, "revoked_before_use": True}}, + {"waiting_on_release": "missing API"}): + s = scenario(**overrides) + r = HarnessRunner().run(s, observed_at=100) + snap = EvidenceSnapshot.capture(s, r, self.context, captured_at=100) + result = self.evaluate(snapshot=snap, scenario=s) + self.assertEqual(result.historical_status, r.status) + self.assertIn("historical_result_not_passed", result.reasons) + self.assertEqual(result.applicability, "reassessment_required") + + def test_result_is_detached_from_mutable_runner_output(self): + retained = self.snapshot.result_json + self.result.boundary_outcomes.clear() + self.result.observations.clear() + self.result.status = "refused" + self.assertEqual(self.snapshot.result_json, retained) + self.assertEqual(self.evaluate().historical_status, "passed") + + def test_capture_rejects_mismatched_or_weakened_results(self): + self.result.transaction_id = "other" + with self.assertRaises(ValueError): + EvidenceSnapshot.capture(self.scenario, self.result, self.context, captured_at=100) + mutated = replace(self.scenario, mutation="disable_authorization_gate") + with self.assertRaisesRegex(ValueError, "weakened"): + EvidenceSnapshot.capture(mutated, HarnessRunner().run(mutated, observed_at=100), + self.context, captured_at=100) + + def test_capture_rejects_inconsistent_summary_and_non_synthetic_evidence(self): + self.result.status = "unknown" + with self.assertRaisesRegex(ValueError, "conflicts"): + EvidenceSnapshot.capture(self.scenario, self.result, self.context, captured_at=100) + self.result = HarnessRunner().run(self.scenario, observed_at=100) + self.result.observations[0] = replace( + self.result.observations[0], evidence_class=EvidenceClass.HOSTED_LIVE) + with self.assertRaisesRegex(ValueError, "synthetic"): + EvidenceSnapshot.capture(self.scenario, self.result, self.context, captured_at=100) + + def test_context_rejects_missing_duplicate_mutable_or_invalid_dependencies(self): + for deps in ((), [("lock", A)], (("lock", A), ("lock", B)), + (("", A),), (("lock", "unknown"),), (["lock", A],)): + with self.subTest(deps=deps), self.assertRaises(ValueError): + EvaluationContext(A, B, deps) + with self.assertRaises(ValueError): + EvaluationContext("bad", B, (("lock", A),)) + + def test_invalid_times_and_missing_current_context_rejected(self): + for value in (True, -1, 1.5, float("nan"), float("inf"), "109", None): + with self.subTest(value=value), self.assertRaises(ValueError): + self.evaluate(evaluated_at=value) + for value in (True, 0, -1, 1.5, None): + with self.subTest(budget=value), self.assertRaises(ValueError): + self.evaluate(max_age_seconds=value) + with self.assertRaises(ValueError): + self.evaluate(current=None) + + def test_tampered_snapshot_content_rejected(self): + data = json.loads(self.snapshot.result_json) + data["claim"] = "changed" + with self.assertRaisesRegex(ValueError, "identity mismatch"): + self.evaluate(snapshot=replace(self.snapshot, result_json=json.dumps(data))) + + def test_reappraisal_does_not_execute_adapter_or_use_wall_clock(self): + with patch.object(HarnessRunner, "run", side_effect=AssertionError("dispatch")), \ + patch("time.time", side_effect=AssertionError("wall clock")): + self.assertEqual(self.evaluate().applicability, "applicable") + + +if __name__ == "__main__": + unittest.main() diff --git a/integrations/agentrust-confidential-workflow-harness/tests/test_result_schema.py b/integrations/agentrust-confidential-workflow-harness/tests/test_result_schema.py new file mode 100644 index 00000000..6e290678 --- /dev/null +++ b/integrations/agentrust-confidential-workflow-harness/tests/test_result_schema.py @@ -0,0 +1,67 @@ +"""Check actual runner serialization against the published result contract.""" +from dataclasses import replace +import json +from pathlib import Path +import unittest + +import jsonschema + +from harness import HarnessRunner +from harness.reappraisal import EvidenceSnapshot, EvaluationContext +from test_harness import scenario + + +SCHEMA = json.loads((Path(__file__).resolve().parents[1] / "result.schema.json").read_text()) +VALIDATOR = jsonschema.Draft202012Validator(SCHEMA) + + +class ResultSchemaTests(unittest.TestCase): + def test_schema_is_well_formed(self): + jsonschema.Draft202012Validator.check_schema(SCHEMA) + + def test_emitted_results_validate_for_every_terminal_status(self): + cases = [ + (scenario(), "passed"), + (scenario(inputs={**scenario().inputs, "revoked_before_use": True}), "refused"), + (scenario(inputs={**scenario().inputs, "timeout_after_dispatch": True}), "unknown"), + (scenario(waiting_on_release="missing API"), "incomplete"), + ] + for selected, expected in cases: + with self.subTest(status=expected): + emitted = HarnessRunner().run(selected, observed_at=100).to_dict() + self.assertEqual(emitted["status"], expected) + self.assertNotIn("scenario_digest", emitted) + self.assertNotIn("observed_at", emitted) + self.assertTrue(VALIDATOR.is_valid(emitted)) + + def test_legacy_output_omits_metadata_but_is_not_reappraisable(self): + selected = scenario() + result = replace(HarnessRunner().run(selected, observed_at=100), + scenario_digest=None, observed_at=None) + legacy = result.to_dict() + self.assertNotIn("scenario_digest", legacy) + self.assertNotIn("observed_at", legacy) + self.assertTrue(VALIDATOR.is_valid(legacy)) + digest = "sha256:" + "a" * 64 + context = EvaluationContext(digest, digest, (("checker", digest),)) + with self.assertRaises(ValueError): + EvidenceSnapshot.capture(selected, result, context, captured_at=100) + + def test_internal_metadata_rejected_by_public_schema(self): + emitted = HarnessRunner().run(scenario(), observed_at=100).to_dict() + metadata = {"scenario_digest": scenario().digest(), "observed_at": 100} + for additions in ({"scenario_digest": metadata["scenario_digest"]}, + {"observed_at": 100}, metadata): + with self.subTest(fields=tuple(additions)): + with self.assertRaises(jsonschema.ValidationError): + VALIDATOR.validate({**emitted, **additions}) + + def test_zero_observation_time_and_unknown_field_boundary(self): + emitted = HarnessRunner().run(scenario(), observed_at=0).to_dict() + self.assertTrue(VALIDATOR.is_valid(emitted)) + with self.assertRaises(jsonschema.ValidationError): + VALIDATOR.validate({**emitted, "unregistered_metadata": True}) + + +if __name__ == "__main__": + unittest.main()