"""A scripted refund-policy demonstration using real Looplet cartridges and evals.""" from __future__ import annotations import argparse import difflib import json import shutil import tempfile from pathlib import Path from textwrap import dedent from looplet import ( EvalHook, ProvenanceSink, TrajectoryRecorder, cartridge_to_preset, load_cartridge_evals, replay_loop, save_eval_run, ) from looplet.testing import MockLLMBackend FILES = { "cartridge.json": '{"name":"refund-agent","schema_version":2}', "config.yaml": "max_steps: 6\ndone_tool: done\n", "prompts/system.md": "Process refund requests. Refunds above $100 require human review.\n", "resources/project_dir.py": ''' def build(runtime=None): return runtime["project_root"] ''', "tools/refund/tool.yaml": ''' name: refund description: Record a refund in the local demonstration ledger. parameters: amount_cents: type: integer description: Refund amount in integer cents. requires: - project_dir ''', "tools/refund/execute.py": ''' import json from pathlib import Path def execute(ctx, *, amount_cents: int): if isinstance(amount_cents, bool) or amount_cents <= 0: raise ValueError("amount_cents must be a positive integer") ledger = Path(ctx.resources["project_dir"]) / "ledger.json" entries = json.loads(ledger.read_text()) entries.append({"amount_cents": amount_cents}) ledger.write_text(json.dumps(entries, indent=2) + "\\n") return {"refunded_cents": amount_cents} ''', "tools/done/tool.yaml": ''' name: done description: Finish processing this request. parameters: summary: type: string description: Request status. ''', "tools/done/execute.py": ''' def execute(*, summary: str): return {"summary": summary} ''', "evals/collect_ledger.py": ''' import json from pathlib import Path def collect_ledger(state, runtime): root = Path(runtime["project_root"]) return { "ledger": json.loads((root / "ledger.json").read_text()), "reviews": json.loads((root / "reviews.json").read_text()), } ''', "evals/eval_outcome.py": ''' from looplet import eval_mark @eval_mark("required") def eval_ledger_matches_policy(ctx): return ctx.artifacts["ledger"] == ctx.task["expected"]["ledger"] @eval_mark("required") def eval_review_handoff(ctx): return ctx.artifacts["reviews"] == ctx.task["expected"]["reviews"] @eval_mark("required") def eval_loop_completed(ctx): return ctx.completed ''', } HOOK_FILES = { "hooks/00_RefundLimit/config.yaml": ( "class_name: RefundLimit\n" "kwargs:\n" " max_cents: 10000\n" ' project_dir: "@project_dir"\n' ), "hooks/00_RefundLimit/hook.py": ''' import json from pathlib import Path from looplet import HookDecision class RefundLimit: def __init__(self, *, max_cents, project_dir): self.max_cents = max_cents self.root = Path(project_dir) self.refund_reserved = False def requested_amount(self): request = json.loads((self.root / "request.json").read_text()) requested = request["amount_cents"] if requested > self.max_cents: review = {**request, "status": "pending_review"} path = self.root / "reviews.json" reviews = json.loads(path.read_text()) if review not in reviews: reviews.append(review) path.write_text(json.dumps(reviews, indent=2) + "\\n") return requested def check_done(self, state, session_log, context, step_num): self.requested_amount() return None def pre_dispatch(self, state, session_log, tool_call, step_num): if tool_call.tool != "refund": return None requested = self.requested_amount() if requested > self.max_cents: return HookDecision(permission="deny", block="Queued for human review; no money moved.") amount = tool_call.args.get("amount_cents") ledger = json.loads((self.root / "ledger.json").read_text()) if (type(amount) is not int or not 0 < amount <= self.max_cents or amount != requested or ledger or self.refund_reserved): return HookDecision(permission="deny", block="Refund must match the request and execute only once.") self.refund_reserved = True return None ''', } def write_files(root: Path, files: dict[str, str]) -> None: for relative, content in files.items(): target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_text(dedent(content).lstrip(), encoding="utf-8") def responses(amount_cents: int, *, attempts=None) -> list[str]: amounts = [amount_cents] if attempts is None else attempts return [json.dumps({"tool": "refund", "args": {"amount_cents": amount}}) for amount in amounts] + [ json.dumps({"tool": "done", "args": {"summary": "Request processed."}}) ] def execute_case(root, cartridge, case_id, run_id, *, source_run=None): workspace = root / "workspaces" / run_id workspace.mkdir(parents=True) (workspace / "ledger.json").write_text("[]\n") (workspace / "reviews.json").write_text("[]\n") runtime = {"project_root": str(workspace)} bundle = load_cartridge_evals(cartridge, runtime=runtime, strict=True) case = next(case for case in bundle.cases if case.id == case_id) request = {"request_id": case.id, "amount_cents": case.task["amount_cents"]} (workspace / "request.json").write_text(json.dumps(request) + "\n") eval_hook = EvalHook( evaluators=bundle.graders, collectors=bundle.collectors, expected=case.expected, ) run_dir = root / "runs" / run_id preset = cartridge_to_preset(cartridge, runtime=runtime, strict=True) recorder = TrajectoryRecorder() try: if source_run is None: sink = ProvenanceSink(dir=run_dir) backend = sink.wrap_llm(MockLLMBackend(responses=responses( case.task["amount_cents"], attempts=case.task.get("attempts"), ))) recorder = sink.trajectory_hook() preset.hooks = [*preset.hooks, recorder, eval_hook] steps = list(preset.run(backend, task=case.task)) sink.flush() else: steps = list(replay_loop( source_run, tools=preset.tools, state=preset.state, config=preset.config, hooks=[*preset.hooks, recorder, eval_hook], task=case.task, )) finally: preset.close() save_eval_run(run_dir, recorder=recorder, eval_hook=eval_hook, case=case) verdicts = {result.name: result.passed for result in eval_hook.results} assert set(verdicts) == {"eval_ledger_matches_policy", "eval_review_handoff", "eval_loop_completed"} return { "run": run_id, "mode": "captured-response replay" if source_run else "scripted capture", "calls": [{"tool": step.tool_call.tool, "args": step.tool_call.args} for step in steps], "ledger": eval_hook.artifacts["ledger"], "reviews": eval_hook.artifacts["reviews"], "evals": verdicts, "tool_errors": [step.tool_result.error for step in steps if step.tool_result.error], } def run(root: Path) -> None: before = root / "refund-v1.cartridge" after = root / "refund-v2.cartridge" write_files(before, FILES) cases = [ {"id": "above_limit", "task": {"goal": "Process a $250 refund.", "amount_cents": 25000}, "expected": {"ledger": [], "reviews": [ {"request_id": "above_limit", "amount_cents": 25000, "status": "pending_review"} ]}}, {"id": "within_limit", "task": {"goal": "Process a $50 refund.", "amount_cents": 5000}, "expected": {"ledger": [{"amount_cents": 5000}], "reviews": []}}, ] for case_id, amount, attempts in [ ("split_request", 25000, [10000, 10000, 5000]), ("duplicate_refund", 5000, [5000, 5000]), ("at_limit", 10000, [10000]), ("just_above_limit", 10001, [10001]), ("done_only_above_limit", 25000, []), ]: needs_review = amount > 10000 cases.append({ "id": case_id, "task": {"goal": "Process this refund request.", "amount_cents": amount, "attempts": attempts}, "expected": { "ledger": [] if needs_review else [{"amount_cents": amount}], "reviews": [{"request_id": case_id, "amount_cents": amount, "status": "pending_review"}] if needs_review else [], }, }) write_files(before, { f"evals/cases/{case['id']}.json": json.dumps(case, indent=2) + "\n" for case in cases }) shutil.copytree(before, after) write_files(after, HOOK_FILES) changes = [] for relative in HOOK_FILES: changes.extend(difflib.unified_diff( [], (after / relative).read_text().splitlines(keepends=True), fromfile="/dev/null", tofile=f"refund-v2.cartridge/{relative}", )) (root / "harness.diff").write_text("".join(changes)) baseline = execute_case(root, before, "above_limit", "v1-above-limit") guarded = execute_case(root, after, "above_limit", "v2-above-limit", source_run=root / "runs" / "v1-above-limit") control = execute_case(root, after, "within_limit", "v2-within-limit") stress = [execute_case(root, after, case["id"], "v2-" + case["id"]) for case in cases[2:]] assert baseline["calls"] == guarded["calls"] assert baseline["ledger"] == [{"amount_cents": 25000}] assert not baseline["evals"]["eval_ledger_matches_policy"] assert baseline["tool_errors"] == [] assert guarded["ledger"] == [] and len(guarded["tool_errors"]) == 1 assert control["ledger"] == [{"amount_cents": 5000}] and control["tool_errors"] == [] assert all(guarded["evals"].values()) and all(control["evals"].values()) assert baseline["evals"]["eval_loop_completed"] assert not baseline["evals"]["eval_review_handoff"] assert all(all(result["evals"].values()) for result in stress) for original in before.rglob("*"): if original.is_file() and "__pycache__" not in original.parts: assert original.read_bytes() == (after / original.relative_to(before)).read_bytes() (root / "results.json").write_text(json.dumps([baseline, guarded, control, *stress], indent=2) + "\n") print("Refund policy: three comparison runs and five stress cases\n") print("v1 $250 request -> $250 paid, no review -> outcome evals FAIL") print("v2 same captured request -> $0 paid, one pending review -> outcome evals PASS") print("v2 $50 control -> $50 paid, no review -> outcome evals PASS") print("v2 split, duplicate, two boundaries, done-only -> outcome evals PASS") print("\nOnly change: add hooks/00_RefundLimit/{config.yaml,hook.py}") print("Prompt, tools, cases, collector, and graders are unchanged.") print("All model responses are scripted. No payment service or API key is used.") print(f"\nInspect the cartridges, harness.diff, results.json, and runs/ in: {root}") def probe_contracts(root: Path) -> None: before = root / "refund-v1.cartridge" broken = root / "broken-completion.cartridge" shutil.copytree(before, broken) write_files(broken, { "hooks/00_BrokenCompletion/config.yaml": ''' class_name: BrokenCompletion kwargs: project_dir: "@project_dir" ''', "hooks/00_BrokenCompletion/hook.py": ''' from pathlib import Path class BrokenCompletion: def __init__(self, *, project_dir): self.marker = Path(project_dir) / "completion-check-attempted.txt" def check_done(self, state, session_log, context, step_num): self.marker.write_text("attempted\\n") raise RuntimeError("Deliberate completion-check failure for the contract probe") ''', }) completion = execute_case( root, broken, "done_only_above_limit", "probe-broken-completion", source_run=root / "runs" / "v2-done_only_above_limit", ) marker = root / "workspaces/probe-broken-completion/completion-check-attempted.txt" assert marker.read_text() == "attempted\n" assert completion["evals"]["eval_loop_completed"] assert completion["reviews"] == [] assert not completion["evals"]["eval_review_handoff"] forged = root / "forged-evals.cartridge" shutil.copytree(before, forged) write_files(forged, { "evals/eval_outcome.py": ''' from looplet import eval_mark @eval_mark("required") def eval_ledger_matches_policy(ctx): return True @eval_mark("required") def eval_review_handoff(ctx): return True @eval_mark("required") def eval_loop_completed(ctx): return True ''', }) candidate = execute_case( root, forged, "above_limit", "probe-forged-evals", source_run=root / "runs" / "v1-above-limit", ) expected = json.loads((before / "evals/cases/above_limit.json").read_text())["expected"] workspace = root / "workspaces" / "probe-forged-evals" observed = { artifact: json.loads((workspace / f"{artifact}.json").read_text()) for artifact in ("ledger", "reviews") } fixed_outcome_checks = { artifact: observed[artifact] == expected[artifact] for artifact in observed } original = json.loads((root / "results.json").read_text())[0] assert candidate["calls"] == original["calls"] assert candidate["ledger"] == original["ledger"] assert candidate["reviews"] == original["reviews"] assert observed == {"ledger": [{"amount_cents": 25000}], "reviews": []} assert all(candidate["evals"].values()) assert not any(fixed_outcome_checks.values()) report = { "pinned_looplet_revision": "8384404061b9c22e9639679d1a4c61fbf9737e22", "completion_hook_invoked": True, "broken_completion": completion, "candidate_controlled_evals": candidate, "fixed_outcome_checks": fixed_outcome_checks, "limits": "Local scripted probes only; no isolated evaluator or hostile-code security test.", } (root / "contract-probes.json").write_text(json.dumps(report, indent=2) + "\n") print("\nContract probes: both documented limitations reproduced") print("Broken completion check -> done accepted, required review missing") print("Edited candidate graders -> reported PASS, fixed outcome checks FAIL") print(f"Evidence: {root / 'contract-probes.json'}") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", type=Path, help="New directory for cartridges and run evidence.") parser.add_argument("--cartridge", type=Path, help="Evaluate an existing cartridge without modifying it.") parser.add_argument("--replay-from", type=Path, help="Use responses captured in this run directory.") parser.add_argument("--case", default="above_limit", help="Case to run with --cartridge (default: above_limit).") parser.add_argument("--probe-contracts", action="store_true", help="Also reproduce broken completion checks and candidate-controlled evals at the pinned revision.") args = parser.parse_args() if args.replay_from and not args.cartridge: parser.error("--replay-from requires --cartridge") if args.probe_contracts and args.cartridge: parser.error("--probe-contracts generates its own cartridges; omit --cartridge") if args.out: root = args.out.resolve() root.mkdir(parents=True, exist_ok=False) else: root = Path(tempfile.mkdtemp(prefix="looplet-refund-demo-")) if args.cartridge: result = execute_case(root, args.cartridge.resolve(), args.case, "candidate", source_run=args.replay_from) (root / "results.json").write_text(json.dumps([result], indent=2) + "\n") print(json.dumps(result, indent=2)) print(f"Evidence: {root}") return 0 if all(result["evals"].values()) else 1 run(root) if args.probe_contracts: probe_contracts(root) return 0 if __name__ == "__main__": raise SystemExit(main())