diff --git a/README.md b/README.md index 049fa27..1de061f 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,8 @@ python -m venv .venv The lifecycle foundation supports one-pass inbox processing, resumable package checkpoints and a deterministic generic CaseBuilder. An async LLM gateway and CaseInterpreter -provide interpretation and bounded initial hypothesis generation. Investigation tools, diagnosis, +provide interpretation and bounded initial hypothesis generation, with durable +per-hypothesis budget accounting. Investigation tools, diagnosis, Change Requests and a continuous analysis worker remain future milestones. Expected-behavior resolution and the read-only provider contract: [expected-behavior-v1.md](docs/expected-behavior-v1.md). @@ -31,3 +32,6 @@ Configurable LLM transport, failover and durable retries: [llm-gateway-v1.md](do The first domain LLM phase and its source-reference contracts: [case-interpretation-v1.md](docs/case-interpretation-v1.md). Initial hypotheses, revision-wide creation limits and equal budget allocations: [hypothesis-builder-v1.md](docs/hypothesis-builder-v1.md), with [analysis configuration](config/analysis.example.toml). + +Durable investigation budget accounting, configurable costs/rewards and stopping +rules: [budget-manager-v1.md](docs/budget-manager-v1.md), with [budget configuration](config/budget.example.toml). diff --git a/config/budget.example.toml b/config/budget.example.toml new file mode 100644 index 0000000..45b3167 --- /dev/null +++ b/config/budget.example.toml @@ -0,0 +1,15 @@ +# Copy to ignored config/budget.local.toml. Frozen per analysis revision. +# Integer investigation units, independent of model tokens and money. +[budget] +default_action_cost = 10 +max_reward_bps = 8000 # At most 80% refund; must remain below 10000 +new_evidence_score = 100 # Deterministic score for unseen nonempty JSON +low_information_threshold = 20 # min(novelty, semantic) <= threshold is low gain +max_low_information_steps = 3 # Consecutive settled low-gain actions + +# Examples for future tools; these do not implement or authorize tool execution. +[budget.action_costs] +"source.find_text" = 10 +"source.read_context" = 5 +"database.lookup" = 10 +"research.lookup" = 20 diff --git a/docs/budget-manager-v1.md b/docs/budget-manager-v1.md new file mode 100644 index 0000000..00861c7 --- /dev/null +++ b/docs/budget-manager-v1.md @@ -0,0 +1,192 @@ +# Budget Manager v1 + +This milestone implements deterministic accounting for the initial hypotheses. +It does not run investigation tools, invoke an LLM for evidence assessment, build +diagnoses, or start an investigation loop. `docs/architecture.md` remains the +authoritative specification, particularly sections 16, 18, 19, 25 and 31. + +## Configuration and deployment tuning + +Copy `config/budget.example.toml` to the ignored `config/budget.local.toml` and +load it with `BudgetConfig.load(path)`. This is a separate config from the initial +hypothesis allocations in `config/analysis.example.toml`. These are conservative +starting values, not format constants or empirically calibrated scoring rules: + +| Setting | Default | Meaning | +| --- | --- | --- | +| `default_action_cost` | 10 | Positive integer cost for an action without an override | +| `action_costs` | empty in Python | Per-action positive integer costs; example TOML includes future tool names | +| `max_reward_bps` | 8000 | Maximum refund in basis points (8000 = 80%); range 0–9999 | +| `new_evidence_score` | 100 | Deterministic score for unseen, nonempty evidence; range 0–100 | +| `low_information_threshold` | 20 | Information scores at or below this value count as low gain | +| `max_low_information_steps` | 3 | Consecutive low-gain settlements before abandonment | + +The example config assigns `source.read_context` a cost of 5, `source.find_text` +and `database.lookup` 10, and `research.lookup` 20. These names describe accounting +overrides only: they neither implement tools nor authorize arbitrary operations. +The future Tool Router must independently validate every action. + +The complete effective policy is frozen in the first budget event of an analysis +revision. Changing the config cannot alter an existing policy, refill balances, +undo expenses, or reopen terminal accounts. An existing snapshot or identical +operation can be replayed with a different config; new operations reject with +`BUDGET_POLICY_CHANGED`. Use the frozen config to continue an existing revision; +new revisions can adopt the tuned config. Every event includes the effective +policy, so historical accounting remains reproducible after deployments change. + +Tune costs, refund ceiling, novelty score, threshold and consecutive-step limit +using actual result quality and cost data. The initial hypothesis allocation +remains separately configurable. No automatic online adaptation is implemented. +The invariants below deliberately cannot be disabled by configuration. + +## Calculation and invariants + +The unit is `investigation_units`, unrelated to LLM token counts, money or +provider retry counts. The LLM can supply only a bounded semantic score and a +reason through the future trusted controller. It cannot choose actual costs, +refunds, remaining balances, or budget policy. + +For each completed action: + +```text +information = min(deterministic_novelty, semantic_score) +reward = floor(cost * max_reward_bps * information / 1_000_000) +reward = min(reward, cost - 1) +remaining = initial - total_spent + total_rewarded +``` + +Both scores range from 0 to 100. Full configured reimbursement requires both +scores to reach 100. A score is an ordinal accounting input, not a confidence +percentage. For example, cost 10 and both scores 100 produce reward 8 under the +default policy; a semantic score of 50 produces reward 4. Even the most generous +allowed policy always has `0 <= reward < cost`: each settled action consumes at +least one whole unit. Thus a hypothesis cannot fund infinite investigation. + +Novelty v1 hashes canonical JSON evidence (sorted object keys, preserved array +order). New nonempty evidence gets `new_evidence_score`; empty or already seen +evidence gets zero. Deduplication spans hypotheses and prior analysis revisions +in the same case package. Fresh revision budgets do not make old evidence novel. +Transport metadata and query IDs should be kept outside the evidence payload. +This detects exact normalized content reuse, not semantic equivalence. Future +EvidenceStore/scoring work may provide richer deterministic novelty signals; +this milestone does not pretend to assess factual correctness or entailment. + +The controller supplies JSON evidence and safe references to existing package +sources. Source hashes, evidence, computed novelty, semantic assessment reason, +cost and reward are all retained. Accounting checks references and integrity, not +whether an assessment is scientifically justified. Tests simulate assessments; +no live model or paid provider is used. + +## Durable controller API + +```python +from viano_bug_analysis.budget import BudgetConfig, BudgetManager + +budget = BudgetManager(store, BudgetConfig.load("config/budget.local.toml")) +budget.initialize(case_id, package_id) +budget.reserve(case_id, package_id, "H1", "query-0001", "source.read_context") +# A future controller invokes the allowed read-only tool after reservation. +budget.record_result( + case_id, package_id, "H1", "query-0001", + {"finding": "Example stored observation"}, source_refs=[source_ref], +) +# In the next iteration the controller validates a bounded semantic assessment. +result = budget.settle( + case_id, package_id, "H1", "query-0001", + semantic_score=50, reason="Example assessment of this stored result", +) +``` + +- `initialize` verifies a completed current-revision hypothesis batch, registry, + allocations, context and original successful LLM response. It creates equal + balances from the controller-owned allocations; it cannot allocate more IDs. +- `reserve` durably debits the configured cost before external work. Unknown + hypotheses, blank IDs, unaffordable costs and terminal accounts are rejected. + Only one unsettled action per hypothesis is allowed. Different hypotheses can + have their own pending actions; action IDs are scoped to revision/hypothesis. +- `record_result` persists evidence, source references and their hashes. It + changes the action to `awaiting_assessment`, with no immediate reimbursement. +- `settle` accepts only an integer score 0–100 and a nonblank reason. The score is + bound to the already stored action result, and the manager computes the reward. +- `fail` closes a reserved action without reimbursement. Failed or uncertain + executions still cost budget and count as low information. A pending action + survives restart and blocks another reservation; the manager never silently + cancels it or retries external work. The controller decides whether the same + external operation can safely resume or must be closed with `fail`. +- `reject` marks an account terminal without refunding or deleting its history; + a pending action must first be settled or failed. +- `snapshot` validates/replays the ledger without changing budgets. + +Identical repeated operations return the current package/state with `reused=True`. +Changed inputs under the same operation/hypothesis/action identity are rejected. +Repeated initialization cannot refill balances, including after index-write +failure, service restart, transport under another filename or config changes. +`BudgetResult` returns the package path, revision, event reference and current +state. Its `investigable` property lists nonterminal accounts without pending +actions; `ready_for_evaluation` is true only when all accounts are terminal and +no action awaits completion. An empty investigable list alone is not sufficient +while asynchronous actions are pending. + +## Status and lifecycle boundaries + +Budget account status is separate from qualitative hypothesis confidence and +the immutable hypothesis birth/creation registry. The manager owns `new`, +`investigating`, `budget_exhausted`, `abandoned_low_information_gain` and the +controller-requested `rejected` budget eligibility marker. It does not infer +`supported`/`plausible`, diagnose causes, create replacements or restore slots. +The birth document's `budget_accounting = not_started` describes generation +time and remains immutable; `artifacts.budget_ledger` is authoritative for the +subsequent accounting state. + +The cheapest cost in the configured override/default table is the affordability +floor. Immediately after initialization and after settlement, a balance below +that floor becomes `budget_exhausted`. Pending actions are not marked exhausted +before their possible reward is evaluated. If affordable but the consecutive +low-gain limit is reached, the account becomes `abandoned_low_information_gain`. +A higher information score resets that counter. Exhaustion takes precedence if +both conditions occur. Request-specific insufficient funds reject that request; +a cheaper permitted action can still be considered. + +Mutations are allowed only in `GENERATING_HYPOTHESES` or +`INVESTIGATING_HYPOTHESES`. Budget initialization does not start investigation +or transition the case. An exhausted set exposes eligibility for final +evaluation, but does not create a diagnosis or route the case to Human Review. +Those decisions belong to the future controller/evaluator. Reopening retains +old artifacts and starts accounting from the new revision's accepted allocations. + +## Persistence, validation and limits + +Each operation appends one immutable event/snapshot: + +```text +analysis/budgets/0001/events/00000001.json # policy + initial balances +analysis/budgets/0001/events/00000002.json # reservation +analysis/budgets/0001/events/00000003.json # result +analysis/budgets/0001/events/00000004.json # assessment + computed reimbursement +``` + +The manifest's `artifacts.budget_ledger` points to the latest event. Each event +has format/version, identities, original hypothesis/registry hashes, sequence, +parent/hash, effective policy, request and fully derived accounting state. The +existing CheckpointStore publishes event, checkpoint and pointer atomically, +increases `package_generation`, preserves unknown files and checks the prepared +generation. A per-case budget lock prevents duplicate dispatch; short existing +case/worker locks protect publication. No lock waits on an external tool or LLM. + +Replay verifies event schemas, checkpoint bindings, immutable origins, lineage, +source hashes and every computed state. Pointer deletion, changed allocations, +costs, rewards, balances or broken lineage are controlled errors, never a reset. +Checks detect inconsistent artifacts; hashes are not digital signatures and do +not authenticate a maliciously rewritten complete case history. + +`budget-policy-v1`, `budget-request-v1` and `budget-event-v1` JSON schemas are +bundled with the Python package. Byte/event limits in `BudgetLimits` are adjustable +operational safeguards, independent of format v1: defaults are 2 MiB per event, +1 MiB per evidence payload and 10,000 events across retained budget revisions. +Limits reject explicitly rather than truncating history or dropping expenses. +Snapshots contain cumulative state, so long-lived cases may need larger limits; +incremental replay optimizations are a later performance task. + +The CaseAssembler excludes checkpoint-declared budget events from raw case input +identity, preserving replay of earlier phases. Undeclared/unknown files remain +preserved and part of raw input inventory as before. diff --git a/src/viano_bug_analysis/budget/__init__.py b/src/viano_bug_analysis/budget/__init__.py new file mode 100644 index 0000000..fa996d6 --- /dev/null +++ b/src/viano_bug_analysis/budget/__init__.py @@ -0,0 +1,6 @@ +"""Deterministic, durable investigation budget accounting.""" + +from .config import BudgetConfig, BudgetError, BudgetLimits +from .manager import BudgetManager, BudgetResult + +__all__ = ["BudgetConfig", "BudgetError", "BudgetLimits", "BudgetManager", "BudgetResult"] diff --git a/src/viano_bug_analysis/budget/config.py b/src/viano_bug_analysis/budget/config.py new file mode 100644 index 0000000..1804c4e --- /dev/null +++ b/src/viano_bug_analysis/budget/config.py @@ -0,0 +1,69 @@ +import tomllib +from dataclasses import asdict, dataclass, field +from pathlib import Path + + +class BudgetError(Exception): + """Controlled accounting rejection without case or provider contents.""" + + def __init__(self, code): + self.code = code + super().__init__(code) + + +def integer(value, minimum, maximum=None): + return (type(value) is int and value >= minimum + and (maximum is None or value <= maximum)) + + +@dataclass(frozen=True) +class BudgetConfig: + default_action_cost: int = 10 + action_costs: dict[str, int] = field(default_factory=dict) + max_reward_bps: int = 8000 + new_evidence_score: int = 100 + low_information_threshold: int = 20 + max_low_information_steps: int = 3 + + def __post_init__(self): + if (not integer(self.default_action_cost, 1) + or not integer(self.max_reward_bps, 0, 9999) + or not integer(self.new_evidence_score, 0, 100) + or not integer(self.low_information_threshold, 0, 100) + or not integer(self.max_low_information_steps, 1) + or type(self.action_costs) is not dict + or any(not isinstance(name, str) or not name.strip() or not integer(cost, 1) + for name, cost in self.action_costs.items())): + raise BudgetError("BUDGET_CONFIG") + # Detach the policy from caller-owned dictionaries. + object.__setattr__(self, "action_costs", dict(self.action_costs)) + + @classmethod + def load(cls, path): + try: + with Path(path).open("rb") as stream: + document = tomllib.load(stream) + if set(document) != {"budget"}: + raise BudgetError("BUDGET_CONFIG") + return cls(**document["budget"]) + except (OSError, ValueError, TypeError, KeyError): + raise BudgetError("BUDGET_CONFIG") from None + + def policy(self): + # Revalidate also after accidental mutation of action_costs. + BudgetConfig(**asdict(self)) + return {"format": "viano-budget-policy", "format_version": 1, + "unit": "investigation_units", **asdict(self)} + + +@dataclass(frozen=True) +class BudgetLimits: + """Operational safeguards, not immutable format boundaries.""" + + max_event_bytes: int = 2 * 1024 * 1024 + max_result_bytes: int = 1024 * 1024 + max_events: int = 10000 + + def __post_init__(self): + if any(not integer(value, 1) for value in asdict(self).values()): + raise BudgetError("BUDGET_LIMIT_CONFIG") diff --git a/src/viano_bug_analysis/budget/engine.py b/src/viano_bug_analysis/budget/engine.py new file mode 100644 index 0000000..77871e4 --- /dev/null +++ b/src/viano_bug_analysis/budget/engine.py @@ -0,0 +1,88 @@ +"""Pure replayable rules; no tools, network, model calls or package writes.""" + +import copy + +from .config import BudgetError +from .scoring import evidence_digest, reward_for + +TERMINAL = {"rejected", "budget_exhausted", "abandoned_low_information_gain"} + + +def initial_state(registry, policy, prior_seen=None): + state = {"accounts": {}, "seen_evidence": list(prior_seen or [])} + minimum = min([policy["default_action_cost"], *policy["action_costs"].values()]) + for creation in registry["creations"]: + initial = creation["allocated_budget"] + state["accounts"][creation["id"]] = { + "initial": initial, "remaining": initial, "spent": 0, "rewarded": 0, + "low_information_steps": 0, + "status": "budget_exhausted" if initial < minimum else "new", "actions": {}, + } + return state + + +def apply(state, request, policy): + state = copy.deepcopy(state) + account = state["accounts"].get(request["hypothesis_id"]) + if account is None: + raise BudgetError("BUDGET_HYPOTHESIS_UNKNOWN") + operation = request["operation"] + actions = account["actions"] + pending = any(action["status"] != "settled" for action in actions.values()) + if operation == "reserve": + if account["status"] in TERMINAL: + raise BudgetError("BUDGET_TERMINAL") + if pending: + raise BudgetError("BUDGET_PENDING_ACTION") + if request["action_id"] in actions: + raise BudgetError("BUDGET_ACTION_CONFLICT") + cost = policy["action_costs"].get(request["action"], policy["default_action_cost"]) + if cost > account["remaining"]: + raise BudgetError("BUDGET_INSUFFICIENT") + account["remaining"] -= cost + account["spent"] += cost + account["status"] = "investigating" + actions[request["action_id"]] = {"action": request["action"], "cost": cost, "status": "reserved"} + elif operation == "terminate": + if pending: + raise BudgetError("BUDGET_PENDING_ACTION") + if account["status"] in TERMINAL: + raise BudgetError("BUDGET_TERMINAL") + account["status"] = "rejected" + account["termination_reason"] = request["reason"] + else: + action = actions.get(request["action_id"]) + if action is None: + raise BudgetError("BUDGET_ACTION_UNKNOWN") + if operation == "result": + if action["status"] != "reserved": + raise BudgetError("BUDGET_ACTION_CONFLICT") + digest = evidence_digest(request["evidence"]) + novelty = (policy["new_evidence_score"] if request["evidence"] + and digest not in state["seen_evidence"] else 0) + if digest not in state["seen_evidence"]: + state["seen_evidence"].append(digest) + action.update(status="awaiting_assessment", evidence_sha256=digest, + novelty_score=novelty, result_ref=request["result_ref"]) + elif operation in {"settle", "fail"}: + if operation == "settle" and action["status"] != "awaiting_assessment": + raise BudgetError("BUDGET_RESULT_REQUIRED") + if operation == "fail" and action["status"] != "reserved": + raise BudgetError("BUDGET_ACTION_CONFLICT") + novelty = action.get("novelty_score", 0) + score = request["semantic_score"] if operation == "settle" else 0 + reward, information = reward_for(action["cost"], novelty, score, policy) + action.update(status="settled", reward=reward, information_score=information, + semantic_score=score, assessment_reason=request["reason"], failed=operation == "fail") + account["remaining"] += reward + account["rewarded"] += reward + account["low_information_steps"] = (account["low_information_steps"] + 1 + if information <= policy["low_information_threshold"] else 0) + minimum = min([policy["default_action_cost"], *policy["action_costs"].values()]) + if account["remaining"] < minimum: + account["status"] = "budget_exhausted" + elif account["low_information_steps"] >= policy["max_low_information_steps"]: + account["status"] = "abandoned_low_information_gain" + else: + raise BudgetError("BUDGET_OPERATION") + return state diff --git a/src/viano_bug_analysis/budget/manager.py b/src/viano_bug_analysis/budget/manager.py new file mode 100644 index 0000000..97f6912 --- /dev/null +++ b/src/viano_bug_analysis/budget/manager.py @@ -0,0 +1,282 @@ +import copy +import hashlib +from dataclasses import dataclass +from pathlib import Path + +from viano_bug_analysis.analysis.hypothesis_contracts import validate_generation +from viano_bug_analysis.analysis.hypothesis_config import HypothesisError +from viano_bug_analysis.case.contracts import read_json, require_file, validate +from viano_bug_analysis.case.errors import CaseBuildError +from viano_bug_analysis.lifecycle import LifecycleError, State +from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain +from viano_bug_analysis.lifecycle.errors import LockBusy +from viano_bug_analysis.lifecycle.locking import FileLock +from viano_bug_analysis.lifecycle.storage import atomic_json, json_bytes, key +from viano_bug_analysis.package import PackageReader, PackageReadError +from viano_bug_analysis.package.validator import ARCHIVE_ERRORS + +from .config import BudgetConfig, BudgetError, BudgetLimits, integer +from .engine import apply, initial_state + + +def digest(value): + return hashlib.sha256(json_bytes(value)).hexdigest() + + +@dataclass(frozen=True) +class BudgetResult: + package: Path + analysis_revision: int + event_ref: str + state: dict + reused: bool = False + + @property + def investigable(self): + return [name for name, account in self.state["accounts"].items() + if account["status"] not in {"rejected", "budget_exhausted", "abandoned_low_information_gain"} + and all(action["status"] == "settled" for action in account["actions"].values())] + + @property + def ready_for_evaluation(self): + return all(account["status"] in {"rejected", "budget_exhausted", "abandoned_low_information_gain"} + and all(action["status"] == "settled" for action in account["actions"].values()) + for account in self.state["accounts"].values()) + + +class BudgetManager: + """Persist each accepted accounting operation atomically in the case package. + + Callers are the future trusted controller, never direct model tool calls. + Reservation is a durable debit; uncertain or failed actions are not refunded. + """ + + def __init__(self, store, config=None, limits=None): + self.store = store + self.config = config or BudgetConfig() + self.limits = limits or BudgetLimits() + + def initialize(self, case_id, package_id): + return self._execute(case_id, package_id, {"operation": "initialize"}) + + def snapshot(self, case_id, package_id): + return self._execute(case_id, package_id, None) + + def reserve(self, case_id, package_id, hypothesis_id, action_id, action): + return self._execute(case_id, package_id, {"operation": "reserve", "hypothesis_id": hypothesis_id, + "action_id": action_id, "action": action}) + + def record_result(self, case_id, package_id, hypothesis_id, action_id, evidence, *, source_refs): + return self._execute(case_id, package_id, {"operation": "result", "hypothesis_id": hypothesis_id, + "action_id": action_id, "evidence": evidence, "source_refs": source_refs}) + + def settle(self, case_id, package_id, hypothesis_id, action_id, *, semantic_score, reason): + return self._execute(case_id, package_id, {"operation": "settle", "hypothesis_id": hypothesis_id, + "action_id": action_id, "semantic_score": semantic_score, "reason": reason}) + + def fail(self, case_id, package_id, hypothesis_id, action_id, *, reason): + return self._execute(case_id, package_id, {"operation": "fail", "hypothesis_id": hypothesis_id, + "action_id": action_id, "reason": reason}) + + def reject(self, case_id, package_id, hypothesis_id, *, reason): + return self._execute(case_id, package_id, {"operation": "terminate", "hypothesis_id": hypothesis_id, + "action_id": "terminal", "reason": reason}) + + def _execute(self, case_id, package_id, request): + try: + with FileLock(self.store.workspace.runtime / "locks" / f"budget-{key(case_id)}.lock"): + return self._locked(case_id, package_id, copy.deepcopy(request)) + except LockBusy: + raise + except (CaseBuildError, HypothesisError, LifecycleError, PackageReadError) as error: + raise BudgetError(getattr(error, "code", "BUDGET_PACKAGE")) from None + except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, UnicodeError, RecursionError): + raise BudgetError("BUDGET_INPUT") from None + + def _origins(self, reader, chain, revision=None): + manifest = reader.manifest + revision = manifest["analysis_revision"] if revision is None else revision + current = revision == manifest["analysis_revision"] + identity = {"case_id": manifest["case_id"], "package_id": manifest["package_id"], "analysis_revision": revision} + completed = next((cp for _, cp in chain if cp["step_id"] == + f"hypothesis-builder:v1:revision:{revision}:complete"), None) + if completed is None or completed["details"].get("operation") != "hypotheses-built": + raise BudgetError("BUDGET_HYPOTHESES_NOT_READY") + details = completed["details"] + ref = details["hypotheses_ref"] + document = read_json(reader, ref, "hypotheses", self.limits.max_event_bytes) + registry = read_json(reader, details["registry_ref"], "hypothesis-registry", self.limits.max_event_bytes) + context = read_json(reader, details["context_ref"], "hypothesis-context", self.limits.max_event_bytes) + validate_generation(document, registry, context, ref) + if (any(document[field] != value or registry[field] != value for field, value in identity.items()) + or (current and manifest["artifacts"].get("hypotheses") != ref) + or (current and manifest["artifacts"].get("hypothesis_registry") != details["registry_ref"]) + or document["context_sha256"] != digest(context) + or document["policy_sha256"] != digest(context["policy"]) + or document["registry_ref"] != details["registry_ref"]): + raise BudgetError("BUDGET_ORIGIN_CHANGED") + call = read_json(reader, document["llm_result_ref"], "llm-call", self.limits.max_event_bytes) + if (call["status"] != "SUCCESS" or call["phase"] != State.GENERATING_HYPOTHESES + or any(call[field] != value for field, value in identity.items()) + or call["outcome"].get("document") != {"hypotheses": [h["proposal"] for h in document["hypotheses"]]} + or call["outcome"].get("metadata") != document["llm_metadata"] + or details["created_count"] != registry["created_count"] + or details["context_sha256"] != digest(context) + or details["policy_sha256"] != digest(context["policy"])): + raise BudgetError("BUDGET_ORIGIN_CHANGED") + return registry, {"case_id": manifest["case_id"], "package_id": manifest["package_id"], + "analysis_revision": revision, "hypotheses_ref": ref, + "hypotheses_sha256": digest(document), "registry_ref": details["registry_ref"], + "registry_sha256": digest(registry)} + + def _sources(self, reader, request): + if request["operation"] != "result": + return + refs = request["source_refs"] + if (not isinstance(refs, list) or not refs or len(refs) != len(set(refs))): + raise BudgetError("BUDGET_SOURCE_REFERENCES") + hashes = {} + for ref in refs: + require_file(reader, ref) + with reader._archive.open(ref) as stream: + source_hash = hashlib.sha256() + while chunk := stream.read(65536): + source_hash.update(chunk) + hashes[ref] = source_hash.hexdigest() + if "source_hashes" in request and request["source_hashes"] != hashes: + raise BudgetError("BUDGET_SOURCE_CHANGED") + request["source_hashes"] = hashes + if len(json_bytes(request["evidence"])) > self.limits.max_result_bytes: + raise BudgetError("BUDGET_RESULT_LIMIT") + + def _load(self, reader, chain, registry, common, *, prior_seen=None, check_pointer=True): + checkpoints = [cp for _, cp in reversed(chain) if cp["analysis_revision"] == common["analysis_revision"] + and cp["details"].get("operation") == "budget-event"] + if len(checkpoints) > self.limits.max_events: + raise BudgetError("BUDGET_HISTORY_LIMIT") + documents = [] + previous_ref = previous_hash = None + state = policy = None + for sequence, cp in enumerate(checkpoints, 1): + ref = cp["details"]["event_ref"] + expected_ref = f"analysis/budgets/{common['analysis_revision']:04d}/events/{sequence:08d}.json" + if ref != expected_ref: + raise BudgetError("BUDGET_HISTORY_INVALID") + event = read_json(reader, ref, "budget-event", self.limits.max_event_bytes) + if (any(event[field] != value for field, value in common.items()) + or event["sequence"] != sequence or event["parent_ref"] != previous_ref + or event["parent_sha256"] != previous_hash + or digest(event) != cp["details"]["event_sha256"]): + raise BudgetError("BUDGET_HISTORY_INVALID") + if sequence == 1: + policy = event["policy"] + validate(policy, "budget-policy") + BudgetConfig(**{name: value for name, value in policy.items() + if name not in {"format", "format_version", "unit"}}) + if event["request"] != {"operation": "initialize"}: + raise BudgetError("BUDGET_HISTORY_INVALID") + state = initial_state(registry, policy, prior_seen) + else: + if event["policy"] != policy or event["request"]["operation"] == "initialize": + raise BudgetError("BUDGET_HISTORY_INVALID") + request = copy.deepcopy(event["request"]) + self._sources(reader, request) + if request["operation"] == "result" and request["result_ref"] != ref: + raise BudgetError("BUDGET_HISTORY_INVALID") + state = apply(state, request, policy) + if event["state"] != state: + raise BudgetError("BUDGET_ACCOUNTING_CHANGED") + previous_ref, previous_hash = ref, digest(event) + documents.append(event) + pointer = reader.manifest["artifacts"].get("budget_ledger") + if not check_pointer: + return documents, state, policy, previous_ref, previous_hash + if documents: + if pointer != previous_ref: + raise BudgetError("BUDGET_POINTER_CHANGED") + elif pointer: + # An older revision is retained; a current/future declaration cannot be reset. + declared = read_json(reader, pointer, "budget-event", self.limits.max_event_bytes) + if (declared["case_id"] != common["case_id"] or declared["package_id"] != common["package_id"] + or declared["analysis_revision"] >= common["analysis_revision"]): + raise BudgetError("BUDGET_HISTORY_MISSING") + return documents, state, policy, previous_ref, previous_hash + + def _locked(self, case_id, package_id, request): + latest = self.store.latest(case_id, package_id) + if latest is None: + raise BudgetError("BUDGET_CASE_NOT_FOUND") + with PackageReader(latest, self.store.limits) as reader: + manifest = reader.manifest + chain = checkpoint_chain(reader) + registry, common = self._origins(reader, chain) + budget_checkpoints = [cp for _, cp in chain if cp["details"].get("operation") == "budget-event"] + if len(budget_checkpoints) > self.limits.max_events: + raise BudgetError("BUDGET_HISTORY_LIMIT") + prior_seen = [] + for old_revision in sorted({cp["analysis_revision"] for cp in budget_checkpoints + if cp["analysis_revision"] < common["analysis_revision"]}): + old_registry, old_common = self._origins(reader, chain, old_revision) + _, old_state, _, _, _ = self._load(reader, chain, old_registry, old_common, + prior_seen=prior_seen, check_pointer=False) + prior_seen = old_state["seen_evidence"] + documents, state, policy, parent_ref, parent_hash = self._load(reader, chain, registry, common, prior_seen=prior_seen) + if request is None: + if not documents: + raise BudgetError("BUDGET_NOT_INITIALIZED") + self._repair_index(latest, manifest) + return BudgetResult(latest, common["analysis_revision"], parent_ref, state, True) + self._sources(reader, request) + validation_request = dict(request) + if request["operation"] == "result": + validation_request["result_ref"] = "analysis/budgets/pending.json" + validate(validation_request, "budget-request") + if request["operation"] == "settle" and not integer(request["semantic_score"], 0, 100): + raise BudgetError("BUDGET_SEMANTIC_SCORE") + for event in documents: + saved = {name: value for name, value in event["request"].items() if name != "result_ref"} + identity = lambda item: (item["operation"], item.get("hypothesis_id"), item.get("action_id")) + if identity(saved) == identity(request): + if saved != request: + raise BudgetError("BUDGET_STEP_CONFLICT") + self._repair_index(latest, manifest) + return BudgetResult(latest, common["analysis_revision"], parent_ref, state, True) + if manifest["state"] not in {State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES}: + raise BudgetError("BUDGET_PHASE") + if len(budget_checkpoints) >= self.limits.max_events: + raise BudgetError("BUDGET_HISTORY_LIMIT") + if not documents: + if request != {"operation": "initialize"}: + raise BudgetError("BUDGET_NOT_INITIALIZED") + policy = self.config.policy() + state = initial_state(registry, policy, prior_seen) + elif policy != self.config.policy(): + raise BudgetError("BUDGET_POLICY_CHANGED") + sequence = len(documents) + 1 + ref = f"analysis/budgets/{common['analysis_revision']:04d}/events/{sequence:08d}.json" + if request["operation"] == "result": + request["result_ref"] = ref + # Validate request before any rules are executed, including Python bool/int distinctions. + validate(request, "budget-request") + if documents: + state = apply(state, request, policy) + event = {"format": "viano-budget-event", "format_version": 1, **common, + "sequence": sequence, "parent_ref": parent_ref, "parent_sha256": parent_hash, + "policy": policy, "request": request, "state": state} + validate(event, "budget-event") + encoded = json_bytes(event) + if len(encoded) > self.limits.max_event_bytes: + raise BudgetError("BUDGET_EVENT_LIMIT") + path = self.store.advance(case_id, package_id, manifest["state"], + step_id=f"budget:v1:revision:{common['analysis_revision']}:event:{sequence}", + details={"operation": "budget-event", "event_ref": ref, "event_sha256": digest(event)}, + additions={ref: encoded}, artifacts={"budget_ledger": ref}, + expected_generation=manifest["package_generation"]) + return BudgetResult(path, common["analysis_revision"], ref, state) + + def _repair_index(self, path, manifest): + with self.store._locks(manifest["case_id"]): + if self.store.latest(manifest["case_id"], manifest["package_id"]) != path: + raise BudgetError("BUDGET_GENERATION_CHANGED") + self.store._index(path, manifest) + atomic_json(self.store.workspace.runtime / "current_job.json", None) diff --git a/src/viano_bug_analysis/budget/scoring.py b/src/viano_bug_analysis/budget/scoring.py new file mode 100644 index 0000000..3b8cc90 --- /dev/null +++ b/src/viano_bug_analysis/budget/scoring.py @@ -0,0 +1,35 @@ +import hashlib + +from viano_bug_analysis.lifecycle.storage import json_bytes + +from .config import BudgetError, integer + + +def evidence_digest(evidence): + """Canonical JSON identity, without query IDs or transport metadata.""" + if not isinstance(evidence, (dict, list)): + raise BudgetError("BUDGET_EVIDENCE") + def check(value): + if isinstance(value, dict): + if any(not isinstance(name, str) for name in value): + raise BudgetError("BUDGET_EVIDENCE") + for child in value.values(): + check(child) + elif isinstance(value, list): + for child in value: + check(child) + elif value is not None and type(value) not in {str, int, float, bool}: + raise BudgetError("BUDGET_EVIDENCE") + try: + check(evidence) + return hashlib.sha256(json_bytes(evidence)).hexdigest() + except (ValueError, TypeError, RecursionError): + raise BudgetError("BUDGET_EVIDENCE") from None + + +def reward_for(cost, novelty, semantic, policy): + if not integer(semantic, 0, 100): + raise BudgetError("BUDGET_SEMANTIC_SCORE") + information = min(novelty, semantic) + reward = cost * policy["max_reward_bps"] * information // 1_000_000 + return min(reward, cost - 1), information diff --git a/src/viano_bug_analysis/case/assembler.py b/src/viano_bug_analysis/case/assembler.py index e135c88..5a2fcd1 100644 --- a/src/viano_bug_analysis/case/assembler.py +++ b/src/viano_bug_analysis/case/assembler.py @@ -84,6 +84,12 @@ class CaseAssembler: prior_refs = [] for _, checkpoint in reversed(chain): details = checkpoint["details"] + if details.get("operation") == "budget-event": + ref = details.get("event_ref") + if not isinstance(ref, str) or not ref.startswith("analysis/budgets/"): + raise CaseBuildError("PREVIOUS_CASE", "Invalid budget artifact checkpoint") + require_file(reader, ref) + excluded.add(ref) if details.get("operation") in {"hypotheses-generating", "hypotheses-built", "hypotheses-rejected"}: for field in ("context_ref", "policy_ref", "hypotheses_ref", "registry_ref", "summary_ref", "error_ref"): if field not in details: diff --git a/src/viano_bug_analysis/schemas/budget-event-v1.schema.json b/src/viano_bug_analysis/schemas/budget-event-v1.schema.json new file mode 100644 index 0000000..d5848c1 --- /dev/null +++ b/src/viano_bug_analysis/schemas/budget-event-v1.schema.json @@ -0,0 +1,484 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "properties": { + "format": { + "const": "viano-budget-event" + }, + "format_version": { + "const": 1 + }, + "case_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "package_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "analysis_revision": { + "type": "integer", + "minimum": 1 + }, + "hypotheses_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "hypotheses_sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + "registry_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "registry_sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + "sequence": { + "type": "integer", + "minimum": 1 + }, + "parent_ref": { + "anyOf": [ + { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + { + "type": "null" + } + ] + }, + "parent_sha256": { + "anyOf": [ + { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + { + "type": "null" + } + ] + }, + "policy": { + "type": "object", + "additionalProperties": false, + "properties": { + "format": { + "const": "viano-budget-policy" + }, + "format_version": { + "const": 1 + }, + "unit": { + "const": "investigation_units" + }, + "default_action_cost": { + "type": "integer", + "minimum": 1 + }, + "action_costs": { + "type": "object", + "propertyNames": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "additionalProperties": { + "type": "integer", + "minimum": 1 + } + }, + "max_reward_bps": { + "type": "integer", + "minimum": 0, + "maximum": 9999 + }, + "new_evidence_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "low_information_threshold": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "max_low_information_steps": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "format", + "format_version", + "unit", + "default_action_cost", + "action_costs", + "max_reward_bps", + "new_evidence_score", + "low_information_threshold", + "max_low_information_steps" + ] + }, + "request": { + "oneOf": [ + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "initialize" + } + }, + "required": [ + "operation" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "reserve" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "action": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "action" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "result" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "evidence": { + "type": [ + "object", + "array" + ] + }, + "source_refs": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "source_hashes": { + "type": "object", + "minProperties": 1, + "additionalProperties": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + }, + "result_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "evidence", + "source_refs", + "source_hashes", + "result_ref" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "settle" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "semantic_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "semantic_score", + "reason" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "fail" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "reason" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "terminate" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "reason" + ] + } + ] + }, + "state": { + "type": "object", + "additionalProperties": false, + "properties": { + "accounts": { + "type": "object", + "minProperties": 1, + "propertyNames": { + "pattern": "^H[1-9][0-9]*$" + }, + "additionalProperties": { + "type": "object", + "additionalProperties": false, + "properties": { + "initial": { + "type": "integer", + "minimum": 1 + }, + "remaining": { + "type": "integer", + "minimum": 0 + }, + "spent": { + "type": "integer", + "minimum": 0 + }, + "rewarded": { + "type": "integer", + "minimum": 0 + }, + "low_information_steps": { + "type": "integer", + "minimum": 0 + }, + "status": { + "enum": [ + "new", + "investigating", + "rejected", + "budget_exhausted", + "abandoned_low_information_gain" + ] + }, + "actions": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": false, + "properties": { + "action": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "cost": { + "type": "integer", + "minimum": 1 + }, + "status": { + "enum": [ + "reserved", + "awaiting_assessment", + "settled" + ] + }, + "evidence_sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + "novelty_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "result_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "reward": { + "type": "integer", + "minimum": 0 + }, + "information_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "semantic_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "assessment_reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "failed": { + "type": "boolean" + } + }, + "required": [ + "action", + "cost", + "status" + ] + } + }, + "termination_reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "initial", + "remaining", + "spent", + "rewarded", + "low_information_steps", + "status", + "actions" + ] + } + }, + "seen_evidence": { + "type": "array", + "uniqueItems": true, + "items": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + } + }, + "required": [ + "accounts", + "seen_evidence" + ] + } + }, + "required": [ + "format", + "format_version", + "case_id", + "package_id", + "analysis_revision", + "hypotheses_ref", + "hypotheses_sha256", + "registry_ref", + "registry_sha256", + "sequence", + "parent_ref", + "parent_sha256", + "policy", + "request", + "state" + ] +} diff --git a/src/viano_bug_analysis/schemas/budget-policy-v1.schema.json b/src/viano_bug_analysis/schemas/budget-policy-v1.schema.json new file mode 100644 index 0000000..6e3b3af --- /dev/null +++ b/src/viano_bug_analysis/schemas/budget-policy-v1.schema.json @@ -0,0 +1,62 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "properties": { + "format": { + "const": "viano-budget-policy" + }, + "format_version": { + "const": 1 + }, + "unit": { + "const": "investigation_units" + }, + "default_action_cost": { + "type": "integer", + "minimum": 1 + }, + "action_costs": { + "type": "object", + "propertyNames": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "additionalProperties": { + "type": "integer", + "minimum": 1 + } + }, + "max_reward_bps": { + "type": "integer", + "minimum": 0, + "maximum": 9999 + }, + "new_evidence_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "low_information_threshold": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "max_low_information_steps": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "format", + "format_version", + "unit", + "default_action_cost", + "action_costs", + "max_reward_bps", + "new_evidence_score", + "low_information_threshold", + "max_low_information_steps" + ] +} diff --git a/src/viano_bug_analysis/schemas/budget-request-v1.schema.json b/src/viano_bug_analysis/schemas/budget-request-v1.schema.json new file mode 100644 index 0000000..20c1006 --- /dev/null +++ b/src/viano_bug_analysis/schemas/budget-request-v1.schema.json @@ -0,0 +1,195 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "oneOf": [ + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "initialize" + } + }, + "required": [ + "operation" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "reserve" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "action": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "action" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "result" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "evidence": { + "type": [ + "object", + "array" + ] + }, + "source_refs": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "source_hashes": { + "type": "object", + "minProperties": 1, + "additionalProperties": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + }, + "result_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "evidence", + "source_refs", + "source_hashes", + "result_ref" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "settle" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "semantic_score": { + "type": "integer", + "minimum": 0, + "maximum": 100 + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "semantic_score", + "reason" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "fail" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "reason" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "operation": { + "const": "terminate" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "reason": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "required": [ + "operation", + "hypothesis_id", + "action_id", + "reason" + ] + } + ] +} diff --git a/tests/profiles/vcore/test_real_package.py b/tests/profiles/vcore/test_real_package.py index e2ef38e..b264e34 100644 --- a/tests/profiles/vcore/test_real_package.py +++ b/tests/profiles/vcore/test_real_package.py @@ -78,4 +78,19 @@ def test_full_bug20_package_pair(tmp_path): assert before.read_bytes(name) == after.read_bytes(name) assert asyncio.run(HypothesisBuilder(store, gateway(hypotheses_adapter)).build("regression-case", "regression-package")).reused assert len(hypotheses_adapter.requests) == 1 + from viano_bug_analysis.budget import BudgetManager + budget = BudgetManager(store) + identity = ("regression-case", "regression-package") + budget.initialize(*identity) + budget.reserve(*identity, "H1", "offline-query", "source.read_context") + source_ref = case["source_context"]["refs"][0]["path"] + budget.record_result(*identity, "H1", "offline-query", {"fixture_observation":"Stored context is available"}, source_refs=[source_ref]) + accounted = budget.settle(*identity, "H1", "offline-query", semantic_score=50, reason="Simulated assessment, no diagnosis") + assert accounted.state["accounts"]["H1"]["remaining"] == 94 + assert PackageValidator().validate(accounted.package).valid + with PackageReader(generated.package) as before, PackageReader(accounted.package) as after: + for name in before.files: + if name != "manifest.json": + assert before.read_bytes(name) == after.read_bytes(name) + assert CaseBuilder(store).build(*identity).reused assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] == original_hashes diff --git a/tests/test_budget_manager.py b/tests/test_budget_manager.py new file mode 100644 index 0000000..c7989a9 --- /dev/null +++ b/tests/test_budget_manager.py @@ -0,0 +1,435 @@ +import asyncio +import hashlib +import json +from importlib.resources import files +from zipfile import ZipFile + +import pytest +from jsonschema import Draft202012Validator + +from viano_bug_analysis.analysis import HypothesisBuilder, HypothesisConfig +from viano_bug_analysis.budget import BudgetConfig, BudgetError, BudgetLimits, BudgetManager +from viano_bug_analysis.budget.engine import apply, initial_state +from viano_bug_analysis.budget.scoring import reward_for +from viano_bug_analysis.lifecycle import CheckpointStore, State, Workspace +from viano_bug_analysis.lifecycle.errors import LockBusy +from viano_bug_analysis.lifecycle.locking import FileLock +from viano_bug_analysis.lifecycle.storage import json_bytes, key +from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter +from test_case_builder import case_setup +from test_case_interpreter import built_case, gateway +from test_hypothesis_builder import interpreted_case, ProposalAdapter + + +CASE = ("case-A", "stable-package") + + +@pytest.fixture +def hypotheses(interpreted_case): + store, case_builder, _, _, _ = interpreted_case + adapter = ProposalAdapter() + result = asyncio.run(HypothesisBuilder(store, gateway(adapter)).build(*CASE)) + return store, case_builder, result, adapter + + +def manager(hypotheses, config=None): + store, _, result, _ = hypotheses + budget = BudgetManager(store, config) + budget.initialize(*CASE) + with PackageReader(result.package) as reader: + source = json.loads(reader.read_bytes(reader.manifest["artifacts"]["case"]))["inputs"]["records"][0]["path"] + return budget, source + + +def result_and_settle(budget, source, action_id, evidence=None, score=100, hypothesis="H1"): + budget.record_result(*CASE, hypothesis, action_id, {"finding": action_id} if evidence is None else evidence, source_refs=[source]) + return budget.settle(*CASE, hypothesis, action_id, semantic_score=score, reason="Controller assessment of stored result") + + +def test_durable_debit_reward_preservation_and_portability(hypotheses, tmp_path): + budget, source = manager(hypotheses) + initial = budget.snapshot(*CASE) + assert set(initial.state["accounts"]) == {"H1", "H2", "H3"} + assert {a["remaining"] for a in initial.state["accounts"].values()} == {100} + assert budget.reserve(*CASE, "H1", "query1", "source.find_text").state["accounts"]["H1"]["remaining"] == 90 + settled = result_and_settle(budget, source, "query1") + account = settled.state["accounts"]["H1"] + assert (account["remaining"], account["spent"], account["rewarded"]) == (98, 10, 8) + assert account["actions"]["query1"]["status"] == "settled" + assert account["actions"]["query1"]["reward"] < account["actions"]["query1"]["cost"] + assert settled.investigable == ["H1", "H2", "H3"] + assert PackageValidator().validate(settled.package).valid + with PackageReader(hypotheses[2].package) as before, PackageReader(settled.package) as after: + assert after.manifest["package_generation"] == before.manifest["package_generation"] + 4 + assert after.manifest["analysis_revision"] == 1 + for name in before.files: + if name != "manifest.json": + assert before.read_bytes(name) == after.read_bytes(name) + event = json.loads(after.read_bytes(settled.event_ref)) + assert event["request"]["semantic_score"] == 100 + assert event["state"] == settled.state + assert after.manifest["artifacts"]["diagnosis"] is None + portable = CheckpointStore(Workspace(tmp_path / "fresh-service")) + renamed = tmp_path / "renamed.zip" + renamed.write_bytes(settled.package.read_bytes()) + portable.intake(renamed) + assert BudgetManager(portable).snapshot(*CASE).state == settled.state + assert hypotheses[1].build(*CASE).reused + assert asyncio.run(HypothesisBuilder(hypotheses[0], gateway(hypotheses[3])).build(*CASE)).reused + + +def test_idempotent_operations_and_conflict(hypotheses): + budget, source = manager(hypotheses) + reserved = budget.reserve(*CASE, "H1", "q", "read") + assert budget.reserve(*CASE, "H1", "q", "read").reused + assert budget.snapshot(*CASE).package == reserved.package + with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"): + budget.reserve(*CASE, "H1", "q", "different") + first = budget.record_result(*CASE, "H1", "q", {"b": 2, "a": 1}, source_refs=[source]) + assert budget.record_result(*CASE, "H1", "q", {"a": 1, "b": 2}, source_refs=[source]).package == first.package + with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"): + budget.record_result(*CASE, "H1", "q", {"a": 3}, source_refs=[source]) + settled = budget.settle(*CASE, "H1", "q", semantic_score=50, reason="Useful") + assert budget.settle(*CASE, "H1", "q", semantic_score=50, reason="Useful").package == settled.package + with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"): + budget.settle(*CASE, "H1", "q", semantic_score=100, reason="Useful") + assert budget.initialize(*CASE).state["accounts"]["H1"]["remaining"] == 94 + # Python equality must not let a float masquerade as a valid replayed score. + with pytest.raises(BudgetError): + budget.settle(*CASE, "H1", "q", semantic_score=50.0, reason="Useful") + + +@pytest.mark.parametrize("evidence", [{1:"invalid JSON object key"}, {"value":float("nan")}, "unstructured"]) +def test_invalid_evidence_does_not_publish_partial_result(hypotheses, evidence): + budget, source = manager(hypotheses) + before = budget.reserve(*CASE, "H1", "q", "read") + with pytest.raises(BudgetError): + budget.record_result(*CASE, "H1", "q", evidence, source_refs=[source]) + assert budget.snapshot(*CASE).package == before.package + + +def test_duplicate_evidence_no_reward_even_across_hypotheses(hypotheses): + budget, source = manager(hypotheses) + budget.reserve(*CASE, "H1", "first", "read") + result_and_settle(budget, source, "first", {"finding": "same"}) + budget.reserve(*CASE, "H2", "other", "different_query") + settled = result_and_settle(budget, source, "other", {"finding": "same"}, hypothesis="H2") + action = settled.state["accounts"]["H2"]["actions"]["other"] + assert action["novelty_score"] == action["reward"] == 0 + assert len(settled.state["seen_evidence"]) == 1 + + +@pytest.mark.parametrize("score", [-1, 101, True, 50.5, 100.0, "100", None]) +def test_invalid_semantic_scores_atomic(hypotheses, score): + budget, source = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + before = budget.record_result(*CASE, "H1", "q", {"finding": "new"}, source_refs=[source]) + with pytest.raises(BudgetError): + budget.settle(*CASE, "H1", "q", semantic_score=score, reason="Value") + assert budget.snapshot(*CASE).package == before.package + assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 90 + + +def test_low_information_stops_and_cannot_be_refilled(hypotheses): + budget, source = manager(hypotheses) + for number in range(3): + budget.reserve(*CASE, "H1", f"q{number}", "read") + result_and_settle(budget, source, f"q{number}", score=20) + state = budget.snapshot(*CASE).state + assert state["accounts"]["H1"]["status"] == "abandoned_low_information_gain" + assert state["accounts"]["H1"]["low_information_steps"] == 3 + with pytest.raises(BudgetError, match="BUDGET_TERMINAL"): + budget.reserve(*CASE, "H1", "q4", "read") + assert BudgetManager(hypotheses[0], BudgetConfig(max_low_information_steps=99)).initialize(*CASE).state == state + + +def test_high_information_resets_consecutive_counter(hypotheses): + budget, source = manager(hypotheses) + for number, score in enumerate([0, 0, 100, 0, 0]): + budget.reserve(*CASE, "H1", str(number), "read") + result_and_settle(budget, source, str(number), score=score) + account = budget.snapshot(*CASE).state["accounts"]["H1"] + assert account["status"] == "investigating" and account["low_information_steps"] == 2 + + +def test_failed_action_consumes_cost_and_pending_survives_restart(hypotheses): + budget, _ = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + restarted = BudgetManager(hypotheses[0]) + assert "H1" not in restarted.snapshot(*CASE).investigable + with pytest.raises(BudgetError, match="BUDGET_PENDING_ACTION"): + restarted.reserve(*CASE, "H1", "q2", "read") + with pytest.raises(BudgetError, match="BUDGET_PENDING_ACTION"): + restarted.reject(*CASE, "H1", reason="Rejected") + failed = restarted.fail(*CASE, "H1", "q", reason="Timeout; outcome uncertain") + account = failed.state["accounts"]["H1"] + assert account["remaining"] == 90 and account["rewarded"] == 0 + assert restarted.fail(*CASE, "H1", "q", reason="Timeout; outcome uncertain").reused + + +def test_cost_overrides_and_exhaustion(hypotheses): + config = BudgetConfig(default_action_cost=40, action_costs={"expensive": 101}, max_reward_bps=0, max_low_information_steps=50) + budget, source = manager(hypotheses, config) + before = budget.snapshot(*CASE) + with pytest.raises(BudgetError, match="BUDGET_INSUFFICIENT"): + budget.reserve(*CASE, "H1", "too_much", "expensive") + assert budget.snapshot(*CASE).package == before.package + for action_id in ["a", "b"]: + budget.reserve(*CASE, "H1", action_id, "read") + result_and_settle(budget, source, action_id) + assert budget.snapshot(*CASE).state["accounts"]["H1"]["status"] == "budget_exhausted" + assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 20 + + +def test_reward_can_restore_affordability_after_reservation(hypotheses): + budget, source = manager(hypotheses, BudgetConfig(default_action_cost=60, max_reward_bps=8000)) + budget.reserve(*CASE, "H1", "q", "read") + assert budget.snapshot(*CASE).state["accounts"]["H1"]["status"] == "investigating" + account = result_and_settle(budget, source, "q").state["accounts"]["H1"] + assert account["remaining"] == 88 and account["status"] == "investigating" + + +@pytest.mark.parametrize("evidence", [{}, []]) +def test_empty_results_not_rewarded(hypotheses, evidence): + budget, source = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + assert result_and_settle(budget, source, "q", evidence).state["accounts"]["H1"]["rewarded"] == 0 + + +def test_terminal_state_is_separate_from_confidence_and_creation_registry(hypotheses): + budget, _ = manager(hypotheses) + terminal = budget.reject(*CASE, "H1", reason="Evidence refutes mechanism") + assert terminal.state["accounts"]["H1"]["status"] == "rejected" + assert "H1" not in terminal.investigable + with pytest.raises(BudgetError, match="BUDGET_TERMINAL"): + budget.reserve(*CASE, "H1", "q", "read") + with PackageReader(hypotheses[2].package) as before, PackageReader(terminal.package) as after: + ref = after.manifest["artifacts"]["hypothesis_registry"] + assert after.read_bytes(ref) == before.read_bytes(ref) + assert json.loads(after.read_bytes(after.manifest["artifacts"]["hypotheses"]))["hypotheses"][0]["confidence"] == "unassessed" + + +def test_policy_is_frozen_and_mutation_rejected(hypotheses): + budget, _ = manager(hypotheses) + changed = BudgetManager(hypotheses[0], BudgetConfig(max_reward_bps=9999)) + assert changed.initialize(*CASE).reused + with pytest.raises(BudgetError, match="BUDGET_POLICY_CHANGED"): + changed.reserve(*CASE, "H1", "q", "read") + assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 100 + + +@pytest.mark.parametrize("config", [{"default_action_cost":0}, {"max_reward_bps":10000}, {"max_reward_bps":True}, + {"new_evidence_score":101}, {"low_information_threshold":-1}, {"max_low_information_steps":0}, + {"action_costs":{"x":False}}, {"action_costs":{" ":1}}, {"default_action_cost":1.5}]) +def test_configuration_rejected(config): + with pytest.raises(BudgetError): + BudgetConfig(**config) + + +def test_config_toml_and_dictionary_detachment(tmp_path): + config = BudgetConfig.load("config/budget.example.toml") + assert config.action_costs["source.read_context"] == 5 + values = {"read":5} + config = BudgetConfig(action_costs=values) + values["read"] = 0 + assert config.policy()["action_costs"]["read"] == 5 + invalid = tmp_path / "bad.toml" + invalid.write_text('[budget]\nunknown=1') + with pytest.raises(BudgetError): + BudgetConfig.load(invalid) + with pytest.raises(BudgetError): + BudgetLimits(max_events=True) + + +def test_reward_strictly_below_cost_and_both_scores_required(): + for cost in range(1, 101): + for novelty in [0, 20, 100]: + for semantic in [0, 20, 100]: + reward, information = reward_for(cost, novelty, semantic, BudgetConfig(max_reward_bps=9999).policy()) + assert 0 <= reward < cost + assert information == min(novelty, semantic) + if min(novelty, semantic) == 0: + assert reward == 0 + assert reward_for(10, 50, 100, BudgetConfig().policy()) == (4, 50) + + +def test_finite_steps_even_with_maximum_rewards(): + policy = BudgetConfig(default_action_cost=1, max_reward_bps=9999).policy() + state = initial_state({"creations":[{"id":"H1","allocated_budget":10}]}, policy) + for index in range(10): + action_id = str(index) + state = apply(state, {"operation":"reserve","hypothesis_id":"H1","action_id":action_id,"action":"read"}, policy) + state = apply(state, {"operation":"result","hypothesis_id":"H1","action_id":action_id,"evidence":{"n":index},"result_ref":"test"}, policy) + state = apply(state, {"operation":"settle","hypothesis_id":"H1","action_id":action_id,"semantic_score":100,"reason":"New"}, policy) + assert state["accounts"]["H1"]["remaining"] == 0 + assert state["accounts"]["H1"]["status"] == "budget_exhausted" + + +def test_lifecycle_preconditions(interpreted_case): + with pytest.raises(BudgetError, match="BUDGET_HYPOTHESES_NOT_READY"): + BudgetManager(interpreted_case[0]).initialize(*CASE) + + +def test_missing_init_result_and_bad_references_rejected(hypotheses): + budget = BudgetManager(hypotheses[0]) + with pytest.raises(BudgetError, match="BUDGET_NOT_INITIALIZED"): + budget.reserve(*CASE, "H1", "q", "read") + budget.initialize(*CASE) + budget.reserve(*CASE, "H1", "q", "read") + with pytest.raises(BudgetError, match="BUDGET_RESULT_REQUIRED"): + budget.settle(*CASE, "H1", "q", semantic_score=100, reason="Missing") + for refs in [[], ["../escape"], ["missing.json"], ["manifest.json"]]: + with pytest.raises(BudgetError): + budget.record_result(*CASE, "H1", "q", {"x":1}, source_refs=refs) + with pytest.raises(BudgetError, match="BUDGET_HYPOTHESIS_UNKNOWN"): + budget.reserve(*CASE, "H99", "q", "read") + + +def test_limits_do_not_truncate_or_drop_accounting(hypotheses): + budget, source = manager(hypotheses) + small = BudgetManager(hypotheses[0], limits=BudgetLimits(max_events=1)) + with pytest.raises(BudgetError, match="BUDGET_HISTORY_LIMIT"): + small.reserve(*CASE, "H1", "q", "read") + budget.reserve(*CASE, "H1", "q", "read") + small = BudgetManager(hypotheses[0], limits=BudgetLimits(max_result_bytes=5)) + with pytest.raises(BudgetError, match="BUDGET_RESULT_LIMIT"): + small.record_result(*CASE, "H1", "q", {"long":"content"}, source_refs=[source]) + + +def test_publish_index_crash_recovered_without_double_debit(hypotheses, monkeypatch): + budget, _ = manager(hypotheses) + original = budget.store._index + def fail_index(*args): + raise OSError("Simulated index failure after publication") + monkeypatch.setattr(budget.store, "_index", fail_index) + with pytest.raises(BudgetError): + budget.reserve(*CASE, "H1", "q", "read") + monkeypatch.setattr(budget.store, "_index", original) + recovered = budget.reserve(*CASE, "H1", "q", "read") + assert recovered.reused and recovered.state["accounts"]["H1"]["remaining"] == 90 + index = json.loads((recovered.package.parent / "current.json").read_bytes()) + assert index["package"] == recovered.package.name + + +def test_same_case_lock_and_concurrent_generation_guard(hypotheses, monkeypatch): + budget, _ = manager(hypotheses) + with FileLock(budget.store.workspace.runtime / "locks" / f"budget-{key(CASE[0])}.lock"): + with pytest.raises(LockBusy): + budget.reserve(*CASE, "H1", "q", "read") + original = budget.store.advance + def race(*args, **kwargs): + original(*CASE, State.GENERATING_HYPOTHESES, step_id="concurrent-note") + return original(*args, **kwargs) + monkeypatch.setattr(budget.store, "advance", race) + with pytest.raises(BudgetError, match="GENERATION_CHANGED"): + budget.reserve(*CASE, "H1", "q", "read") + assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 100 + + +def test_pointer_deletion_does_not_reset_accounting(hypotheses): + budget, _ = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + budget.store.advance(*CASE, State.GENERATING_HYPOTHESES, step_id="bad-pointer", artifacts={"budget_ledger":None}) + with pytest.raises(BudgetError, match="BUDGET_POINTER_CHANGED"): + budget.initialize(*CASE) + + +def rewrite_managed_fixture(path, mutate): + """Simulate an imported corrupt ledger whose archive integrity was recomputed.""" + with ZipFile(path) as archive: + payloads = {name: archive.read(name) for name in archive.namelist()} + manifest = json.loads(payloads["manifest.json"]) + event_ref = manifest["artifacts"]["budget_ledger"] + cp_ref = manifest["artifacts"]["lifecycle_checkpoint"] + event, checkpoint = json.loads(payloads[event_ref]), json.loads(payloads[cp_ref]) + mutate(event, checkpoint) + payloads[event_ref] = json_bytes(event) + checkpoint["details"]["event_sha256"] = hashlib.sha256(payloads[event_ref]).hexdigest() + payloads[cp_ref] = json_bytes(checkpoint) + for ref in (event_ref, cp_ref): + manifest["integrity"][ref] = {"sha256": hashlib.sha256(payloads[ref]).hexdigest(), "size": len(payloads[ref])} + payloads["manifest.json"] = json_bytes(manifest) + temporary = path.with_suffix(".tampered.zip") + with ZipFile(temporary, "w") as archive: + for name, content in payloads.items(): + archive.writestr(name, content) + temporary.replace(path) + assert PackageValidator().validate(path).valid + + +@pytest.mark.parametrize("mutate", [ + lambda e,c: e["state"]["accounts"]["H1"].update(remaining=100), + lambda e,c: e["state"]["accounts"]["H1"]["actions"]["q"].update(cost=1), + lambda e,c: e.update(parent_sha256="0"*64), + lambda e,c: e["policy"].update(max_reward_bps=9999), + lambda e,c: e.update(registry_sha256="0"*64), +]) +def test_recomputed_archive_hashes_cannot_hide_ledger_corruption(hypotheses, mutate): + budget, _ = manager(hypotheses) + reserved = budget.reserve(*CASE, "H1", "q", "read") + rewrite_managed_fixture(reserved.package, mutate) + with pytest.raises(BudgetError): + budget.snapshot(*CASE) + + +def test_reward_tampering_rejected_by_replay(hypotheses): + budget, source = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + settled = result_and_settle(budget, source, "q") + def mutate(event, checkpoint): + account = event["state"]["accounts"]["H1"] + account.update(remaining=100, rewarded=10) + account["actions"]["q"]["reward"] = 10 + rewrite_managed_fixture(settled.package, mutate) + with pytest.raises(BudgetError, match="BUDGET_ACCOUNTING_CHANGED"): + budget.snapshot(*CASE) + + +def test_zero_affordable_actions_ready_for_evaluation(hypotheses): + budget = BudgetManager(hypotheses[0], BudgetConfig(default_action_cost=101)) + snapshot = budget.initialize(*CASE) + assert snapshot.ready_for_evaluation + assert not snapshot.investigable + assert {a["status"] for a in snapshot.state["accounts"].values()} == {"budget_exhausted"} + + +def test_pending_actions_not_ready_for_evaluation(hypotheses): + budget, _ = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + budget.reject(*CASE, "H2", reason="Refuted") + budget.reject(*CASE, "H3", reason="Refuted") + assert not budget.snapshot(*CASE).investigable + assert not budget.snapshot(*CASE).ready_for_evaluation + + +def test_new_revision_new_policy_without_losing_old_bytes(hypotheses): + budget, source = manager(hypotheses) + budget.reserve(*CASE, "H1", "q", "read") + result_and_settle(budget, source, "q", {"finding":"same"}) + before = budget.snapshot(*CASE) + store, case_builder, _, _ = hypotheses + store.advance(*CASE, State.FAILED_TO_ANALYZE, step_id="review", reason="Needs evidence") + store.advance(*CASE, State.REOPENED, step_id="new-revision") + case_builder.build(*CASE) + from test_case_interpreter import FixtureAdapter + from viano_bug_analysis.analysis import CaseInterpreter + asyncio.run(CaseInterpreter(store, gateway(FixtureAdapter())).interpret(*CASE)) + asyncio.run(HypothesisBuilder(store, gateway(ProposalAdapter()), HypothesisConfig(initial_budget=23)).build(*CASE)) + new_budget = BudgetManager(store, BudgetConfig(default_action_cost=3)) + after = new_budget.initialize(*CASE) + assert after.analysis_revision == 2 + assert {a["remaining"] for a in after.state["accounts"].values()} == {23} + with PackageReader(before.package) as old, PackageReader(after.package) as new: + for name in old.files: + if name != "manifest.json": + assert old.read_bytes(name) == new.read_bytes(name) + new_budget.reserve(*CASE, "H1", "q", "read") + duplicate = result_and_settle(new_budget, source, "q", {"finding":"same"}) + assert duplicate.state["accounts"]["H1"]["rewarded"] == 0 + assert BudgetManager(store, BudgetConfig(default_action_cost=3)).snapshot(*CASE).state == duplicate.state + + +@pytest.mark.parametrize("name", ["budget-policy", "budget-request", "budget-event"]) +def test_schemas_well_formed(name): + schema = json.loads(files("viano_bug_analysis.schemas").joinpath(name + "-v1.schema.json").read_text()) + Draft202012Validator.check_schema(schema)