feat: optional prebind assurance and replayable receipts

plutus_agent/db.py

@@ -151,6 +151,8 @@ def micros_to_usd(micros) -> float:
     "action_receipt_hash",
     "resource_constraints_version",
     "resource_constraints_hash",
+    "prebind_json",
+    "prebind_hash",
 )
 
 
@@ -572,6 +574,8 @@ def verify_checkpoints(conn, checkpoints, hmac_key: Optional[bytes] = None) -> d
     action_receipt_hash TEXT,
     resource_constraints_version TEXT,
     resource_constraints_hash TEXT,
+    prebind_json TEXT,
+    prebind_hash TEXT,
     estimated         INTEGER NOT NULL DEFAULT 1,
     source            TEXT NOT NULL DEFAULT 'api',
     ts                REAL NOT NULL,
@@ -886,6 +890,8 @@ def _migrate_add_columns(conn) -> None:
         ("usage_events", "action_receipt_hash", "TEXT"),
         ("usage_events", "resource_constraints_version", "TEXT"),
         ("usage_events", "resource_constraints_hash", "TEXT"),
+        ("usage_events", "prebind_json", "TEXT"),
+        ("usage_events", "prebind_hash", "TEXT"),
     ]
     for table, col, defn in additions:
         cols = _table_columns(conn, table)

plutus_agent/metering.py

@@ -25,6 +25,7 @@
 from typing import Optional
 
 from . import db, pricing
+from .prebind import validate_prebind
 
 DAY = 86400
 _SHA256_HEX = re.compile(r"^[0-9a-fA-F]{64}$")
@@ -166,6 +167,7 @@ def record_usage(conn, org_id: str, provider: str,
                  block_over_limit: bool = False,
                  block_over_balance: bool = False,
                  chain_hmac_key: Optional[bytes] = None,
+                 prebind: Optional[dict] = None,
                  commit: bool = True) -> MeterResult:
     """Meter one LLM/agent call. Returns a :class:`MeterResult`.
 
@@ -206,6 +208,10 @@ def record_usage(conn, org_id: str, provider: str,
     follows the actual ``cost_usd``.
     """
     ts = ts if ts is not None else time.time()
+    if prebind is not None:
+        valid, errors = validate_prebind(prebind)
+        if not valid:
+            raise ValueError("invalid prebind: " + ", ".join(errors))
     evidence_hashes_json = _canonical_evidence_hashes(evidence_hashes)
     policy_version = _optional_text(policy_version, "policy_version")
     correction_ref = _optional_text(correction_ref, "correction_ref")
