feat: reappraise retained synthetic evidence for #199

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
 ```
 

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,
+        })

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))

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,