From 3e23c8d9c97e36f208aefe6030f681e056259ef3 Mon Sep 17 00:00:00 2001 From: Thomas Connally Date: Mon, 17 Aug 2026 18:52:12 +0000 Subject: [PATCH] feat: add acceptance campaign integrity and spend guards --- docs/acceptance-campaigns.md | 142 ++++++ docs/evidence-receipts.md | 2 +- docs/schema.md | 3 +- ledger_agent/campaigns.py | 680 +++++++++++++++++++++++++++ ledger_agent/client.py | 103 ++-- ledger_agent/db.py | 284 ++++++++++- ledger_agent/mcp_server.py | 6 + ledger_agent/metering.py | 52 +- ledger_agent/server/api.py | 53 ++- ledger_agent/server/app.py | 96 +++- openapi.yaml | 196 ++++++++ server.json | 4 + tests/test_cache_write.py | 2 +- tests/test_campaign_persistence.py | 153 ++++++ tests/test_campaign_usage_binding.py | 135 ++++++ tests/test_campaigns.py | 200 ++++++++ tests/test_external_ref.py | 4 +- tests/test_mcp.py | 25 + tests/test_schema_version.py | 6 +- tests/test_server.py | 89 +++- 20 files changed, 2174 insertions(+), 61 deletions(-) create mode 100644 docs/acceptance-campaigns.md create mode 100644 ledger_agent/campaigns.py create mode 100644 tests/test_campaign_persistence.py create mode 100644 tests/test_campaign_usage_binding.py create mode 100644 tests/test_campaigns.py diff --git a/docs/acceptance-campaigns.md b/docs/acceptance-campaigns.md new file mode 100644 index 0000000..066041f --- /dev/null +++ b/docs/acceptance-campaigns.md @@ -0,0 +1,142 @@ +# Acceptance Campaigns and Benchmark Guard Contract + +Status: implemented contract +Date: 2026-08-17 +Resolves: ledger#256 · ledger#257 +Related: `docs/evidence-receipts.md` (#235), `docs/authorized-action-receipts.md` (#197), `docs/schema.md` + +## Overview + +Ledger usage receipts describe individual actions. An acceptance campaign describes +whether a multi-check runner completed with intact evidence and what the tested +target did. These are independent facts: a healthy framework can record a real +target failure, while a broken or evidence-losing framework cannot produce a +verified pass. + +Benchmark campaigns also carry an economic and continuation boundary. Planned +cells, provider/model lanes, configuration and fixture commitments, spend +limits, checkpoint lineage, and correction attempts are recorded as hash-only +projections. Individual usage receipts remain the accounting and tamper-evident +source of spend. + +Ledger records supplied evidence. It does not certify a target, authorize paid +work, or infer facts that a runner did not send. + +## Compact public-safe example + +The following projection contains identifiers, statuses, counts, costs, and +commitments only: + +```json +{ + "campaign_id": "campaign:recon-1061", + "framework_status": "completed", + "target_status": "fail", + "budget_status": "within_guard", + "evidence_status": "complete", + "counts": {"planned": 4, "executed": 4, "passed": 3, "failed": 1, "skipped": 0}, + "spent_micros": 184200, + "manifest_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "receipt_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +} +``` + +## Versioned public objects + +The schemas are implemented in `ledger_agent/campaigns.py`: + +| Object | Schema | Purpose | +|---|---|---| +| Manifest | `perseus-ledger-acceptance-campaign/v1` | Immutable plan and commitments | +| Check | `perseus-ledger-acceptance-check/v1` | One cell result/skip/error | +| Binding | `perseus-ledger-campaign-binding/v1` | Usage-to-cell attribution | +| Receipt | `perseus-ledger-acceptance-receipt/v1` | Final campaign envelope | + +Every object has a canonical SHA-256 digest. Unknown fields and forbidden raw +material are rejected. Forbidden material includes prompts, memory bodies, +provider payloads, credentials, authorization values, and tool arguments. + +## Independent status axes + +`framework_status` is one of `completed`, `error`, `interrupted`, or `cancelled`. +`target_status` is one of `pass`, `fail`, `inconclusive`, or `not_run`. + +The required interpretation is: + +- `completed` + `fail` is a valid completed failure campaign. +- A framework error before any check produces `target_status=not_run`. +- A completed campaign with no executed checks is `inconclusive`, never pass. +- A check-level provider/schema error is retained as an explicit failed evidence + path; it cannot become a verified pass. + +`budget_status` is `not_configured`, `within_guard`, `stopped`, or `overrun`. +`evidence_status` is `pending`, `complete`, `incomplete`, or `invalid`. +`finalization_status` is `pending`, `complete`, or `failed`. + +`verification.verified_pass` is true only when the receipt digest, manifest, +checks, framework, target, budget, evidence, and finalization all verify. A +valid receipt with `target_status=fail` remains useful evidence but has +`verified_pass=false`. + +## Manifest and check commitments + +A manifest binds: + +- unique planned cell IDs and provider/model lane labels; +- configuration and fixture SHA-256 commitments; +- optional target commit/build/runtime identity; +- expected spend range, integer-micro-dollar hard stop, and runaway guard; +- retry policy and whether continuation is allowed; +- optional action-intent commitment and whether evidence is required. + +A check binds its cell, lane, configuration, status, result/evidence hashes, +usage-event IDs, checkpoint reference, and attempt lineage. A second attempt +requires `continuation=true`, the immediately prior attempt, a new configuration +commitment, and a new action-intent commitment. Completed cells cannot be silently +replayed. + +## Spend and durable state + +Campaign manifests, checks, and final receipts are stored in the additive +`acceptance_campaigns` and `acceptance_checks` SQLite tables. A campaign-bound +usage event stores nullable `campaign_id` and a canonical +`campaign_binding_json`/`campaign_binding_hash`; those fields extend the existing +per-organization usage hash chain only when present, preserving historical +canonical bytes. + +Budget admission is integer-micro-dollar arithmetic. The spend read and event +insert run under `db.immediate()`. A proposed event crossing the runaway guard or +hard stop is rejected before insertion. The campaign is durably marked +`budget_status=stopped` with a bounded reason code and remaining guard. No +partial overrun usage event is accepted. + +The normal lifecycle is: + +1. `POST /v1/campaigns` stores or idempotently replays the manifest. +2. `POST /v1/usage` records bound usage events. +3. `POST /v1/campaigns/checks` stores immutable cell results. +4. `POST /v1/campaigns/finalize` stores the final receipt. +5. `GET /v1/campaigns?campaign_id=...` returns the public projection. + +The local SDK and MCP `ledger_record` accept the same optional binding. Existing +usage calls without a binding retain their prior behavior. + +## Verification and non-goals + +`ledger_agent.server.api.campaign_json` rehydrates the manifest, checks, receipt, +spend count, and independent verification after restart. `/api/audit` accepts a +campaign selector, while an event-scoped audit includes the hash-only campaign +binding when present. + +This contract does not: + +- certify the target system or its external environment; +- authorize provider calls, paid benchmarks, or AAR actions; +- store prompts, memory bodies, provider responses, or secrets; +- replace individual action receipts or the existing usage hash chain; +- make a failed target result positive because the framework completed. + +The complete regression battery covers clean completion, target failure, +all-skipped/inconclusive, not-run, interruption, cancellation, failed +finalization, budget stops, malformed bindings, duplicate cells, correction +lineage, restart/readback, HTTP, SDK, MCP, and legacy receipt compatibility. diff --git a/docs/evidence-receipts.md b/docs/evidence-receipts.md index 149fe91..1718810 100644 --- a/docs/evidence-receipts.md +++ b/docs/evidence-receipts.md @@ -5,7 +5,7 @@ hash-only fields, see [Memory governance and Ledger provenance](memory-governanc For a local Perseus + Vault + Ledger walkthrough, see [the local integration guide](local-perseus-vault-ledger.md). -An evidence receipt is a task-scoped, machine-readable view of hash-chained Ledger events. It answers a bounded question: +An evidence receipt is a task-scoped, machine-readable view of hash-chained Ledger events. For multi-check acceptance and benchmark runs, compose it with the hash-only [Acceptance Campaign contract](acceptance-campaigns.md): campaign framework integrity, target outcome, budget state, and evidence state remain separate. It answers a bounded question: > For this externally identified task or artifact, which recorded autonomous-system actions exist, what resource allocation accompanied them, and does their containing organization ledger verify? diff --git a/docs/schema.md b/docs/schema.md index b178dd5..e8bff97 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -9,7 +9,7 @@ schema — the database half of the frozen contract whose API half is `ledger_agent.db.SCHEMA_VERSION` is an integer bumped on every schema change and stamped into the `meta` table key `schema_version` on `init_schema()`. The -runtime currently declares **`SCHEMA_VERSION=22`**. Read the stored value with +runtime currently declares **`SCHEMA_VERSION=23`**. Read the stored value with `db.get_schema_version(conn)`; a fresh database is stamped with the runtime value after all additive migrations have run. @@ -34,6 +34,7 @@ value after all additive migrations have run. | 20 | Adds nullable hash-covered governance self-cost (#239): `governance_cost_json`/`governance_cost_hash` — internal telemetry (wall/cpu/mem/storage/tokens/model_calls/approval waits), excluded from customer-facing usage and billing totals. | | 21 | Adds nullable hash-covered behavior-snapshot receipt pin (#238): `behavior_snapshot_json`/`behavior_snapshot_hash` — the sha256 of a canonical agent-run snapshot, re-verifiable with `ledger diff --require-target-digest`. | | 22 | Adds nullable hash-covered custody disclosure for the referenced authority manifest (#241): `authority_manifest_custody` — 1f916 taxonomy label; missing/unknown custody renders as labeled uncertainty in verification output. | +| 23 | Adds `acceptance_campaigns`/`acceptance_checks` plus nullable hash-covered `usage_events` campaign bindings (`campaign_id`, `campaign_binding_json`, `campaign_binding_hash`) for acceptance integrity, budget guards, and correction/resume lineage (#256/#257). | ## The contract (within the 1.0 major line) diff --git a/ledger_agent/campaigns.py b/ledger_agent/campaigns.py new file mode 100644 index 0000000..081cd88 --- /dev/null +++ b/ledger_agent/campaigns.py @@ -0,0 +1,680 @@ +"""Hash-only acceptance campaign contracts and fail-closed budget helpers. + +A campaign is an envelope around individual Ledger usage/action receipts. The +module is deliberately persistence-agnostic: it validates the public contract, +computes deterministic commitments, and keeps framework integrity separate from +the target's acceptance result. ``ledger_agent.db`` owns durable storage. +""" +from __future__ import annotations + +import hashlib +import json +from typing import Any, Optional + +CAMPAIGN_MANIFEST_SCHEMA = "perseus-ledger-acceptance-campaign/v1" +CAMPAIGN_CHECK_SCHEMA = "perseus-ledger-acceptance-check/v1" +CAMPAIGN_RECEIPT_SCHEMA = "perseus-ledger-acceptance-receipt/v1" +CAMPAIGN_BINDING_SCHEMA = "perseus-ledger-campaign-binding/v1" + +FRAMEWORK_STATUS_VALUES = {"completed", "error", "interrupted", "cancelled"} +TARGET_STATUS_VALUES = {"pass", "fail", "inconclusive", "not_run"} +CHECK_STATUS_VALUES = {"pass", "fail", "skip", "error"} +BUDGET_STATUS_VALUES = {"not_configured", "within_guard", "stopped", "overrun"} +EVIDENCE_STATUS_VALUES = {"pending", "complete", "incomplete", "invalid"} +FINALIZATION_STATUS_VALUES = {"pending", "complete", "failed"} +RETRY_POLICY_VALUES = {"none", "same_config", "new_version_only", "explicit_continuation"} + + +class CampaignBudgetError(ValueError): + """A proposed campaign usage event crossed a durable spend guard.""" + + def __init__(self, campaign_id: str, reason: str, remaining_micros: int): + self.campaign_id = campaign_id + self.reason = reason + self.remaining_micros = remaining_micros + super().__init__( + f"campaign budget guard: {reason} (remaining_micros={remaining_micros})" + ) + + +_SHA256_HEX = set("0123456789abcdef") +_FORBIDDEN_KEYS = { + "prompt", "memory_body", "memory_bodies", "provider_payload", + "provider_response", "raw_payload", "tool_arguments", "tool_output", + "api_key", "authorization", "password", "credential", "secret", + "private_key", "access_token", "refresh_token", "bearer_token", +} + +_MANIFEST_FIELDS = { + "schema", "campaign_id", "planned_cells", "provider_lanes", "config_hash", + "fixture_hash", "target_ref", "target_commit_hash", "target_build_hash", + "runtime_identity", "expected_spend_min_micros", "expected_spend_max_micros", + "hard_stop_micros", "runaway_guard_micros", "retry_policy", + "continuation_allowed", "action_intent_hash", "evidence_required", + "manifest_hash", +} +_CHECK_FIELDS = { + "schema", "campaign_id", "cell_id", "lane", "status", "config_hash", + "attempt", "continuation", "parent_attempt", "action_intent_hash", + "result_hash", "evidence_hashes", "usage_event_ids", "checkpoint_ref", + "reason_code", "check_hash", +} +_RECEIPT_FIELDS = { + "schema", "campaign_id", "manifest_hash", "framework_status", "target_status", + "budget_status", "evidence_status", "finalization_status", "finalization_reason", + "counts", "check_hashes", "target_identity", "evidence_bundle_hash", + "protected_state_before_hash", "protected_state_after_hash", "cleanup_manifest_hash", + "spent_micros", "remaining_micros", "stop_reason", "last_checkpoint_ref", + "verification", "receipt_hash", +} +_BINDING_FIELDS = { + "schema", "campaign_id", "cell_id", "lane", "config_hash", "attempt", + "continuation", "parent_attempt", "action_intent_hash", "checkpoint_ref", + "binding_hash", +} + + +def _sha(value: Any) -> str: + return hashlib.sha256( + json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + ).hexdigest() + + +def _is_sha256(value: Any) -> bool: + return ( + isinstance(value, str) and len(value) == 64 + and set(value.lower()) <= _SHA256_HEX + ) + + +def _text(value: Any, field: str, *, required: bool = False, max_len: int = 256) -> list[str]: + if value is None and not required: + return [] + if not isinstance(value, str) or not value.strip(): + return [field] + if len(value) > max_len: + return [f"{field}_too_long"] + return [] + + +def _nonnegative(value: Any, field: str, *, required: bool = False) -> list[str]: + if value is None and not required: + return [] + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return [field] + return [] + + +def _list_of_text(value: Any, field: str, *, required: bool = False) -> list[str]: + if value is None and not required: + return [] + if not isinstance(value, list) or not value: + return [field] + errors: list[str] = [] + normalized: set[str] = set() + for item in value: + errors.extend(_text(item, field)) + if isinstance(item, str): + normalized.add(item) + if len(normalized) != len(value): + errors.append(f"{field}_duplicate") + return errors + + +def _forbidden(value: Any, path: str = "") -> list[str]: + errors: list[str] = [] + if isinstance(value, dict): + for key, child in value.items(): + key_text = str(key).lower() + child_path = f"{path}.{key}" if path else str(key) + if key_text in _FORBIDDEN_KEYS: + errors.append(f"forbidden_field:{child_path}") + errors.extend(_forbidden(child, child_path)) + elif isinstance(value, list): + for index, child in enumerate(value): + errors.extend(_forbidden(child, f"{path}[{index}]")) + return errors + + +def _unknown(value: dict[str, Any], allowed: set[str]) -> list[str]: + return [f"unknown:{key}" for key in sorted(set(value) - allowed)] + + +def _digest_without(value: dict[str, Any], field: str) -> str: + return _sha({key: item for key, item in value.items() if key != field}) + + +def manifest_digest(manifest: dict[str, Any]) -> str: + return _digest_without(manifest, "manifest_hash") + + +def check_digest(check: dict[str, Any]) -> str: + return _digest_without(check, "check_hash") + + +def receipt_digest(receipt: dict[str, Any]) -> str: + return _sha({ + key: value for key, value in receipt.items() + if key not in {"receipt_hash", "verification"} + }) + + +def binding_digest(binding: dict[str, Any]) -> str: + return _digest_without(binding, "binding_hash") + + +def build_binding(*, campaign_id: str, cell_id: str, lane: str, + config_hash: str, attempt: int = 1, + continuation: bool = False, + parent_attempt: Optional[int] = None, + action_intent_hash: Optional[str] = None, + checkpoint_ref: Optional[str] = None, + **extra: Any) -> dict[str, Any]: + """Build the hash-only attribution carried by a usage event.""" + binding: dict[str, Any] = { + "schema": CAMPAIGN_BINDING_SCHEMA, + "campaign_id": campaign_id, + "cell_id": cell_id, + "lane": lane, + "config_hash": config_hash.lower() if isinstance(config_hash, str) else config_hash, + "attempt": attempt, + "continuation": continuation, + "parent_attempt": parent_attempt, + "action_intent_hash": action_intent_hash.lower() if isinstance(action_intent_hash, str) else action_intent_hash, + "checkpoint_ref": checkpoint_ref, + } + binding.update(extra) + binding["binding_hash"] = binding_digest(binding) + valid, errors = validate_binding(binding) + if not valid: + raise ValueError("invalid campaign binding: " + ", ".join(errors)) + return binding + + +def validate_binding(binding: dict[str, Any]) -> tuple[bool, list[str]]: + if not isinstance(binding, dict): + return False, ["binding"] + errors = _unknown(binding, _BINDING_FIELDS) + _forbidden(binding) + if binding.get("schema") != CAMPAIGN_BINDING_SCHEMA: + errors.append("schema") + for field in ("campaign_id", "cell_id", "lane"): + errors.extend(_text(binding.get(field), field, required=True)) + if not _is_sha256(binding.get("config_hash")): + errors.append("config_hash") + errors.extend(_nonnegative(binding.get("attempt"), "attempt", required=True)) + if binding.get("attempt") == 0: + errors.append("attempt") + if not isinstance(binding.get("continuation"), bool): + errors.append("continuation") + if binding.get("parent_attempt") is not None: + errors.extend(_nonnegative(binding.get("parent_attempt"), "parent_attempt")) + for field in ("action_intent_hash",): + value = binding.get(field) + if value is not None and not _is_sha256(value): + errors.append(field) + errors.extend(_text(binding.get("checkpoint_ref"), "checkpoint_ref")) + if binding.get("attempt") == 1 and (binding.get("continuation") or binding.get("parent_attempt") is not None): + errors.append("initial_attempt_lineage") + if binding.get("attempt", 0) > 1: + if not binding.get("continuation") or binding.get("parent_attempt") != binding["attempt"] - 1: + errors.append("continuation_lineage") + if not binding.get("action_intent_hash"): + errors.append("continuation_action_intent_required") + digest = binding.get("binding_hash") + if not _is_sha256(digest): + errors.append("binding_hash") + elif digest != binding_digest(binding): + errors.append("binding_hash_mismatch") + return not errors, sorted(set(errors)) + + +def build_manifest(*, campaign_id: str, planned_cells: list[str], + provider_lanes: list[str], config_hash: str, fixture_hash: str, + target_ref: Optional[str] = None, + target_commit_hash: Optional[str] = None, + target_build_hash: Optional[str] = None, + runtime_identity: Optional[str] = None, + expected_spend_min_micros: Optional[int] = None, + expected_spend_max_micros: Optional[int] = None, + hard_stop_micros: Optional[int] = None, + runaway_guard_micros: Optional[int] = None, + retry_policy: str = "none", continuation_allowed: bool = False, + action_intent_hash: Optional[str] = None, + evidence_required: bool = True) -> dict[str, Any]: + manifest: dict[str, Any] = { + "schema": CAMPAIGN_MANIFEST_SCHEMA, + "campaign_id": campaign_id, + "planned_cells": list(planned_cells), + "provider_lanes": list(provider_lanes), + "config_hash": config_hash.lower() if isinstance(config_hash, str) else config_hash, + "fixture_hash": fixture_hash.lower() if isinstance(fixture_hash, str) else fixture_hash, + "target_ref": target_ref, + "target_commit_hash": target_commit_hash.lower() if isinstance(target_commit_hash, str) else target_commit_hash, + "target_build_hash": target_build_hash.lower() if isinstance(target_build_hash, str) else target_build_hash, + "runtime_identity": runtime_identity, + "expected_spend_min_micros": expected_spend_min_micros, + "expected_spend_max_micros": expected_spend_max_micros, + "hard_stop_micros": hard_stop_micros, + "runaway_guard_micros": runaway_guard_micros, + "retry_policy": retry_policy, + "continuation_allowed": continuation_allowed, + "action_intent_hash": action_intent_hash.lower() if isinstance(action_intent_hash, str) else action_intent_hash, + "evidence_required": evidence_required, + } + manifest["manifest_hash"] = manifest_digest(manifest) + valid, errors = validate_manifest(manifest) + if not valid: + raise ValueError("invalid campaign manifest: " + ", ".join(errors)) + return manifest + + +def validate_manifest(manifest: dict[str, Any]) -> tuple[bool, list[str]]: + if not isinstance(manifest, dict): + return False, ["manifest"] + errors = _unknown(manifest, _MANIFEST_FIELDS) + _forbidden(manifest) + if manifest.get("schema") != CAMPAIGN_MANIFEST_SCHEMA: + errors.append("schema") + errors.extend(_text(manifest.get("campaign_id"), "campaign_id", required=True)) + errors.extend(_list_of_text(manifest.get("planned_cells"), "planned_cells", required=True)) + errors.extend(_list_of_text(manifest.get("provider_lanes"), "provider_lanes", required=True)) + for field in ("config_hash", "fixture_hash", "target_commit_hash", "target_build_hash", "action_intent_hash"): + value = manifest.get(field) + if value is not None and not _is_sha256(value): + errors.append(field) + for field in ("target_ref", "runtime_identity"): + errors.extend(_text(manifest.get(field), field)) + for field in ("expected_spend_min_micros", "expected_spend_max_micros", "hard_stop_micros", "runaway_guard_micros"): + errors.extend(_nonnegative(manifest.get(field), field)) + low = manifest.get("expected_spend_min_micros") + high = manifest.get("expected_spend_max_micros") + hard = manifest.get("hard_stop_micros") + runaway = manifest.get("runaway_guard_micros") + if low is not None and high is not None and low > high: + errors.append("expected_spend_range") + if high is not None and hard is not None and hard < high: + errors.append("hard_stop_below_expected_max") + if runaway is not None and hard is not None and runaway > hard: + errors.append("runaway_guard_above_hard_stop") + if manifest.get("retry_policy") not in RETRY_POLICY_VALUES: + errors.append("retry_policy") + if not isinstance(manifest.get("continuation_allowed"), bool): + errors.append("continuation_allowed") + if not isinstance(manifest.get("evidence_required"), bool): + errors.append("evidence_required") + digest = manifest.get("manifest_hash") + if not _is_sha256(digest): + errors.append("manifest_hash") + elif digest != manifest_digest(manifest): + errors.append("manifest_hash_mismatch") + return not errors, sorted(set(errors)) + + +def build_check(*, campaign_id: str, cell_id: str, lane: str, status: str, + config_hash: str, attempt: int = 1, continuation: bool = False, + parent_attempt: Optional[int] = None, + action_intent_hash: Optional[str] = None, + result_hash: Optional[str] = None, + evidence_hashes: Optional[list[str]] = None, + usage_event_ids: Optional[list[str]] = None, + checkpoint_ref: Optional[str] = None, + reason_code: Optional[str] = None) -> dict[str, Any]: + check: dict[str, Any] = { + "schema": CAMPAIGN_CHECK_SCHEMA, + "campaign_id": campaign_id, + "cell_id": cell_id, + "lane": lane, + "status": status, + "config_hash": config_hash.lower() if isinstance(config_hash, str) else config_hash, + "attempt": attempt, + "continuation": continuation, + "parent_attempt": parent_attempt, + "action_intent_hash": action_intent_hash.lower() if isinstance(action_intent_hash, str) else action_intent_hash, + "result_hash": result_hash.lower() if isinstance(result_hash, str) else result_hash, + "evidence_hashes": sorted(set(evidence_hashes or [])), + "usage_event_ids": list(usage_event_ids or []), + "checkpoint_ref": checkpoint_ref, + "reason_code": reason_code, + } + check["check_hash"] = check_digest(check) + valid, errors = validate_check(check) + if not valid: + raise ValueError("invalid campaign check: " + ", ".join(errors)) + return check + + +def validate_check(check: dict[str, Any]) -> tuple[bool, list[str]]: + if not isinstance(check, dict): + return False, ["check"] + errors = _unknown(check, _CHECK_FIELDS) + _forbidden(check) + if check.get("schema") != CAMPAIGN_CHECK_SCHEMA: + errors.append("schema") + for field in ("campaign_id", "cell_id", "lane"): + errors.extend(_text(check.get(field), field, required=True)) + if check.get("status") not in CHECK_STATUS_VALUES: + errors.append("status") + if not _is_sha256(check.get("config_hash")): + errors.append("config_hash") + errors.extend(_nonnegative(check.get("attempt"), "attempt", required=True)) + if check.get("attempt") == 0: + errors.append("attempt") + if not isinstance(check.get("continuation"), bool): + errors.append("continuation") + parent = check.get("parent_attempt") + if parent is not None: + errors.extend(_nonnegative(parent, "parent_attempt")) + for field in ("action_intent_hash", "result_hash"): + value = check.get(field) + if value is not None and not _is_sha256(value): + errors.append(field) + evidence = check.get("evidence_hashes") + if not isinstance(evidence, list) or any(not _is_sha256(item) for item in evidence): + errors.append("evidence_hashes") + usage = check.get("usage_event_ids") + if not isinstance(usage, list) or any(_text(item, "usage_event_id") for item in usage): + errors.append("usage_event_ids") + for field in ("checkpoint_ref", "reason_code"): + errors.extend(_text(check.get(field), field)) + if check.get("status") in {"pass", "fail"} and not _is_sha256(check.get("result_hash")): + errors.append("result_hash_required") + if check.get("status") == "error" and not check.get("reason_code"): + errors.append("reason_code_required") + if check.get("attempt") == 1 and (check.get("continuation") or check.get("parent_attempt") is not None): + errors.append("initial_attempt_lineage") + if check.get("attempt", 0) > 1: + if not check.get("continuation") or check.get("parent_attempt") != check["attempt"] - 1: + errors.append("continuation_lineage") + if not check.get("action_intent_hash"): + errors.append("continuation_action_intent_required") + digest = check.get("check_hash") + if not _is_sha256(digest): + errors.append("check_hash") + elif digest != check_digest(check): + errors.append("check_hash_mismatch") + return not errors, sorted(set(errors)) + + +def validate_attempt_lineage(checks: list[dict[str, Any]]) -> tuple[bool, list[str]]: + errors: list[str] = [] + by_cell: dict[str, list[dict[str, Any]]] = {} + for item in checks: + valid, item_errors = validate_check(item) + if not valid: + errors.extend(item_errors) + by_cell.setdefault(item.get("cell_id", ""), []).append(item) + for items in by_cell.values(): + items.sort(key=lambda item: item.get("attempt", 0)) + for index, item in enumerate(items): + if index == 0: + if item.get("attempt") != 1: + errors.append("attempt_sequence") + continue + prior = items[index - 1] + if item.get("attempt") != prior.get("attempt", 0) + 1: + errors.append("attempt_sequence") + if prior.get("status") == "pass" and not item.get("continuation"): + errors.append("completed_cell_duplicate") + if item.get("continuation"): + if item.get("config_hash") == prior.get("config_hash"): + errors.append("continuation_config_unchanged") + if item.get("action_intent_hash") == prior.get("action_intent_hash"): + errors.append("continuation_action_unchanged") + return not errors, sorted(set(errors)) + + +def _target_status(framework_status: str, checks: list[dict[str, Any]]) -> str: + executed = [item for item in checks if item.get("status") != "skip"] + if framework_status != "completed": + return "not_run" if not executed else "inconclusive" + if not executed: + return "inconclusive" + if any(item.get("status") == "fail" for item in executed): + return "fail" + if any(item.get("status") == "error" for item in executed): + return "inconclusive" + return "pass" + + +def build_receipt(*, manifest: dict[str, Any], checks: list[dict[str, Any]], + framework_status: str, target_status: Optional[str] = None, + budget_status: Optional[str] = None, + evidence_status: str = "pending", + finalization_status: str = "complete", + finalization_reason: Optional[str] = None, + target_identity: Optional[dict[str, Any]] = None, + evidence_bundle_hash: Optional[str] = None, + protected_state_before_hash: Optional[str] = None, + protected_state_after_hash: Optional[str] = None, + cleanup_manifest_hash: Optional[str] = None, + spent_micros: int = 0, remaining_micros: Optional[int] = None, + stop_reason: Optional[str] = None, + last_checkpoint_ref: Optional[str] = None) -> dict[str, Any]: + valid, errors = validate_manifest(manifest) + if not valid: + raise ValueError("invalid campaign manifest: " + ", ".join(errors)) + if framework_status not in FRAMEWORK_STATUS_VALUES: + raise ValueError("invalid framework_status") + if _nonnegative(spent_micros, "spent_micros", required=True): + raise ValueError("spent_micros must be a non-negative integer") + if remaining_micros is not None and _nonnegative(remaining_micros, "remaining_micros"): + raise ValueError("remaining_micros must be a non-negative integer") + if evidence_status not in EVIDENCE_STATUS_VALUES: + raise ValueError("invalid evidence_status") + if finalization_status not in FINALIZATION_STATUS_VALUES: + raise ValueError("invalid finalization_status") + for item in checks: + valid, errors = validate_check(item) + if not valid: + raise ValueError("invalid campaign check: " + ", ".join(errors)) + if item["campaign_id"] != manifest["campaign_id"]: + raise ValueError("check campaign_id does not match manifest") + if item["cell_id"] not in manifest["planned_cells"]: + raise ValueError("check cell_id is not planned") + checks = sorted(checks, key=lambda item: (item["cell_id"], item["attempt"])) + valid, errors = validate_attempt_lineage(checks) + if not valid: + raise ValueError("invalid campaign attempt lineage: " + ", ".join(errors)) + if any(item.get("attempt", 1) > 1 for item in checks) and not manifest.get("continuation_allowed"): + raise ValueError("campaign continuation is not allowed by the manifest") + derived_target_status = _target_status(framework_status, checks) + if target_status is None: + target_status = derived_target_status + elif target_status != derived_target_status: + raise ValueError("target_status does not match framework/check outcomes") + if target_status not in TARGET_STATUS_VALUES: + raise ValueError("invalid target_status") + if budget_status is None: + if manifest.get("hard_stop_micros") is None: + budget_status = "not_configured" + elif spent_micros > manifest["hard_stop_micros"]: + budget_status = "overrun" + else: + budget_status = "within_guard" + if budget_status not in BUDGET_STATUS_VALUES: + raise ValueError("invalid budget_status") + if remaining_micros is None and manifest.get("hard_stop_micros") is not None: + remaining_micros = max(0, manifest["hard_stop_micros"] - spent_micros) + counts = { + "planned": len(manifest["planned_cells"]), + "executed": sum(item["status"] != "skip" for item in checks), + "passed": sum(item["status"] == "pass" for item in checks), + "failed": sum(item["status"] in {"fail", "error"} for item in checks), + "skipped": sum(item["status"] == "skip" for item in checks), + } + receipt: dict[str, Any] = { + "schema": CAMPAIGN_RECEIPT_SCHEMA, + "campaign_id": manifest["campaign_id"], + "manifest_hash": manifest["manifest_hash"], + "framework_status": framework_status, + "target_status": target_status, + "budget_status": budget_status, + "evidence_status": evidence_status, + "finalization_status": finalization_status, + "finalization_reason": finalization_reason, + "counts": counts, + "check_hashes": [item["check_hash"] for item in checks], + "target_identity": target_identity or {}, + "evidence_bundle_hash": evidence_bundle_hash, + "protected_state_before_hash": protected_state_before_hash, + "protected_state_after_hash": protected_state_after_hash, + "cleanup_manifest_hash": cleanup_manifest_hash, + "spent_micros": spent_micros, + "remaining_micros": remaining_micros, + "stop_reason": stop_reason, + "last_checkpoint_ref": last_checkpoint_ref, + } + receipt["receipt_hash"] = receipt_digest(receipt) + receipt["verification"] = verify_campaign_receipt(receipt, manifest=manifest, checks=checks) + return receipt + + +def verify_campaign_receipt(receipt: dict[str, Any], *, manifest: Optional[dict[str, Any]] = None, + checks: Optional[list[dict[str, Any]]] = None) -> dict[str, Any]: + reasons: list[str] = [] + if not isinstance(receipt, dict): + return {"valid": False, "verified_pass": False, "reasons": ["receipt"]} + reasons.extend(_unknown(receipt, _RECEIPT_FIELDS)) + reasons.extend(_forbidden(receipt)) + if receipt.get("schema") != CAMPAIGN_RECEIPT_SCHEMA: + reasons.append("schema") + digest = receipt.get("receipt_hash") + if not _is_sha256(digest): + reasons.append("receipt_hash") + elif digest != receipt_digest(receipt): + reasons.append("receipt_hash_mismatch") + if manifest is not None: + ok, manifest_errors = validate_manifest(manifest) + if not ok: + reasons.extend("manifest:" + item for item in manifest_errors) + elif receipt.get("manifest_hash") != manifest.get("manifest_hash"): + reasons.append("manifest_hash_mismatch") + if checks is not None: + ok, check_errors = validate_attempt_lineage(checks) + if not ok: + reasons.extend("checks:" + item for item in check_errors) + expected = [item.get("check_hash") for item in checks] + if receipt.get("check_hashes") != expected: + reasons.append("check_hashes_mismatch") + framework = receipt.get("framework_status") + target = receipt.get("target_status") + budget = receipt.get("budget_status") + evidence = receipt.get("evidence_status") + finalization = receipt.get("finalization_status") + if framework not in FRAMEWORK_STATUS_VALUES: + reasons.append("framework_status") + if target not in TARGET_STATUS_VALUES: + reasons.append("target_status") + if budget not in BUDGET_STATUS_VALUES: + reasons.append("budget_status") + if evidence not in EVIDENCE_STATUS_VALUES: + reasons.append("evidence_status") + if finalization not in FINALIZATION_STATUS_VALUES: + reasons.append("finalization_status") + if _nonnegative(receipt.get("spent_micros"), "spent_micros", required=True): + reasons.append("spent_micros") + if receipt.get("remaining_micros") is not None and _nonnegative(receipt.get("remaining_micros"), "remaining_micros"): + reasons.append("remaining_micros") + counts = receipt.get("counts") + count_keys = ("planned", "executed", "passed", "failed", "skipped") + if not isinstance(counts, dict) or any(not isinstance(counts.get(key), int) for key in count_keys): + reasons.append("counts") + counts = {} + if checks is not None: + expected_counts = { + "planned": len(manifest["planned_cells"]) if manifest is not None else counts.get("planned", 0), + "executed": sum(item.get("status") != "skip" for item in checks), + "passed": sum(item.get("status") == "pass" for item in checks), + "failed": sum(item.get("status") in {"fail", "error"} for item in checks), + "skipped": sum(item.get("status") == "skip" for item in checks), + } + if any(counts.get(key) != value for key, value in expected_counts.items()): + reasons.append("counts_mismatch") + expected_target = _target_status(framework, checks) + if target != expected_target: + reasons.append("target_status_mismatch") + if target == "pass": + for item in checks: + if item.get("status") != "pass": + continue + if manifest is not None and manifest.get("evidence_required") and not item.get("evidence_hashes"): + reasons.append("evidence_missing") + if not item.get("usage_event_ids"): + reasons.append("usage_binding_missing") + status_reasons: list[str] = [] + if framework != "completed": + status_reasons.append("framework_not_completed") + if counts.get("executed", 0) == 0: + status_reasons.append("no_executed_checks") + if target != "pass": + status_reasons.append("target_not_pass") + if budget in {"stopped", "overrun"}: + status_reasons.append("budget_not_within_guard") + if evidence != "complete": + status_reasons.append("evidence_not_complete") + if finalization != "complete": + status_reasons.append("finalization_failed" if finalization == "failed" else "finalization_pending") + reasons_all = sorted(set(reasons + status_reasons)) + return { + "valid": not reasons, + "verified_pass": not reasons and not status_reasons, + "reasons": reasons_all, + "integrity_ok": not reasons, + } + + +def admit_spend(manifest: dict[str, Any], *, spent_micros: int, proposed_micros: int) -> dict[str, Any]: + valid, errors = validate_manifest(manifest) + if not valid: + raise ValueError("invalid campaign manifest: " + ", ".join(errors)) + if spent_micros < 0 or proposed_micros < 0: + raise ValueError("spend values must be non-negative") + if manifest.get("runaway_guard_micros") is not None: + remaining = manifest["runaway_guard_micros"] - spent_micros + if proposed_micros > remaining: + return {"allowed": False, "reason": "runaway_guard_exceeded", "remaining_micros": max(0, remaining)} + if manifest.get("hard_stop_micros") is not None: + remaining = manifest["hard_stop_micros"] - spent_micros + if proposed_micros > remaining: + return {"allowed": False, "reason": "hard_stop_exceeded", "remaining_micros": max(0, remaining)} + hard = manifest.get("hard_stop_micros") + return { + "allowed": True, + "reason": "within_guard", + "remaining_micros": (hard - spent_micros - proposed_micros) if hard is not None else None, + } + + +def record_usage(conn, org_id: str, *, campaign_binding: dict, **kwargs): + """Record one campaign-bound usage event with atomic budget-stop custody. + + The normal metering function remains the low-level writer. This wrapper owns + the transaction and, on a rejected spend, records the durable stop state in + a second transaction so the failed attempt cannot disappear with the rolled + back usage insert. + """ + from . import db, metering + try: + with db.immediate(conn): + return metering.record_usage( + conn, org_id, campaign_binding=campaign_binding, + commit=False, **kwargs, + ) + except CampaignBudgetError as exc: + with db.immediate(conn): + db.mark_campaign_budget_stop( + conn, org_id, exc.campaign_id, + remaining_micros=exc.remaining_micros, + reason=exc.reason, + ) + raise + + +__all__ = [ + "CAMPAIGN_MANIFEST_SCHEMA", "CAMPAIGN_CHECK_SCHEMA", "CAMPAIGN_RECEIPT_SCHEMA", + "CampaignBudgetError", + "FRAMEWORK_STATUS_VALUES", "TARGET_STATUS_VALUES", "CHECK_STATUS_VALUES", + "BUDGET_STATUS_VALUES", "EVIDENCE_STATUS_VALUES", "FINALIZATION_STATUS_VALUES", + "CAMPAIGN_BINDING_SCHEMA", "build_binding", "validate_binding", "binding_digest", + "build_manifest", "validate_manifest", "manifest_digest", "build_check", + "validate_check", "check_digest", "validate_attempt_lineage", "build_receipt", + "verify_campaign_receipt", "receipt_digest", "admit_spend", +] diff --git a/ledger_agent/client.py b/ledger_agent/client.py index 943b1d1..f82f9ae 100644 --- a/ledger_agent/client.py +++ b/ledger_agent/client.py @@ -25,13 +25,14 @@ """ from __future__ import annotations +import contextlib import json import os import urllib.error import urllib.request from typing import Optional -from . import __version__, config as cfgmod, db, metering +from . import __version__, campaigns, config as cfgmod, db, metering # A real User-Agent — some CDNs/WAFs (e.g. Cloudflare, error 1010) hard-block the # default "Python-urllib/x.y" signature, which would break ingest from behind a @@ -128,6 +129,7 @@ def track(self, provider: str, *, model: Optional[str] = None, belief_context: Optional[dict] = None, governance_cost: Optional[dict] = None, behavior_snapshot: Optional[dict] = None, + campaign_binding: Optional[dict] = None, user_id: Optional[str] = None, source: str = "sdk"): """Meter one call. Returns a :class:`metering.MeterResult`. @@ -201,6 +203,7 @@ def track(self, provider: str, *, model: Optional[str] = None, "belief_context": belief_context, "governance_cost": governance_cost, "behavior_snapshot": behavior_snapshot, + "campaign_binding": campaign_binding, } event.update({key: value for key, value in optional_fields.items() if value is not None}) @@ -208,49 +211,60 @@ def track(self, provider: str, *, model: Optional[str] = None, event["user_id"] = user_id return self._track_remote(event) - return metering.record_usage( - self.conn, self.org_id, provider=provider, model=model, - task_type=task_type, workspace=workspace, - input_tokens=input_tokens, output_tokens=output_tokens, - cache_read_tokens=cache_read_tokens, reasoning_tokens=reasoning_tokens, - cache_write_tokens=cache_write_tokens, - cost_usd=cost_usd, - baseline_cost_usd=baseline_cost_usd, baseline_model=baseline_model, - baseline_input_tokens=baseline_input_tokens, - baseline_output_tokens=baseline_output_tokens, - external_ref=external_ref, - evidence_hashes=evidence_hashes, - policy_version=policy_version, - result_hash=result_hash, - human_review=human_review, - correction_ref=correction_ref, - agent_id=agent_id, - authority_manifest_ref=authority_manifest_ref, - authority_manifest_custody=authority_manifest_custody, - scope_anchor=scope_anchor, - action_intent_hash=action_intent_hash, - action_status=action_status, - approval_ref=approval_ref, - context_render_schema=context_render_schema, - context_render_hash=context_render_hash, - served_memory_provenance_hash=served_memory_provenance_hash, - action_receipt_hash=action_receipt_hash, - resource_constraints_version=resource_constraints_version, - resource_constraints_hash=resource_constraints_hash, - belief_context=belief_context, - prebind=prebind, - governance_cost=governance_cost, - behavior_snapshot=behavior_snapshot, - user_id=user_id, - source=source, - pricing_overrides=self.cfg.get("pricing", {}).get("overrides"), - alert_cfg=self.cfg.get("alerts", {}), - block_over_limit=bool(self.cfg.get("pricing", {}).get("block_over_free_limit")), - # P1 fix: enforce the prepaid hard-stop on the embedded path too, not - # just the hosted API. Default on (matches DEFAULT_CONFIG); only bites - # orgs that actually hold prepaid credit. - block_over_balance=bool(self.cfg.get("pricing", {}).get("block_over_balance", True)), - ) + local_tx = (db.immediate(self.conn) + if campaign_binding is not None else contextlib.nullcontext()) + try: + with local_tx: + return metering.record_usage( + self.conn, self.org_id, provider=provider, model=model, + task_type=task_type, workspace=workspace, + input_tokens=input_tokens, output_tokens=output_tokens, + cache_read_tokens=cache_read_tokens, reasoning_tokens=reasoning_tokens, + cache_write_tokens=cache_write_tokens, + cost_usd=cost_usd, + baseline_cost_usd=baseline_cost_usd, baseline_model=baseline_model, + baseline_input_tokens=baseline_input_tokens, + baseline_output_tokens=baseline_output_tokens, + external_ref=external_ref, + evidence_hashes=evidence_hashes, + policy_version=policy_version, + result_hash=result_hash, + human_review=human_review, + correction_ref=correction_ref, + agent_id=agent_id, + authority_manifest_ref=authority_manifest_ref, + authority_manifest_custody=authority_manifest_custody, + scope_anchor=scope_anchor, + action_intent_hash=action_intent_hash, + action_status=action_status, + approval_ref=approval_ref, + context_render_schema=context_render_schema, + context_render_hash=context_render_hash, + served_memory_provenance_hash=served_memory_provenance_hash, + action_receipt_hash=action_receipt_hash, + resource_constraints_version=resource_constraints_version, + resource_constraints_hash=resource_constraints_hash, + belief_context=belief_context, + prebind=prebind, + governance_cost=governance_cost, + behavior_snapshot=behavior_snapshot, + campaign_binding=campaign_binding, + user_id=user_id, + source=source, + pricing_overrides=self.cfg.get("pricing", {}).get("overrides"), + alert_cfg=self.cfg.get("alerts", {}), + block_over_limit=bool(self.cfg.get("pricing", {}).get("block_over_free_limit")), + # P1 fix: enforce the prepaid hard-stop on the embedded path too. + block_over_balance=bool(self.cfg.get("pricing", {}).get("block_over_balance", True)), + commit=campaign_binding is None, + ) + except campaigns.CampaignBudgetError as exc: + with db.immediate(self.conn): + db.mark_campaign_budget_stop( + self.conn, self.org_id, exc.campaign_id, + remaining_micros=exc.remaining_micros, reason=exc.reason, + ) + raise def _track_remote(self, event: dict) -> "metering.MeterResult": req = urllib.request.Request( @@ -302,6 +316,7 @@ def _track_remote(self, event: dict) -> "metering.MeterResult": over_free_limit=bool(body.get("over_free_limit", False)), over_balance=bool(body.get("over_balance", False)), unpriced=bool(body.get("unpriced", False)), + campaign_id=body.get("campaign_id") or event.get("campaign_binding", {}).get("campaign_id"), ) def _local_only(self, what: str): diff --git a/ledger_agent/db.py b/ledger_agent/db.py index 22a6375..f142094 100644 --- a/ledger_agent/db.py +++ b/ledger_agent/db.py @@ -61,7 +61,9 @@ # 18 = adds stage-aware action receipts and evidence bindings (#219–#224): # served_claim_json/hash (#221), evidence_status (#222), # runtime_manifest_json/hash (#223), external_artifact_json/hash (#224). -SCHEMA_VERSION = 22 +# 23 = adds acceptance campaign manifests/checks/receipts and optional campaign +# usage-event bindings for #256/#257. +SCHEMA_VERSION = 23 # ---- money: integer micro-dollars ------------------------------------------ # All money is stored as integer micro-dollars (1 USD == MICROS_PER_USD micros). @@ -177,6 +179,11 @@ def micros_to_usd(micros) -> float: # v21 (#238): behavior-snapshot receipt pin. Trailing and optional. "behavior_snapshot_json", "behavior_snapshot_hash", + # v23 (#256/#257): hash-only campaign/cell attribution and continuation + # lineage. Optional so every historical event keeps its canonical bytes. + "campaign_id", + "campaign_binding_json", + "campaign_binding_hash", ) @@ -643,6 +650,12 @@ def verify_checkpoints(conn, checkpoints, hmac_key: Optional[bytes] = None) -> d -- `ledger diff --require-target-digest`. behavior_snapshot_json TEXT, behavior_snapshot_hash TEXT, + -- v23 (#256/#257): optional hash-only acceptance-campaign binding. The + -- campaign_id is indexed for integer-exact spend guards; the JSON/hash pair + -- carries cell, lane, config, checkpoint, and attempt lineage. + campaign_id TEXT, + campaign_binding_json TEXT, + campaign_binding_hash TEXT, estimated INTEGER NOT NULL DEFAULT 1, source TEXT NOT NULL DEFAULT 'api', ts REAL NOT NULL, @@ -812,6 +825,44 @@ def verify_checkpoints(conn, checkpoints, hmac_key: Optional[bytes] = None) -> d ts REAL NOT NULL ); CREATE INDEX IF NOT EXISTS ix_recon_event ON reconciliation_events(event_id); + +-- Acceptance campaigns (#256/#257). The manifest, checks, and final receipt are +-- hash-bound public-safe projections; raw benchmark inputs remain out of Ledger. +CREATE TABLE IF NOT EXISTS acceptance_campaigns ( + id TEXT PRIMARY KEY, + org_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + manifest_json TEXT NOT NULL, + manifest_hash TEXT NOT NULL, + framework_status TEXT NOT NULL DEFAULT 'error', + target_status TEXT NOT NULL DEFAULT 'not_run', + budget_status TEXT NOT NULL DEFAULT 'not_configured', + evidence_status TEXT NOT NULL DEFAULT 'pending', + finalization_status TEXT NOT NULL DEFAULT 'pending', + receipt_json TEXT, + receipt_hash TEXT, + spent_micros INTEGER NOT NULL DEFAULT 0, + remaining_micros INTEGER, + stop_reason TEXT, + last_checkpoint_ref TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(org_id, id) +); +CREATE INDEX IF NOT EXISTS ix_campaign_org ON acceptance_campaigns(org_id, updated_at); + +CREATE TABLE IF NOT EXISTS acceptance_checks ( + id TEXT PRIMARY KEY, + campaign_id TEXT NOT NULL REFERENCES acceptance_campaigns(id) ON DELETE CASCADE, + org_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + cell_id TEXT NOT NULL, + attempt INTEGER NOT NULL, + status TEXT NOT NULL, + check_json TEXT NOT NULL, + check_hash TEXT NOT NULL, + created_at REAL NOT NULL, + UNIQUE(campaign_id, cell_id, attempt) +); +CREATE INDEX IF NOT EXISTS ix_campaign_checks ON acceptance_checks(org_id, campaign_id, cell_id, attempt); """ # Public prefix for ingest API keys. The secret is `ledger_sk_`; only its @@ -1005,6 +1056,10 @@ def _migrate_add_columns(conn) -> None: # existing rows and their historical hash canonical form. ("usage_events", "behavior_snapshot_json", "TEXT"), ("usage_events", "behavior_snapshot_hash", "TEXT"), + # v23 (#256/#257): campaign binding; nullable preserves existing chains. + ("usage_events", "campaign_id", "TEXT"), + ("usage_events", "campaign_binding_json", "TEXT"), + ("usage_events", "campaign_binding_hash", "TEXT"), ] for table, col, defn in additions: cols = _table_columns(conn, table) @@ -1025,6 +1080,230 @@ def _migrate_add_columns(conn) -> None: ) +def _campaign_row(row): + return dict(row) if row is not None else None + + +def create_campaign(conn, org_id: str, manifest: dict, *, commit: bool = True) -> dict: + """Persist a hash-bound campaign manifest idempotently within one org.""" + from . import campaigns + valid, errors = campaigns.validate_manifest(manifest) + if not valid: + raise ValueError("invalid campaign manifest: " + ", ".join(errors)) + if not get_org(conn, org_id): + raise ValueError("organization not found") + cid = manifest["campaign_id"] + existing = conn.execute( + "SELECT * FROM acceptance_campaigns WHERE org_id=? AND id=?", (org_id, cid) + ).fetchone() + if existing: + if existing["manifest_hash"] != manifest["manifest_hash"]: + raise ValueError("manifest conflict") + return _campaign_row(existing) + now = time.time() + conn.execute( + "INSERT INTO acceptance_campaigns(id,org_id,manifest_json,manifest_hash,created_at,updated_at) " + "VALUES(?,?,?,?,?,?)", + (cid, org_id, json.dumps(manifest, sort_keys=True, separators=(",", ":")), + manifest["manifest_hash"], now, now), + ) + if commit: + conn.commit() + return get_campaign(conn, org_id, cid) + + +def get_campaign(conn, org_id: str, campaign_id: str) -> Optional[dict]: + row = conn.execute( + "SELECT * FROM acceptance_campaigns WHERE org_id=? AND id=?", + (org_id, campaign_id), + ).fetchone() + return _campaign_row(row) + + +def list_campaign_checks(conn, org_id: str, campaign_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM acceptance_checks WHERE org_id=? AND campaign_id=? " + "ORDER BY cell_id, attempt", + (org_id, campaign_id), + ).fetchall() + return [dict(row) for row in rows] + + +def _campaign_check_objects(conn, org_id: str, campaign_id: str) -> list[dict]: + return [json.loads(row["check_json"]) + for row in list_campaign_checks(conn, org_id, campaign_id)] + + +def campaign_usage_errors(conn, org_id: str, campaign_id: str, + checks: list[dict]) -> list[str]: + """Validate that executed check references resolve to bound usage rows. + + This is intentionally read-only and is applied at finalization/readback. A + runner may persist a check before its usage event arrives, but such a check + can never become a verified pass until the event exists and its binding + agrees with the check lineage. + """ + from . import campaigns + errors: list[str] = [] + seen: set[str] = set() + for check in checks: + if check.get("status") == "skip": + continue + event_ids = check.get("usage_event_ids") or [] + for event_id in event_ids: + if event_id in seen: + errors.append(f"usage_duplicate:{event_id}") + seen.add(event_id) + row = conn.execute( + "SELECT id,campaign_id,campaign_binding_json FROM usage_events " + "WHERE org_id=? AND id=?", (org_id, event_id) + ).fetchone() + if row is None: + errors.append(f"usage_missing:{event_id}") + continue + if row["campaign_id"] != campaign_id: + errors.append(f"usage_campaign_mismatch:{event_id}") + continue + try: + binding = json.loads(row["campaign_binding_json"]) + except (TypeError, ValueError, json.JSONDecodeError): + errors.append(f"usage_binding_invalid:{event_id}") + continue + valid, binding_errors = campaigns.validate_binding(binding) + if not valid: + errors.extend(f"usage_binding:{event_id}:{item}" for item in binding_errors) + continue + for field in ("campaign_id", "cell_id", "lane", "config_hash", + "attempt", "continuation", "parent_attempt", + "action_intent_hash", "checkpoint_ref"): + if binding.get(field) != check.get(field): + errors.append(f"usage_binding_mismatch:{event_id}:{field}") + return sorted(set(errors)) + + +def record_campaign_check(conn, org_id: str, check: dict, *, commit: bool = True) -> dict: + """Store one immutable campaign check; conflicting hashes never overwrite.""" + from . import campaigns + valid, errors = campaigns.validate_check(check) + if not valid: + raise ValueError("invalid campaign check: " + ", ".join(errors)) + campaign = get_campaign(conn, org_id, check["campaign_id"]) + if campaign is None: + raise ValueError("campaign not found") + manifest = json.loads(campaign["manifest_json"]) + if check["cell_id"] not in manifest["planned_cells"]: + raise ValueError("check cell_id is not planned") + if check["attempt"] > 1 and not manifest.get("continuation_allowed"): + raise ValueError("campaign continuation is not allowed by the manifest") + prior = _campaign_check_objects(conn, org_id, check["campaign_id"]) + valid, errors = campaigns.validate_attempt_lineage(prior + [check]) + if not valid: + raise ValueError("invalid campaign attempt lineage: " + ", ".join(errors)) + existing = conn.execute( + "SELECT * FROM acceptance_checks WHERE org_id=? AND campaign_id=? AND cell_id=? AND attempt=?", + (org_id, check["campaign_id"], check["cell_id"], check["attempt"]), + ).fetchone() + if existing: + if existing["check_hash"] != check["check_hash"]: + raise ValueError("check conflict") + return dict(existing) + now = time.time() + conn.execute( + "INSERT INTO acceptance_checks(id,campaign_id,org_id,cell_id,attempt,status,check_json,check_hash,created_at) " + "VALUES(?,?,?,?,?,?,?,?,?)", + (new_id("chk"), check["campaign_id"], org_id, check["cell_id"], + check["attempt"], check["status"], + json.dumps(check, sort_keys=True, separators=(",", ":")), + check["check_hash"], now), + ) + conn.execute("UPDATE acceptance_campaigns SET updated_at=? WHERE id=? AND org_id=?", + (now, check["campaign_id"], org_id)) + if commit: + conn.commit() + return dict(conn.execute( + "SELECT * FROM acceptance_checks WHERE org_id=? AND campaign_id=? AND cell_id=? AND attempt=?", + (org_id, check["campaign_id"], check["cell_id"], check["attempt"]), + ).fetchone()) + + +def finalize_campaign(conn, org_id: str, receipt: dict, *, commit: bool = True) -> dict: + """Persist the final campaign receipt without mutating prior checks.""" + from . import campaigns + campaign = get_campaign(conn, org_id, receipt.get("campaign_id")) + if campaign is None: + raise ValueError("campaign not found") + manifest = json.loads(campaign["manifest_json"]) + checks = _campaign_check_objects(conn, org_id, receipt["campaign_id"]) + verification = campaigns.verify_campaign_receipt(receipt, manifest=manifest, checks=checks) + actual_spend = campaign_spend_micros(conn, receipt["campaign_id"]) + if receipt.get("spent_micros") != actual_spend: + verification["valid"] = False + verification["verified_pass"] = False + verification["integrity_ok"] = False + verification["reasons"] = sorted(set(verification["reasons"] + ["spent_micros_mismatch"])) + if receipt.get("target_status") == "pass": + usage_errors = campaign_usage_errors(conn, org_id, receipt["campaign_id"], checks) + if usage_errors: + verification["valid"] = False + verification["verified_pass"] = False + verification["integrity_ok"] = False + verification["reasons"] = sorted(set(verification["reasons"] + usage_errors)) + if not verification["valid"]: + raise ValueError("invalid campaign receipt: " + ", ".join(verification["reasons"])) + if campaign["receipt_hash"] is not None: + if campaign["receipt_hash"] != receipt["receipt_hash"]: + raise ValueError("receipt conflict") + return campaign + now = time.time() + conn.execute( + "UPDATE acceptance_campaigns SET framework_status=?,target_status=?,budget_status=?," + "evidence_status=?,finalization_status=?,receipt_json=?,receipt_hash=?,spent_micros=?," + "remaining_micros=?,stop_reason=?,last_checkpoint_ref=?,updated_at=? WHERE id=? AND org_id=?", + (receipt["framework_status"], receipt["target_status"], receipt["budget_status"], + receipt["evidence_status"], receipt["finalization_status"], + json.dumps(receipt, sort_keys=True, separators=(",", ":")), receipt["receipt_hash"], + receipt["spent_micros"], receipt["remaining_micros"], receipt["stop_reason"], + receipt["last_checkpoint_ref"], now, receipt["campaign_id"], org_id), + ) + if commit: + conn.commit() + return get_campaign(conn, org_id, receipt["campaign_id"]) + + +def campaign_spend_micros(conn, campaign_id: str) -> int: + row = conn.execute( + "SELECT COALESCE(SUM(cost_micros),0) AS spend FROM usage_events WHERE campaign_id=?", + (campaign_id,), + ).fetchone() + return int(row["spend"] or 0) + + +def campaign_usage_count(conn, campaign_id: str) -> int: + row = conn.execute( + "SELECT COUNT(*) AS count FROM usage_events WHERE campaign_id=?", + (campaign_id,), + ).fetchone() + return int(row["count"] or 0) + + +def mark_campaign_budget_stop(conn, org_id: str, campaign_id: str, *, + remaining_micros: int, reason: str, + commit: bool = True) -> dict: + """Durably mark a campaign stopped after a rejected spend admission.""" + row = get_campaign(conn, org_id, campaign_id) + if row is None: + raise ValueError("campaign not found") + now = time.time() + conn.execute( + "UPDATE acceptance_campaigns SET budget_status='stopped',remaining_micros=?," + "stop_reason=?,updated_at=? WHERE org_id=? AND id=?", + (max(0, int(remaining_micros)), reason, now, org_id, campaign_id), + ) + if commit: + conn.commit() + return get_campaign(conn, org_id, campaign_id) + + def _table_columns(conn, table: str) -> set: try: return {r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()} @@ -1693,6 +1972,7 @@ def export_events(conn, org_id: str, since: Optional[float] = None, "ue.cache_read_tokens, ue.cache_write_tokens, ue.reasoning_tokens, " "ue.user_id, ue.cost_micros, " "ue.baseline_micros, ue.optimal_micros, ue.external_ref, " + "ue.campaign_id, ue.campaign_binding_json, " "ue.estimated, ue.source " "FROM usage_events ue " "LEFT JOIN workspaces w ON w.id=ue.workspace_id WHERE ue.org_id=?") @@ -1715,6 +1995,8 @@ def export_events(conn, org_id: str, since: Optional[float] = None, d["baseline_usd"] = None if bm is None else micros_to_usd(int(bm)) om = d.pop("optimal_micros", None) d["optimal_usd"] = None if om is None else micros_to_usd(int(om)) + binding = d.pop("campaign_binding_json", None) + d["campaign_binding"] = json.loads(binding) if binding is not None else None d["estimated"] = bool(d["estimated"]) out.append(d) return out diff --git a/ledger_agent/mcp_server.py b/ledger_agent/mcp_server.py index 435e23a..6d3a692 100644 --- a/ledger_agent/mcp_server.py +++ b/ledger_agent/mcp_server.py @@ -154,6 +154,11 @@ def _tools() -> list[dict[str, Any]]: "taxonomy); unknown custody renders as " "labeled uncertainty (#241).", }, + "campaign_binding": { + "type": "object", + "description": "Hash-only acceptance-campaign cell/lane/config/" + "checkpoint binding (#256/#257).", + }, }, required=["provider"], ), @@ -264,6 +269,7 @@ def _handle_record(meter: client.Meter, args: dict) -> dict: governance_cost=args.get("governance_cost"), behavior_snapshot=args.get("behavior_snapshot"), authority_manifest_custody=args.get("authority_manifest_custody"), + campaign_binding=args.get("campaign_binding"), source="mcp", ) return asdict(res) diff --git a/ledger_agent/metering.py b/ledger_agent/metering.py index 1df77b2..bf51a94 100644 --- a/ledger_agent/metering.py +++ b/ledger_agent/metering.py @@ -24,7 +24,7 @@ from dataclasses import dataclass, asdict from typing import Optional -from . import db, pricing +from . import db, pricing, campaigns from .prebind import validate_prebind from .receipts import ( build_served_claim, validate_served_claim, @@ -99,6 +99,7 @@ class MeterResult: # (Free = 1) forced the event into the org's earliest workspace instead of # creating a new one — the recorded attribution differs from what was sent. workspace_folded: bool = False + campaign_id: Optional[str] = None def _resolve_workspace(conn, org_id: str, workspace: Optional[str], @@ -185,6 +186,7 @@ def record_usage(conn, org_id: str, provider: str, belief_context: Optional[dict] = None, governance_cost: Optional[dict] = None, behavior_snapshot: Optional[dict] = None, + campaign_binding: Optional[dict] = None, commit: bool = True) -> MeterResult: """Meter one LLM/agent call. Returns a :class:`MeterResult`. @@ -302,6 +304,31 @@ def record_usage(conn, org_id: str, provider: str, raise ValueError("resource_constraints_hash must be a 64-character SHA-256 hex digest") resource_constraints_hash = resource_constraints_hash.lower() if resource_constraints_hash else None + # v23: acceptance-campaign binding. It is persisted as a canonical JSON/hash + # pair and checked against the org-scoped durable campaign before pricing or + # writing the event. + campaign_binding_json: Optional[str] = None + campaign_binding_hash: Optional[str] = None + campaign_id: Optional[str] = None + campaign_manifest: Optional[dict] = None + if campaign_binding is not None: + if not isinstance(campaign_binding, dict): + raise ValueError("campaign_binding must be a dict") + valid, errors = campaigns.validate_binding(campaign_binding) + if not valid: + raise ValueError("invalid campaign_binding: " + ", ".join(errors)) + campaign_id = campaign_binding["campaign_id"] + campaign_binding_json = json.dumps(campaign_binding, sort_keys=True, separators=(",", ":")) + campaign_binding_hash = campaign_binding["binding_hash"] + campaign_row = db.get_campaign(conn, org_id, campaign_id) + if campaign_row is None: + raise ValueError("campaign_binding references an unknown campaign") + campaign_manifest = json.loads(campaign_row["manifest_json"]) + if campaign_binding["cell_id"] not in campaign_manifest["planned_cells"]: + raise ValueError("campaign_binding cell_id is not planned") + if campaign_binding["lane"] not in campaign_manifest["provider_lanes"]: + raise ValueError("campaign_binding lane is not planned") + # v18: stage-aware action receipts and evidence bindings (#219–#224) served_claim_json: Optional[str] = None served_claim_hash: Optional[str] = None @@ -547,6 +574,20 @@ def record_usage(conn, org_id: str, provider: str, optimal_micros = db.usd_to_micros(optimal_cost_usd) leaked_usd = round(max(0.0, cost_usd - optimal_cost_usd), 6) + # #257: campaign economics are checked in integer micros before the event + # insert. Campaign callers use db.immediate(), so the spend read and insert + # share one SQLite write lock and an overrun leaves no usage row. + if campaign_binding is not None: + decision = campaigns.admit_spend( + campaign_manifest, + spent_micros=db.campaign_spend_micros(conn, campaign_id), + proposed_micros=db.usd_to_micros(cost_usd), + ) + if not decision["allowed"]: + raise campaigns.CampaignBudgetError( + campaign_id, decision["reason"], decision["remaining_micros"] + ) + # Fix #28: prepaid credit hard-stop. Skipped for orgs explicitly flagged # allow_negative_balance (trusted/internal track-only mode) so they keep # full tracking even past zero. @@ -632,6 +673,10 @@ def record_usage(conn, org_id: str, provider: str, # v21 (#238) "behavior_snapshot_json": behavior_snapshot_json, "behavior_snapshot_hash": behavior_snapshot_hash, + # v23 (#256/#257) + "campaign_id": campaign_id, + "campaign_binding_json": campaign_binding_json, + "campaign_binding_hash": campaign_binding_hash, } row_hash = db.compute_row_hash(prev_hash, row_fields, hmac_key=chain_hmac_key) conn.execute( @@ -645,8 +690,9 @@ def record_usage(conn, org_id: str, provider: str, "belief_context_json,belief_context_hash," "governance_cost_json,governance_cost_hash," "behavior_snapshot_json,behavior_snapshot_hash," + "campaign_id,campaign_binding_json,campaign_binding_hash," "estimated,source,ts,prev_hash,row_hash) " - "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + "VALUES(" + ",".join("?" for _ in range(57)) + ")", (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), @@ -665,6 +711,7 @@ def record_usage(conn, org_id: str, provider: str, belief_context_json, belief_context_hash, governance_cost_json, governance_cost_hash, behavior_snapshot_json, behavior_snapshot_hash, + campaign_id, campaign_binding_json, campaign_binding_hash, int(estimated), source, ts, prev_hash, row_hash), ) @@ -695,6 +742,7 @@ def record_usage(conn, org_id: str, provider: str, external_ref=external_ref, user_id=user_id, workspace_folded=workspace_folded, + campaign_id=campaign_id, ) diff --git a/ledger_agent/server/api.py b/ledger_agent/server/api.py index f71590b..40498e3 100644 --- a/ledger_agent/server/api.py +++ b/ledger_agent/server/api.py @@ -5,7 +5,7 @@ import io import json -from .. import db, evidence_levels, metering, pricing, savings +from .. import campaigns, db, evidence_levels, metering, pricing, savings from ..prebind import validate_prebind @@ -82,6 +82,52 @@ def replay_receipt_prebind(conn, org_id: str, external_ref: str, **kwargs) -> di return replay_prebind(prior, **kwargs) +def campaign_json(conn, org_id: str, campaign_id: str) -> dict: + """Return a public-safe campaign manifest/check/receipt projection.""" + row = db.get_campaign(conn, org_id, campaign_id) + if row is None: + raise ValueError("campaign not found") + manifest = json.loads(row["manifest_json"]) + check_rows = db.list_campaign_checks(conn, org_id, campaign_id) + checks = [json.loads(item["check_json"]) for item in check_rows] + receipt = json.loads(row["receipt_json"]) if row["receipt_json"] else None + if receipt is None: + verification = { + "valid": False, "verified_pass": False, + "reasons": ["receipt_missing"], "integrity_ok": False, + } + else: + verification = campaigns.verify_campaign_receipt( + receipt, manifest=manifest, checks=checks, + ) + actual_spend = db.campaign_spend_micros(conn, campaign_id) + if receipt.get("spent_micros") != actual_spend: + verification["valid"] = False + verification["verified_pass"] = False + verification["integrity_ok"] = False + verification["reasons"] = sorted(set(verification["reasons"] + ["spent_micros_mismatch"])) + if receipt.get("target_status") == "pass": + usage_errors = db.campaign_usage_errors(conn, org_id, campaign_id, checks) + if usage_errors: + verification["valid"] = False + verification["verified_pass"] = False + verification["integrity_ok"] = False + verification["reasons"] = sorted(set(verification["reasons"] + usage_errors)) + return { + "org_id": org_id, + "campaign_id": campaign_id, + "manifest": manifest, + "checks": checks, + "receipt": receipt, + "verification": verification, + "spend_micros": db.campaign_spend_micros(conn, campaign_id), + "usage_events": db.campaign_usage_count(conn, campaign_id), + "budget_status": row["budget_status"], + "stop_reason": row["stop_reason"], + "updated_at": row["updated_at"], + } + + _EXPORT_COLUMNS = ["id", "ts", "provider", "model", "task_type", "workspace", "input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", "user_id", @@ -100,6 +146,7 @@ def _csv_safe(value): def audit_json(conn, org_id: str, *, hmac_key: bytes | None = None, external_ref: str | None = None, + campaign_id: str | None = None, key_registry: dict[str, bytes] | None = None, sign_key_id: str | None = None) -> dict: """Return either an organization audit summary or one task evidence receipt. @@ -116,6 +163,8 @@ def audit_json(conn, org_id: str, *, hmac_key: bytes | None = None, ``verification.evidence`` block reporting the highest evidence level the retained objects actually verify (#235). """ + if campaign_id is not None: + return campaign_json(conn, org_id, campaign_id) if external_ref is not None: org = db.get_org(conn, org_id) integrity = db.verify_chain(conn, org_id=org_id, hmac_key=hmac_key) @@ -161,6 +210,8 @@ def audit_json(conn, org_id: str, *, hmac_key: bytes | None = None, if row["governance_cost_json"] is not None else None, "behavior_snapshot": json.loads(row["behavior_snapshot_json"]) if row["behavior_snapshot_json"] is not None else None, + "campaign_binding": json.loads(row["campaign_binding_json"]) + if row["campaign_binding_json"] is not None else None, "action_authorization": { "agent_id": row["agent_id"], "authority_manifest_ref": row["authority_manifest_ref"], diff --git a/ledger_agent/server/app.py b/ledger_agent/server/app.py index ae3049c..0faac85 100644 --- a/ledger_agent/server/app.py +++ b/ledger_agent/server/app.py @@ -19,7 +19,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs -from .. import __version__, bridge, config as cfgmod, db, pricing +from .. import __version__, bridge, campaigns, config as cfgmod, db, pricing from ..billing import StripeClient, BillingError, handle_webhook_event from ..utils import strict_int from ..prebind import validate_prebind @@ -31,6 +31,7 @@ "/v1/usage", # authenticated by its own Bearer API key, not a session "/v1/usage/export.csv", "/v1/usage/export.json", # Bearer-auth (#66) "/v1/checkpoints", # Bearer-auth tamper-evidence anchors (#121) + "/v1/campaigns", "/v1/campaigns/checks", "/v1/campaigns/finalize", "/api/audit", # Bearer-auth evidence receipts (org-scoped API key) "/v1/admin/orgs", "/v1/admin/credits", "/v1/admin/keys", # admin-token (#66) "/v1/admin/keys/rotate", "/v1/admin/keys/revoke", # admin-token key lifecycle @@ -434,7 +435,8 @@ def do_GET(self): return self._json(404, {"error": "no organizations"}) return self._json(200, api.audit_json( conn, org_id, hmac_key=cfgmod.chain_hmac_key(self.ctx.cfg), - external_ref=q.get("external_ref", [None])[0])) + external_ref=q.get("external_ref", [None])[0], + campaign_id=q.get("campaign_id", [None])[0])) if path == "/api/ledger": org_id = self._authz_org(conn, q.get("org", [None])[0]) if not org_id: @@ -453,6 +455,8 @@ def do_GET(self): return self._usage_export(conn, path, q) if path == "/v1/checkpoints": return self._checkpoints_get(conn, q) + if path == "/v1/campaigns": + return self._campaign_get(conn, q) if path.startswith("/v1/admin/"): return self._admin_get(conn, path, q) if path in ("/billing/success", "/billing/cancel"): @@ -502,7 +506,8 @@ def do_POST(self): needs_csrf_check = ( self.ctx.auth_on and path not in {"/v1/usage", "/webhook/stripe", - "/v1/checkpoints"} and # Bearer-auth (#121) + "/v1/checkpoints", "/v1/campaigns", + "/v1/campaigns/checks", "/v1/campaigns/finalize"} and # Bearer-auth (#121) self._user is not None # Cookie-authenticated ) if needs_csrf_check and not (self._same_origin() or self._csrf_token_ok()): @@ -514,6 +519,8 @@ def do_POST(self): return self._ingest_usage(conn) if path == "/v1/checkpoints": return self._checkpoints_post(conn) + if path in {"/v1/campaigns", "/v1/campaigns/checks", "/v1/campaigns/finalize"}: + return self._campaign_post(conn, path) if path == "/api/users": return self._users_create(conn) if path == "/api/users/deactivate": @@ -646,6 +653,62 @@ def _usage_export(self, conn, path, q): return self._send(200, csv_text, "text/csv; charset=utf-8", {"Content-Disposition": 'attachment; filename="usage.csv"'}) + def _campaign_org(self, conn): + org_id = self._bearer_org(conn) + if not org_id: + self._json(401, {"error": "invalid or missing API key"}) + return None + return org_id + + def _campaign_get(self, conn, q): + org_id = self._campaign_org(conn) + if not org_id: + return + campaign_id = (q.get("campaign_id") or [None])[0] + if not campaign_id: + return self._json(400, {"error": "campaign_id is required"}) + try: + return self._json(200, api.campaign_json(conn, org_id, campaign_id)) + except ValueError: + return self._json(404, {"error": "campaign not found"}) + + def _campaign_post(self, conn, path): + org_id = self._campaign_org(conn) + if not org_id: + return + try: + payload = json.loads(self._body() or b"{}") + except (TypeError, ValueError): + return self._json(400, {"error": "body must be JSON"}) + if not isinstance(payload, dict): + return self._json(400, {"error": "body must be an object"}) + try: + with db.immediate(conn): + if path == "/v1/campaigns": + manifest = payload.get("manifest") + if not isinstance(manifest, dict): + return self._json(400, {"error": "manifest must be an object"}) + db.create_campaign(conn, org_id, manifest, commit=False) + result = api.campaign_json(conn, org_id, manifest["campaign_id"]) + return self._json(201, result) + if path == "/v1/campaigns/checks": + check = payload.get("check") + if not isinstance(check, dict): + return self._json(400, {"error": "check must be an object"}) + db.record_campaign_check(conn, org_id, check, commit=False) + result = api.campaign_json(conn, org_id, check["campaign_id"]) + return self._json(201, result) + receipt = payload.get("receipt") + if not isinstance(receipt, dict): + return self._json(400, {"error": "receipt must be an object"}) + db.finalize_campaign(conn, org_id, receipt, commit=False) + result = api.campaign_json(conn, org_id, receipt["campaign_id"]) + return self._json(200, result) + except ValueError as exc: + message = str(exc) + status = 404 if message == "campaign not found" else 400 + return self._json(status, {"error": message}) + def _checkpoints_get(self, conn, q): """GET /v1/checkpoints — the org's retained anchors (#121). Bearer- authenticated for API callers; falls back to the signed-in session for @@ -858,6 +921,12 @@ def _ingest_usage(self, conn): return self._json(400, {"error": "governance_cost must be an object"}) if ev.get("behavior_snapshot") is not None and not isinstance(ev["behavior_snapshot"], dict): return self._json(400, {"error": "behavior_snapshot must be an object"}) + if ev.get("campaign_binding") is not None: + if not isinstance(ev["campaign_binding"], dict): + return self._json(400, {"error": "campaign_binding must be an object"}) + valid, errors = campaigns.validate_binding(ev["campaign_binding"]) + if not valid: + return self._json(400, {"error": "invalid campaign_binding", "fields": errors}) # All valid — record the whole batch as one serialized transaction. # Fix #27/#30: db.immediate() takes the write lock up front (BEGIN @@ -926,6 +995,7 @@ def _ingest_usage(self, conn): belief_context=ev.get("belief_context"), governance_cost=ev.get("governance_cost"), behavior_snapshot=ev.get("behavior_snapshot"), + campaign_binding=ev.get("campaign_binding"), user_id=ev.get("user_id"), source=ev.get("source", "api"), pricing_overrides=cfg.get("pricing", {}).get("overrides"), @@ -959,12 +1029,32 @@ def _ingest_usage(self, conn): # collapse. "workspace_id": res.workspace_id, "workspace_folded": res.workspace_folded, + "campaign_id": res.campaign_id, }) code, body = self._usage_response( conn, org_id, out, n_blocked, n_over_balance, cfg) if idem_key: db.store_idempotency_response( conn, org_id, idem_key, code, json.dumps(body), commit=False) + except campaigns.CampaignBudgetError as e: + # The usage transaction has rolled back. Persist only the bounded + # campaign stop marker in a fresh transaction; no overrun event or + # partial batch may survive the rejected admission. + try: + with db.immediate(conn): + db.mark_campaign_budget_stop( + conn, org_id, e.campaign_id, + remaining_micros=e.remaining_micros, reason=e.reason, + commit=False, + ) + except Exception: + self._log_exc("POST", "/v1/usage campaign stop", e) + return self._json(402, { + "error": "campaign budget guard stopped the campaign", + "campaign_id": e.campaign_id, + "reason": e.reason, + "remaining_micros": e.remaining_micros, + }) except sqlite3.OperationalError as e: # Transient write contention (e.g. "database is locked" after the # busy_timeout). The batch did NOT commit — BEGIN IMMEDIATE rolled it diff --git a/openapi.yaml b/openapi.yaml index 9cefe3c..4b46bfa 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -101,6 +101,92 @@ paths: description: A request with the same `Idempotency-Key` is still in flight. content: { application/json: { schema: { $ref: '#/components/schemas/Error' } } } + /v1/campaigns: + get: + tags: [usage] + summary: Read one org-scoped acceptance campaign. + security: + - apiKey: [] + parameters: + - name: campaign_id + in: query + required: true + schema: { type: string } + responses: + '200': + description: Public-safe manifest, checks, receipt, and verification projection. + content: { application/json: { schema: { $ref: '#/components/schemas/CampaignProjection' } } } + '400': { description: Missing campaign_id. } + '401': { $ref: '#/components/responses/Unauthorized' } + '404': { description: Campaign not found in the authenticated org. } + post: + tags: [usage] + summary: Create or idempotently recover an acceptance campaign manifest. + security: + - apiKey: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [manifest] + properties: + manifest: { $ref: '#/components/schemas/CampaignManifest' } + additionalProperties: false + responses: + '201': + description: Campaign created or the same manifest was replayed. + content: { application/json: { schema: { $ref: '#/components/schemas/CampaignProjection' } } } + '400': { description: Invalid or conflicting manifest. } + '401': { $ref: '#/components/responses/Unauthorized' } + /v1/campaigns/checks: + post: + tags: [usage] + summary: Persist one immutable campaign cell/check result. + security: + - apiKey: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [check] + properties: + check: { $ref: '#/components/schemas/CampaignCheck' } + additionalProperties: false + responses: + '201': + description: Check stored or the same check was replayed. + content: { application/json: { schema: { $ref: '#/components/schemas/CampaignProjection' } } } + '400': { description: Invalid, conflicting, or out-of-lineage check. } + '401': { $ref: '#/components/responses/Unauthorized' } + '404': { description: Campaign not found in the authenticated org. } + /v1/campaigns/finalize: + post: + tags: [usage] + summary: Persist a final campaign receipt and independent verification state. + security: + - apiKey: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [receipt] + properties: + receipt: { $ref: '#/components/schemas/CampaignReceipt' } + additionalProperties: false + responses: + '200': + description: Final receipt stored or the same receipt was replayed. + content: { application/json: { schema: { $ref: '#/components/schemas/CampaignProjection' } } } + '400': { description: Invalid or conflicting receipt. } + '401': { $ref: '#/components/responses/Unauthorized' } + '404': { description: Campaign not found in the authenticated org. } + /v1/usage/export.csv: get: tags: [usage] @@ -369,7 +455,114 @@ components: governance_cost: { type: object, nullable: true, description: "Governance self-cost block (wall_ms/cpu_ms/mem_bytes/storage_bytes/tokens/model_calls/approval_waits_ms); internal telemetry excluded from customer-facing totals (#239)." } behavior_snapshot: { type: object, nullable: true, description: "Behavior-snapshot receipt pin carrying the sha256 of a canonical agent-run snapshot (#238)." } authority_manifest_custody: { type: string, nullable: true, description: "Custody disclosure label for the referenced authority manifest (1f916 taxonomy); unknown custody renders as labeled uncertainty (#241)." } + campaign_binding: { $ref: '#/components/schemas/CampaignBinding' } source: { type: string, default: api } + CampaignBinding: + type: object + additionalProperties: false + description: "Hash-only campaign/cell/lane/config/checkpoint binding." + required: [schema, campaign_id, cell_id, lane, config_hash, attempt, continuation, binding_hash] + properties: + schema: { type: string, const: perseus-ledger-campaign-binding/v1 } + campaign_id: { type: string } + cell_id: { type: string } + lane: { type: string } + config_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + attempt: { type: integer, minimum: 1 } + continuation: { type: boolean } + parent_attempt: { type: integer, minimum: 1, nullable: true } + action_intent_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + checkpoint_ref: { type: string, nullable: true } + binding_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + CampaignManifest: + type: object + additionalProperties: false + description: "Hash-only acceptance campaign plan; raw benchmark inputs remain out of band." + required: [schema, campaign_id, planned_cells, provider_lanes, config_hash, fixture_hash, retry_policy, continuation_allowed, evidence_required, manifest_hash] + properties: + schema: { type: string, const: perseus-ledger-acceptance-campaign/v1 } + campaign_id: { type: string } + planned_cells: { type: array, minItems: 1, items: { type: string } } + provider_lanes: { type: array, minItems: 1, items: { type: string } } + config_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + fixture_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + target_ref: { type: string, nullable: true } + target_commit_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + target_build_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + runtime_identity: { type: string, nullable: true } + expected_spend_min_micros: { type: integer, minimum: 0, nullable: true } + expected_spend_max_micros: { type: integer, minimum: 0, nullable: true } + hard_stop_micros: { type: integer, minimum: 0, nullable: true } + runaway_guard_micros: { type: integer, minimum: 0, nullable: true } + retry_policy: { type: string, enum: [none, same_config, new_version_only, explicit_continuation] } + continuation_allowed: { type: boolean } + action_intent_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + evidence_required: { type: boolean } + manifest_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + CampaignCheck: + type: object + additionalProperties: false + description: "One hash-bound campaign cell result or explicit skip/error." + required: [schema, campaign_id, cell_id, lane, status, config_hash, attempt, continuation, evidence_hashes, usage_event_ids, check_hash] + properties: + schema: { type: string, const: perseus-ledger-acceptance-check/v1 } + campaign_id: { type: string } + cell_id: { type: string } + lane: { type: string } + status: { type: string, enum: [pass, fail, skip, error] } + config_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + attempt: { type: integer, minimum: 1 } + continuation: { type: boolean } + parent_attempt: { type: integer, minimum: 1, nullable: true } + action_intent_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + result_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + evidence_hashes: { type: array, items: { type: string, pattern: '^[0-9a-fA-F]{64}$' } } + usage_event_ids: { type: array, items: { type: string } } + checkpoint_ref: { type: string, nullable: true } + reason_code: { type: string, nullable: true } + check_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + CampaignReceipt: + type: object + additionalProperties: false + description: "Final hash-bound campaign envelope; verified_pass is independent of target_status." + required: [schema, campaign_id, manifest_hash, framework_status, target_status, budget_status, evidence_status, finalization_status, counts, check_hashes, spent_micros, receipt_hash] + properties: + schema: { type: string, const: perseus-ledger-acceptance-receipt/v1 } + campaign_id: { type: string } + manifest_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + framework_status: { type: string, enum: [completed, error, interrupted, cancelled] } + target_status: { type: string, enum: [pass, fail, inconclusive, not_run] } + budget_status: { type: string, enum: [not_configured, within_guard, stopped, overrun] } + evidence_status: { type: string, enum: [pending, complete, incomplete, invalid] } + finalization_status: { type: string, enum: [pending, complete, failed] } + finalization_reason: { type: string, nullable: true } + counts: { type: object, additionalProperties: { type: integer, minimum: 0 } } + check_hashes: { type: array, items: { type: string, pattern: '^[0-9a-fA-F]{64}$' } } + target_identity: { type: object } + evidence_bundle_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + protected_state_before_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + protected_state_after_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + cleanup_manifest_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$', nullable: true } + spent_micros: { type: integer, minimum: 0 } + remaining_micros: { type: integer, minimum: 0, nullable: true } + stop_reason: { type: string, nullable: true } + last_checkpoint_ref: { type: string, nullable: true } + verification: { type: object } + receipt_hash: { type: string, pattern: '^[0-9a-fA-F]{64}$' } + CampaignProjection: + type: object + properties: + org_id: { type: string } + campaign_id: { type: string } + manifest: { $ref: '#/components/schemas/CampaignManifest' } + checks: { type: array, items: { $ref: '#/components/schemas/CampaignCheck' } } + receipt: { allOf: [{ $ref: '#/components/schemas/CampaignReceipt' }], nullable: true } + verification: { type: object } + spend_micros: { type: integer, minimum: 0 } + usage_events: { type: integer, minimum: 0 } + budget_status: { type: string } + stop_reason: { type: string, nullable: true } + updated_at: { type: number } Prebind: type: object additionalProperties: false @@ -412,6 +605,7 @@ components: leaked_usd: { type: number, description: "Per-event efficiency leakage vs the on-policy optimal (#8); 0 when no optimal was recorded." } workspace_id: { type: string, nullable: true, description: "The workspace the event was actually attributed to." } workspace_folded: { type: boolean, description: "True when the client-sent workspace did not exist and the tier's workspace cap (Free = 1) forced the event into the org's earliest workspace instead of creating a new one — the recorded attribution then differs from what was sent." } + campaign_id: { type: string, nullable: true, description: "Campaign ID when the event carries a validated acceptance-campaign binding (#256/#257)." } UsageResponse: type: object description: | @@ -451,6 +645,8 @@ components: cost_usd: { type: number } baseline_usd: { type: number, nullable: true, description: "Counterfactual cost recorded for savings-share; null when none." } external_ref: { type: string, nullable: true, description: "Per-task attribution id (e.g. an Invarium task_id)." } + campaign_id: { type: string, nullable: true, description: "Acceptance campaign attribution ID when present." } + campaign_binding: { $ref: '#/components/schemas/CampaignBinding' } estimated: { type: boolean } source: { type: string } Org: diff --git a/server.json b/server.json index 5224186..cf8d525 100644 --- a/server.json +++ b/server.json @@ -173,6 +173,10 @@ "authority_manifest_custody": { "type": "string", "description": "Custody disclosure label for the referenced authority manifest (1f916 taxonomy); unknown custody renders as labeled uncertainty (#241)." + }, + "campaign_binding": { + "type": "object", + "description": "Hash-only acceptance-campaign cell/lane/config/checkpoint binding (#256/#257)." } }, "required": [ diff --git a/tests/test_cache_write.py b/tests/test_cache_write.py index 8f8a083..95b9e5a 100644 --- a/tests/test_cache_write.py +++ b/tests/test_cache_write.py @@ -16,7 +16,7 @@ def _org(tmp_path): def test_schema_v11_and_cache_write_is_stored_and_chained(tmp_path): conn, org = _org(tmp_path) - assert db.get_schema_version(conn) == 22 + assert db.get_schema_version(conn) == 23 cols = {r["name"] for r in conn.execute("PRAGMA table_info(usage_events)")} assert "cache_write_tokens" in cols diff --git a/tests/test_campaign_persistence.py b/tests/test_campaign_persistence.py new file mode 100644 index 0000000..1283878 --- /dev/null +++ b/tests/test_campaign_persistence.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import hashlib +import json + +import pytest + +from ledger_agent import campaigns, db, metering +from ledger_agent.server.api import campaign_json + + +def h(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def make_manifest(campaign_id="campaign:persist-1"): + return campaigns.build_manifest( + campaign_id=campaign_id, + planned_cells=["cell:a", "cell:b"], + provider_lanes=["fixture/lane"], + config_hash=h("config"), + fixture_hash=h("fixture"), + hard_stop_micros=1000, + continuation_allowed=True, + retry_policy="new_version_only", + action_intent_hash=h("intent"), + ) + + +def make_check(status="pass", campaign_id="campaign:persist-1", usage_event_ids=None): + if usage_event_ids is None: + usage_event_ids = ["evt_1"] if status == "pass" else [] + return campaigns.build_check( + campaign_id=campaign_id, cell_id="cell:a", lane="fixture/lane", + status=status, config_hash=h("config"), result_hash=h("result") if status == "pass" else None, + evidence_hashes=[h("evidence")] if status == "pass" else [], + usage_event_ids=usage_event_ids, + reason_code=None if status == "pass" else "skipped_by_fixture", + ) + + +def record_bound_usage(conn, org_id, campaign_id): + binding = campaigns.build_binding( + campaign_id=campaign_id, cell_id="cell:a", lane="fixture/lane", + config_hash=h("config"), + ) + result = metering.record_usage( + conn, org_id, provider="fixture", model="fixture", cost_usd=0.0001, + campaign_binding=binding, + ) + return result.event_id + + +def test_campaign_manifest_check_and_receipt_survive_restart(tmp_path): + path = str(tmp_path / "campaign.db") + conn = db.connect(path) + db.init_schema(conn) + org_id = db.create_org(conn, "campaign-org", tier="pro")["id"] + manifest = make_manifest() + stored = db.create_campaign(conn, org_id, manifest) + assert stored["id"] == manifest["campaign_id"] + assert stored["manifest_hash"] == manifest["manifest_hash"] + assert json.loads(stored["manifest_json"]) == manifest + event_id = record_bound_usage(conn, org_id, manifest["campaign_id"]) + check = make_check(campaign_id=manifest["campaign_id"], usage_event_ids=[event_id]) + db.record_campaign_check(conn, org_id, check) + receipt = campaigns.build_receipt( + manifest=manifest, checks=[check], framework_status="completed", + target_status="pass", evidence_status="complete", spent_micros=100, + ) + db.finalize_campaign(conn, org_id, receipt) + conn.close() + + conn = db.connect(path) + db.init_schema(conn) + loaded = db.get_campaign(conn, org_id, manifest["campaign_id"]) + assert loaded["framework_status"] == "completed" + assert loaded["target_status"] == "pass" + assert loaded["receipt_hash"] == receipt["receipt_hash"] + assert json.loads(loaded["receipt_json"]) == receipt + assert db.list_campaign_checks(conn, org_id, manifest["campaign_id"])[0]["check_hash"] == check["check_hash"] + assert db.campaign_spend_micros(conn, manifest["campaign_id"]) == 100 + assert db.get_schema_version(conn) == 23 + conn.close() + + +def test_campaign_create_and_finalize_are_idempotent_but_conflicts_fail_closed(tmp_path): + conn = db.connect(str(tmp_path / "idempotent.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "idempotent-org", tier="free")["id"] + manifest = make_manifest("campaign:idempotent") + assert db.create_campaign(conn, org_id, manifest)["manifest_hash"] == manifest["manifest_hash"] + assert db.create_campaign(conn, org_id, manifest)["manifest_hash"] == manifest["manifest_hash"] + changed = make_manifest("campaign:idempotent") + changed["fixture_hash"] = h("other-fixture") + changed["manifest_hash"] = campaigns.manifest_digest(changed) + with pytest.raises(ValueError, match="manifest conflict"): + db.create_campaign(conn, org_id, changed) + conn.close() + + +def test_missing_usage_cannot_finalize_a_verified_pass(tmp_path): + conn = db.connect(str(tmp_path / "missing-usage.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "missing-usage-org", tier="pro")["id"] + manifest = make_manifest("campaign:missing-usage") + db.create_campaign(conn, org_id, manifest) + check = make_check(campaign_id=manifest["campaign_id"], usage_event_ids=["evt_missing"]) + db.record_campaign_check(conn, org_id, check) + receipt = campaigns.build_receipt( + manifest=manifest, checks=[check], framework_status="completed", + target_status="pass", evidence_status="complete", + ) + with pytest.raises(ValueError, match="usage_missing"): + db.finalize_campaign(conn, org_id, receipt) + assert db.get_campaign(conn, org_id, manifest["campaign_id"])["receipt_hash"] is None + conn.close() + + +def test_campaign_scope_is_enforced_for_checks_and_reads(tmp_path): + conn = db.connect(str(tmp_path / "scope.db")) + db.init_schema(conn) + org_a = db.create_org(conn, "scope-a", tier="free")["id"] + org_b = db.create_org(conn, "scope-b", tier="free")["id"] + manifest = make_manifest("campaign:scope") + db.create_campaign(conn, org_a, manifest) + with pytest.raises(ValueError, match="campaign not found"): + db.record_campaign_check(conn, org_b, make_check()) + assert db.get_campaign(conn, org_b, manifest["campaign_id"]) is None + conn.close() + + +def test_campaign_json_is_public_safe_and_verifiable(tmp_path): + conn = db.connect(str(tmp_path / "summary.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "summary-org", tier="pro")["id"] + manifest = make_manifest("campaign:summary") + db.create_campaign(conn, org_id, manifest) + event_id = record_bound_usage(conn, org_id, manifest["campaign_id"]) + check = make_check(campaign_id=manifest["campaign_id"], usage_event_ids=[event_id]) + db.record_campaign_check(conn, org_id, check) + receipt = campaigns.build_receipt( + manifest=manifest, checks=[check], framework_status="completed", + target_status="pass", evidence_status="complete", spent_micros=100, + ) + db.finalize_campaign(conn, org_id, receipt) + summary = campaign_json(conn, org_id, manifest["campaign_id"]) + assert summary["manifest"]["manifest_hash"] == manifest["manifest_hash"] + assert summary["checks"][0]["check_hash"] == check["check_hash"] + assert summary["receipt"]["receipt_hash"] == receipt["receipt_hash"] + assert summary["verification"]["verified_pass"] is True + assert "prompt" not in json.dumps(summary) + conn.close() diff --git a/tests/test_campaign_usage_binding.py b/tests/test_campaign_usage_binding.py new file mode 100644 index 0000000..8bbe6d7 --- /dev/null +++ b/tests/test_campaign_usage_binding.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import hashlib + +import pytest + +from ledger_agent import campaigns, db, metering +from ledger_agent.server.api import audit_json + + +def h(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def make_manifest(): + return campaigns.build_manifest( + campaign_id="campaign:usage-1", planned_cells=["cell:a"], + provider_lanes=["fixture/lane"], config_hash=h("config-v1"), + fixture_hash=h("fixture"), hard_stop_micros=100, + runaway_guard_micros=100, continuation_allowed=True, + action_intent_hash=h("intent-v1"), + ) + + +def make_binding(attempt=1, config="config-v1", **kwargs): + return campaigns.build_binding( + campaign_id="campaign:usage-1", cell_id="cell:a", lane="fixture/lane", + config_hash=h(config), attempt=attempt, **kwargs, + ) + + +def test_campaign_binding_is_hash_bound_and_raw_payload_free(): + binding = make_binding() + assert campaigns.validate_binding(binding) == (True, []) + assert campaigns.binding_digest(binding) == binding["binding_hash"] + with pytest.raises(ValueError, match="forbidden"): + campaigns.build_binding( + campaign_id="campaign:usage-1", cell_id="cell:a", lane="fixture/lane", + config_hash=h("config-v1"), metadata={"prompt": "hidden"}, + ) + + +def test_campaign_usage_round_trips_through_chain_and_audit(tmp_path): + conn = db.connect(str(tmp_path / "usage.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "usage-org", tier="pro")["id"] + manifest = make_manifest() + db.create_campaign(conn, org_id, manifest) + binding = make_binding() + with db.immediate(conn): + result = metering.record_usage( + conn, org_id, provider="openai", model="fixture", task_type="benchmark", + external_ref="cell:a", input_tokens=2, output_tokens=1, cost_usd=0.00005, + campaign_binding=binding, commit=False, + ) + assert result.recorded is True + assert result.campaign_id == manifest["campaign_id"] + assert db.campaign_spend_micros(conn, manifest["campaign_id"]) == 50 + receipt = audit_json(conn, org_id, external_ref="cell:a") + assert receipt["events"][0]["campaign_binding"] == binding + assert receipt["verification"]["chain_ok"] is True + conn.close() + + +def test_campaign_budget_stop_writes_no_overrun_event(tmp_path): + conn = db.connect(str(tmp_path / "budget.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "budget-org", tier="pro")["id"] + manifest = make_manifest() + db.create_campaign(conn, org_id, manifest) + with db.immediate(conn): + metering.record_usage( + conn, org_id, provider="openai", model="fixture", task_type="benchmark", + input_tokens=1, output_tokens=1, cost_usd=0.0001, + campaign_binding=make_binding(), commit=False, + ) + before = conn.execute("SELECT COUNT(*) AS n FROM usage_events").fetchone()["n"] + with pytest.raises(ValueError, match="campaign budget guard"): + with db.immediate(conn): + metering.record_usage( + conn, org_id, provider="openai", model="fixture", task_type="benchmark", + input_tokens=1, output_tokens=1, cost_usd=0.000001, + campaign_binding=make_binding(), commit=False, + ) + after = conn.execute("SELECT COUNT(*) AS n FROM usage_events").fetchone()["n"] + assert after == before == 1 + assert db.campaign_spend_micros(conn, manifest["campaign_id"]) == 100 + conn.close() + + +def test_sdk_local_campaign_binding_uses_serialized_budget_guard(tmp_path): + from ledger_agent.client import Meter + meter = Meter(org="sdk-campaign-org", tier="pro", db_path=str(tmp_path / "sdk.db")) + manifest = campaigns.build_manifest( + campaign_id="campaign:sdk-1", planned_cells=["cell:a"], + provider_lanes=["fixture/lane"], config_hash=h("config-v1"), + fixture_hash=h("fixture"), hard_stop_micros=100, + action_intent_hash=h("intent-v1"), + ) + db.create_campaign(meter.conn, meter.org_id, manifest) + result = meter.track( + "openai", model="fixture", task_type="benchmark", input_tokens=1, + output_tokens=1, cost_usd=0.0001, + campaign_binding=campaigns.build_binding( + campaign_id="campaign:sdk-1", cell_id="cell:a", lane="fixture/lane", + config_hash=h("config-v1"), + ), + ) + assert result.recorded is True + assert db.campaign_spend_micros(meter.conn, "campaign:sdk-1") == 100 + meter.close() + + +def test_campaign_wrapper_durably_records_budget_stop(tmp_path): + conn = db.connect(str(tmp_path / "stop.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "stop-org", tier="pro")["id"] + manifest = make_manifest() + db.create_campaign(conn, org_id, manifest) + campaigns.record_usage( + conn, org_id, campaign_binding=make_binding(), provider="openai", + model="fixture", task_type="benchmark", input_tokens=1, + output_tokens=1, cost_usd=0.0001, + ) + with pytest.raises(campaigns.CampaignBudgetError): + campaigns.record_usage( + conn, org_id, campaign_binding=make_binding(), provider="openai", + model="fixture", task_type="benchmark", input_tokens=1, + output_tokens=1, cost_usd=0.000001, + ) + row = db.get_campaign(conn, org_id, manifest["campaign_id"]) + assert row["budget_status"] == "stopped" + assert row["stop_reason"] == "runaway_guard_exceeded" + assert db.campaign_usage_count(conn, manifest["campaign_id"]) == 1 + conn.close() diff --git a/tests/test_campaigns.py b/tests/test_campaigns.py new file mode 100644 index 0000000..85e4ac7 --- /dev/null +++ b/tests/test_campaigns.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import copy +import hashlib +import json + +import pytest + +from ledger_agent import campaigns + + +def h(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def manifest(**overrides): + value = campaigns.build_manifest( + campaign_id="campaign:acceptance-1", + planned_cells=["cell:a", "cell:b"], + provider_lanes=["openai/gpt-fixture"], + config_hash=h("config-v1"), + fixture_hash=h("fixture-v1"), + expected_spend_min_micros=100, + expected_spend_max_micros=500, + hard_stop_micros=600, + runaway_guard_micros=550, + retry_policy="new_version_only", + continuation_allowed=True, + action_intent_hash=h("intent-v1"), + target_commit_hash=h("target-commit"), + ) + value.update(overrides) + value["manifest_hash"] = campaigns.manifest_digest(value) + return value + + +def check(**overrides): + value = campaigns.build_check( + campaign_id="campaign:acceptance-1", + cell_id="cell:a", + lane="openai/gpt-fixture", + status="pass", + config_hash=h("config-v1"), + result_hash=h("result-a"), + evidence_hashes=[h("evidence-a")], + usage_event_ids=["evt_a"], + ) + value.update(overrides) + value["check_hash"] = campaigns.check_digest(value) + return value + + +def test_manifest_and_check_are_hash_bound_and_public_safe(): + m = manifest() + c = check() + assert campaigns.validate_manifest(m) == (True, []) + assert campaigns.validate_check(c) == (True, []) + assert campaigns.manifest_digest(m) == m["manifest_hash"] + assert campaigns.check_digest(c) == c["check_hash"] + raw = json.dumps({"manifest": m, "check": c}, sort_keys=True) + assert "prompt" not in raw + assert "memory_body" not in raw + assert "provider_payload" not in raw + assert "api_key" not in raw + + +def test_receipt_separates_framework_completion_from_target_failure(): + receipt = campaigns.build_receipt( + manifest=manifest(), + checks=[check(status="fail", result_hash=h("failed-result"))], + framework_status="completed", + target_status="fail", + evidence_status="complete", + finalization_status="complete", + spent_micros=300, + remaining_micros=300, + ) + assert receipt["framework_status"] == "completed" + assert receipt["target_status"] == "fail" + assert receipt["verification"]["valid"] is True + assert receipt["verification"]["verified_pass"] is False + assert receipt["counts"] == {"planned": 2, "executed": 1, "passed": 0, "failed": 1, "skipped": 0} + + +def test_all_skipped_is_inconclusive_not_pass(): + receipt = campaigns.build_receipt( + manifest=manifest(), + checks=[check(status="skip", result_hash=None, evidence_hashes=[], usage_event_ids=[])], + framework_status="completed", + evidence_status="complete", + finalization_status="complete", + spent_micros=0, + remaining_micros=600, + ) + assert receipt["target_status"] == "inconclusive" + assert receipt["verification"]["verified_pass"] is False + assert "no_executed_checks" in receipt["verification"]["reasons"] + + +def test_framework_error_before_checks_is_not_run(): + receipt = campaigns.build_receipt( + manifest=manifest(), checks=[], framework_status="error", + finalization_status="failed", finalization_reason="runner_error", + ) + assert receipt["target_status"] == "not_run" + assert receipt["verification"]["verified_pass"] is False + assert "framework_not_completed" in receipt["verification"]["reasons"] + + +def test_failed_finalization_downgrades_a_target_pass(): + receipt = campaigns.build_receipt( + manifest=manifest(), checks=[check()], framework_status="completed", + target_status="pass", evidence_status="complete", + finalization_status="failed", finalization_reason="evidence_write_failed", + spent_micros=200, remaining_micros=400, + ) + assert receipt["target_status"] == "pass" + assert receipt["verification"]["valid"] is True + assert receipt["verification"]["verified_pass"] is False + assert "finalization_failed" in receipt["verification"]["reasons"] + + +def test_validation_rejects_tampering_unknown_fields_and_raw_payloads(): + tampered = manifest() + tampered["planned_cells"] = ["cell:forged"] + assert campaigns.validate_manifest(tampered)[0] is False + unknown = manifest() + unknown["unexpected"] = "nope" + assert "unknown:unexpected" in campaigns.validate_manifest(unknown)[1] + leaked = manifest() + leaked["prompt"] = "do not persist" + valid, errors = campaigns.validate_manifest(leaked) + assert not valid + assert "forbidden_field:prompt" in errors + + +def test_correction_attempt_requires_new_lineage_and_config(): + prior = check(status="fail", attempt=1, result_hash=h("failed")) + resumed = check( + cell_id="cell:a", status="pass", attempt=2, + continuation=True, parent_attempt=1, + config_hash=h("config-v2"), action_intent_hash=h("intent-v2"), + result_hash=h("repaired"), usage_event_ids=["evt_b"], + ) + assert campaigns.validate_check(prior)[0] + assert campaigns.validate_check(resumed)[0] + assert campaigns.validate_attempt_lineage([prior, resumed]) == (True, []) + invalid = copy.deepcopy(resumed) + invalid["config_hash"] = prior["config_hash"] + invalid["check_hash"] = campaigns.check_digest(invalid) + assert "continuation_config_unchanged" in campaigns.validate_attempt_lineage([prior, invalid])[1] + + +def test_budget_admission_is_fail_closed_at_runaway_and_hard_stop(): + assert campaigns.admit_spend(manifest(), spent_micros=500, proposed_micros=50)["allowed"] is True + runaway = campaigns.admit_spend(manifest(), spent_micros=500, proposed_micros=51) + assert runaway == {"allowed": False, "reason": "runaway_guard_exceeded", "remaining_micros": 50} + hard = campaigns.admit_spend( + manifest(runaway_guard_micros=None), spent_micros=500, proposed_micros=101, + ) + assert hard == {"allowed": False, "reason": "hard_stop_exceeded", "remaining_micros": 100} + with pytest.raises(ValueError, match="non-negative"): + campaigns.admit_spend(manifest(), spent_micros=0, proposed_micros=-1) + + +def test_verified_pass_rejects_forged_target_status_and_counts(): + failed = check(status="fail", result_hash=h("failed")) + receipt = campaigns.build_receipt( + manifest=manifest(), checks=[failed], framework_status="completed", + target_status="fail", evidence_status="complete", + finalization_status="complete", spent_micros=1, + ) + forged = copy.deepcopy(receipt) + forged["target_status"] = "pass" + forged["counts"]["failed"] = 0 + forged["counts"]["passed"] = 1 + forged["receipt_hash"] = campaigns.receipt_digest(forged) + verification = campaigns.verify_campaign_receipt( + forged, manifest=manifest(), checks=[failed], + ) + assert verification["valid"] is False + assert verification["verified_pass"] is False + assert "target_status_mismatch" in verification["reasons"] + assert "counts_mismatch" in verification["reasons"] + + +def test_continuation_requires_manifest_permission(): + prior = check(status="fail", result_hash=h("failed")) + resumed = check( + cell_id="cell:a", status="pass", attempt=2, continuation=True, + parent_attempt=1, config_hash=h("config-v2"), + action_intent_hash=h("intent-v2"), result_hash=h("repaired"), + usage_event_ids=["evt_b"], + ) + with pytest.raises(ValueError, match="continuation is not allowed"): + campaigns.build_receipt( + manifest=manifest(continuation_allowed=False), + checks=[prior, resumed], framework_status="completed", + evidence_status="complete", + ) diff --git a/tests/test_external_ref.py b/tests/test_external_ref.py index f2f08c4..8d0b572 100644 --- a/tests/test_external_ref.py +++ b/tests/test_external_ref.py @@ -34,7 +34,7 @@ def _meter(conn, org_id, cost, external_ref=None, baseline=None, ts=None): # --- schema / migration ----------------------------------------------------- def test_schema_is_v10_and_has_external_ref(tmp_path): conn, _ = _org(tmp_path) - assert db.get_schema_version(conn) == 22 + assert db.get_schema_version(conn) == 23 cols = {r["name"] for r in conn.execute("PRAGMA table_info(usage_events)")} assert "external_ref" in cols @@ -48,7 +48,7 @@ def test_external_ref_index_exists(tmp_path): def test_init_schema_is_idempotent(tmp_path): conn, _ = _org(tmp_path) db.init_schema(conn) # second run must not raise or double-add - assert db.get_schema_version(conn) == 22 + assert db.get_schema_version(conn) == 23 # --- round-trip + join ------------------------------------------------------ diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 2cf382a..cd6a66b 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -327,3 +327,28 @@ def test_stdio_protocol_roundtrip(tmp_path): proc.stdin.close() proc.terminate() proc.wait(timeout=10) + + +def test_record_campaign_binding_round_trip(tmp_path, monkeypatch): + from ledger_agent import campaigns, db + import hashlib + digest = lambda value: hashlib.sha256(value.encode()).hexdigest() + meter = _meter(tmp_path, monkeypatch) + manifest = campaigns.build_manifest( + campaign_id="campaign:mcp-1", planned_cells=["cell:mcp"], + provider_lanes=["fixture/lane"], config_hash=digest("config"), + fixture_hash=digest("fixture"), hard_stop_micros=1000, + ) + db.create_campaign(meter.conn, meter.org_id, manifest) + binding = campaigns.build_binding( + campaign_id="campaign:mcp-1", cell_id="cell:mcp", lane="fixture/lane", + config_hash=digest("config"), + ) + text, result = _call("ledger_record", { + "provider": "openai", "model": "fixture", "task_type": "benchmark", + "input_tokens": 1, "output_tokens": 1, "cost_usd": 0.0001, + "external_ref": "cell:mcp", "campaign_binding": binding, + }, meter) + assert result.get("isError") is not True + assert text["recorded"] is True + assert text["campaign_id"] == "campaign:mcp-1" diff --git a/tests/test_schema_version.py b/tests/test_schema_version.py index 0cd2ca8..e0d4921 100644 --- a/tests/test_schema_version.py +++ b/tests/test_schema_version.py @@ -31,6 +31,7 @@ "20": {"governance_cost_json", "governance_cost_hash"}, "21": {"behavior_snapshot_json", "behavior_snapshot_hash"}, "22": {"authority_manifest_custody"}, + "23": {"acceptance_campaigns", "acceptance_checks", "campaign_id", "campaign_binding_json", "campaign_binding_hash"}, } @@ -80,8 +81,8 @@ def test_schema_docs_match_runtime_migration_contract(self): docs_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "schema.md") with open(docs_path, encoding="utf-8") as handle: docs = handle.read() - self.assertEqual(db.SCHEMA_VERSION, 22) - self.assertIn("SCHEMA_VERSION=22", docs) + self.assertEqual(db.SCHEMA_VERSION, 23) + self.assertIn("SCHEMA_VERSION=23", docs) table_rows = {} for line in docs.splitlines(): @@ -102,6 +103,7 @@ def test_schema_docs_match_runtime_migration_contract(self): "served_memory_provenance_hash", "action_receipt_hash", "resource_constraints_version", "resource_constraints_hash", "prebind_json", "prebind_hash", "reconciliation_note", + "campaign_id", "campaign_binding_json", "campaign_binding_hash", } <= columns) conn.close() diff --git a/tests/test_server.py b/tests/test_server.py index 00a1466..109d401 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 """Integration smoke test: boot the HTTP server on an ephemeral port.""" +import hashlib import io import json import os @@ -13,7 +14,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from ledger_agent import Meter, db, demo, metering +from ledger_agent import Meter, campaigns, db, demo, metering from ledger_agent.client import LedgerAuthError, LedgerError from ledger_agent.config import DEFAULT_CONFIG from ledger_agent.server import app @@ -45,9 +46,11 @@ def tearDownClass(cls): except OSError: pass - def _get(self, path): + def _get(self, path, token=None): url = f"http://127.0.0.1:{self.port}{path}" - with urllib.request.urlopen(url, timeout=5) as r: + headers = {"Authorization": f"Bearer {token}"} if token else {} + req = urllib.request.Request(url, headers=headers) + with urllib.request.urlopen(req, timeout=5) as r: return r.status, r.read().decode() def _post(self, path, payload, token=None): @@ -293,6 +296,86 @@ def read(self): self.assertTrue(captured["ua"].startswith("ledger-agent")) self.assertNotIn("urllib", captured["ua"].lower()) + def test_campaign_api_round_trip_and_scope(self): + digest = lambda value: hashlib.sha256(value.encode()).hexdigest() + campaign_id = "campaign:http-256-257" + manifest = campaigns.build_manifest( + campaign_id=campaign_id, planned_cells=["cell:http"], + provider_lanes=["fixture/lane"], config_hash=digest("config"), + fixture_hash=digest("fixture"), hard_stop_micros=1000, + action_intent_hash=digest("intent"), continuation_allowed=True, + ) + status, body = self._post("/v1/campaigns", {"manifest": manifest}, token=self.key) + self.assertEqual(status, 201) + self.assertEqual(body["campaign_id"], campaign_id) + binding = campaigns.build_binding( + campaign_id=campaign_id, cell_id="cell:http", lane="fixture/lane", + config_hash=digest("config"), + ) + status, usage = self._post("/v1/usage", { + "provider": "openai", "model": "fixture", "task_type": "benchmark", + "input_tokens": 1, "output_tokens": 1, "cost_usd": 0.0001, + "external_ref": "cell:http", "campaign_binding": binding, + }, token=self.key) + self.assertEqual(status, 200) + check = campaigns.build_check( + campaign_id=campaign_id, cell_id="cell:http", lane="fixture/lane", + status="pass", config_hash=digest("config"), result_hash=digest("result"), + evidence_hashes=[digest("evidence")], usage_event_ids=[usage["event_id"]], + ) + status, body = self._post("/v1/campaigns/checks", {"check": check}, token=self.key) + self.assertEqual(status, 201) + receipt = campaigns.build_receipt( + manifest=manifest, checks=[check], framework_status="completed", + target_status="pass", evidence_status="complete", spent_micros=100, + ) + status, body = self._post("/v1/campaigns/finalize", {"receipt": receipt}, token=self.key) + self.assertEqual(status, 200) + status, body = self._get( + f"/v1/campaigns?campaign_id={campaign_id}", token=self.key) + self.assertEqual(status, 200) + self.assertTrue(json.loads(body)["verification"]["verified_pass"]) + + + + status, body = self._get( + f"/api/audit?campaign_id={campaign_id}", token=self.key) + self.assertEqual(status, 200) + self.assertEqual(json.loads(body)["campaign_id"], campaign_id) + + status, body = self._get("/v1/usage/export.json", token=self.key) + self.assertEqual(status, 200) + exported = json.loads(body)["events"] + bound = next(item for item in exported if item["campaign_id"] == campaign_id) + self.assertEqual(bound["campaign_binding"]["campaign_id"], campaign_id) + + def test_campaign_budget_stop_is_durable(self): + digest = lambda value: hashlib.sha256(value.encode()).hexdigest() + campaign_id = "campaign:http-budget-stop" + manifest = campaigns.build_manifest( + campaign_id=campaign_id, planned_cells=["cell:stop"], + provider_lanes=["fixture/lane"], config_hash=digest("config-stop"), + fixture_hash=digest("fixture-stop"), hard_stop_micros=1, + ) + status, _ = self._post("/v1/campaigns", {"manifest": manifest}, token=self.key) + self.assertEqual(status, 201) + binding = campaigns.build_binding( + campaign_id=campaign_id, cell_id="cell:stop", lane="fixture/lane", + config_hash=digest("config-stop"), + ) + status, body = self._post("/v1/usage", { + "provider": "openai", "model": "fixture", "cost_usd": 0.000002, + "campaign_binding": binding, + }, token=self.key) + self.assertEqual(status, 402) + self.assertEqual(body["campaign_id"], campaign_id) + status, body = self._get( + f"/v1/campaigns?campaign_id={campaign_id}", token=self.key) + self.assertEqual(status, 200) + projection = json.loads(body) + self.assertEqual(projection["budget_status"], "stopped") + self.assertEqual(projection["usage_events"], 0) + self.assertEqual(projection["verification"]["verified_pass"], False) class TestIngestQuota(unittest.TestCase): """Free org past its cap with hard-blocking on → 402."""