@@ -475,6 +481,8 @@ def record_usage(conn, org_id: str, provider: str,
         "action_receipt_hash": action_receipt_hash,
         "resource_constraints_version": resource_constraints_version,
         "resource_constraints_hash": resource_constraints_hash,
+        "prebind_json": json.dumps(prebind, sort_keys=True, separators=(",", ":")) if prebind else None,
+        "prebind_hash": prebind.get("prebind_hash") if prebind else None,
     }
     row_hash = db.compute_row_hash(prev_hash, row_fields, hmac_key=chain_hmac_key)
     conn.execute(
@@ -483,8 +491,8 @@ def record_usage(conn, org_id: str, provider: str,
         "baseline_micros,optimal_micros,external_ref,user_id,evidence_hashes,policy_version,"
         "result_hash,human_review,correction_ref,agent_id,authority_manifest_ref,scope_anchor,"
         "action_intent_hash,action_status,approval_ref,context_render_schema,context_render_hash,"
-        "served_memory_provenance_hash,action_receipt_hash,resource_constraints_version,resource_constraints_hash,estimated,source,ts,prev_hash,row_hash) "
-        "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        "served_memory_provenance_hash,action_receipt_hash,resource_constraints_version,resource_constraints_hash,prebind_json,prebind_hash,estimated,source,ts,prev_hash,row_hash) "
+        "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
         (eid, org_id, workspace_id, provider, model, task_type,
          int(input_tokens), int(output_tokens), int(cache_read_tokens),
          (int(cache_write_tokens) if cache_write_tokens is not None else None),
@@ -494,6 +502,8 @@ def record_usage(conn, org_id: str, provider: str,
          action_intent_hash, action_status, approval_ref, context_render_schema,
          context_render_hash, served_memory_provenance_hash, action_receipt_hash,
          resource_constraints_version, resource_constraints_hash,
+         (json.dumps(prebind, sort_keys=True, separators=(",", ":")) if prebind else None),
+         (prebind.get("prebind_hash") if prebind else None),
          int(estimated), source, ts, prev_hash, row_hash),
     )
 

plutus_agent/prebind.py

@@ -0,0 +1,147 @@
+"""Optional hash-bound pre-action assurance and pure replay comparison."""
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from typing import Any
+
+PREBIND_SCHEMA = "perseus-ledger-prebind/v1"
+OUTCOMES = {"allow", "hold", "deny", "abstain", "interrupt", "recover"}
+NON_EFFECTIVE_RESULTS = {"not_executed", "held", "denied", "abstained", "cancelled", "failed"}
+FORBIDDEN_KEYS = {"prompt", "context", "content", "body", "body_json", "tool_arguments", "arguments", "result", "response", "token", "secret", "password", "api_key"}
+_REQUIRED = (
+    "attempted_action", "actor_ref", "authority_ref", "trusted_scope", "policy_version",
+    "evidence_hashes", "selected_context_digest", "resource_ref", "boundary_outcome",
+    "non_effective_result", "replay_id",
+)
+
+
+def _sha(value: Any) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
+
+
+def _is_hash(value: Any) -> bool:
+    return isinstance(value, str) and len(value) == 64 and all(char in "0123456789abcdef" for char in value.lower())
+
+
+def _scan(value: Any, errors: list[str]) -> None:
+    if isinstance(value, Mapping):
+        for key, child in value.items():
+            lowered = str(key).lower()
+            if lowered in FORBIDDEN_KEYS or lowered.startswith("raw_"):
+                errors.append(f"forbidden_field:{key}")
+            _scan(child, errors)
+    elif isinstance(value, list):
+        for child in value:
+            _scan(child, errors)
+
+
+def prebind_digest(block: Mapping[str, Any]) -> str:
+    return _sha({key: value for key, value in block.items() if key != "prebind_hash"})
+
+
+def build_prebind(*, attempted_action: str, actor_ref: str, authority_ref: str, trusted_scope: str,
+                  policy_version: str, evidence_hashes: list[str], selected_context_digest: str,
+                  resource_ref: str, boundary_outcome: str, non_effective_result: str,
+                  replay_id: str, approval_ref: str | None = None,
+                  stage_refs: list[str] | None = None) -> dict[str, Any]:
+    block: dict[str, Any] = {
+        "schema_version": PREBIND_SCHEMA,
+        "attempted_action": attempted_action,
+        "actor_ref": actor_ref,
+        "authority_ref": authority_ref,
+        "trusted_scope": trusted_scope,
+        "policy_version": policy_version,
+        "evidence_hashes": sorted(set(evidence_hashes)),
+        "selected_context_digest": selected_context_digest,
+        "resource_ref": resource_ref,
+        "boundary_outcome": boundary_outcome,
+        "non_effective_result": non_effective_result,
+        "replay_id": replay_id,
+        "approval_ref": approval_ref,
+        "stage_refs": list(stage_refs or []),
+    }
+    block["prebind_hash"] = prebind_digest(block)
+    return block
+
+
+def validate_prebind(block: Mapping[str, Any]) -> tuple[bool, list[str]]:
+    errors: list[str] = []
+    if not isinstance(block, Mapping):
+        return False, ["prebind"]
+    _scan(block, errors)
+    if block.get("schema_version") != PREBIND_SCHEMA:
+        errors.append("schema_version")
+    for field in _REQUIRED:
+        if field not in block or not isinstance(block[field], str) or not block[field].strip():
+            if field != "evidence_hashes":
+                errors.append(field)
+    hashes = block.get("evidence_hashes")
+    if not isinstance(hashes, list) or not hashes or any(not _is_hash(value) for value in hashes):
+        errors.append("evidence_hashes")
+    if not _is_hash(block.get("selected_context_digest")):
+        errors.append("selected_context_digest")
+    if block.get("boundary_outcome") not in OUTCOMES:
+        errors.append("boundary_outcome")
+    if block.get("non_effective_result") not in NON_EFFECTIVE_RESULTS:
+        errors.append("non_effective_result")
+    if block.get("boundary_outcome") == "allow" and block.get("non_effective_result") == "executed":
+        errors.append("outcome_result_mismatch")
+    if block.get("boundary_outcome") != "allow" and block.get("non_effective_result") == "executed":
+        errors.append("outcome_result_mismatch")
+    stage_refs = block.get("stage_refs")
+    if not isinstance(stage_refs, list) or any(not isinstance(value, str) or not value for value in stage_refs):
+        errors.append("stage_refs")
+    supplied = block.get("prebind_hash")
+    if not _is_hash(supplied) or supplied != prebind_digest(block):
+        errors.append("prebind_hash")
+    allowed = set(_REQUIRED) | {"schema_version", "approval_ref", "stage_refs", "prebind_hash"}
+    for key in set(block) - allowed:
+        errors.append(f"unknown_field:{key}")
+    return not errors, sorted(set(errors))
+
+
+def replay_prebind(prior: Mapping[str, Any], *, current_authority_ref: str | None = None,
+                   current_trusted_scope: str | None = None, current_evidence_hashes: list[str] | None = None,
+                   current_policy_version: str | None = None, current_state: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    valid, errors = validate_prebind(prior)
+    if not valid:
+        raise ValueError("invalid prebind: " + ", ".join(errors))
+    state = dict(current_state or {})
+    changed: list[str] = []
+    if current_authority_ref is not None and current_authority_ref != prior["authority_ref"]:
+        changed.append("authority_ref")
+    if current_trusted_scope is not None and current_trusted_scope != prior["trusted_scope"]:
+        changed.append("trusted_scope")
+    normalized = None if current_evidence_hashes is None else sorted(set(current_evidence_hashes))
+    if normalized is not None and normalized != prior["evidence_hashes"]:
+        changed.append("evidence_hashes")
+    if current_policy_version is not None and current_policy_version != prior["policy_version"]:
+        changed.append("policy_version")
+    authority_ok = bool(state.get("authority_ok", not any(field in changed for field in ("authority_ref", "trusted_scope"))))
+    evidence_current = bool(state.get("evidence_current", not bool(state.get("evidence_stale"))))
+    approved = bool(state.get("approval_granted", prior.get("approval_ref") is not None))
+    action_allowed = bool(state.get("action_allowed", authority_ok and evidence_current))
+    admitted = authority_ok and evidence_current and approved and action_allowed
+    if admitted:
+        admission = "admitted_after_correction" if prior["boundary_outcome"] in {"hold", "deny", "abstain", "recover", "interrupt"} else "admitted"
+        outcome = "allow"
+    else:
+        admission = "not_admitted"
+        outcome = "hold" if not authority_ok or not evidence_current else "deny"
+    return {
+        "schema_version": "perseus-ledger-replay/v1",
+        "replay_id": prior["replay_id"],
+        "prior_prebind_hash": prior["prebind_hash"],
+        "changed_fields": changed,
+        "admission": admission,
+        "replayed_boundary_outcome": outcome,
+        "non_mutating": True,
+        "reason_codes": (["authority_changed"] if "authority_ref" in changed or "trusted_scope" in changed else [])
+        + (["evidence_changed"] if "evidence_hashes" in changed else [])
+        + (["policy_changed"] if "policy_version" in changed else []),
+    }
+
+
+__all__ = ["PREBIND_SCHEMA", "build_prebind", "prebind_digest", "replay_prebind", "validate_prebind"]

plutus_agent/server/api.py

@@ -6,6 +6,7 @@
 import json
 
 from .. import db, metering, pricing, savings
+from ..prebind import validate_prebind
 
 
 def default_org_id(conn) -> str | None:
@@ -64,6 +65,23 @@ def events_json(conn, org_id: str, limit: int = 50, before=None) -> dict:
     return _page(metering.recent_events(conn, org_id, limit=limit, before=before), limit)
 
 
+def replay_receipt_prebind(conn, org_id: str, external_ref: str, **kwargs) -> dict:
+    """Re-evaluate a stored prebind without mutating the usage history."""
+    from ..prebind import replay_prebind
+
+    rows = db.events_by_ref(conn, org_id, external_ref)
+    if not rows:
+        raise ValueError("receipt not found")
+    payload = rows[0]["prebind_json"]
+    if payload is None:
+        raise ValueError("receipt has no prebind block")
+    prior = json.loads(payload)
+    valid, errors = validate_prebind(prior)
+    if not valid:
+        raise ValueError("stored prebind is invalid: " + ", ".join(errors))
+    return replay_prebind(prior, **kwargs)
+
+
 _EXPORT_COLUMNS = ["id", "ts", "provider", "model", "task_type", "workspace",
                    "input_tokens", "output_tokens", "cache_read_tokens",
                    "cache_write_tokens", "reasoning_tokens", "user_id",
@@ -119,6 +137,8 @@ def audit_json(conn, org_id: str, *, hmac_key: bytes | None = None,
                     "served_memory_provenance_hash": row["served_memory_provenance_hash"],
                     "action_receipt_hash": row["action_receipt_hash"],
                 },
+                "prebind": json.loads(row["prebind_json"])
+                if row["prebind_json"] is not None else None,
                 "action_authorization": {
                     "agent_id": row["agent_id"],
                     "authority_manifest_ref": row["authority_manifest_ref"],

plutus_agent/server/app.py

@@ -22,6 +22,7 @@
 from .. import __version__, bridge, config as cfgmod, db, pricing
 from ..billing import StripeClient, BillingError, handle_webhook_event
 from ..utils import strict_int
+from ..prebind import validate_prebind
 from . import api, views, auth as authmod
 
 # Paths reachable without a session when auth is enabled.
@@ -830,6 +831,12 @@ def _ingest_usage(self, conn):
                           "resource_constraints_version", "resource_constraints_hash"):
                 if ev.get(field) is not None and not isinstance(ev[field], str):
                     return self._json(400, {"error": f"{field} must be a string"})
+            if ev.get("prebind") is not None:
+                if not isinstance(ev["prebind"], dict):
+                    return self._json(400, {"error": "prebind must be an object"})
+                valid, errors = validate_prebind(ev["prebind"])
+                if not valid:
+                    return self._json(400, {"error": "invalid prebind", "fields": errors})
 
         # All valid — record the whole batch as one serialized transaction.
         # Fix #27/#30: db.immediate() takes the write lock up front (BEGIN
@@ -893,6 +900,7 @@ def _ingest_usage(self, conn):
                             action_receipt_hash=ev.get("action_receipt_hash"),
                             resource_constraints_version=ev.get("resource_constraints_version"),
                             resource_constraints_hash=ev.get("resource_constraints_hash"),
+                            prebind=ev.get("prebind"),
                             user_id=ev.get("user_id"),
                             source=ev.get("source", "api"),
                             pricing_overrides=cfg.get("pricing", {}).get("overrides"),