From c381e724c78fd5cdd9c73088906fbc3931cb2f15 Mon Sep 17 00:00:00 2001 From: Rhein1 / Agoragentic <140637602+rhein1@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:37:47 -0400 Subject: [PATCH 1/2] feat: add bounded offline trace-guidance replay --- examples/rrsi-trace-guidance-fixture.py | 62 +++ .../fable5-codex/experiments/rrsi/README.md | 61 +++ .../experiments/rrsi/trace_guidance.py | 420 ++++++++++++++++++ test/rrsi.test.mjs | 3 +- test/rrsi_trace_guidance_test.py | 284 ++++++++++++ 5 files changed, 829 insertions(+), 1 deletion(-) create mode 100644 examples/rrsi-trace-guidance-fixture.py create mode 100644 plugins/fable5-codex/experiments/rrsi/trace_guidance.py create mode 100644 test/rrsi_trace_guidance_test.py diff --git a/examples/rrsi-trace-guidance-fixture.py b/examples/rrsi-trace-guidance-fixture.py new file mode 100644 index 0000000..df3a82f --- /dev/null +++ b/examples/rrsi-trace-guidance-fixture.py @@ -0,0 +1,62 @@ +"""Small offline TraceSession demonstration; no provider or execution.""" +import json +from pathlib import Path +import sys + +ROOT = Path(__file__).resolve().parents[1] +BASE = ROOT / "plugins/fable5-codex/experiments/rrsi" +sys.path.insert(0, str(BASE)) + +import evaluator as e +import trace_guidance as g + +source = "export const enabled = false;" +line = {"path": "worker.js", "line": 1, "content_digest": e.digest(source.encode())} +tasks = [{"id": "evolve-a", "repository": "repo-a", "family": "family-a", "split": "evolve", + "prompt": "Check worker startup state.", "sources": {"worker.js": source + "\n"}, + "assertions": [{"id": "enabled", "path": "worker.enabled", "equals": False, "evidence": line}]}, + {"id": "heldout-b", "repository": "repo-b", "family": "family-b", "split": "heldout", + "prompt": "Determine dispatch gating.", "sources": {"worker.js": source + "\n"}, + "assertions": [{"id": "enabled", "path": "worker.enabled", "equals": False, "evidence": line}]}, + {"id": "ood-c", "repository": "repo-c", "family": "family-c", "split": "ood", + "prompt": "Audit activation evidence.", "sources": {"worker.js": source + "\n"}, + "assertions": [{"id": "enabled", "path": "worker.enabled", "equals": False, "evidence": line}]}] +config = e.freeze_evaluation_config( + suite_id="trace-fixture-demo", tasks=tasks, weights={"uniform": 1}, + model={"provider": "synthetic", "name": "fixture", "effort": "low"}, + runtime={"workers": 1, "timeout_ms": 5000, "max_retries": 1, "heldout_attempt_cap": 4}, + split={"tasks": {"evolve": ["evolve-a"], "heldout": ["heldout-b"], "ood": ["ood-c"]}}) + +# Synthetic binding placeholders, not claims about reviewed Git source. +disabled = g.TraceSession(config, baseline_commit="a" * 40, snapshot_digest="b" * 64) +assert disabled.propose(b"[]")["status"] == "disabled" +session = g.TraceSession(config, baseline_commit="a" * 40, snapshot_digest="b" * 64, enabled=True) +candidate = session.propose(g.encode([{"task_id": "evolve-a", "failure": "missing-evidence", + "output_digest": e.digest("synthetic failure")}])) + + +def attempt(arm, number, correct): + # Prewritten observations exercise replay accounting, not a model's response + # to guidance. Neither their scores nor their costs measure real efficacy. + return {"task_id": "heldout-b", "attempt": number, + "input_digest": session.baseline_digest if arm == "baseline" else candidate["candidate_digest"], + "status": "completed", "result": {"facts": {"worker": {"enabled": not correct}}, + "evidence": {"worker": {"enabled": line}}}, + "safety": {key: True for key in e.SAFETY_GUARDS}, + "estimated": {"tokens": 20, "cost_microusd": 100, "duration_ms": 50}} + + +receipt = session.replay(g.encode({ + "binding_digest": session.binding_digest, "candidate_digest": candidate["candidate_digest"], + "split": "heldout", "attempts_per_task": 2, + "lanes": {"baseline": [attempt("baseline", 0, False), attempt("baseline", 1, True)], + "policy": [attempt("policy", 0, True), attempt("policy", 1, True)]}})) +assert receipt["body"]["comparable_fixture"] is True +assert receipt["accepted"] is False and receipt["promotion_allowed"] is False +assert g.verify_archive(session.export_archive(), expected_tip=session.archive_tip) +print(json.dumps({"default_off_verified": True, "synthetic": True, + "effectiveness_measured": False, "provider_calls": 0, + "guidance": session.render(candidate["candidate_digest"]), + "fixture_results": receipt["body"]["results"], + "estimated": receipt["body"]["estimated_attempt_totals"], + "actual": receipt["body"]["actual"], "archive_tip": session.archive_tip}, sort_keys=True)) diff --git a/plugins/fable5-codex/experiments/rrsi/README.md b/plugins/fable5-codex/experiments/rrsi/README.md index 3c1f12e..6776e4d 100644 --- a/plugins/fable5-codex/experiments/rrsi/README.md +++ b/plugins/fable5-codex/experiments/rrsi/README.md @@ -139,3 +139,64 @@ real Codex evaluation, host containment, Memory ingestion, owner promotion, installation, reload, rollback observation, merge or deployment is claimed. P2/P4 orchestration and Memory source contracts remain in the companion integrations workstream. OpenShell remains optional and independently gated. + +## Declarative trace guidance (offline, opt-in) + +`trace_guidance.py` adds `TraceSession` beside the existing frozen evaluator. +It does not invoke a model/provider, run generated code, install guidance, or +select a promotion. It defaults off. An explicitly enabled **synthetic** session +freezes the baseline commit, snapshot, evaluator source, guidance source, task +splits, model/runtime configuration and common arm budgets before proposals. +These hashes bind fixture inputs; they do not authenticate a live observation. + +`propose()` accepts bounded JSON bytes containing development task IDs, finite +failure codes and output digests. It maps these codes to the existing host-owned +`OPTIONS` text. Arbitrary instructions, executable hooks, extra authority fields +and held-out task IDs are rejected. A candidate has at most two context-only +layers. Parent candidates must already exist in that session. Rendered guidance +is returned separately; this module never changes governing instructions. + +`replay()` grades supplied structured results with the unchanged factual grader. +It reports the unchanged baseline's first attempt, the final flat retry, the +flat best-of-N **oracle**, and the final attempt under one fixed policy separately. +The first-attempt baseline is not an equal-compute arm; the complete flat and +policy lanes share the same attempt, estimated token/cost and wall-time limits. +Every attempt is counted, including failures. Policy token estimates include a +conservative UTF-8-byte context allowance, not a measured tokenizer count. +Missing results, unsafe attempts, unknown/zero consumption estimates, deadline +overruns, cancellation and excess budgets prevent a comparable fixture result. +Per-task regressions and policy violations remain explicit; no winner is promoted. + +Held-out/OOD replay seals the selected candidate before grading, rejects further +proposals or a different held-out policy, and charges a cumulative query cap even +for unsuccessful admitted runs. Development traces never receive held-out results. +Public fixtures labelled held-out test this separation mechanism; they are not a +secret representative benchmark. A caller can fabricate fixture results, so the +output always remains `synthetic_unverified`, with `accepted` and +`promotion_allowed` false. A higher fixture score cannot demonstrate real benefit. + +Inputs are limited to 256 KiB, 8,192 nodes, depth 12, bounded strings/collections, +20 traces, 16 suite tasks and 20 archive records. Context is capped at 4,096 bytes; +replay deadlines are at most five seconds and no larger than the frozen runtime +deadline. `cancel()` permanently stops the session and can be called by the host +from another thread. Checkpoints bound cooperative work; this is not an OS quota. + +The archive holds canonical immutable byte records in memory, with versioned +candidate/evidence hashes, provenance and a hash chain. Exports contain finite +guidance and recomputed summaries, never raw traces, task prompts or answers. +`verify_archive(bytes, expected_tip=host_tip)` requires a separately retained tip; +a caller-supplied self-hash is not authentication. Actual offline grading latency +and zero provider tokens/cost are separate from fixture **estimates**. Recreating +a session resets its cap. Durable global accounting, cross-process isolation, +authenticated observation import and production promotion remain separate work. + +```sh +python -I -B test/rrsi_trace_guidance_test.py -v +python -I -B examples/rrsi-trace-guidance-fixture.py +``` + +This clean implementation is informed by the bounded trace/archive and comparison +ideas in [Meta^n](https://arxiv.org/html/2608.24735v1) (Kim, Lee, Jwa and Kang) and +the [MIT-licensed reference project at b708184](https://github.com/minnesotanlp/meta-n/tree/b7081843d3c7b0e0f418ca10aaf2ccbff856e7f8). +No upstream code, paper text or figures are copied. Its generated executable +preprocessing design is excluded. Its benchmark results are not our results. diff --git a/plugins/fable5-codex/experiments/rrsi/trace_guidance.py b/plugins/fable5-codex/experiments/rrsi/trace_guidance.py new file mode 100644 index 0000000..f1711a1 --- /dev/null +++ b/plugins/fable5-codex/experiments/rrsi/trace_guidance.py @@ -0,0 +1,420 @@ +"""Opt-in fixture replay beside RRSI's frozen evaluator; no generated execution. + +The caller owns this session and its archive tip outside candidate reach. Python +objects and hashes do not establish process isolation, authentication or storage +durability. Only the explicit byte-input methods accept candidate/fixture data. +""" +from __future__ import annotations + +import json +import math +from pathlib import Path +from threading import Event +import time + +import evaluator as e +import source as source_contract +from source import OPTIONS + +SCHEMA = "fable-declarative-replay-v1" +MAX_BYTES = 262144 +MAX_TRACES = 20 +MAX_RECORDS = 20 +MAX_TASKS = 16 +MAX_CONTEXT_BYTES = 4096 # conservative token upper bound, not a tokenizer +ZERO = "0" * 64 +FAILURES = { + "missing-evidence": ("seek-refutation", "retain-evidence"), + "unsupported-claim": ("seek-refutation", "retain-unknowns"), + "missed-boundary": ("test-edge-cases", "retain-evidence"), + "unknown-cost": ("prioritize-uncertainty", "retain-costs"), +} +BOUNDARY = {key: False for key in ( + "execution", "provider_dispatch", "evaluator_mutation", "tests_mutation", + "reward_mutation", "permissions_mutation", "approval_mutation", "spending_mutation", + "identity_mutation", "promotion_mutation", "authority_mutation")} + + +class GuidanceDenied(ValueError): + pass + + +class _Stopped(Exception): + pass + + +def encode(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=True, allow_nan=False).encode() + + +def _copy(value): + return json.loads(encode(value)) + + +def _fields(value, keys): + if type(value) is not dict or set(value) != set(keys): + raise GuidanceDenied("unexpected fields") + + +def _integer(value, low, high): + if type(value) is not int or not low <= value <= high: + raise GuidanceDenied("bounded integer required") + + +def _hash(value, length=64): + if type(value) is not str or len(value) != length or any(c not in "0123456789abcdef" for c in value): + raise GuidanceDenied("exact digest required") + + +def _pairs(items): + result = {} + for key, value in items: + if key in result: + raise GuidanceDenied("duplicate JSON field") + result[key] = value + return result + + +def _bounded(value, depth=0, nodes=None): + nodes = [0] if nodes is None else nodes + nodes[0] += 1 + if depth > 12 or nodes[0] > 8192: + raise GuidanceDenied("input complexity exceeded") + if type(value) is str: + if len(value) > 2048: + raise GuidanceDenied("input string exceeded") + elif value is None or type(value) is bool: + pass + elif type(value) is int: + if abs(value) > 10**12: + raise GuidanceDenied("input integer exceeded") + elif type(value) is float: + if not math.isfinite(value) or abs(value) > 10**12: + raise GuidanceDenied("invalid input number") + elif type(value) in (list, dict): + if len(value) > 256: + raise GuidanceDenied("input collection exceeded") + if type(value) is dict: + for key in value: + if type(key) is not str or len(key) > 128: + raise GuidanceDenied("invalid input key") + for child in value.values() if type(value) is dict else value: + _bounded(child, depth + 1, nodes) + else: + raise GuidanceDenied("plain JSON required") + return value + + +def decode(raw): + if type(raw) is not bytes or not 0 < len(raw) <= MAX_BYTES: + raise GuidanceDenied("bounded JSON bytes required") + try: + return _bounded(json.loads(raw, object_pairs_hook=_pairs)) + except (ValueError, UnicodeError, RecursionError) as error: + raise GuidanceDenied("invalid bounded JSON") from error + + +def _envelope(kind, body, previous, sequence): + record = {"schema": SCHEMA, "kind": kind, "sequence": sequence, + "previous": previous, "body": body, "authority_boundary": dict(BOUNDARY), + "accepted": False, "promotion_allowed": False, + "evidence_class": "synthetic_unverified"} + record["digest"] = e.digest(record) + return record + + +def verify_archive(raw, *, expected_tip): + """Check an export against a separately retained host tip, never self-trust.""" + archive = decode(raw) + _fields(archive, ("schema", "records", "tip")) + _hash(expected_tip) + if archive["schema"] != SCHEMA or archive["tip"] != expected_tip: + raise GuidanceDenied("archive anchor mismatch") + records = archive["records"] + if type(records) is not list or not 0 <= len(records) <= MAX_RECORDS: + raise GuidanceDenied("archive record limit") + previous = ZERO + for index, record in enumerate(records): + _fields(record, ("schema", "kind", "sequence", "previous", "body", "authority_boundary", + "accepted", "promotion_allowed", "evidence_class", "digest")) + if (record["kind"] not in ("binding", "candidate", "replay") + or (index == 0) != (record["kind"] == "binding") + or type(record["body"]) is not dict + or record != _envelope(record["kind"], record["body"], previous, index)): + raise GuidanceDenied("archive record or chain mismatch") + previous = record["digest"] + if previous != expected_tip: + raise GuidanceDenied("archive tip mismatch") + return True + + +class TraceSession: + """Host-owned fixture session, frozen before proposals; never executes policy. + + An external host can call cancel() from another thread. Cancellation is + permanent. Reconstructing a session resets offline accounting; this API is + not a durable global query quota or a qualified live-observation importer. + """ + + def __init__(self, config, *, baseline_commit, snapshot_digest, enabled=False, + attempts_per_task=2, token_budget=100000, cost_budget_microusd=100000, + wall_ms=1000): + self.enabled = enabled is True + self._cancelled = Event() + self._records = () # canonical bytes, never a caller-mutable dict + self._candidates = {} + self._sealed_policy = None + self._heldout_used = 0 + if not self.enabled: + return + _hash(baseline_commit, 40); _hash(snapshot_digest) + _integer(attempts_per_task, 1, 5) + _integer(token_budget, 1, 10**7); _integer(cost_budget_microusd, 1, 10**9) + _integer(wall_ms, 1, 5000) + # Reconstruct, instead of trusting a caller-forged dataclass digest. + rebuilt = e.freeze_evaluation_config( + suite_id=config.suite_id, tasks=e._plain(config.tasks), weights=e._plain(config.weights), + model=e._plain(config.model), runtime=e._plain(config.runtime), split=e._plain(config.split)) + if (rebuilt != config or config.model["provider"] != "synthetic" + or len(config.tasks) > MAX_TASKS + or attempts_per_task > config.runtime["max_retries"] + 1 + or wall_ms > config.runtime["timeout_ms"]): + raise GuidanceDenied("fixture-only frozen configuration required") + self._config = rebuilt + self._budget = dict(attempts_per_task=attempts_per_task, token_budget=token_budget, + cost_budget_microusd=cost_budget_microusd, wall_ms=wall_ms) + self._binding = {key: getattr(config, key) for key in e.DIGEST_FIELDS if key != "snapshot_digest"} + self._binding.update(baseline_commit=baseline_commit, snapshot_digest=snapshot_digest, + guidance_digest=e.digest(Path(__file__).read_bytes()), + source_digest=e.digest(Path(source_contract.__file__).read_bytes()), + options_digest=e.digest(OPTIONS), budget=self._budget) + self._options = e._freeze(_copy(OPTIONS)) + self._binding_digest = e.digest(self._binding) + self._baseline_digest = e.digest({"binding": self._binding_digest, "guidance": []}) + self._append("binding", self._binding) + + @property + def binding_digest(self): + if not self.enabled: + raise GuidanceDenied("feature disabled") + return self._binding_digest + + @property + def baseline_digest(self): + if not self.enabled: + raise GuidanceDenied("feature disabled") + return self._baseline_digest + + @property + def archive_tip(self): + return json.loads(self._records[-1])["digest"] if self._records else ZERO + + def cancel(self): + self._cancelled.set() + + def _check(self, started): + if self._cancelled.is_set(): + raise _Stopped("cancelled") + if (time.monotonic() - started) * 1000 >= self._budget["wall_ms"]: + raise _Stopped("timeout") + + def _current(self, *, append=True): + if not self.enabled: + raise GuidanceDenied("feature disabled") + if (e.digest(Path(e.__file__).read_bytes()) != self._config.evaluator_digest + or e.digest(Path(__file__).read_bytes()) != self._binding["guidance_digest"] + or e.digest(Path(source_contract.__file__).read_bytes()) != self._binding["source_digest"] + or e.digest(OPTIONS) != self._binding["options_digest"]): + raise GuidanceDenied("frozen implementation drift") + if append and len(self._records) >= MAX_RECORDS: + raise GuidanceDenied("archive capacity exceeded") + + def _append(self, kind, body): + record = _envelope(kind, _copy(body), self.archive_tip, len(self._records)) + self._records += (encode(record),) + return _copy(record) + + def export_archive(self): + return encode({"schema": SCHEMA, "records": [json.loads(raw) for raw in self._records], + "tip": self.archive_tip}) + + def propose(self, raw_traces, *, parent_digest=None): + if not self.enabled: + return {"status": "disabled", "accepted": False, "promotion_allowed": False} + self._current() + if self._sealed_policy is not None: + raise GuidanceDenied("proposals are sealed after heldout admission") + started = time.monotonic() + try: + self._check(started) + traces = decode(raw_traces) + if type(traces) is not list or not 1 <= len(traces) <= MAX_TRACES: + raise GuidanceDenied("trace count exceeded") + evolve = set(self._config.split["tasks"]["evolve"]) + strategies, contexts, refs = set(), set(), [] + for trace in traces: + self._check(started) + _fields(trace, ("task_id", "failure", "output_digest")) + if type(trace["task_id"]) is not str or trace["task_id"] not in evolve: + raise GuidanceDenied("trace is outside development split") + if type(trace["failure"]) is not str or trace["failure"] not in FAILURES: + raise GuidanceDenied("finite failure code required") + _hash(trace["output_digest"]) + strategy, context = FAILURES[trace["failure"]] + strategies.add(strategy); contexts.add(context) + refs.append(e.digest(trace)) # never raw IDs, answers or instructions + layers = [] + if parent_digest is not None: + _hash(parent_digest) + if parent_digest not in self._candidates: + raise GuidanceDenied("unknown parent") + layers = _copy(self._candidates[parent_digest]["layers"]) + if len(layers) >= 2: + raise GuidanceDenied("maximum guidance depth reached") + layers.append({"strategy": sorted(strategies), "context": sorted(contexts)}) + rendered = [self._options[kind][item] for layer in layers + for kind, items in layer.items() for item in items] + if len(encode(rendered)) > MAX_CONTEXT_BYTES: + raise GuidanceDenied("context token upper bound exceeded") + candidate = dict(binding_digest=self._binding_digest, parent_digest=parent_digest, + layers=layers, trace_refs=sorted(set(refs)), + context_token_upper_bound=len(encode(rendered))) + candidate["candidate_digest"] = e.digest(candidate) + self._check(started) + if candidate["candidate_digest"] not in self._candidates: + self._candidates[candidate["candidate_digest"]] = _copy(candidate) + self._append("candidate", candidate) + return _copy(candidate) + except _Stopped as stop: + return {"status": str(stop), "accepted": False, "promotion_allowed": False} + + def render(self, candidate_digest): + self._current(append=False) + if self._cancelled.is_set(): + raise GuidanceDenied("session cancelled") + _hash(candidate_digest) + if candidate_digest not in self._candidates: + raise GuidanceDenied("unknown candidate") + self._candidate(candidate_digest) + return [[self._options[kind][item] for kind, items in layer.items() for item in items] + for layer in self._candidates[candidate_digest]["layers"]] + + def _candidate(self, candidate_digest): + candidate = self._candidates[candidate_digest] + if (candidate["candidate_digest"] != candidate_digest + or candidate["binding_digest"] != self._binding_digest + or e.digest({k: v for k, v in candidate.items() if k != "candidate_digest"}) != candidate_digest): + raise GuidanceDenied("host candidate record drift") + return candidate + + def replay(self, raw_request): + if not self.enabled: + return {"status": "disabled", "accepted": False, "promotion_allowed": False} + self._current() + # Even an already-cancelled request is hashed into its receipt. Bound it + # before cancellation/deadline paths, rather than hashing unchecked data. + if type(raw_request) is not bytes or not 0 < len(raw_request) <= MAX_BYTES: + raise GuidanceDenied("bounded JSON bytes required") + started = time.monotonic() + rows, summaries = {}, [] + status = "completed" + candidate_id, split = None, None + metrics = {arm: dict(tokens=0, cost_microusd=0, duration_ms=0) for arm in ("baseline", "policy")} + violations = 0 + comparable = True + try: + self._check(started) + request = decode(raw_request) + _fields(request, ("binding_digest", "candidate_digest", "split", "attempts_per_task", "lanes")) + candidate_id, split = request["candidate_digest"], request["split"] + _hash(candidate_id) + if (request["binding_digest"] != self._binding_digest or candidate_id not in self._candidates + or split not in ("evolve", "heldout", "ood") + or type(request["attempts_per_task"]) is not int + or request["attempts_per_task"] != self._budget["attempts_per_task"]): + raise GuidanceDenied("frozen request binding mismatch") + _fields(request["lanes"], ("baseline", "policy")) + self._candidate(candidate_id) + tasks = [e._plain(task) for task in self._config.tasks if task["split"] == split] + task_ids = {task["id"] for task in tasks} + attempts = self._budget["attempts_per_task"] + expected = {(identifier, attempt) for identifier in task_ids for attempt in range(attempts)} + if split != "evolve": + if self._sealed_policy not in (None, candidate_id): + raise GuidanceDenied("heldout policy is already fixed") + reserve = len(expected) * 2 + if self._heldout_used + reserve > self._config.runtime["heldout_attempt_cap"]: + raise GuidanceDenied("cumulative heldout budget exceeded") + self._heldout_used += reserve # failures and cancellations still consume admission + self._sealed_policy = candidate_id + for arm in ("baseline", "policy"): + lane = request["lanes"][arm] + if type(lane) is not list or len(lane) > len(expected): + raise GuidanceDenied("attempt count exceeded") + groups = rows[arm] = {} + expected_input = self._baseline_digest if arm == "baseline" else candidate_id + for row in lane: + self._check(started) + _fields(row, ("task_id", "attempt", "input_digest", "status", "result", "safety", "estimated")) + _integer(row["attempt"], 0, attempts - 1) + if type(row["task_id"]) is not str: + raise GuidanceDenied("invalid task identifier") + slot = (row["task_id"], row["attempt"]) + if slot not in expected or slot in groups or row["input_digest"] != expected_input: + raise GuidanceDenied("task/attempt/policy mismatch") + if row["status"] not in ("completed", "failed", "cancelled", "timeout", "unknown"): + raise GuidanceDenied("invalid attempt status") + _fields(row["safety"], e.SAFETY_GUARDS) + if any(type(flag) is not bool for flag in row["safety"].values()): + raise GuidanceDenied("explicit fixture safety flags required") + _fields(row["estimated"], ("tokens", "cost_microusd", "duration_ms")) + for key, value in row["estimated"].items(): + _integer(value, 0, 10**9) + metrics[arm][key] += value + # Charge conservative context overhead to every policy attempt. + if arm == "policy": + metrics[arm]["tokens"] += self._candidates[candidate_id]["context_token_upper_bound"] + unsafe = not all(row["safety"].values()) + violations += int(unsafe) + comparable &= (not unsafe and row["status"] == "completed" + and row["estimated"]["tokens"] > 0 + and row["estimated"]["cost_microusd"] > 0 + and row["estimated"]["duration_ms"] <= self._config.runtime["timeout_ms"]) + groups[slot] = row + comparable &= set(groups) == expected + if (metrics[arm]["tokens"] > self._budget["token_budget"] + or metrics[arm]["cost_microusd"] > self._budget["cost_budget_microusd"]): + comparable = False + status = "budget_exceeded" + for task in tasks: + self._check(started) + counts = {} + for arm in ("baseline", "policy"): + values = [] + for attempt in range(attempts): + self._check(started) + row = rows[arm].get((task["id"], attempt)) + passed = 0 + if row and row["status"] == "completed" and all(row["safety"].values()): + passed = sum(a["passed"] for a in e.grade_factual_task(task, row["result"])["assertions"]) + values.append(passed) + counts[arm] = values + summaries.append({"task_digest": e.digest(task["id"]), "assertions": len(task["assertions"]), + "baseline_first": counts["baseline"][0], "flat_retry": counts["baseline"][-1], + "flat_best_of_n_oracle": max(counts["baseline"]), + "single_policy": counts["policy"][-1]}) + self._check(started) + except _Stopped as stop: + status, comparable = str(stop), False + # No winner is selected. Scores are fixture facts, never efficacy evidence. + regressions = [row["task_digest"] for row in summaries if row["single_policy"] < row["baseline_first"]] + receipt = dict(binding_digest=self._binding_digest, candidate_digest=candidate_id, split=split, + request_digest=e.digest(raw_request), status=status, comparable_fixture=comparable, + qualification="incomplete" if not comparable else "regressed" if regressions else "fixture_only", + results=summaries, regressions=regressions, policy_violations=violations, + fixed_per_arm_budget=self._budget, estimated_attempt_totals=metrics, + actual={"offline_elapsed_ms": max(0, int((time.monotonic() - started) * 1000)), + "provider_tokens": 0, "provider_cost_microusd": 0}, + effectiveness_measured=False) + return self._append("replay", receipt) diff --git a/test/rrsi.test.mjs b/test/rrsi.test.mjs index cbb0958..0601a96 100644 --- a/test/rrsi.test.mjs +++ b/test/rrsi.test.mjs @@ -11,7 +11,8 @@ const python = [['python3'], ['python'], ['py', '-3']].find(([command, ...args]) 'import sys;sys.exit(0 if sys.version_info >= (3,10) else 1)'], { timeout: 10000, stdio: 'ignore', shell: false }).status === 0); -for (const file of ['rrsi_source_test.py', 'rrsi_evaluator_test.py', '../examples/rrsi-synthetic-comparison.py']) { +for (const file of ['rrsi_source_test.py', 'rrsi_evaluator_test.py', 'rrsi_trace_guidance_test.py', + '../examples/rrsi-synthetic-comparison.py', '../examples/rrsi-trace-guidance-fixture.py']) { test(`RRSI offline ${file}`, (t) => { assert.ok(python, 'Python 3.10+ is required for RRSI checks; coverage must not silently skip'); const [command, ...args] = python; diff --git a/test/rrsi_trace_guidance_test.py b/test/rrsi_trace_guidance_test.py new file mode 100644 index 0000000..43b6e57 --- /dev/null +++ b/test/rrsi_trace_guidance_test.py @@ -0,0 +1,284 @@ +"""Adversarial offline contract tests for TraceSession.""" +import copy +from dataclasses import replace +import importlib.util +import json +from pathlib import Path +import sys +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +BASE = ROOT / "plugins/fable5-codex/experiments/rrsi" +sys.path.insert(0, str(BASE)) + +def load(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec); sys.modules[name] = module + spec.loader.exec_module(module); return module + +e = load("evaluator", BASE / "evaluator.py") +g = load("trace_session_guidance", BASE / "trace_guidance.py") + +def task(identifier, split): + source = "export const enabled = false;" + prompts = {"a": "Check worker startup state.", "b": "Determine asynchronous dispatch gating.", "c": "Audit evidence for activation claims."} + return {"id": identifier, "repository": "repo-" + identifier, "family": "family-" + identifier, + "split": split, "prompt": prompts[identifier], "sources": {"worker.js": source + "\n"}, + "assertions": [{"id": "enabled", "path": "worker.enabled", "equals": False, + "evidence": {"path": "worker.js", "line": 1, "content_digest": e.digest(source.encode())}}]} + +def config(**changes): + args = {"suite_id": "trace-session-suite", "tasks": [task("a", "evolve"), task("b", "heldout"), task("c", "ood")], + "weights": {"uniform": 1}, "model": {"provider": "synthetic", "name": "fixture", "effort": "low"}, + "runtime": {"workers": 1, "timeout_ms": 5000, "max_retries": 1, "heldout_attempt_cap": 4}, + "split": {"tasks": {"evolve": ["a"], "heldout": ["b"], "ood": ["c"]}}} + args.update(changes); return e.freeze_evaluation_config(**args) + +def raw(value): return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() +def traces(*failures): return raw([{"task_id": "a", "failure": f, "output_digest": "1" * 64} for f in failures]) +def result(correct=True): + return {"facts": {"worker": {"enabled": not correct}}, "evidence": {"worker": {"enabled": { + "path": "worker.js", "line": 1, "content_digest": e.digest(b"export const enabled = false;")}}}} + +def row(session, task_id, attempt, arm="baseline", status="completed", correct=True, **changes): + digest = session.baseline_digest if arm == "baseline" else session._candidate_for_test + value = {"task_id": task_id, "attempt": attempt, "input_digest": digest, "status": status, + "result": result(correct) if status == "completed" else None, + "safety": {key: True for key in e.SAFETY_GUARDS}, + "estimated": {"tokens": 10, "cost_microusd": 10, "duration_ms": 1}} + value.update(changes); return value + +class TraceSessionTests(unittest.TestCase): + def setUp(self): + self.session = g.TraceSession(config(), baseline_commit="a" * 40, snapshot_digest="b" * 64, + enabled=True, attempts_per_task=2, token_budget=10000, + cost_budget_microusd=10000, wall_ms=1000) + self.candidate = self.session.propose(traces("missing-evidence")) + self.session._candidate_for_test = self.candidate["candidate_digest"] + + def request(self, split="evolve", candidate=None, baseline_rows=None, policy_rows=None, attempts=2): + return raw({"binding_digest": self.session.binding_digest, "candidate_digest": candidate or self.candidate["candidate_digest"], + "split": split, "attempts_per_task": attempts, + "lanes": {"baseline": baseline_rows if baseline_rows is not None else [row(self.session, "a", i) for i in range(attempts)], + "policy": policy_rows if policy_rows is not None else [row(self.session, "a", i, "policy") for i in range(attempts)]}}) + + def test_disabled_is_default_off(self): + disabled = g.TraceSession(config(), baseline_commit="a" * 40, snapshot_digest="b" * 64) + self.assertEqual(disabled.propose(traces("missing-evidence"))["status"], "disabled") + self.assertEqual(disabled.replay(b"{}")["status"], "disabled") + with self.assertRaises(g.GuidanceDenied): disabled.binding_digest + with self.assertRaises(g.GuidanceDenied): disabled.baseline_digest + with self.assertRaises(g.GuidanceDenied): disabled.render("a" * 64) + self.assertTrue(g.verify_archive(disabled.export_archive(), expected_tip=g.ZERO)) + + def test_binding_baseline_and_snapshot_are_frozen(self): + self.assertEqual(len(self.session.binding_digest), 64) + changed = g.TraceSession(config(), baseline_commit="a" * 39 + "c", snapshot_digest="b" * 64, enabled=True) + self.assertNotEqual(changed.binding_digest, self.session.binding_digest) + with self.assertRaises(g.GuidanceDenied): self.session.replay(raw({"binding_digest": "c" * 64, "candidate_digest": self.candidate["candidate_digest"], "split": "evolve", "attempts_per_task": 2, "lanes": {"baseline": [], "policy": []}})) + + def test_config_rebuilt_and_provider_must_be_synthetic(self): + altered = config(model={"provider": "fixture", "name": "fixture", "effort": "low"}) + with self.assertRaises(g.GuidanceDenied): g.TraceSession(altered, baseline_commit="a" * 40, snapshot_digest="b" * 64, enabled=True) + with self.assertRaises(g.GuidanceDenied): + g.TraceSession(replace(config(), config_digest="d" * 64), baseline_commit="a" * 40, + snapshot_digest="b" * 64, enabled=True) + + def test_trace_proposal_maps_finite_dev_failures_to_options(self): + candidate = self.session.propose(traces("unsupported-claim", "missed-boundary")) + self.assertEqual(candidate["binding_digest"], self.session.binding_digest) + self.assertIn("Try to refute", " ".join(self.session.render(candidate["candidate_digest"])[0])) + self.assertTrue(all(isinstance(ref, str) and len(ref) == 64 for ref in candidate["trace_refs"])) + + def test_malformed_or_unknown_failure_and_heldout_trace_denied(self): + for trace in ([{"task_id": "a", "failure": "free prose", "output_digest": "1" * 64}], + [{"task_id": "b", "failure": "missing-evidence", "output_digest": "1" * 64}], + [{"task_id": "a", "failure": "missing-evidence", "output_digest": "bad"}]): + with self.subTest(trace=trace), self.assertRaises(g.GuidanceDenied): self.session.propose(raw(trace)) + + def test_raw_trace_fields_and_duplicate_fields_denied(self): + with self.assertRaises(g.GuidanceDenied): self.session.propose(raw([{"task_id": "a", "failure": "missing-evidence", "output_digest": "1" * 64, "prompt": "secret"}])) + duplicate = b'[{"task_id":"a","task_id":"a","failure":"missing-evidence","output_digest":"' + b"1" * 64 + b'"}]' + with self.assertRaises(g.GuidanceDenied): self.session.propose(duplicate) + + def test_proposal_bounds_and_depth_parent(self): + with self.assertRaises(g.GuidanceDenied): self.session.propose(raw([])) + with self.assertRaises(g.GuidanceDenied): self.session.propose(raw([{"task_id": "a", "failure": "missing-evidence", "output_digest": "1" * 64}] * 21)) + with self.assertRaises(g.GuidanceDenied): self.session.propose(traces("missing-evidence"), parent_digest="2" * 64) + child = self.session.propose(traces("unsupported-claim"), parent_digest=self.candidate["candidate_digest"]) + self.assertEqual(len(child["layers"]), 2) + with self.assertRaises(g.GuidanceDenied): self.session.propose(traces("missed-boundary"), parent_digest=child["candidate_digest"]) + + def test_candidate_render_is_digest_bound(self): + self.assertEqual(len(self.session.render(self.candidate["candidate_digest"])), 1) + with self.assertRaises(g.GuidanceDenied): self.session.render("0" * 64) + self.session._candidates["3" * 64] = copy.deepcopy(self.candidate) + with self.assertRaises(g.GuidanceDenied): self.session.render("3" * 64) + + def test_replay_strict_binding_and_legitimate_fixture(self): + receipt = self.session.replay(self.request()) + self.assertEqual(receipt["kind"], "replay") + self.assertEqual(receipt["body"]["qualification"], "fixture_only") + self.assertEqual(receipt["body"]["actual"]["provider_tokens"], 0) + self.assertEqual(receipt["body"]["actual"]["provider_cost_microusd"], 0) + self.assertFalse(receipt["body"]["effectiveness_measured"]) + + def test_replay_binding_candidate_split_and_attempts_denied(self): + for changes in ({"binding_digest": "c" * 64}, {"candidate_digest": "d" * 64}, {"split": "ood"}): + request = json.loads(self.request()); request.update(changes) + with self.subTest(changes=changes), self.assertRaises(g.GuidanceDenied): self.session.replay(raw(request)) + + def test_duplicate_missing_and_unknown_attempt_rows_denied(self): + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(baseline_rows=[row(self.session, "a", 0)] * 2)) + receipt = self.session.replay(self.request(baseline_rows=[row(self.session, "a", 0)])) + self.assertFalse(receipt["body"]["comparable_fixture"]) + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(baseline_rows=[row(self.session, "x", i) for i in range(2)])) + + def test_unsafe_unknown_zero_and_bad_estimates_invalidate(self): + for changes in ({"status": "unknown"}, {"estimated": {"tokens": 0, "cost_microusd": 10, "duration_ms": 1}}, + {"safety": {key: False for key in e.SAFETY_GUARDS}}): + receipt = self.session.replay(self.request(baseline_rows=[row(self.session, "a", 0, **changes), row(self.session, "a", 1)])) + self.assertFalse(receipt["body"]["comparable_fixture"]) + + def test_policy_input_hash_is_candidate_not_baseline(self): + bad = [row(self.session, "a", i, "policy") for i in range(2)]; bad[0]["input_digest"] = self.session.baseline_digest + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(policy_rows=bad)) + + def test_heldout_seals_policy_and_cumulative_cap(self): + request = {"binding_digest": self.session.binding_digest, "candidate_digest": self.candidate["candidate_digest"], "split": "heldout", "attempts_per_task": 2, "lanes": {"baseline": [], "policy": []}} + first = self.session.replay(raw(request)); self.assertEqual(first["body"]["status"], "completed") + with self.assertRaises(g.GuidanceDenied): self.session.propose(traces("unknown-cost")) + with self.assertRaises(g.GuidanceDenied): self.session.replay(raw({**request, "candidate_digest": "f" * 64})) + + def test_heldout_attempt_cap_is_cumulative(self): + small = g.TraceSession(config(runtime={"workers": 1, "timeout_ms": 5000, "max_retries": 1, "heldout_attempt_cap": 2}), baseline_commit="a" * 40, snapshot_digest="b" * 64, enabled=True, attempts_per_task=1) + candidate = small.propose(traces("missing-evidence")); request = {"binding_digest": small.binding_digest, "candidate_digest": candidate["candidate_digest"], "split": "heldout", "attempts_per_task": 1, "lanes": {"baseline": [], "policy": []}} + small.replay(raw(request)) + with self.assertRaises(g.GuidanceDenied): small.replay(raw(request)) + + def test_cancel_is_permanent_and_no_winner(self): + self.session.cancel(); receipt = self.session.replay(self.request()) + self.assertEqual(receipt["body"]["status"], "cancelled"); self.assertFalse(receipt["body"]["comparable_fixture"]); self.assertNotIn("winner", receipt["body"]) + with self.assertRaises(g.GuidanceDenied): self.session.replay(b"x" * (g.MAX_BYTES + 1)) + + def test_deadline_monotonic_mid_grade(self): + with patch.object(g.time, "monotonic", side_effect=[0.0, 0.0, 2.0, 2.0]): receipt = self.session.replay(self.request()) + self.assertEqual(receipt["body"]["status"], "timeout"); self.assertFalse(receipt["body"]["comparable_fixture"]) + + def test_budget_exceeded_is_not_comparable(self): + low = g.TraceSession(config(), baseline_commit="a" * 40, snapshot_digest="b" * 64, enabled=True, token_budget=1, cost_budget_microusd=1) + candidate = low.propose(traces("missing-evidence")); low._candidate_for_test = candidate["candidate_digest"] + request = {"binding_digest": low.binding_digest, "candidate_digest": candidate["candidate_digest"], "split": "evolve", "attempts_per_task": 2, "lanes": {"baseline": [row(low, "a", i) for i in range(2)], "policy": [row(low, "a", i, "policy") for i in range(2)]}} + self.assertFalse(low.replay(raw(request))["body"]["comparable_fixture"]) + + def test_archive_export_verifies_against_external_tip(self): + self.session.replay(self.request()); archive = self.session.export_archive() + self.assertTrue(g.verify_archive(archive, expected_tip=self.session.archive_tip)) + with self.assertRaises(g.GuidanceDenied): g.verify_archive(archive, expected_tip="0" * 64) + + def test_archive_old_record_tamper_and_rehashed_semantics_denied(self): + self.session.replay(self.request()); archive = json.loads(self.session.export_archive()); archive["records"][0]["body"]["baseline_commit"] = "secrets" + with self.assertRaises(g.GuidanceDenied): g.verify_archive(raw(archive), expected_tip=self.session.archive_tip) + archive = json.loads(self.session.export_archive()); archive["records"][0]["digest"] = "f" * 64; archive["tip"] = "f" * 64 + with self.assertRaises(g.GuidanceDenied): g.verify_archive(raw(archive), expected_tip="f" * 64) + + def test_archive_records_are_bounded_and_authority_fixed(self): + record = json.loads(self.session.export_archive())["records"][0] + self.assertEqual(record["authority_boundary"], g.BOUNDARY); self.assertFalse(record["accepted"]); self.assertFalse(record["promotion_allowed"]) + self.assertTrue(g.verify_archive(raw({"schema": g.SCHEMA, "records": [], "tip": g.ZERO}), expected_tip=g.ZERO)) + + def test_unknown_request_fields_and_oversized_json_denied(self): + request = json.loads(self.request()); request["private_payload"] = "x" + with self.assertRaises(g.GuidanceDenied): self.session.replay(raw(request)) + with self.assertRaises(g.GuidanceDenied): self.session.replay(b"{" + b"x" * 262144 + b"}") + + def test_grader_and_vocabulary_drift_refuse_work(self): + with patch.object(g.Path, "read_bytes", return_value=b"changed source"): + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request()) + with patch.dict(g.OPTIONS["strategy"], {"seek-refutation": "new instruction"}): + with self.assertRaises(g.GuidanceDenied): self.session.render(self.candidate["candidate_digest"]) + + def test_json_depth_nodes_strings_numbers_and_unknown_authority_are_bounded(self): + nested = [] + for _ in range(14): nested = [nested] + for value in (nested, [[0] * 40] * 220, "x" * 2049, 10**13, float("nan"), {"x" * 129: 1}): + with self.subTest(kind=type(value).__name__), self.assertRaises(g.GuidanceDenied): + self.session.propose(raw(value)) + request = json.loads(self.request()) + request["lanes"]["policy"][0]["authority"] = {"spend": True} + with self.assertRaises(g.GuidanceDenied): self.session.replay(raw(request)) + + def test_attempt_count_and_integer_estimates_cannot_be_forged(self): + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(attempts=3)) + for field, value in (("attempt", True), ("attempt", 2)): + rows = [row(self.session, "a", i) for i in range(2)]; rows[0][field] = value + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(baseline_rows=rows)) + for value in (True, 1.5, -1): + rows = [row(self.session, "a", i) for i in range(2)] + rows[0]["estimated"]["tokens"] = value + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request(baseline_rows=rows)) + + def test_fixed_policy_flat_last_and_oracle_max_are_distinct(self): + baseline = [row(self.session, "a", 0, correct=True), row(self.session, "a", 1, correct=False)] + policy = [row(self.session, "a", 0, "policy", correct=True), row(self.session, "a", 1, "policy", correct=False)] + body = self.session.replay(self.request(baseline_rows=baseline, policy_rows=policy))["body"] + scores = body["results"][0] + self.assertEqual((scores["baseline_first"], scores["flat_retry"], scores["flat_best_of_n_oracle"], scores["single_policy"]), (1, 0, 1, 0)) + self.assertEqual(body["qualification"], "regressed") + self.assertEqual(body["regressions"], [e.digest("a")]) + self.assertEqual(body["estimated_attempt_totals"]["baseline"]["cost_microusd"], 20) + self.assertEqual(body["estimated_attempt_totals"]["policy"]["tokens"], + 20 + 2 * self.candidate["context_token_upper_bound"]) + + def test_heldout_evaluation_rejects_known_alternative_policy(self): + alternative = self.session.propose(traces("unknown-cost")) + heldout = self.request(split="heldout", baseline_rows=[row(self.session, "b", i) for i in range(2)], + policy_rows=[row(self.session, "b", i, "policy") for i in range(2)]) + self.assertTrue(self.session.replay(heldout)["body"]["comparable_fixture"]) + request = json.loads(heldout); request["candidate_digest"] = alternative["candidate_digest"] + with self.assertRaisesRegex(g.GuidanceDenied, "policy is already fixed"): + self.session.replay(raw(request)) + + def test_cancel_during_grading_is_terminal(self): + original = e.grade_factual_task + def stop_after_first(*args): + result = original(*args); self.session.cancel(); return result + with patch.object(e, "grade_factual_task", side_effect=stop_after_first): + body = self.session.replay(self.request())["body"] + self.assertEqual(body["status"], "cancelled") + self.assertEqual(body["qualification"], "incomplete") + self.assertFalse(body["comparable_fixture"]) + self.assertEqual(self.session.propose(traces("unknown-cost"))["status"], "cancelled") + + def test_candidate_export_copies_do_not_mutate_future_work(self): + exported = self.session.propose(traces("missing-evidence")) + exported["layers"][0]["strategy"] = ["run-shell"] + rendered = self.session.render(self.candidate["candidate_digest"]) + self.assertNotIn("run-shell", str(rendered)) + archive = json.loads(self.session.export_archive()); archive["records"].clear() + self.assertEqual(len(json.loads(self.session.export_archive())["records"]), 2) + + def test_authority_flags_cannot_be_rehashed_into_archive(self): + archive = json.loads(self.session.export_archive()) + last = archive["records"][-1]; last["accepted"] = True + last["digest"] = e.digest({key: value for key, value in last.items() if key != "digest"}) + archive["tip"] = last["digest"] + with self.assertRaises(g.GuidanceDenied): + g.verify_archive(raw(archive), expected_tip=archive["tip"]) + + def test_archive_capacity_refuses_further_work(self): + for _ in range(g.MAX_RECORDS - 2): self.session.replay(self.request()) + tip = self.session.archive_tip + with self.assertRaises(g.GuidanceDenied): self.session.replay(self.request()) + self.assertEqual(self.session.archive_tip, tip) + self.assertTrue(g.verify_archive(self.session.export_archive(), expected_tip=tip)) + + def test_reproduction_preserves_scores_binding_and_estimates(self): + left = self.session.replay(self.request())["body"] + self.setUp() + right = self.session.replay(self.request())["body"] + left["actual"].pop("offline_elapsed_ms"); right["actual"].pop("offline_elapsed_ms") + self.assertEqual(left, right) + +if __name__ == "__main__": unittest.main() From 95280698faed6c29555325815ce860eda7527695 Mon Sep 17 00:00:00 2001 From: Rhein1 / Agoragentic <140637602+rhein1@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:51:54 -0400 Subject: [PATCH 2/2] fix: bind archive verification to canonical received values --- .../experiments/rrsi/trace_guidance.py | 7 +++- test/rrsi_trace_guidance_test.py | 32 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/plugins/fable5-codex/experiments/rrsi/trace_guidance.py b/plugins/fable5-codex/experiments/rrsi/trace_guidance.py index f1711a1..046ba3e 100644 --- a/plugins/fable5-codex/experiments/rrsi/trace_guidance.py +++ b/plugins/fable5-codex/experiments/rrsi/trace_guidance.py @@ -138,10 +138,15 @@ def verify_archive(raw, *, expected_tip): for index, record in enumerate(records): _fields(record, ("schema", "kind", "sequence", "previous", "body", "authority_boundary", "accepted", "promotion_allowed", "evidence_class", "digest")) + # Python equality treats False == 0 and True == 1. Recompute from the + # received fields first, then compare canonical encodings, so neither + # envelope substitutions nor numeric type changes retain an old tip. + received_digest = e.digest({key: value for key, value in record.items() if key != "digest"}) if (record["kind"] not in ("binding", "candidate", "replay") or (index == 0) != (record["kind"] == "binding") or type(record["body"]) is not dict - or record != _envelope(record["kind"], record["body"], previous, index)): + or record["digest"] != received_digest + or encode(record) != encode(_envelope(record["kind"], record["body"], previous, index))): raise GuidanceDenied("archive record or chain mismatch") previous = record["digest"] if previous != expected_tip: diff --git a/test/rrsi_trace_guidance_test.py b/test/rrsi_trace_guidance_test.py index 43b6e57..40fc820 100644 --- a/test/rrsi_trace_guidance_test.py +++ b/test/rrsi_trace_guidance_test.py @@ -267,6 +267,38 @@ def test_authority_flags_cannot_be_rehashed_into_archive(self): with self.assertRaises(g.GuidanceDenied): g.verify_archive(raw(archive), expected_tip=archive["tip"]) + def test_numeric_bool_substitutions_cannot_retain_original_tip(self): + original = self.session.export_archive() + for field, value in (("accepted", 0), ("promotion_allowed", 0), + ("sequence", True), ("sequence", 1.0)): + changed = json.loads(original) + changed["records"][-1][field] = value + with self.subTest(field=field, value=value), self.assertRaises(g.GuidanceDenied): + g.verify_archive(raw(changed), expected_tip=self.session.archive_tip) + changed = json.loads(original) + changed["records"][-1]["authority_boundary"]["execution"] = 0 + with self.assertRaises(g.GuidanceDenied): + g.verify_archive(raw(changed), expected_tip=self.session.archive_tip) + # Whitespace and object-key order do not change canonical data. + self.assertTrue(g.verify_archive(json.dumps(json.loads(original), indent=2).encode(), + expected_tip=self.session.archive_tip)) + + def test_rehashing_mistyped_envelope_cannot_make_it_valid(self): + for field, value in (("accepted", 0), ("promotion_allowed", 0), ("sequence", True), ("sequence", 1.0)): + changed = json.loads(self.session.export_archive()) + record = changed["records"][-1]; record[field] = value + record["digest"] = e.digest({key: item for key, item in record.items() if key != "digest"}) + changed["tip"] = record["digest"] + with self.subTest(field=field, value=value), self.assertRaises(g.GuidanceDenied): + g.verify_archive(raw(changed), expected_tip=changed["tip"]) + + def test_nested_numeric_value_changes_are_bound_to_received_record_digest(self): + archive = json.loads(self.session.export_archive()) + record = archive["records"][-1] + record["body"]["context_token_upper_bound"] = float(record["body"]["context_token_upper_bound"]) + with self.assertRaises(g.GuidanceDenied): + g.verify_archive(raw(archive), expected_tip=self.session.archive_tip) + def test_archive_capacity_refuses_further_work(self): for _ in range(g.MAX_RECORDS - 2): self.session.replay(self.request()) tip = self.session.archive_tip