diff --git a/src/fi/alk/harness/authoring_entrypoint.py b/src/fi/alk/harness/authoring_entrypoint.py index 402e82ce1..22a328907 100644 --- a/src/fi/alk/harness/authoring_entrypoint.py +++ b/src/fi/alk/harness/authoring_entrypoint.py @@ -428,6 +428,8 @@ async def refuse_repair(phase: str, guidance: str) -> int: raise RuntimeValidationError( "environment", f"Certify-only validation failed: {_one_line(exc)}" ) from exc + if getattr(args, "prove_scenarios", False): + _prove_scenarios(output, getattr(args, "deadline_at", None)) try: # Written before sealing: the platform's fill gate keys on the sealed certificate. report_path.write_text( @@ -458,6 +460,19 @@ async def refuse_repair(phase: str, guidance: str) -> int: return 0 +SCENARIO_PROOFS = "scenario-proofs.json" + + +def _prove_scenarios(output: Path, deadline_at: float | None) -> None: + """Prove every scenario on this certification's authoring tree; never fails the certify.""" + from .prove_suite import prove_suite + + try: + prove_suite(output, output / SCENARIO_PROOFS, deadline_at=deadline_at) + except Exception as exc: # noqa: BLE001 - a missing proofs file reads as "no new proofs" + print(f"certify-only: scenario proofs not written: {_one_line(exc)}", file=sys.stderr, flush=True) + + def _load_provider_import_profile( job: HarnessJob, secrets_path: Path | None, @@ -674,6 +689,17 @@ def main(argv: list[str] | None = None, *, validate_runtime: bool = False) -> in type=Path, help="Directory the sealed bundle is written to in --certify-only mode", ) + parser.add_argument( + "--prove-scenarios", + action="store_true", + help="With --certify-only: prove every scenario after validation and before sealing", + ) + parser.add_argument( + "--deadline-at", + type=float, + default=None, + help="Epoch seconds the validation job is stopped at; scenario proving keeps a margin before it", + ) args = parser.parse_args(argv) if args.certify_only: if args.bundle_output is None: diff --git a/src/fi/alk/harness/authoring_runtime_validation.py b/src/fi/alk/harness/authoring_runtime_validation.py index 8d608d511..57ba529ba 100644 --- a/src/fi/alk/harness/authoring_runtime_validation.py +++ b/src/fi/alk/harness/authoring_runtime_validation.py @@ -1333,6 +1333,27 @@ async def check_setup_once(scenario, invariants, against, keep_processes): f"invariants, complete in {time.monotonic() - started:.0f}s", flush=True, ) + if not readiness_only: + # While the stack is up, play each case's reference and controls through the + # service-backed tools for real; never fails validation. + from .live_proof import prove_live + + try: + live = await prove_live( + authoring=authoring, + contract=AgentContract.model_validate_json( + (bundle / "contract.json").read_text(encoding="utf-8") + ), + runtime=runtime, + provider=provider, + factory=factory, + work=work, + seed=job.seed or 0, + ) + if live: + print(f"runtime validation: proved {len(live)} cases on the live runtime", flush=True) + except Exception as exc: # noqa: BLE001 - live proofs only add evidence + print(f"runtime validation: live proofs not run: {type(exc).__name__}: {exc}", flush=True) if generic: await provider.reset(runtime, work_directory=work) reset_world = await factory.create( diff --git a/src/fi/alk/harness/builder.py b/src/fi/alk/harness/builder.py index 21fe0200f..ff7737c68 100644 --- a/src/fi/alk/harness/builder.py +++ b/src/fi/alk/harness/builder.py @@ -210,7 +210,12 @@ async def prove_scenarios(args: dict[str, Any]) -> dict[str, Any]: lines.append( f"{name}: {'holds' if proof.holds else 'does not hold'} ({gates})" + ("" if proof.holds else f". {proof.why()}") - + (" Judged-only: nothing observable to check." if proof.judged_only else "") + + ( + " Judged-only: nothing observable to check." + + ("" if proof.controls_failed else f" {proof.why()}") + if proof.judged_only + else "" + ) ) return _ok("\n".join(lines) or "name at least one scenario") diff --git a/src/fi/alk/harness/folder.py b/src/fi/alk/harness/folder.py index 42cd2d0be..26d5bc602 100644 --- a/src/fi/alk/harness/folder.py +++ b/src/fi/alk/harness/folder.py @@ -144,6 +144,9 @@ def document_for(scenario: Scenario) -> dict: body = scenario.model_dump() body.pop("setup_code", None) body.pop("ready_code", None) + # Written only when set, so a suite saved before must_pass existed reads as unchanged. + if not body.get("must_pass"): + body.pop("must_pass", None) return body diff --git a/src/fi/alk/harness/grounding.py b/src/fi/alk/harness/grounding.py new file mode 100644 index 000000000..29da270db --- /dev/null +++ b/src/fi/alk/harness/grounding.py @@ -0,0 +1,197 @@ +"""Whether each source a scenario cites says what the scenario claims it says. + +A proof does not block a case with no sources (the save path refuses a new or changed one, so a +guess is written as an `assumption` source). A source that does not resolve is blocking, because a false citation is worse than none. Chat messages and evidence cases +live on the platform and are resolved there. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import tempfile +from pathlib import Path +from typing import Any, Callable + +from .leaks import normalise + +# Kinds whose resolution ALK decides; the rest are left to the platform. +ALK_RESOLVED_KINDS = frozenset({"rule", "file", "requirement", "assumption"}) +_FILE_ROOTS = ("knowledge", "") +POLICY_CHECKS = ".futureagi-policy-checks.json" +logger = logging.getLogger(__name__) + + +def _file(world_root: Path, ref: str) -> Path | None: + name = ref.removeprefix("file:").strip().lstrip("/") + if not name or ".." in name.split("/"): + return None + for root in _FILE_ROOTS: + candidate = Path(world_root) / root / name if root else Path(world_root) / name + if candidate.is_file(): + return candidate + return None + + +def resolve_sources(scenario: Any, contract: Any, world_root: Path) -> list[dict[str, Any]]: + """One entry per cited source: `{kind, ref, resolved, problem, ...}`; `resolved` None means + the platform decides it.""" + rules = list(getattr(contract, "hard_constraints", None) or []) if contract is not None else [] + resolved: list[dict[str, Any]] = [] + for source in getattr(scenario, "sources", None) or []: + kind, ref, quote = source.kind, source.ref, source.quote + entry: dict[str, Any] = {"kind": kind, "ref": ref, "resolved": True, "problem": ""} + if kind == "rule": + wanted = normalise(quote) + matched = next((at for at, rule in enumerate(rules) if wanted and wanted in normalise(rule)), None) + if matched is None: + entry.update(resolved=False, problem="no rule of the agent contains the quoted words") + else: + entry["resolved_ref"] = f"rule:{matched}" + elif kind == "file": + path = _file(world_root, ref) + if path is None: + # Uploaded files may live only on the platform; it resolves them there. + entry.update(resolved=None, problem="the file is not in the authoring tree") + else: + text = path.read_bytes() + entry["matched_digest"] = "sha256:" + hashlib.sha256(text).hexdigest() + if normalise(quote) not in normalise(text.decode("utf-8", errors="replace")): + entry.update(resolved=False, problem=f"{path.name} does not contain the quoted words") + elif kind == "requirement": + if not quote.strip(): + entry.update(resolved=False, problem="a requirement needs the person's own words") + elif kind == "assumption": + pass + else: + entry.update(resolved=None, problem="resolved on the platform") + resolved.append(entry) + return resolved + + +def unresolved(entries: list[dict[str, Any]]) -> list[dict[str, Any]]: + """The entries ALK resolved and found false.""" + return [one for one in entries if one["kind"] in ALK_RESOLVED_KINDS and one.get("resolved") is False] + + +def expected_outcome(scenario: Any, catalogue: Any) -> str: + """What the case expects, in words: each of its sub-goals' `what`.""" + lines = [] + for name in getattr(scenario, "sub_goals", None) or []: + goal = catalogue.named(name) if catalogue is not None else None + what = str(getattr(goal, "what", "") or "").strip() + if what: + lines.append(f"- {what}") + return "\n".join(lines) + + +_CONFLICT_PROMPT = ( + "A test case for an agent expects the outcome below, and cites the agent's rule below as its source. " + "Decide whether the expected outcome contradicts the rule: expecting the agent to do what the rule " + "forbids, or a number, limit, window or condition the rule states differently. A case that only " + "covers part of the rule, or adds details the rule does not mention, does not contradict it.\n\n" + "Expected outcome:\n{expected}\n\nRule:\n{rule}\n\nCall decide." +) + + +def _harness_conflict_judge(expected: str, rule: str) -> tuple[bool | None, str]: + """One forced judge call at temperature 0: (conflict, why). None when no judge can run here.""" + from .backends import tool + from .judge import NoGateway, _complete + from .tools import schema + + @tool( + "decide", + "Whether the expected outcome contradicts the rule.", + schema({"conflict": bool, "why": str}, ["conflict", "why"]), + ) + async def decide(args: dict[str, Any]) -> dict[str, Any]: # pragma: no cover - never executed + return {} + + conversation = [{"role": "user", "content": _CONFLICT_PROMPT.format(expected=expected, rule=rule)}] + try: + message = _complete(conversation, [decide], True, temperature=0) + except NoGateway: + return None, "" + for call in message.get("tool_calls") or []: + function = call.get("function") or {} + if function.get("name") != "decide": + continue + try: + args = json.loads(function.get("arguments") or "{}") + except ValueError: + continue + if isinstance(args.get("conflict"), bool): + return args["conflict"], str(args.get("why") or "").strip()[:300] + return None, "" + + +def _load_checks(world_root: Path) -> dict[str, dict[str, Any]]: + try: + value = json.loads((Path(world_root) / POLICY_CHECKS).read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return value if isinstance(value, dict) else {} + + +def _save_checks(world_root: Path, checks: dict[str, dict[str, Any]]) -> None: + root = Path(world_root) + try: + root.mkdir(parents=True, exist_ok=True) + handle, temporary = tempfile.mkstemp(dir=root, prefix=".policy-checks-", suffix=".json") + with os.fdopen(handle, "w", encoding="utf-8") as out: + json.dump(checks, out, indent=2, sort_keys=True, ensure_ascii=False) + os.replace(temporary, root / POLICY_CHECKS) + except OSError as exc: # a cache that cannot be written only costs a repeated call + logger.warning("policy check cache not written: %s", exc) + + +def policy_conflicts( + scenario: Any, + catalogue: Any, + rules: list[str], + world_root: Path, + *, + judge: Callable[[str, str], tuple[bool | None, str]] | None = None, +) -> list[dict[str, Any]]: + """For each cited rule that resolves, whether the case's expected outcome contradicts it: + `{rule, ref, conflict, why}`. `conflict` None means no judge could decide it here.""" + expected = expected_outcome(scenario, catalogue) + if not expected: + return [] + decide = judge or _harness_conflict_judge + cache = _load_checks(world_root) + changed = False + found: list[dict[str, Any]] = [] + for entry in resolve_sources(scenario, None if not rules else _Rules(rules), world_root): + if entry["kind"] != "rule" or entry.get("resolved") is not True: + continue + at = int(str(entry.get("resolved_ref", "rule:-1")).split(":")[1]) + rule = rules[at] if 0 <= at < len(rules) else "" + if not rule: + continue + key = hashlib.sha256(f"{expected}\n---\n{rule}".encode()).hexdigest() + if key in cache: + conflict, why = cache[key].get("conflict"), str(cache[key].get("why") or "") + else: + try: + conflict, why = decide(expected, rule) + except Exception as exc: # noqa: BLE001 - a judge that could not run decides nothing + logger.warning("policy conflict judge could not run: %s", type(exc).__name__) + conflict, why = None, "" + if conflict is not None: + cache[key] = {"conflict": conflict, "why": why} + changed = True + found.append({"rule": rule, "ref": entry["ref"], "conflict": conflict, "why": why}) + if changed: + _save_checks(world_root, cache) + return found + + +class _Rules: + """A contract stand-in carrying only the rules, for resolving rule sources.""" + + def __init__(self, rules: list[str]) -> None: + self.hard_constraints = rules diff --git a/src/fi/alk/harness/judge.py b/src/fi/alk/harness/judge.py index a816bba56..aeaefe388 100644 --- a/src/fi/alk/harness/judge.py +++ b/src/fi/alk/harness/judge.py @@ -557,7 +557,7 @@ def _transcript(messages: Sequence[Any]) -> str: def _complete( - conversation: list[dict[str, Any]], tools: Sequence[Any], forced: bool + conversation: list[dict[str, Any]], tools: Sequence[Any], forced: bool, *, temperature: float | None = None ) -> dict[str, Any]: """One turn of the judge on the gateway's OpenAI-compatible chat completions.""" env = provider_env(judge_model()) @@ -578,6 +578,8 @@ def _complete( ] if forced: body["tool_choice"] = {"type": "function", "function": {"name": "decide"}} + if temperature is not None: + body["temperature"] = temperature response = requests.post( f"{base}/v1/chat/completions", headers={"Authorization": f"Bearer {key}"}, diff --git a/src/fi/alk/harness/leaks.py b/src/fi/alk/harness/leaks.py new file mode 100644 index 000000000..53a91dcda --- /dev/null +++ b/src/fi/alk/harness/leaks.py @@ -0,0 +1,315 @@ +"""Whether what the agent or the simulated people see gives a scenario's answer away. + +Structural, with no phrase list. Grader-only text (what the checks and judges expect) is compared +with the text each audience is given: an 8-word span repeated, a literal a check compares against +that the world does not hold, or a sub-goal's multi-word name. Whether the agent's process can +*open* the grader files is a separate question, answered by the runtime boundary, not here. +""" + +from __future__ import annotations + +import ast +import json +import re +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Iterable + +from .catalogue import Catalogue + +SHINGLE = 8 +EXCERPT = 120 + +# Fields only the grader reads. Importable by the runtime-boundary probe so both name one list. +GRADER_ONLY_FIELDS = ( + "sub_goal.what", + "sub_goal.judged", + "sub_goal.reference", + "sub_goal.verifier", + "sub_goal.comparison", + "sub_goal.check (compared literals)", + "scenario.tests", + "scenario.negative_controls", + "scenario.solution (argument values)", + "evaluator_only files", +) +# What each audience is given. +VISIBLE_FIELDS = { + "simulated person": ("instruction", "variables", "persona"), + "tested agent": ("instruction (when not conversational)", "persona.initial_message", "variables (opening message)"), + "actor": ("actor instructions", "actor knowledge"), +} + + +@dataclass +class Leak: + who: str + field: str + grader_field: str + excerpt: str + + def as_dict(self) -> dict[str, str]: + return asdict(self) + + +def normalise(text: str) -> str: + text = str(text or "").casefold().replace("_", " ").replace("-", " ") + text = re.sub(r"[^\w\s.]", " ", text) + # A decimal point inside a number is kept; any other full stop is punctuation. + text = re.sub(r"(? list[str]: + return normalise(text).split() + + +def _contains(haystack: list[str], needle: list[str]) -> int: + """Index of `needle` as a contiguous token run in `haystack`, or -1.""" + if not needle or len(needle) > len(haystack): + return -1 + for at in range(len(haystack) - len(needle) + 1): + if haystack[at : at + len(needle)] == needle: + return at + return -1 + + +def _excerpt(tokens: list[str]) -> str: + return " ".join(tokens)[:EXCERPT] + + +def _returned(value: ast.expr) -> list[ast.expr]: + """The expressions a return can hand back, looking through conditional expressions.""" + if isinstance(value, ast.IfExp): + return _returned(value.body) + _returned(value.orelse) + if isinstance(value, ast.BoolOp): + return [leaf for one in value.values for leaf in _returned(one)] + return [value] + + +def _check_literals(source: str) -> set[str]: + """String literals with a digit or 3+ tokens, and decimal numbers, from comparison and argument + positions of a check. Never from a return, a raise or an f-string: a check returns a sentence + saying what was wrong, and that sentence is ordinary instruction text.""" + try: + tree = ast.parse(source) + except SyntaxError: + return set() + excluded: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Raise, ast.JoinedStr)): + excluded.update(id(inner) for inner in ast.walk(node)) + elif isinstance(node, ast.Return) and node.value is not None: + # `return None if world... == 42.5 else "wrong"`: the condition is a comparison and + # its literals count; only what the check can hand back is a sentence. + for value in _returned(node.value): + excluded.update(id(inner) for inner in ast.walk(value)) + positions: list[ast.AST] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Compare): + positions.append(node.left) + positions.extend(node.comparators) + elif isinstance(node, ast.Call): + positions.extend(node.args) + positions.extend(keyword.value for keyword in node.keywords) + elif isinstance(node, ast.Subscript): + positions.append(node.slice) + found: set[str] = set() + for position in positions: + for inner in ast.walk(position): + if id(inner) in excluded or not isinstance(inner, ast.Constant): + continue + value = inner.value + if isinstance(value, bool): + continue + if isinstance(value, str): + if any(ch.isdigit() for ch in value) or len(_tokens(value)) >= 3: + found.add(value) + elif isinstance(value, float): + found.add(f"{value:.2f}" if round(value, 2) == value else repr(value)) + return found + + +def _solution_literals(scenario: Any) -> set[str]: + """Argument values of the reference steps a caller would not know: strings with a digit or 3+ + tokens, and decimal numbers. Values the caller already holds are dropped by the caller.""" + found: set[str] = set() + + def visit(value: Any) -> None: + if isinstance(value, bool): + return + if isinstance(value, dict): + for item in value.values(): + visit(item) + elif isinstance(value, (list, tuple)): + for item in value: + visit(item) + elif isinstance(value, str) and not value.startswith("$"): + if any(ch.isdigit() for ch in value) or len(_tokens(value)) >= 3: + found.add(value) + elif isinstance(value, float): + # As a person would write it: 37.5 and 37.50 are the same answer. + found.add(repr(value)) + if round(value, 2) == value: + found.add(f"{value:.2f}") + + for step in getattr(scenario, "solution", None) or []: + visit(dict(getattr(step, "arguments", None) or {})) + return found + + +def _strings(value: Any) -> Iterable[str]: + if isinstance(value, str): + yield value + elif isinstance(value, dict): + for key, item in value.items(): + yield str(key) + yield from _strings(item) + elif isinstance(value, (list, tuple, set)): + for item in value: + yield from _strings(item) + elif isinstance(value, (int, float)) and not isinstance(value, bool): + yield str(value) + + +def _in_world(literal: str, world_text: list[str]) -> bool: + needle = _tokens(literal) + return any(_contains(_tokens(text), needle) >= 0 for text in world_text) + + +def _evaluator_only_texts(world_root: Path | None) -> list[tuple[str, str]]: + if world_root is None: + return [] + try: + document = json.loads((Path(world_root) / "actors.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + return [] + rules = document.get("rules") if isinstance(document, dict) else None + names = ((rules or {}).get("evaluator_only") or {}).get("files") or [] + texts = [] + for name in names: + for candidate in (Path(world_root) / "knowledge" / str(name), Path(world_root) / str(name)): + try: + texts.append((f"evaluator_only:{name}", candidate.read_text(encoding="utf-8"))) + break + except (OSError, UnicodeDecodeError): + continue + return texts + + +def _takes_part(scenario: Any, actor: dict[str, Any]) -> bool: + actor_id = actor.get("id") + start = getattr(scenario, "starts_with", None) + if isinstance(start, dict) and start.get("actor") == actor_id: + return True + return any(isinstance(one, dict) and one.get("id") == actor_id for one in getattr(scenario, "actors", None) or []) + + +def leaks( + scenario: Any, + catalogue: Catalogue, + contract: Any = None, + world_state: Any = None, + actors: list[dict[str, Any]] | None = None, + *, + world_root: Path | None = None, +) -> list[Leak]: + """Every place a visible field repeats grader-only material. `world_state` None skips rule (b).""" + grader: list[tuple[str, str]] = [] + literals: set[str] = set() + names: list[str] = [] + for name in getattr(scenario, "sub_goals", None) or []: + sub_goal = catalogue.named(name) + if sub_goal is None: + continue + names.append(name) + for field_name in ("what", "judged", "reference", "verifier", "comparison"): + text = getattr(sub_goal, field_name, "") or "" + if text.strip(): + grader.append((f"sub_goal:{name}.{field_name}", text)) + if sub_goal.check: + literals |= _check_literals(sub_goal.check) + if str(getattr(scenario, "tests", "") or "").strip(): + grader.append(("tests", scenario.tests)) + for control in getattr(scenario, "negative_controls", None) or []: + grader.append((f"negative_control:{control.name}.why", control.why)) + grader.extend(_evaluator_only_texts(world_root)) + + rule_b = world_state is not None + if rule_b: + world_text = list(_strings(world_state)) + [str(getattr(scenario, "setup_code", "") or "")] + literals = {one for one in literals if not _in_world(one, world_text)} + + persona = getattr(scenario, "persona", None) + persona_values = persona.model_dump() if hasattr(persona, "model_dump") else (persona or {}) + variables = getattr(scenario, "variables", None) or {} + instruction = str(getattr(scenario, "instruction", "") or "") + initial = str((persona_values or {}).get("initial_message") or "") + # The caller's own facts (their variables, persona, instruction and the records the world holds) + # are theirs to say; any other value the reference sends is the answer. + caller_text = [instruction, *_strings(variables)] + caller_text += [ + text for key, value in (persona_values or {}).items() if key != "initial_message" for text in _strings(value) + ] + if rule_b: + caller_text += world_text + answers = {one for one in _solution_literals(scenario) if not _in_world(one, caller_text)} + visible: list[tuple[str, str, str]] = [("simulated person", "instruction", instruction)] + visible += [("simulated person", f"variables.{key}", text) for key, value in variables.items() for text in _strings(value)] + visible += [ + ("simulated person", f"persona.{key}", text) + for key, value in (persona_values or {}).items() + for text in _strings(value) + ] + conversational = getattr(contract, "conversational", True) if contract is not None else True + agent_fields = [("persona.initial_message", initial)] + agent_fields += [(f"variables.{key}", text) for key, value in variables.items() for text in _strings(value)] + if not conversational: + agent_fields.append(("instruction", instruction)) + visible += [("tested agent", field_name, text) for field_name, text in agent_fields] + for actor in actors or []: + if not isinstance(actor, dict) or not _takes_part(scenario, actor): + continue + label = str(actor.get("name") or actor.get("id") or "actor") + for key in ("instructions", "instruction", "brief", "knowledge"): + for text in _strings(actor.get(key)): + visible.append(("actor", f"{label}.{key}", text)) + + found: list[Leak] = [] + seen: set[tuple[str, str, str]] = set() + + def add(who: str, field_name: str, grader_field: str, excerpt: str) -> None: + if (who, field_name, grader_field) not in seen: + seen.add((who, field_name, grader_field)) + found.append(Leak(who, field_name, grader_field, excerpt)) + + for who, field_name, text in visible: + tokens = _tokens(text) + if not tokens: + continue + sources = list(grader) + if who == "tested agent" and conversational and field_name == "persona.initial_message": + # On a conversational agent the instruction is the simulated person's; repeating it as the + # opening message hands the whole hidden instruction to the tested agent. + sources.append(("instruction", instruction)) + for grader_field, grader_text in sources: + grader_tokens = _tokens(grader_text) + for at in range(len(grader_tokens) - SHINGLE + 1): + span = grader_tokens[at : at + SHINGLE] + if _contains(tokens, span) >= 0: + add(who, field_name, grader_field, _excerpt(span)) + break + if rule_b: + for literal in sorted(literals): + needle = _tokens(literal) + if _contains(tokens, needle) >= 0: + add(who, field_name, "check literal", _excerpt(needle)) + for answer in sorted(answers): + needle = _tokens(answer) + if _contains(tokens, needle) >= 0: + add(who, field_name, "scenario.solution", _excerpt(needle)) + for name in names: + needle = _tokens(name) + if len(needle) >= 2 and _contains(tokens, needle) >= 0: + add(who, field_name, f"sub_goal:{name}.name", _excerpt(needle)) + return found diff --git a/src/fi/alk/harness/live_proof.py b/src/fi/alk/harness/live_proof.py new file mode 100644 index 000000000..6a50ef1ec --- /dev/null +++ b/src/fi/alk/harness/live_proof.py @@ -0,0 +1,222 @@ +"""Proving cases on the live runtime for tools the agent serves over HTTP. + +Offline, every hosted tool is runtime-owned, so a case's reference solution and its negative +controls are recorded without running and the case reads "kept as written". While certification +has the agent's own stack up, a tool whose contract entry is `service` can be played for real: +the bundle's tool proxy forwards a JSON POST to the agent's tools service, records the call, and +the change lands in the real datastore. So, per case and with a reset between each: set up, play +the reference and run the checks; set up, play nothing and run the checks; set up, play each +control and run the checks. A tool the agent runs inside its own process has no endpoint and stays +assumed. The result per case is written beside the authoring tree and laid over the offline proof +by `prove_suite`, so leaks, sources and rule checks still come from the offline proof. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import random +import tempfile +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +LIVE_PROOFS = ".futureagi-live-proofs.json" +TOOL_PROXY = "tool_proxy" + + +def service_endpoints(contract: Any, endpoints: dict[str, Any]) -> dict[str, str]: + """Contract tool -> route behind the tool proxy, for tools whose entry mode is `service`. + + Empty when the runtime has no tool proxy: nothing could be played.""" + if TOOL_PROXY not in (endpoints or {}): + return {} + bound: dict[str, str] = {} + for name in contract.tool_names(): + entry = contract.entry_for(name) + if entry is not None and entry.mode == "service": + bound[name] = str(entry.endpoint or name).strip("/") + return bound + + +def _address(endpoint: Any) -> str: + return str(getattr(endpoint, "address", endpoint) or "") + + +def _outcomes(scenario: Any, catalogue: Any, world: Any, calls: list[Any]) -> list[Any]: + from .checks import reference_of, run_check + from .prove import _checks_for + + reference = reference_of(scenario.solution) + return [run_check(source, world, calls, name=name, reference=reference) for name, source in _checks_for(scenario, catalogue)] + + +def _play(scenario: Any, world: Any, steps: list[Any]) -> tuple[list[Any], list[str], list[str]]: + """Set up, play `steps`, and say which ran unbound (assumed) and which were refused.""" + from .folder import apply_setup, check_ready + from .prove import executes, play_reference_step + + applied = apply_setup(scenario, world) + if not applied.ok: + raise RuntimeError(f"setup: {applied.said}") + ready = check_ready(scenario, world) + if not ready.ok: + raise RuntimeError(f"ready: {ready.said}") + world.calls = [] + refused = [] + for step in steps: + call = play_reference_step(world, step) + if not call.ok: + refused.append(f"{call.name}({step.arguments}): {call.error}") + assumed = sorted({str(step.tool) for step in steps if not executes(world, str(step.tool))}) + return list(world.calls), assumed, refused + + +def _digest(world: Any) -> str | None: + from .authoring_runtime_validation import _world_state_digest + + try: + return _world_state_digest(world) + except Exception: # noqa: BLE001 - an unreadable state compares as different + return None + + +async def prove_live( + *, + authoring: Path, + contract: Any, + runtime: Any, + provider: Any, + factory: Any, + work: Path, + seed: int = 0, +) -> dict[str, dict[str, Any]]: + """Play each case's reference, an empty run and its controls on the live runtime; write and + return `{scenario_key: live result}`. Returns {} when no tool is service-bound.""" + from .catalogue import load_catalogue + from .proofs import proof_key + from .scenario_tools import load_scenarios + + bound = service_endpoints(contract, runtime.endpoints) + if not bound: + return {} + proxy = _address(runtime.endpoints[TOOL_PROXY]) + scenarios = load_scenarios(authoring) + catalogue = load_catalogue(authoring) + results: dict[str, dict[str, Any]] = {} + + async def fresh() -> Any: + await provider.reset(runtime, work_directory=work) + world = await factory.create(runtime, rng=random.Random(seed)) + world.bind_tools(proxy, bound, contract.tool_names()) + return world + + for scenario in scenarios: + key = proof_key(scenario) + if not any(str(step.tool) in bound for step in scenario.solution) and not any( + str(step.tool) in bound for control in scenario.negative_controls for step in control.steps + ): + continue + try: + world = await fresh() + calls, assumed, refused = await asyncio.to_thread(_play, scenario, world, list(scenario.solution)) + with_solution = await asyncio.to_thread(_outcomes, scenario, catalogue, world, calls) + right = await asyncio.to_thread(_digest, world) + world = await fresh() + nothing_calls, _, _ = await asyncio.to_thread(_play, scenario, world, []) + with_nothing = await asyncio.to_thread(_outcomes, scenario, catalogue, world, nothing_calls) + untouched = await asyncio.to_thread(_digest, world) + controls = [] + for control in scenario.negative_controls: + try: + world = await fresh() + control_calls, control_assumed, _ = await asyncio.to_thread(_play, scenario, world, list(control.steps)) + outcomes = await asyncio.to_thread(_outcomes, scenario, catalogue, world, control_calls) + except Exception as exc: # noqa: BLE001 - a control that cannot be played is no evidence, the case still is + controls.append( + {"name": control.name, "why": control.why, "failed": None, "held": [], "assumed": [], + "error": f"{type(exc).__name__}: {exc}"[:300]} + ) + continue + named = set(control.must_fail) or {one.name for one in outcomes} + controls.append( + { + "name": control.name, + "why": control.why, + "failed": any(not one.held and not one.broken for one in outcomes if one.name in named) + if outcomes + else None, + "held": [one.name for one in outcomes if one.held], + "broken": [one.name for one in outcomes if one.broken and one.name in named], + "assumed": control_assumed, + } + ) + except Exception as exc: # noqa: BLE001 - a case the live lane cannot play keeps its offline proof + logger.warning("live proof skipped for %s: %s: %s", key, type(exc).__name__, exc) + continue + results[key] = { + "assumed": assumed, + "refused": refused, + "with_solution": [{"name": one.name, "held": one.held, "said": one.said, "broken": one.broken} for one in with_solution], + "with_nothing": [{"name": one.name, "held": one.held, "said": one.said, "broken": one.broken} for one in with_nothing], + "outcome": "no_change" if not assumed and right is not None and right == untouched else "change", + "controls": controls, + } + await provider.reset(runtime, work_directory=work) + write_live(authoring, results) + return results + + +def write_live(authoring: Path, results: dict[str, dict[str, Any]]) -> None: + """Merge `results` into the live proofs file by key, so other keys keep theirs.""" + path = Path(authoring) / LIVE_PROOFS + merged = {**load_live(authoring), **results} + handle, temporary = tempfile.mkstemp(dir=Path(authoring), prefix=".live-proofs-", suffix=".json") + try: + with os.fdopen(handle, "w", encoding="utf-8") as out: + json.dump(merged, out, indent=2, sort_keys=True, default=str) + os.replace(temporary, path) + except BaseException: + Path(temporary).unlink(missing_ok=True) + raise + + +def load_live(authoring: Path) -> dict[str, dict[str, Any]]: + try: + value = json.loads((Path(authoring) / LIVE_PROOFS).read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return value if isinstance(value, dict) else {} + + +def apply_live(proof: Any, live: dict[str, Any]) -> Any: + """The offline proof with its gates replaced by what the live runtime showed. + + Leaks, sources and rule checks stay the offline proof's; readiness, the reference, the empty run + and the controls come from the live run, which is the stronger evidence.""" + from .checks import Outcome + from .prove import control_broken + + def outcomes(rows: list[dict[str, Any]]) -> list[Any]: + return [Outcome(row["name"], bool(row["held"]), str(row.get("said") or ""), broken=bool(row.get("broken"))) for row in rows] + + proof.deferred = False + proof.ready = True + proof.why_not_ready = "" + proof.with_solution = outcomes(live.get("with_solution") or []) + proof.with_nothing = outcomes(live.get("with_nothing") or []) + proof.assumed = list(live.get("assumed") or []) + proof.refused = list(live.get("refused") or []) + proof.broken = [one.name for one in proof.with_solution if one.broken] + proof.solvable = all(one.held for one in proof.with_solution) and not proof.broken + proof.controls = [dict(one) for one in live.get("controls") or []] + proof.broken += control_broken(proof.controls) + proof.weak = [one.name for one in proof.with_nothing if one.held] + proof.outcome = str(live.get("outcome") or "change") + proof.vacuous = bool(proof.with_nothing) and len(proof.weak) == len(proof.with_nothing) + if proof.outcome == "no_change" and any(one.get("failed") for one in proof.ran_controls): + proof.vacuous = False + return proof diff --git a/src/fi/alk/harness/proofs.py b/src/fi/alk/harness/proofs.py index 6d9e03ea3..9ae217fb5 100644 --- a/src/fi/alk/harness/proofs.py +++ b/src/fi/alk/harness/proofs.py @@ -54,7 +54,7 @@ def _entry(proof: Proof, solution_steps: int) -> dict[str, Any]: why = "" elif not proof.holds: why = proof.why() - elif proof.assumed: + elif proof.assumed and blocked == "solution_assumed": why = ( "kept as written: these tools have no endpoint here, so their solution steps never ran: " + ", ".join(proof.assumed) @@ -75,6 +75,12 @@ def _entry(proof: Proof, solution_steps: int) -> dict[str, Any]: "solution_steps": solution_steps, "deferred": proof.deferred, "judged_only": proof.judged_only, + "outcome": proof.outcome, + "controls": [dict(one) for one in proof.controls], + "leaks": [dict(one) for one in proof.leaks], + "sources": [dict(one) for one in proof.sources], + "policy_checks": [dict(one) for one in proof.policy_checks], + "evidence": {name: list(kinds) for name, kinds in proof.evidence.items()}, "blocked_reason": blocked, } @@ -101,6 +107,12 @@ def unprovable_entry(why: str, blocked_reason: str = "broken_check") -> dict[str "solution_steps": 0, "deferred": False, "judged_only": False, + "outcome": "change", + "controls": [], + "leaks": [], + "sources": [], + "policy_checks": [], + "evidence": {}, "blocked_reason": blocked_reason, } diff --git a/src/fi/alk/harness/prove.py b/src/fi/alk/harness/prove.py index a33d34b00..d6bf9e7ac 100644 --- a/src/fi/alk/harness/prove.py +++ b/src/fi/alk/harness/prove.py @@ -42,7 +42,8 @@ # The lane could not execute something; the case runs labelled "kept as written", never as proved. -LANE_CODES = frozenset({"proof_deferred", "solution_assumed", "control_not_run"}) +# "judged_only": only a judge grades the case and no control has shown the judge fails a wrong behaviour. +LANE_CODES = frozenset({"proof_deferred", "solution_assumed", "control_not_run", "judged_only"}) @dataclass @@ -77,6 +78,52 @@ class Proof: too_strict: list[str] = field(default_factory=list) # Values the scenario marks fixed that no code check notices when they are wrong. unchecked: list[str] = field(default_factory=list) + # Each negative control played: {name, why, failed, held, assumed}. A control with assumed + # steps did not run: its tools have no endpoint, so the world never changed. + controls: list[dict[str, Any]] = field(default_factory=list) + # "no_change" when the reference leaves the world as doing nothing does: a refusal case. + outcome: str = "change" + # Visible text repeating grader-only material: {who, field, grader_field, excerpt}. + leaks: list[dict[str, str]] = field(default_factory=list) + # Each cited source and whether it resolved: {kind, ref, resolved, problem, ...}. + sources: list[dict[str, Any]] = field(default_factory=list) + # Each cited rule checked against the expected outcome: {rule, ref, conflict, why}. + policy_checks: list[dict[str, Any]] = field(default_factory=list) + # What each deterministic check reads: "tool_calls" when it uses `calls`, "world_state" when it + # uses `world`. The platform compares it with what the agent gives the harness. + evidence: dict[str, list[str]] = field(default_factory=dict) + + @property + def conflicts(self) -> list[dict[str, Any]]: + """Cited rules the judge found the expected outcome contradicts.""" + return [one for one in self.policy_checks if one.get("conflict") is True] + + @property + def ran_controls(self) -> list[dict[str, Any]]: + """Controls played by code to a pass or fail; one that could not be played (no checks, a + crash) is no evidence either way.""" + return [ + one + for one in self.controls + if not one.get("assumed") and not one.get("judged") and one.get("failed") is not None + ] + + @property + def judged_controls(self) -> list[dict[str, Any]]: + """Controls the judge decided: played through the judge, which said fail or pass.""" + return [one for one in self.controls if one.get("judged") and one.get("failed") is not None] + + @property + def controls_failed(self) -> bool: + """Every control that ran failed, and at least one ran: by code, or by the judge for a judged-only case.""" + ran = self.judged_controls if self.judged_only else self.ran_controls + return bool(ran) and all(one.get("failed") for one in ran) + + @property + def unresolved_sources(self) -> list[dict[str, Any]]: + from .grounding import unresolved + + return unresolved(self.sources) @property def holds(self) -> bool: @@ -84,7 +131,7 @@ def holds(self) -> bool: self.ready and self.solvable and not self.vacuous - and not self.weak + and (self.outcome == "no_change" or not self.weak) and not self.failed_attempts and not self.broken and not self.too_strict @@ -94,7 +141,15 @@ def holds(self) -> bool: @property def proved(self) -> bool: """Holds, and the reference solution really ran: nothing assumed, nothing deferred.""" - return self.holds and not self.assumed and not self.deferred + return ( + self.holds + and not self.assumed + and not self.deferred + and self.controls_failed + and not self.leaks + and not self.unresolved_sources + and not self.conflicts + ) @property def kept_as_written(self) -> bool: @@ -115,7 +170,7 @@ def blocked_reason(self) -> str | None: return "not_solvable" if self.vacuous: return "vacuous" - if self.weak: + if self.weak and self.outcome == "change": return "weak_check" if self.failed_attempts: return "credits_failed_attempt" @@ -123,10 +178,27 @@ def blocked_reason(self) -> str | None: return "too_strict" if self.unchecked: return "unchecked" + if not self.judged_only: + if not self.controls: + return "no_negative_control" + if any(one.get("failed") is False for one in self.ran_controls): + return "negative_control_passed" + elif any(one.get("failed") is False for one in self.judged_controls): + return "negative_control_passed" + if self.conflicts: + return "policy_conflict" + if self.leaks: + return "answer_leaked" + if self.unresolved_sources: + return "source_unresolved" if self.deferred: return "proof_deferred" if self.assumed: return "solution_assumed" + if self.controls and not self.ran_controls and not self.judged_only: + return "control_not_run" + if self.judged_only and not self.controls_failed: + return "judged_only" return None def gates(self) -> dict[str, bool]: @@ -139,7 +211,7 @@ def gates(self) -> dict[str, bool]: def why(self) -> str: """What to fix, in the order worth fixing it.""" - if self.deferred and not self.broken: + if self.deferred and not self.broken and not self.leaks and not self.unresolved_sources and not self.conflicts: return "DEFERRED: setup, readiness and outcome checks require the live runtime; file access declarations do not verify fixture contents." if self.broken and not self.why_not_ready: return "these checks are broken, not failing:\n - " + "\n - ".join(self.broken) @@ -212,7 +284,7 @@ def why(self) -> str: "successful outcome or the corresponding world effect. For a refusal, require " "evidence of the intended refusal, not an arbitrary execution error." ) - if self.weak: + if self.weak and self.outcome == "change": return ( "these individual checks pass without the agent doing anything, so keeping them " "would display credit unsupported by evidence:\n - " @@ -221,9 +293,67 @@ def why(self) -> str: "final state. For a refusal, require that the call was attempted and refused; " "an untouched world is not evidence that the agent refused correctly." ) + if self.controls: + passed = [ + one + for one in (self.judged_controls if self.judged_only else self.ran_controls) + if one.get("failed") is False + ] + if passed: + return "the checks pass a wrong behaviour, so they cannot tell it from the right one:\n - " + "\n - ".join( + f"{one['name']} ({one['why']}): held {', '.join(one.get('held') or []) or 'every check'}" + for one in passed + ) + "\n\nTighten the named checks so this wrong behaviour fails them." + if not self.controls and not self.judged_only and not self.deferred: + return ( + "no negative control: add a wrong behaviour written for this task (for example the " + "wrong record, the action done twice, or the forbidden change) that its checks must fail." + ) + if self.conflicts: + return "its expected outcome contradicts a cited rule:\n - " + "\n - ".join( + f"{one['ref'] or 'rule'} \"{one['rule']}\": {one['why'] or 'contradicted'}" for one in self.conflicts + ) + if self.leaks: + return "what the caller or the agent sees gives the answer away:\n - " + "\n - ".join( + f"{one['who']}: {one['field']} repeats \"{one['excerpt']}\" from {one['grader_field']}" + for one in self.leaks + ) + if self.unresolved_sources: + return "a cited source does not resolve:\n - " + "\n - ".join( + f"{one['kind']} {one['ref']}: {one['problem']}" for one in self.unresolved_sources + ) + if self.judged_only and not self.controls_failed: + return JUDGED_ONLY_TEXT + if self.controls and not self.ran_controls: + return "no negative control could run here: " + ", ".join( + f"{one['name']} ({', '.join(one.get('assumed') or [])})" for one in self.controls + ) return "holds" +def check_evidence(source: str) -> list[str]: + """What a check's source reads: "tool_calls" if it uses its `calls` argument, "world_state" if it + uses `world`. An unparsable check reads nothing.""" + try: + tree = ast.parse(source) + except SyntaxError: + return [] + function = next( + (node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "check"), + None, + ) + if function is None: + return [] + params = [arg.arg for arg in function.args.args] + names = {node.id for node in ast.walk(function) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load)} + found = [] + if len(params) > 1 and params[1] in names: + found.append("tool_calls") + if params and params[0] in names: + found.append("world_state") + return found + + def _checks_for(scenario: Scenario, catalogue: Catalogue) -> list[tuple[str, str]]: """The deterministic checks this scenario is graded by, in catalogue order.""" chosen: list[tuple[str, str]] = [] @@ -569,14 +699,192 @@ def _with_one_wrong(scenario: Scenario) -> list[tuple[str, list[Any], bool]]: return wrong +def _controls( + scenario: Scenario, + world_root: Path, + checks: list[tuple[str, str]], + reference: Any, +) -> list[dict[str, Any]]: + """Play each negative control like a solution; a control fails when a check it names fails.""" + played: list[dict[str, Any]] = [] + for control in getattr(scenario, "negative_controls", None) or []: + if not checks: + played.append({"name": control.name, "why": control.why, "failed": None, "held": [], "assumed": []}) + continue + world, calls, _refused, assumed = _run(scenario, world_root, with_solution=True, steps=control.steps) + try: + outcomes = [run_check(source, world, calls, name=name, reference=reference) for name, source in checks] + finally: + world.close() + named = set(control.must_fail) or {name for name, _ in checks} + held = [one.name for one in outcomes if one.held] + played.append( + { + "name": control.name, + "why": control.why, + # A check that raised under the control is broken, not failing: no evidence. + "failed": any(not one.held and not one.broken for one in outcomes if one.name in named), + "held": held, + "broken": [one.name for one in outcomes if one.broken and one.name in named], + "assumed": assumed, + } + ) + return played + + +JUDGED_ONLY_TEXT = ( + "Kept as written: only a judge grades this case, and nothing has shown the judge fails a wrong behaviour." +) + + +def _harness_judge(goal: Any, world: Any, calls: list[Call], messages: list[dict[str, str]], scenario: Any) -> Any: + """The harness judge on one sub-goal, run on its own loop so it works from sync and async callers.""" + import asyncio + from concurrent.futures import ThreadPoolExecutor + + from .judge import judge_sub_goal + + def decide() -> Any: + return asyncio.run(judge_sub_goal(goal, world, calls, messages=messages, scenario=scenario)) + + with ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(decide).result() + + +def _judged_controls( + scenario: Scenario, catalogue: Catalogue, world_root: Path, *, judge: Any = None +) -> list[dict[str, Any]]: + """Play each control of a judged-only case through the judge: its steps on a set-up world and its + transcript as what was said. It fails when the judge fails a sub-goal it names. A control with no + transcript gives the judge nothing the wrong agent said, so it is not played and reads not run.""" + decide = judge or _harness_judge + judged = { + name: goal + for name in scenario.sub_goals + if (goal := catalogue.named(name)) is not None and not goal.deterministic() + } + played: list[dict[str, Any]] = [] + for control in getattr(scenario, "negative_controls", None) or []: + entry: dict[str, Any] = { + "name": control.name, + "why": control.why, + "failed": None, + "held": [], + "assumed": [], + "judged": True, + "verdicts": {}, + } + named = [name for name in (control.must_fail or judged) if name in judged] + if not control.transcript or not named: + played.append(entry) + continue + world, calls, _refused, _assumed = _run(scenario, world_root, with_solution=True, steps=control.steps) + try: + for name in named: + try: + held, explanation = decide(judged[name], world, calls, list(control.transcript), scenario) + except Exception as exc: # noqa: BLE001 - a judge that could not run decides nothing + held, explanation = None, f"the judge could not run: {type(exc).__name__}" + entry["verdicts"][name] = { + "verdict": "pass" if held is True else "fail" if held is False else "inconclusive", + "explanation": str(explanation or "")[:500], + } + finally: + world.close() + verdicts = [one["verdict"] for one in entry["verdicts"].values()] + entry["held"] = [name for name, one in entry["verdicts"].items() if one["verdict"] == "pass"] + if "fail" in verdicts: + entry["failed"] = True + elif verdicts and all(one == "pass" for one in verdicts): + entry["failed"] = False + played.append(entry) + return played + + +def _outcome(scenario: Scenario, world_root: Path) -> str: + """"no_change" when the reference leaves the world as doing nothing does; unreadable reads "change".""" + done, _, _, assumed = _run(scenario, world_root, with_solution=True) + try: + after = _end_state(done) + finally: + done.close() + if assumed: + return "change" + nothing, _, _, _ = _run(scenario, world_root, with_solution=False) + try: + untouched = _end_state(nothing) + finally: + nothing.close() + return "no_change" if after is not None and untouched is not None and after == untouched else "change" + + +def _base_state(world_root: Path) -> Any: + """The world before any setup, or None when it cannot be restored here.""" + try: + world = restore(world_root) + except Exception: # noqa: BLE001 - a world only the live runtime holds has no base state here + return None + try: + return _end_state(world) + finally: + world.close() + + +def _load_contract(world_root: Path) -> Any: + try: + from .understand import load + + return load(world_root) + except Exception: # noqa: BLE001 - no contract means the defaults: conversational, no rules + return None + + +def _load_actors(world_root: Path) -> list[dict[str, Any]]: + try: + from .actors import load_actors # type: ignore[import-not-found] + except ImportError: + return [] + try: + loaded = load_actors(world_root) + except Exception: # noqa: BLE001 - an unreadable cast skips the actor audience + return [] + return [one if isinstance(one, dict) else getattr(one, "model_dump", lambda: {})() for one in loaded or []] + + def prove( scenario: Scenario, catalogue: Catalogue, world_root: Path, + *, + contract: Any = None, + actors: list | None = None, + judge: Any = None, + conflict_judge: Any = None, ) -> Proof: - """Run all three gates and say whether this scenario is worth keeping.""" + """Run all three gates and say whether this scenario is worth keeping. + + `judge(goal, world, calls, messages, scenario) -> (held, explanation)` decides a judged-only + case's controls; None uses the harness judge. `conflict_judge(expected, rule) -> (conflict, why)` + decides whether a cited rule contradicts the expected outcome; None uses the harness judge.""" + from .grounding import policy_conflicts, resolve_sources + from .leaks import leaks + proof = Proof() + if contract is None: + contract = _load_contract(world_root) + if actors is None: + actors = _load_actors(world_root) + # Before any gate or early return, so a leak or a false citation is reported whatever the gates say. + world_state = None if native_runtime(world_root) else _base_state(world_root) + proof.leaks = [ + one.as_dict() + for one in leaks(scenario, catalogue, contract, world_state, actors, world_root=world_root) + ] + proof.sources = resolve_sources(scenario, contract, world_root) + rules = [str(one) for one in (getattr(contract, "hard_constraints", None) or [])] if contract is not None else [] + proof.policy_checks = policy_conflicts(scenario, catalogue, rules, world_root, judge=conflict_judge) if rules else [] checks = _checks_for(scenario, catalogue) + proof.evidence = {name: check_evidence(source) for name, source in checks} if workspace_fixture(scenario, catalogue) and not native_runtime(world_root): proof.deferred = True proof.broken = workspace_fixture_problems( @@ -611,6 +919,9 @@ def prove( proof.solvable = proof.ready proof.vacuous = False proof.judged_only = True + if proof.ready: + proof.outcome = _outcome(scenario, world_root) + proof.controls = _judged_controls(scenario, catalogue, world_root, judge=judge) return proof # Gate 1: is the world ready for this scenario at all? @@ -688,6 +999,7 @@ def prove( # Gate 3: do those same checks fail when nothing is done? untouched, nothing, _, _ = _run(scenario, world_root, with_solution=False) try: + untouched_state = _end_state(untouched) proof.with_nothing = [ run_check(source, untouched, nothing, name=name, reference=reference) for name, source in checks @@ -727,4 +1039,29 @@ def prove( and len(proof.weak) == len(proof.with_nothing) and not judged ) + # A refusal case: the reference leaves the world as doing nothing does. Its checks should hold on + # an untouched world; what proves they can fail is a control performing the forbidden change. + # An assumed step never ran, so an unchanged world says nothing about what the solution does: + # it reads "change", as an unreadable state does, and its weak checks are not excused. + proof.outcome = ( + "no_change" + if not proof.assumed + and right_state is not None + and untouched_state is not None + and right_state == untouched_state + else "change" + ) + proof.controls = _controls(scenario, world_root, checks, reference) + proof.broken += control_broken(proof.controls) + if proof.outcome == "no_change" and any(one.get("failed") for one in proof.ran_controls): + proof.vacuous = False return proof + + +def control_broken(controls: list[dict[str, Any]]) -> list[str]: + """Each check that raised under a negative control, named with the control.""" + return [ + f"{name} (under negative control {one.get('name')})" + for one in controls + for name in one.get("broken") or [] + ] diff --git a/src/fi/alk/harness/prove_suite.py b/src/fi/alk/harness/prove_suite.py new file mode 100644 index 000000000..7004124a2 --- /dev/null +++ b/src/fi/alk/harness/prove_suite.py @@ -0,0 +1,124 @@ +"""Prove every scenario of a suite again, writing each proof as soon as it is made. + +Run at every certification so a case's proof always matches the environment it will run on: + + python -m fi.alk.harness.prove_suite --authoring DIR --output FILE [--budget-seconds N] [--deadline-at EPOCH] + [--keys KEY,KEY] + +The output is `{scenario_key: format-2 proof entry}`, rewritten atomically after every scenario, so a +killed process leaves the keys proved so far. When the time budget runs out, every remaining key is +recorded `validation_incomplete`: nothing is wrong with it, it was not reached. Exit 0 whenever it +ran; exit 3 when the authoring tree cannot be read. `--keys` proves only the named cases; the output +then holds only their keys, and the platform merges it over the proofs it already has. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import tempfile +import time +from pathlib import Path +from typing import Any + +EXIT_UNREADABLE = 3 +DEADLINE_MARGIN_SECONDS = 300 +INCOMPLETE_WHY = "not proved: the validation's time budget ran out before this case" + + +def _write(output: Path, proofs: dict[str, dict[str, Any]]) -> None: + output.parent.mkdir(parents=True, exist_ok=True) + handle, temporary = tempfile.mkstemp(dir=output.parent, prefix=".scenario-proofs-", suffix=".json") + try: + with os.fdopen(handle, "w", encoding="utf-8") as out: + json.dump(proofs, out, indent=2, sort_keys=True, ensure_ascii=False) + os.chmod(temporary, 0o644) + os.replace(temporary, output) + except BaseException: + Path(temporary).unlink(missing_ok=True) + raise + + +def _incomplete() -> dict[str, Any]: + from .proofs import unprovable_entry + + return unprovable_entry(INCOMPLETE_WHY, blocked_reason="validation_incomplete") + + +def budget_seconds(budget: float, deadline_at: float | None, now: float) -> float: + """The budget, cut so the job's own deadline keeps its margin.""" + if deadline_at is None: + return budget + return min(budget, deadline_at - now - DEADLINE_MARGIN_SECONDS) + + +def prove_suite( + authoring: Path, + output: Path, + *, + budget: float = 1200, + deadline_at: float | None = None, + keys: set[str] | None = None, +) -> dict[str, dict[str, Any]]: + from .catalogue import load_catalogue + from .proofs import _entry, proof_key, unprovable_entry + from .live_proof import apply_live, load_live + from .prove import _load_actors, _load_contract, prove + from .scenario_tools import load_scenarios + + scenarios = load_scenarios(authoring) + catalogue = load_catalogue(authoring) + contract = _load_contract(authoring) + actors = _load_actors(authoring) + live = load_live(authoring) + started = time.monotonic() + allowed = budget_seconds(budget, deadline_at, time.time()) + proofs: dict[str, dict[str, Any]] = {} + for scenario in scenarios: + key = proof_key(scenario) + if keys is not None and key not in keys: + continue + if time.monotonic() - started >= allowed: + proofs[key] = _incomplete() + continue + try: + proof = prove(scenario, catalogue, authoring, contract=contract, actors=actors) + if key in live: + # The certification just played this case on the live runtime: its gates win. + proof = apply_live(proof, live[key]) + except Exception as exc: # noqa: BLE001 - a case that cannot be proved is recorded, never fatal + first = (str(exc).splitlines() or [""])[0][:200] + proofs[key] = unprovable_entry(f"{type(exc).__name__}: {first}".rstrip(": ")) + else: + proofs[key] = _entry(proof, len(scenario.solution)) + _write(output, proofs) + _write(output, proofs) + return proofs + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--authoring", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--budget-seconds", type=float, default=1200) + parser.add_argument("--deadline-at", type=float, default=None) + parser.add_argument("--keys", default=None, help="comma-separated scenario keys to prove; default every one") + args = parser.parse_args(argv) + keys = {one.strip() for one in args.keys.split(",") if one.strip()} if args.keys is not None else None + if not args.authoring.is_dir(): + print(f"prove_suite: authoring tree not readable: {args.authoring}", file=sys.stderr) + return EXIT_UNREADABLE + try: + prove_suite( + args.authoring, args.output, budget=args.budget_seconds, deadline_at=args.deadline_at, keys=keys + ) + except OSError as exc: + print(f"prove_suite: authoring tree not readable: {exc}", file=sys.stderr) + return EXIT_UNREADABLE + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/fi/alk/harness/scenario.py b/src/fi/alk/harness/scenario.py index b008ee25b..5b24e0a35 100644 --- a/src/fi/alk/harness/scenario.py +++ b/src/fi/alk/harness/scenario.py @@ -21,7 +21,7 @@ from itertools import combinations from math import ceil, log from pathlib import Path -from typing import Any, ClassVar +from typing import Any, ClassVar, Literal from pydantic import BaseModel, Field, field_validator, model_validator @@ -76,6 +76,54 @@ class Step(BaseModel): fixed: dict[str, str] | None = None +_CONTROL_NAME = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") + + +class NegativeControl(BaseModel): + """A wrong behaviour written for this task, played like a solution; the checks must fail it.""" + + name: str # slug, unique within the scenario + why: str # one sentence: the wrong behaviour, e.g. "refunds the order twice" + steps: list[Step] = Field(default_factory=list) # the wrong trajectory + must_fail: list[str] = Field(default_factory=list) # sub-goals that must fail; empty = any check + # What the wrong agent says, as [{role: "user" | "assistant", content}]: the judge reads it with + # the steps for a case only a judge can grade, and must fail it. + transcript: list[dict[str, str]] = Field(default_factory=list) + + @model_validator(mode="after") + def _shape(self) -> "NegativeControl": + if not _CONTROL_NAME.match(self.name or ""): + raise ValueError(f"negative control name {self.name!r} must be a lowercase slug") + if not self.steps and not self.transcript: + raise ValueError(f"negative control {self.name} has no steps and no transcript") + for turn in self.transcript: + if str(turn.get("role") or "") not in ("user", "assistant") or not str(turn.get("content") or "").strip(): + raise ValueError( + f"negative control {self.name}: each transcript turn needs a role (user or assistant) and content" + ) + return self + + +SOURCE_KINDS = ("rule", "file", "chat_message", "evidence_case", "requirement", "assumption") + + +class ScenarioSource(BaseModel): + """Where one requirement of a case comes from, so it is shown on the case and checked.""" + + kind: Literal["rule", "file", "chat_message", "evidence_case", "requirement", "assumption"] + # rule: | file: | chat: | + # evidence: | "" for requirement and assumption + ref: str = "" + quote: str = Field(default="", max_length=500) # the cited words; required for rule and file + supports: str = "" # which part of the case it justifies, e.g. "the 14-day limit" + + @model_validator(mode="after") + def _quoted(self) -> "ScenarioSource": + if self.kind in ("rule", "file") and not self.quote.strip(): + raise ValueError(f"a {self.kind} source needs the exact quote it cites") + return self + + # The language every accent in the platform's accent vocabulary is a way of speaking. _ACCENTED_LANGUAGE = "english" @@ -262,6 +310,9 @@ class Scenario(BaseModel): # on the pair, so a use case can carry many scenarios without any reading as a duplicate. branch: str = "" tests: str = "" + # A failure here blocks the release: the platform marks the case must-pass and names it on a run + # whose result it failed. Set by the person or the builder; versioned with the case. + must_pass: bool = False # What this scenario changes about the world after it is reset, as code: a file defining # ``setup(world)``. Rows in a table were enough while every world was a database, and they @@ -299,6 +350,11 @@ class Scenario(BaseModel): solution: list[Step] = Field(default_factory=list) # Other ways the agent's instructions allow to the same outcome; every check must pass on each. alternatives: list[list[Step]] = Field(default_factory=list) + # Wrong behaviours written for this task; each must fail the checks. Never shown to the agent. + negative_controls: list[NegativeControl] = Field(default_factory=list) + # Where each requirement comes from: a rule, a file, a chat message, an evidence case, the + # person's own words, or an assumption labelled as such. + sources: list[ScenarioSource] = Field(default_factory=list) # Level per plan axis, e.g. {"task": "book_ride", "overlay": "interruption"}. Optional. coverage: dict[str, str] = Field(default_factory=dict) @@ -351,6 +407,17 @@ def _lift_keywords(cls, data: Any) -> Any: data = {**data, "keywords": list(persona["keywords"])} return data + @model_validator(mode="after") + def _controls_fit(self) -> "Scenario": + names = [one.name for one in self.negative_controls] + if len(names) != len(set(names)): + raise ValueError("negative control names must be unique within a scenario") + for one in self.negative_controls: + unknown = sorted(set(one.must_fail) - set(self.sub_goals)) + if unknown: + raise ValueError(f"negative control {one.name} names sub-goals the scenario lacks: {', '.join(unknown)}") + return self + @model_validator(mode="after") def _identify(self) -> "Scenario": if not self.scenario_key: diff --git a/src/fi/alk/harness/scenario_tools.py b/src/fi/alk/harness/scenario_tools.py index 08653b50c..e64e1776a 100644 --- a/src/fi/alk/harness/scenario_tools.py +++ b/src/fi/alk/harness/scenario_tools.py @@ -16,7 +16,7 @@ import logging from collections import Counter from pathlib import Path -from typing import Any, Callable +from typing import Any, Callable, Sequence from .backends import tool, tool_server @@ -804,8 +804,10 @@ def _identifiers_the_instruction_invents(scenario: Scenario, trial: Any) -> list ] -def _steps_the_contract_refuses(scenario: Scenario, world_root: Path) -> list[str]: - """Solution steps that call a tool, argument or declared value the agent's contract does not have.""" +def _steps_the_contract_refuses( + scenario: Scenario, world_root: Path, *, steps: Sequence[Step] | None = None +) -> list[str]: + """Solution steps (or `steps`) that call a tool, argument or declared value the agent's contract lacks.""" try: contract = json.loads((Path(world_root) / "contract.json").read_text(encoding="utf-8")) except (OSError, ValueError): @@ -814,7 +816,7 @@ def _steps_the_contract_refuses(scenario: Scenario, world_root: Path) -> list[st if not tools: return [] wrong = [] - for step in scenario.solution: + for step in scenario.solution if steps is None else steps: spec = tools.get(step.tool) if spec is None: wrong.append(f"{step.tool} is not one of the agent's tools") @@ -836,6 +838,19 @@ def _steps_the_contract_refuses(scenario: Scenario, world_root: Path) -> list[st ] +def _controls_the_contract_refuses(scenario: Scenario, world_root: Path) -> list[str]: + """The same contract check over each negative control's steps, each problem naming its control.""" + return [ + f"negative control {control.name}: {problem}" + for control in scenario.negative_controls + for problem in _steps_the_contract_refuses(scenario, world_root, steps=control.steps) + ] + + +# A writer's fault: the scenario is not kept. The lane codes (an unbound tool) are kept and labelled. +_REFUSED_CODES = frozenset({"negative_control_passed", "no_negative_control", "answer_leaked", "source_unresolved"}) + + def _fixed_sources_that_do_not_hold(scenario: Scenario, trial: Any, instructions: str) -> list[str]: """Fixed text values whose stated source (caller, record or a quoted rule) does not contain them.""" @@ -939,6 +954,20 @@ def apply(saved: dict[str, Any], given: dict[str, Any]) -> dict[str, Any]: return result +NO_SOURCES = "Cite where this case comes from, or add an assumption source that says what you assumed." + + +def _unchanged(scenario: Scenario, payload: dict[str, Any], kept: list[Scenario]) -> bool: + """Whether a saved scenario with this key (else this name) already holds exactly this content.""" + named = str(payload.get("scenario_key") or "").strip() + previous = next((one for one in kept if named and one.scenario_key == named), None) or next( + (one for one in kept if one.name == scenario.name), None + ) + if previous is None: + return False + return previous.model_dump(exclude={"scenario_key"}) == scenario.model_dump(exclude={"scenario_key"}) + + def accept_scenario( payload: dict[str, Any], *, @@ -955,9 +984,15 @@ def accept_scenario( instructions: str = "", used: set[str] | None = None, renaming: bool = False, + require_sources: bool = False, ) -> dict[str, Any]: """Validate one scenario, then prove it. A plain function so both halves are testable. + ``require_sources`` refuses a new or changed scenario that cites nothing: a requirement the + writer cannot cite is written as an ``assumption`` source, never left as a silent default. A + resubmission identical to the saved scenario (a case written before sources were required) is + exempt. + ``renaming`` keeps a proved scenario under a new name without proving it again: what it tests is unchanged. ``used`` is every key the environment has given a scenario, so a new one never takes a dropped scenario's key and inherits its run history. @@ -995,6 +1030,7 @@ def accept_scenario( ) problems.extend(contract_sequence_problems(scenario, hard_constraints or [])) problems.extend(_steps_the_contract_refuses(scenario, world_root)) + problems.extend(_controls_the_contract_refuses(scenario, world_root)) problems.extend(_overlay_asserts_nothing(scenario, catalogue)) problems.extend(_shared_check_bound_to_one_task(scenario, catalogue, kept)) else: @@ -1015,18 +1051,25 @@ def accept_scenario( problems.extend(_identifiers_the_instruction_invents(scenario, trial)) problems.extend(_fixed_sources_that_do_not_hold(scenario, trial, instructions)) problems.extend(_steps_the_contract_refuses(scenario, world_root)) + problems.extend(_controls_the_contract_refuses(scenario, world_root)) problems.extend(_overlay_asserts_nothing(scenario, catalogue)) problems.extend(_shared_check_bound_to_one_task(scenario, catalogue, kept)) finally: trial.close() + # An assumption that writes down nothing it assumed cites nothing either. + cited = [one for one in scenario.sources if one.kind != "assumption" or (one.quote or one.supports).strip()] + if require_sources and not cited and not _unchanged(scenario, payload, kept): + problems.append(NO_SOURCES) if problems: return _err( "Not kept. Fix these and submit again:\n - " + "\n - ".join(problems) ) proof = prove(scenario, catalogue, world_root) - if not proof.holds and not (proof.deferred and not proof.broken): + if (not proof.holds and not (proof.deferred and not proof.broken)) or ( + proof.blocked_reason() in _REFUSED_CODES + ): said = f"Not kept. {proof.why()}" # Code written against the wrong collection shape is the commonest way setup, ready and a # check fail here, and the exception alone does not say which collections are mappings and @@ -1130,6 +1173,19 @@ def accept_scenario( if proof.assumed else "" ) + if proof is not None and proof.blocked_reason() == "control_not_run": + unproved += ( + "\nNOT PROVED: no negative control could run here, because these tools have no endpoint: " + + ", ".join(sorted({tool for one in proof.controls for tool in one.get("assumed") or []})) + + ". It is kept as written; the checks were never shown to fail a wrong behaviour." + ) + if proof is not None and proof.blocked_reason() == "judged_only": + unproved += ( + f"\nNOT PROVED: {proof.why()} Give each negative control a `transcript` of the wrong behaviour as said; " + "the judge must fail it." + ) + if proof is not None and proof.outcome == "no_change" and proof.proved: + unproved += "\nA refusal case: nothing should change, and each negative control's forbidden change fails its checks." if proof is None: proof_summary = "Nothing it tests changed, so it was not proved again." elif proof.deferred: @@ -1805,6 +1861,22 @@ async def add_sub_goal(args: dict[str, Any]) -> dict[str, Any]: "must still pass. Empty when the instructions allow only one way.", "items": {"type": "array", "items": {"type": "object"}}, }, + "negative_controls": { + "type": "array", + "description": "Wrong behaviours written for this task that its checks must fail, at " + "least one: the action done twice, the wrong record, or the forbidden change on a " + "refusal case. Never 'do nothing'. Each is {name, why, steps (like solution), " + "must_fail (sub-goal names; empty means any check)}.", + "items": {"type": "object"}, + }, + "sources": { + "type": "array", + "description": "Where each requirement comes from: {kind (rule, file, chat_message, " + "evidence_case, requirement, assumption), ref (rule:, file:, chat:, " + "evidence:), quote (the exact words; required for rule and file), supports}. A " + "requirement the person stated is cited in their words; a guess is an assumption.", + "items": {"type": "object"}, + }, "sub_goals": { "type": "array", "items": {"type": "string"}, @@ -1923,6 +1995,7 @@ def _refuse(said: str, as_given: dict[str, Any] | None = None) -> dict[str, Any] [contract.system_prompt_excerpt, *contract.hard_constraints] ), used=used, + require_sources=True, ) if not result.get("is_error"): exploration["since_submit"] = 0 diff --git a/src/fi/alk/harness/skills/builder/SKILL.md b/src/fi/alk/harness/skills/builder/SKILL.md index efc0b1d8e..c573f11a4 100644 --- a/src/fi/alk/harness/skills/builder/SKILL.md +++ b/src/fi/alk/harness/skills/builder/SKILL.md @@ -101,6 +101,8 @@ is what enforces it (the world's behaviour, the check that fails without it), ne to follow it. Tell the person concretely what you will redo and what stays, then redo it, prove again what it affects, and keep everything that still holds. +A case a person describes is cited as a requirement in their words. + ## What you know, and how to answer Everything about how this system works is written down for you, listed under "What you know" at the diff --git a/src/fi/alk/harness/skills/write-scenarios/SKILL.md b/src/fi/alk/harness/skills/write-scenarios/SKILL.md index 3c36349da..08e95197a 100644 --- a/src/fi/alk/harness/skills/write-scenarios/SKILL.md +++ b/src/fi/alk/harness/skills/write-scenarios/SKILL.md @@ -773,6 +773,54 @@ world alone cannot show it: an untouched world looks exactly like one where the refused. Check the calls instead: that the agent tried, and that the attempt was refused rather than succeeding. +## Negative controls (required) + +Every scenario carries at least one negative control in `negative_controls`: a wrong behaviour, +written for this task, that its checks must fail. Each is `{name, why, steps, must_fail}`; `steps` is +played exactly like `solution`. + +- Make it specific to the task: the refund issued twice, a refund on the 20-day order, the wrong + record looked up. A control that fails for a reason unrelated to the task proves nothing. +- "Do nothing" is never a control. The third gate already covers it. +- On a no-change (refusal) scenario, the control performs the forbidden change, and the checks must + fail it. The checks of such a scenario may hold on an untouched world; the control is what shows + they can fail. +- A no-change scenario whose solution looks a record up with a fixed id also checks that the right + record was read (for example `any(c.name == "lookup_order" and c.arguments.get("order_id") == + "O-20" for c in calls)`), or leaves that argument out of `fixed`. A read-only lookup with a wrong id + ends in the same state, so nothing else notices it. +- A control whose checks still all pass is refused with the checks to tighten. +- A case only a judge can grade (no code check reads its world) is kept as written, never proved, + until a control shows the judge fails a wrong behaviour. Give each of its controls a `transcript`: + the wrong exchange as said, `[{role: "user" | "assistant", content}]`, with any `steps` it takes. + The judge reads it like a real call and must fail a sub-goal the control names. + +## Cite where each requirement comes from + +Each scenario lists `sources`: where each of its requirements comes from. Kinds are `rule` (ref +`rule:`, quote the rule's exact words), `file` (ref `file:`, quote the file exactly), +`chat_message` (ref `chat:`), `evidence_case` (ref `evidence:`), `requirement` (the person's +own words in `quote`) and `assumption` (anything you decided yourself). A citation that does not +resolve is refused, so quote exactly. A new or changed scenario with no sources is refused: for any +requirement you cannot cite, add an `assumption` source whose `quote` says what you assumed and whose +`supports` names the part of the case it covers, rather than citing a source that does not say it. + +## Must-pass cases + +Set `must_pass: true` on a case when the person asks for it to block a release, or names it as one +that must never fail. A failed must-pass case is named on the run's result. Leave it false otherwise. + +## Never put the answer where the caller or the agent can read it + +The grader-only material is each sub-goal's `what`, `judged`, `reference`, `verifier` and +`comparison`, the values a check compares against that the world does not hold, `tests`, each +negative control's `why`, and every file only the evaluator reads. None of it may appear in what the +simulated person or the agent is given: the instruction, the variables, the persona, its opening +message, or an actor's brief and knowledge. A scenario that repeats eight words of a sub-goal, a +multi-word sub-goal name or a computed expected value in those fields is refused with the excerpt. +On a conversational agent, the opening message must not repeat the instruction either: the +instruction is the caller's, never the agent's. + ## What will be refused, and what to do Validation runs before the gates, and every problem is reported at once, so fix them together. diff --git a/src/fi/alk/harness/skills/write-scenarios/references/refusals.md b/src/fi/alk/harness/skills/write-scenarios/references/refusals.md index c07c79dd4..52d53e236 100644 --- a/src/fi/alk/harness/skills/write-scenarios/references/refusals.md +++ b/src/fi/alk/harness/skills/write-scenarios/references/refusals.md @@ -24,6 +24,10 @@ often; each is cheaper to avoid while writing than to fix after a refusal. | `setup_code must define setup(world)` / `ready_code must define ready(world)` | Wrong entry point. | Define the function with that exact name. | | `the prompt asks for ..., which this scenario does not supply` | The prompt has a slot nothing fills, and an unfilled slot reaches the person verbatim. | Add it to `variables`. | | ` requires from this call, but the reference solution does not create it first` | A hard rule says a value must come from this conversation, and the solution supplies it from setup or `environment_arguments` instead. | Put the step that produces it earlier in the solution. | +| `no negative control` | Nothing shows the checks can tell a wrong behaviour from the right one. | Add a wrong behaviour written for this task to `negative_controls`: the action done twice, the wrong record, or the forbidden change. Never "do nothing". On a no-change scenario, the control performs the forbidden change. | +| `negative control {name} passed` | The checks hold on a wrong behaviour, so they would pass an agent that does it. | Tighten the named checks until that control fails them. A no-change scenario with a fixed lookup also checks which record was read, or leaves that argument unfixed. | +| `answer leaked: {field} repeats ...` | The instruction, persona, variables, opening message or an actor's brief repeats grader-only text: a sub-goal's words or name, or a value a check expects that the world does not hold. | Move the answer out of the instruction or persona. Say what the person wants, never what the grader checks. | +| `source does not resolve: {kind} {ref}` | A cited rule or file does not contain the quoted words, or a requirement has no quote. | Quote the rule or file exactly, or cite it as an `assumption`. | ## The refusals writers hit most, and how to avoid them first time diff --git a/src/fi/alk/harness/world/handle.py b/src/fi/alk/harness/world/handle.py index ed566da04..b31168aa8 100644 --- a/src/fi/alk/harness/world/handle.py +++ b/src/fi/alk/harness/world/handle.py @@ -90,9 +90,22 @@ def __init__( # re-measured: a live count would make the cap depend on what a scenario already wrote, # and the whole point is that it is decided before any scenario runs. self._baseline_row_counts = dict(baseline_row_counts) + # Tools a live proof may play: bound by `bind_tools` to a route behind the bundle's tool + # proxy. Unbound, `call` refuses as before; `runtime_tools` marks every contract tool so a + # step for an unbound one reads assumed, never executed. + self.runtime_tools: set[str] = set() + self.endpoint_for: dict[str, str] = {} + self.calls: list[Call] = [] + self._tool_proxy = "" if _validate_baseline: self._require_baseline_coverage(self._visible_tables()) + def bind_tools(self, tool_proxy: str, endpoint_for: Mapping[str, str], runtime_tools: Sequence[str]) -> None: + """Let `call` play the service-backed tools in `endpoint_for` through the tool proxy.""" + self._tool_proxy = str(tool_proxy or "").rstrip("/") + self.endpoint_for = {name: str(path).strip("/") for name, path in endpoint_for.items() if path} + self.runtime_tools = set(runtime_tools) | set(self.endpoint_for) + # -- reading ------------------------------------------------------------------------------ def _execution_state(self) -> dict[str, Any]: @@ -284,19 +297,56 @@ def drop(self, collection: str, key: str = "", *, by: str = "") -> int: ) return self._store.remove(collection, key, by=by) + def forward( + self, + endpoint: str, + arguments: Mapping[str, Any] | None = None, + *, + record_as: str = "", + record: bool = True, + ) -> Call: + """POST `arguments` as JSON to `endpoint` behind the bundle's tool proxy, which forwards it to + the agent's own tools service and records the call. The same wire format a provisioned + world uses for a submitted service.""" + import json + import time + + import requests + + if not self._tool_proxy: + raise WorldUnavailable("no tool proxy is bound to this world, so no tool can be played") + args = dict(arguments or {}) + name = record_as or endpoint.strip("/") + at = time.time() + try: + response = requests.post(f"{self._tool_proxy}/{endpoint.strip('/')}", json=args, timeout=30) + try: + body: Any = response.json() + except ValueError: + body = response.text + refused = response.status_code >= 400 + error = (json.dumps(body, default=str) if not isinstance(body, str) else body) if refused else "" + call = Call(name=name, arguments=args, result=body, ok=not refused, refused=refused, error=error, at=at) + except requests.RequestException as exc: + call = Call(name=name, arguments=args, ok=False, error=f"{type(exc).__name__}: {exc}", at=at) + if record: + self.calls.append(call) + return call + def call(self, name: str, arguments: Mapping[str, Any] | None = None) -> Call: """Play one of the agent's own tools against this world. - Not implemented: the `http_tool` evidence seam's wire format is not pinned anywhere in - the contracts yet, and guessing at a shape here would ship one nobody agreed to and - scenario code would end up depending on. Raising unconditionally is the honest answer - until the evidence layer pins it — run-time scenario code should not need this anyway, - since `setup` already runs at proof time and the data verbs cover the rest. + Only a service-backed tool bound by `bind_tools` can be played: its route behind the tool + proxy is the one wire format pinned here. A tool the agent runs in its own process has no + endpoint, and any other is not guessed at, so both refuse. """ - raise WorldUnavailable( - "the http_tool shim wire format is not yet pinned by the contracts — report, " - "don't guess." - ) + endpoint = self.endpoint_for.get(name) + if not endpoint: + raise WorldUnavailable( + f"{name} has no bound service endpoint; only service-backed tools behind the tool " + "proxy can be played — report, don't guess." + ) + return self.forward(endpoint, arguments, record_as=name) def read_only(self) -> "ReadOnlyWorld": """The view `ready` and `check` run against: every write verb refuses outright. diff --git a/tests/harness/proof_world.py b/tests/harness/proof_world.py index a5d290a1e..2d2a35e3f 100644 --- a/tests/harness/proof_world.py +++ b/tests/harness/proof_world.py @@ -93,3 +93,31 @@ def scenario(solution, *, sub_goals, name="refund_case", instruction="", **extra ONE_REFUND = check("n = world.data['refunds'].count('O-10')\nreturn None if n == 1 else f'{n} refunds'") + + +LOOKED_UP_O10 = check( + "return None if any(c.name == 'lookup_order' and c.ok and c.arguments.get('order_id') == 'O-10' for c in calls) " + "else 'O-10 never looked up'" +) +LOOKUP_O10 = {"tool": "lookup_order", "arguments": {"order_id": "O-10"}, "fixed": {"order_id": "the caller"}} +REFUND_O10 = {"tool": "refund_order", "arguments": {"order_id": "O-10"}, "fixed": {"order_id": "the caller"}} + + +def control(name, why, steps, must_fail=()): + from fi.alk.harness.scenario import NegativeControl + + return NegativeControl(name=name, why=why, steps=steps, must_fail=list(must_fail)) + + +def refund_case(**extra): + """A 10-day order refunded once; its control refunds twice.""" + values = dict( + sub_goals=["refund_once", "looked_up"], + instruction="I want my money back for order O-10.", + negative_controls=[control("two_refunds", "refunds the order twice", [LOOKUP_O10, REFUND_O10, REFUND_O10])], + ) + values.update(extra) + return scenario([LOOKUP_O10, REFUND_O10], **values) + + +REFUND_CATALOGUE = dict(refund_once=ONE_REFUND, looked_up=LOOKED_UP_O10) diff --git a/tests/harness/test_builder_chat.py b/tests/harness/test_builder_chat.py index 71a73198b..4562b4c77 100644 --- a/tests/harness/test_builder_chat.py +++ b/tests/harness/test_builder_chat.py @@ -1737,8 +1737,10 @@ def test_a_suite_rebuild_can_add_a_scenario(tmp_path, monkeypatch): async def act(stage): await _tool(stage, "scenarios", "aim_for").handler({"count": 2}) + # A rebuild cites where a new case comes from, as any builder submit does. + assumed = [{"kind": "assumption", "quote": "Customers order one item at a time.", "supports": "one Big Mac"}] seen["added"] = await _tool(stage, "scenarios", "submit_scenario").handler( - _scenario("adds-one-big-mac") + {**_scenario("adds-one-big-mac"), "sources": assumed} ) _rebuild(tmp_path, monkeypatch, "suite", act) @@ -1750,6 +1752,22 @@ async def act(stage): } +def test_the_builder_s_submit_refuses_a_new_case_that_cites_nothing(tmp_path, monkeypatch): + from test_nudge_turn import _scenario + + from fi.alk.harness.scenario_tools import NO_SOURCES + + seen: dict = {} + + async def act(stage): + await _tool(stage, "scenarios", "aim_for").handler({"count": 2}) + seen["added"] = await _tool(stage, "scenarios", "submit_scenario").handler(_scenario("adds-one-big-mac")) + + _rebuild(tmp_path, monkeypatch, "suite", act) + + assert seen["added"].get("is_error") and NO_SOURCES in seen["added"]["content"][0]["text"] + + def test_a_replayed_rebuild_commits_the_workspace_holding_its_result( tmp_path, monkeypatch ): @@ -2043,7 +2061,10 @@ async def overview(self): # Every solution uses only the two in-process tools, which have no endpoint: each step is assumed. kept = [SimpleNamespace(name=f"case-{n}", scenario_key=f"case-{n}") for n in range(10)] for one in kept: - proof = Proof(ready=True, solvable=True, vacuous=False, assumed=["place_order", "search_books"]) + not_run = [{"name": "wrong", "why": "orders twice", "failed": True, "held": [], "assumed": ["place_order"]}] + proof = Proof( + ready=True, solvable=True, vacuous=False, assumed=["place_order", "search_books"], controls=not_run + ) record_proof(workspace, one.scenario_key, proof, solution_steps=2) monkeypatch.setattr(runtime, "_what_exists", lambda: (True, True)) monkeypatch.setattr(runtime, "_after_saved", lambda **kwargs: None) diff --git a/tests/harness/test_grounding.py b/tests/harness/test_grounding.py new file mode 100644 index 000000000..f7887474e --- /dev/null +++ b/tests/harness/test_grounding.py @@ -0,0 +1,121 @@ +"""A cited source must say what the scenario claims; no sources at all is not a fault.""" + +from __future__ import annotations + +from types import SimpleNamespace + +from fi.alk.harness import prove as module +from fi.alk.harness.grounding import resolve_sources +from fi.alk.harness.scenario import ScenarioSource + +from proof_world import REFUND_CATALOGUE, catalogue, install, refund_case + +CONTRACT = SimpleNamespace(hard_constraints=["Never refund an order older than 14 days.", "Refund each order at most once."]) + + +def test_rule_quote_must_match(tmp_path): + wrong = SimpleNamespace(sources=[ScenarioSource(kind="rule", ref="rule:0", quote="older than 30 days")]) + right = SimpleNamespace(sources=[ScenarioSource(kind="rule", ref="rule:0", quote="at most once")]) + assert resolve_sources(wrong, CONTRACT, tmp_path)[0]["resolved"] is False + resolved = resolve_sources(right, CONTRACT, tmp_path)[0] + assert resolved["resolved"] is True and resolved["resolved_ref"] == "rule:1" + + +def test_no_sources_is_not_blocking(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(refund_case(sources=[]), catalogue(**REFUND_CATALOGUE), tmp_path, contract=CONTRACT, actors=[]) + assert proof.sources == [] + assert proof.proved is True + + +def _thirty_day_case(): + from fi.alk.harness.catalogue import Catalogue, SubGoal + + goals = Catalogue( + sub_goals=[SubGoal(name="refunds_late_order", what="The agent refunds an order placed 30 days ago.", judged="x")] + ) + case = SimpleNamespace( + scenario_key="late_refund", + sub_goals=["refunds_late_order"], + sources=[ScenarioSource(kind="rule", ref="rule:0", quote="older than 14 days")], + ) + return case, goals + + +def test_30_days_against_a_14_day_rule_conflicts(monkeypatch, tmp_path): + from fi.alk.harness.grounding import policy_conflicts + + case, goals = _thirty_day_case() + asked: list[tuple[str, str]] = [] + + def judge(expected, rule): + asked.append((expected, rule)) + return True, "the rule allows 14 days, the case expects 30" + + found = policy_conflicts(case, goals, list(CONTRACT.hard_constraints), tmp_path, judge=judge) + + assert asked == [("- The agent refunds an order placed 30 days ago.", "Never refund an order older than 14 days.")] + assert found == [ + { + "rule": "Never refund an order older than 14 days.", + "ref": "rule:0", + "conflict": True, + "why": "the rule allows 14 days, the case expects 30", + } + ] + # The proof blocks on it, before a leak or an unresolved source would. + install(monkeypatch) + proof = module.prove( + refund_case(sources=list(case.sources)), + catalogue(**REFUND_CATALOGUE), + tmp_path / "proof", + contract=CONTRACT, + actors=[], + conflict_judge=judge, + ) + assert proof.blocked_reason() == "policy_conflict" and proof.proved is False + assert "contradicts a cited rule" in proof.why() + no_conflict = module.prove( + refund_case(sources=list(case.sources)), + catalogue(**REFUND_CATALOGUE), + tmp_path / "proof2", + contract=CONTRACT, + actors=[], + conflict_judge=lambda expected, rule: (False, "consistent"), + ) + assert no_conflict.blocked_reason() is None + + +def test_cache_hit_makes_no_call(tmp_path): + from fi.alk.harness.grounding import POLICY_CHECKS, policy_conflicts + + case, goals = _thirty_day_case() + calls: list[str] = [] + + def judge(expected, rule): + calls.append(rule) + return True, "contradicts" + + rules = list(CONTRACT.hard_constraints) + first = policy_conflicts(case, goals, rules, tmp_path, judge=judge) + second = policy_conflicts(case, goals, rules, tmp_path, judge=judge) + + assert len(calls) == 1 + assert first == second + assert (tmp_path / POLICY_CHECKS).is_file() + + +def test_undecided_judge_is_not_a_conflict_and_is_not_cached(tmp_path): + from fi.alk.harness.grounding import policy_conflicts + + case, goals = _thirty_day_case() + calls: list[str] = [] + + def judge(expected, rule): + calls.append(rule) + return None, "" + + rules = list(CONTRACT.hard_constraints) + assert policy_conflicts(case, goals, rules, tmp_path, judge=judge)[0]["conflict"] is None + policy_conflicts(case, goals, rules, tmp_path, judge=judge) + assert len(calls) == 2 diff --git a/tests/harness/test_leaks.py b/tests/harness/test_leaks.py new file mode 100644 index 000000000..e11930f2a --- /dev/null +++ b/tests/harness/test_leaks.py @@ -0,0 +1,156 @@ +"""What the caller, the agent or an actor sees must not give the answer away.""" + +from __future__ import annotations + +from types import SimpleNamespace + +from fi.alk.harness.catalogue import Catalogue, SubGoal +from fi.alk.harness.leaks import leaks + +WORLD = {"orders": {"O-123": {"total": 18.0}}} + + +def _case(instruction="", sub_goals=("refund_once",), **extra): + values = dict( + instruction=instruction, + sub_goals=list(sub_goals), + tests="", + variables={}, + persona=None, + negative_controls=[], + setup_code="", + actors=[], + ) + values.update(extra) + return SimpleNamespace(**values) + + +def _catalogue(**goals): + return Catalogue(sub_goals=[SubGoal(name=name, **fields) for name, fields in goals.items()]) + + +def _check(body): + return "def check(world, calls):\n" + "".join(f" {line}\n" for line in body.splitlines()) + + +def test_instruction_repeating_sub_goal_text_leaks(): + catalogue = _catalogue(refund_once={"what": "The agent issues exactly one refund for the delayed order and confirms it"}) + case = _case("Please make sure you issues exactly one refund for the delayed order and confirms it today.") + found = leaks(case, catalogue, None, WORLD) + assert [(one.who, one.field, one.grader_field) for one in found] == [ + ("simulated person", "instruction", "sub_goal:refund_once.what") + ] + + +def test_order_id_in_world_is_not_a_leak(): + catalogue = _catalogue(refund_once={"check": _check("return None if calls and calls[0].arguments.get('id') == 'O-123' else 'no'")}) + assert leaks(_case("My order is O-123."), catalogue, None, WORLD) == [] + + +def test_expected_total_not_in_world_leaks(): + catalogue = _catalogue(refund_once={"check": _check("return None if world.data['refunded'] == 42.50 else 'wrong'")}) + found = leaks(_case("I expect 42.50 back."), catalogue, None, WORLD) + assert [(one.grader_field, one.excerpt) for one in found] == [("check literal", "42.50")] + + +def test_actor_brief_leak(): + judged = "The agent explains the fourteen day refund window before refusing politely" + catalogue = _catalogue(explains_window={"judged": judged}) + case = _case("Ask for a refund.", sub_goals=("explains_window",), actors=[{"id": "clerk"}]) + actors = [{"id": "clerk", "name": "Clerk", "instructions": f"Make sure {judged}."}] + found = leaks(case, catalogue, None, WORLD, actors) + assert [one.who for one in found] == ["actor"] + assert found[0].grader_field == "sub_goal:explains_window.judged" + + +def test_initial_message_repeating_instruction_leaks(): + instruction = "You bought a lamp twenty days ago and it arrived broken so you want a refund" + persona = {"name": "Dana", "initial_message": "Hi, I bought a lamp twenty days ago and it arrived broken."} + case = _case(instruction, persona=persona, sub_goals=()) + found = leaks(case, Catalogue(), SimpleNamespace(conversational=True), WORLD) + assert ("tested agent", "persona.initial_message", "instruction") in [ + (one.who, one.field, one.grader_field) for one in found + ] + + +def test_single_word_literal_is_not_a_leak(): + single = _catalogue(refund_once={"check": _check("return None if world.data['status'] == 'refunded' else 'no'")}) + assert leaks(_case("I want it refunded."), single, None, WORLD) == [] + coded = _catalogue(refund_once={"check": _check("return None if world.data['ref'] == 'REF-2201' else 'no'")}) + found = leaks(_case("My reference should be REF-2201."), coded, None, WORLD) + assert [one.grader_field for one in found] == ["check literal"] + + +def test_unrestorable_world_skips_world_rule(): + what = "The agent issues exactly one refund for the delayed order and confirms it" + catalogue = _catalogue( + refund_once={"what": what, "check": _check("return None if world.data['ref'] == 'REF-2201' else 'no'")} + ) + case = _case(f"Make sure he {what}; my code is REF-2201 and I want a refund once.") + found = {one.grader_field for one in leaks(case, catalogue, None, None)} + assert found == {"sub_goal:refund_once.what", "sub_goal:refund_once.name"} + + +def test_single_word_sub_goal_name_is_not_a_leak(): + assert leaks(_case("I would like to ask for a refund.", sub_goals=("refund",)), _catalogue(refund={}), None, WORLD) == [] + found = leaks(_case("Just refund once, please.", sub_goals=("refund_once",)), _catalogue(refund_once={}), None, WORLD) + assert [one.grader_field for one in found] == ["sub_goal:refund_once.name"] + + +def test_failure_message_is_not_grader_only(): + said = _catalogue(refund_once={"check": _check("if world.data['refunds']:\n return None\nreturn 'no refund was issued'")}) + assert leaks(_case("I am worried no refund was issued."), said, None, WORLD) == [] + raised = _catalogue(refund_once={"check": _check("if not world.data['refunds']:\n raise AssertionError('no refund was issued')")}) + assert leaks(_case("I am worried no refund was issued."), raised, None, WORLD) == [] + expected = _catalogue(refund_once={"check": _check("if world.data['total'] == '42.50 refunded':\n return None\nreturn 'no'")}) + found = leaks(_case("It should say 42.50 refunded."), expected, None, WORLD) + assert [one.grader_field for one in found] == ["check literal"] + + +def _step(tool, **arguments): + return SimpleNamespace(tool=tool, arguments=arguments) + + +def test_solution_argument_in_the_opening_is_a_leak(): + from fi.alk.harness.leaks import GRADER_ONLY_FIELDS + + assert "scenario.solution (argument values)" in GRADER_ONLY_FIELDS + catalogue = _catalogue(refund_once={"what": "The agent refunds the order."}) + case = _case( + "Ask for your money back.", + persona={"name": "Asha", "initial_message": "Please refund 37.25 as store credit for my order."}, + solution=[_step("refund_order", order_id="O-123", amount=37.25)], + ) + found = leaks(case, catalogue, None, WORLD) + assert [(one.who, one.field, one.grader_field, one.excerpt) for one in found] == [ + ("simulated person", "persona.initial_message", "scenario.solution", "37.25"), + ("tested agent", "persona.initial_message", "scenario.solution", "37.25"), + ] + + +def test_customer_fact_in_both_is_not_a_leak(): + catalogue = _catalogue(refund_once={"what": "The agent refunds the order."}) + case = _case( + "Ask for your money back for the order on your receipt.", + variables={"receipt": "RX-4417"}, + persona={"name": "Asha", "initial_message": "Hi, my receipt number is RX-4417."}, + solution=[_step("refund_order", receipt="RX-4417", note="the customer asked nicely today")], + ) + assert leaks(case, catalogue, None, WORLD) == [] + + +def test_solution_amount_written_either_way_or_nested_is_a_leak(): + catalogue = _catalogue(refund_once={"what": "The agent refunds the order."}) + short = _case( + "Ask for your money back.", + persona={"name": "Asha", "initial_message": "Please refund 37.5 to my card."}, + solution=[_step("refund_order", order_id="O-123", amount=37.5)], + ) + nested = _case( + "Ask for your money back.", + persona={"name": "Asha", "initial_message": "Swap it for item SKU-889 please."}, + solution=[_step("exchange_order", order_id="O-123", items=[{"sku": "SKU-889", "qty": 1}])], + ) + + assert [one.excerpt for one in leaks(short, catalogue, None, WORLD)] == ["37.5", "37.5"] + assert [one.excerpt for one in leaks(nested, catalogue, None, WORLD)] == ["sku 889", "sku 889"] diff --git a/tests/harness/test_negative_controls.py b/tests/harness/test_negative_controls.py new file mode 100644 index 000000000..ead481c78 --- /dev/null +++ b/tests/harness/test_negative_controls.py @@ -0,0 +1,286 @@ +"""Task-specific negative controls must fail the checks, and a refusal case is proved by them.""" + +from __future__ import annotations + +import asyncio +import copy +from types import SimpleNamespace + +from fi.alk.harness import catalogue as catalogue_module +from fi.alk.harness import live_proof +from fi.alk.harness import prove as module +from fi.alk.harness import scenario_tools +from fi.alk.harness.contract import ToolEntry +from fi.alk.harness.prove import Proof +from fi.alk.harness.world.runtime import Call + +from proof_world import ( + BASE, + LOOKUP_O10, + REFUND_CATALOGUE, + REFUND_O10, + catalogue, + check, + control, + install, + refund_case, + scenario, +) + +LOOKUP_O20 = {"tool": "lookup_order", "arguments": {"order_id": "O-20"}, "fixed": {"order_id": "the caller"}} +REFUND_O20 = {"tool": "refund_order", "arguments": {"order_id": "O-20"}, "fixed": {"order_id": "the caller"}} +NO_REFUND = check( + "if world.data['refunds'] or any(c.name == 'refund_order' for c in calls):\n return 'a refund was attempted'\nreturn None" +) +LOOKED_UP_O20 = check( + "return None if any(c.name == 'lookup_order' and c.ok and c.arguments.get('order_id') == 'O-20' for c in calls) " + "else 'O-20 never looked up'" +) + + +def _refusal(sub_goals=("no_refund", "looked_up_o20")): + return scenario( + [LOOKUP_O20], + name="refuse_20_day", + sub_goals=list(sub_goals), + instruction="I bought this 20 days ago and want my money back.", + negative_controls=[control("refund_anyway", "refunds the 20-day order", [LOOKUP_O20, REFUND_O20])], + ) + + +def test_two_refunds_control_fails(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(refund_case(), catalogue(**REFUND_CATALOGUE), tmp_path) + assert proof.controls[0]["name"] == "two_refunds" + assert proof.controls[0]["failed"] is True + assert proof.proved is True and proof.blocked_reason() is None + + +def test_control_that_passes_blocks(monkeypatch, tmp_path): + install(monkeypatch) + loose = check("return None if len(world.data['refunds']) >= 1 else 'no refund'") + proof = module.prove(refund_case(), catalogue(refund_once=loose, looked_up=REFUND_CATALOGUE["looked_up"]), tmp_path) + assert proof.blocked_reason() == "negative_control_passed" + assert "refund_once" in proof.controls[0]["held"] + assert proof.proved is False + + +def test_no_control_blocks(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(refund_case(negative_controls=[]), catalogue(**REFUND_CATALOGUE), tmp_path) + assert proof.blocked_reason() == "no_negative_control" + assert proof.proved is False and proof.kept_as_written is False + + +def test_refusal_case_outcome_no_change(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(_refusal(), catalogue(no_refund=NO_REFUND, looked_up_o20=LOOKED_UP_O20), tmp_path) + assert proof.outcome == "no_change" + assert proof.controls[0]["name"] == "refund_anyway" and proof.controls[0]["failed"] is True + + +def test_no_change_case_holds_despite_weak_checks(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(_refusal(), catalogue(no_refund=NO_REFUND, looked_up_o20=LOOKED_UP_O20), tmp_path) + assert proof.weak == ["no_refund"] + assert proof.holds is True and proof.proved is True + + +def test_control_with_unbound_tool_is_not_run(monkeypatch, tmp_path): + install(monkeypatch, runtime_tools={"place_order"}) + place = {"tool": "place_order", "arguments": {"order_id": "O-10"}} + case = refund_case(negative_controls=[control("places_instead", "places a new order", [LOOKUP_O10, REFUND_O10, place])]) + proof = module.prove(case, catalogue(**REFUND_CATALOGUE), tmp_path) + assert proof.controls[0]["assumed"] == ["place_order"] + assert proof.blocked_reason() == "control_not_run" + assert proof.proved is False and proof.kept_as_written is True + + +def test_refusal_lookup_without_check_is_unchecked(monkeypatch, tmp_path): + install(monkeypatch) + proof = module.prove(_refusal(sub_goals=("no_refund",)), catalogue(no_refund=NO_REFUND), tmp_path) + assert proof.unchecked == ["lookup_order.order_id"] + assert proof.blocked_reason() == "unchecked" + + +def test_unreadable_state_reads_change(monkeypatch, tmp_path): + install(monkeypatch, unreadable=True) + proof = module.prove(_refusal(), catalogue(no_refund=NO_REFUND, looked_up_o20=LOOKED_UP_O20), tmp_path) + assert proof.outcome == "change" + + +def test_assumed_write_never_reads_no_change(monkeypatch, tmp_path): + # The refund step never ran, so the unchanged world says nothing: the case is no refusal case. + install(monkeypatch, runtime_tools={"refund_order"}) + proof = module.prove(refund_case(), catalogue(**REFUND_CATALOGUE), tmp_path, contract=None, actors=[]) + assert proof.assumed == ["refund_order"] + assert proof.outcome == "change" + + +class ServiceWorld: + """The live world over one shared datastore: a bound tool is a POST to the tools service.""" + + def __init__(self, store: dict) -> None: + self.store = store + self.calls: list[Call] = [] + self.runtime_tools: set[str] = set() + self.endpoint_for: dict[str, str] = {} + self.forwarded: list[str] = [] + + @property + def data(self) -> dict: + return self.store + + def bind_tools(self, proxy, endpoint_for, runtime_tools) -> None: + assert proxy == "http://proxy" + self.endpoint_for = dict(endpoint_for) + self.runtime_tools = set(runtime_tools) | set(endpoint_for) + + def forward(self, endpoint, arguments=None, *, record_as="", record=True) -> Call: + self.forwarded.append(endpoint) + args = dict(arguments or {}) + order = self.store["orders"].get(args.get("order_id")) + if endpoint == "lookup": + return Call(endpoint, args, result=order, ok=order is not None) + if order is None or order["age_days"] > 14: + return Call(endpoint, args, ok=False, refused=True, error="refused by policy") + self.store["refunds"].append(args["order_id"]) + return Call(endpoint, args, result={"refunded": args["order_id"]}) + + def call(self, name, arguments=None) -> Call: + call = self.forward(self.endpoint_for[name], arguments) + self.calls.append(call) + return call + + def state(self) -> dict: + return copy.deepcopy(self.store) + + +class Stack: + """A runtime whose reset restores the seeded datastore, and a factory over it.""" + + def __init__(self) -> None: + self.store = copy.deepcopy(BASE) + self.resets = 0 + self.endpoints = {"tool_proxy": SimpleNamespace(address="http://proxy"), "target_http": "x"} + self.worlds: list[ServiceWorld] = [] + + async def reset(self, runtime, *, work_directory) -> None: + self.resets += 1 + self.store.clear() + self.store.update(copy.deepcopy(BASE)) + + async def create(self, runtime, *, rng) -> ServiceWorld: + world = ServiceWorld(self.store) + self.worlds.append(world) + return world + + +def _contract(modes: dict[str, str]): + entries = {name: ToolEntry(tool=name, mode=mode, endpoint={"lookup_order": "lookup", "refund_order": "refund"}[name]) for name, mode in modes.items()} + return SimpleNamespace(tool_names=lambda: list(modes), entry_for=entries.get) + + +def _prove(monkeypatch, tmp_path, modes, goals=REFUND_CATALOGUE): + monkeypatch.setattr(scenario_tools, "load_scenarios", lambda path: [refund_case()]) + monkeypatch.setattr(catalogue_module, "load_catalogue", lambda path: catalogue(**goals)) + monkeypatch.setattr(live_proof, "_digest", lambda world: repr(sorted(world.state()["refunds"]))) + stack = Stack() + results = asyncio.run( + live_proof.prove_live( + authoring=tmp_path, contract=_contract(modes), runtime=stack, provider=stack, factory=stack, work=tmp_path + ) + ) + return results, stack + + +def test_service_tool_control_runs_and_fails_on_the_runtime(monkeypatch, tmp_path): + results, stack = _prove(monkeypatch, tmp_path, {"lookup_order": "service", "refund_order": "service"}) + + live = results["refund_case"] + assert live["assumed"] == [] + assert [(one["name"], one["held"]) for one in live["with_solution"]] == [("refund_once", True), ("looked_up", True)] + control = live["controls"][0] + assert control["name"] == "two_refunds" + assert control["failed"] is True and control["assumed"] == [] + assert "refund" in stack.worlds[0].forwarded + # A reset before the reference, the empty run and the control, and one after. + assert stack.resets == 4 + assert live_proof.load_live(tmp_path)["refund_case"] == live + + proof = live_proof.apply_live(Proof(deferred=True), live) + assert proof.proved is True and proof.blocked_reason() is None + + +CALLED_ONCE = check( + "n = sum(1 for c in calls if c.ok and c.name == 'refund_order' and c.arguments.get('order_id') == 'O-10')\n" + "return None if n == 1 else f'{n} refund calls'" +) + + +def test_in_process_tool_stays_assumed(monkeypatch, tmp_path): + # The refund runs inside the agent's own process: it has no endpoint, so its step is recorded + # unrun and only a check that reads the calls can hold. + goals = dict(refund_once=CALLED_ONCE, looked_up=REFUND_CATALOGUE["looked_up"]) + results, stack = _prove(monkeypatch, tmp_path, {"lookup_order": "service", "refund_order": "import"}, goals) + assert "refund" not in stack.worlds[0].forwarded + + live = results["refund_case"] + assert live["assumed"] == ["refund_order"] + assert live["controls"][0]["assumed"] == ["refund_order"] + proof = live_proof.apply_live(Proof(deferred=True), live) + assert proof.proved is False and proof.kept_as_written is True + + +def test_no_tool_proxy_plays_nothing(monkeypatch, tmp_path): + contract = _contract({"lookup_order": "service", "refund_order": "service"}) + assert live_proof.service_endpoints(contract, {"target_http": "x"}) == {} + assert live_proof.service_endpoints(contract, {"tool_proxy": "y"}) == {"lookup_order": "lookup", "refund_order": "refund"} + + +RAISES_ON_TWO = check( + "if len(world.data['refunds']) > 1:\n raise KeyError('second refund')\n" + "return None if len(world.data['refunds']) == 1 else 'no refund'" +) + + +def test_a_check_that_raises_under_a_control_is_broken_not_failing(monkeypatch, tmp_path): + install(monkeypatch) + goals = dict(refund_once=RAISES_ON_TWO, looked_up=REFUND_CATALOGUE["looked_up"]) + proof = module.prove(refund_case(), catalogue(**goals), tmp_path) + + control = proof.controls[0] + assert control["failed"] is False and control["broken"] == ["refund_once"] + assert proof.proved is False + assert proof.blocked_reason() == "broken_check" + assert proof.broken == ["refund_once (under negative control two_refunds)"] + + +def test_a_control_that_could_not_be_played_is_no_evidence(): + proof = Proof(ready=True, solvable=True, vacuous=False, outcome="change") + proof.controls = [{"name": "two_refunds", "why": "", "failed": None, "held": [], "assumed": []}] + + assert proof.ran_controls == [] + assert proof.proved is False + assert proof.blocked_reason() == "control_not_run" + + +def test_a_control_that_crashes_on_the_runtime_keeps_the_case_live_proof(monkeypatch, tmp_path): + real = live_proof._play + played = [] + + def play(scenario, world, steps): + played.append(len(steps)) + if len(played) == 3: # reference, empty run, then the control + raise ConnectionError("tools service went away") + return real(scenario, world, steps) + + monkeypatch.setattr(live_proof, "_play", play) + results, _stack = _prove(monkeypatch, tmp_path, {"lookup_order": "service", "refund_order": "service"}) + + live = results["refund_case"] + assert [one["held"] for one in live["with_solution"]] == [True, True] + control = live["controls"][0] + assert control["failed"] is None and "tools service went away" in control["error"] + proof = live_proof.apply_live(Proof(deferred=True), live) + assert proof.proved is False and proof.blocked_reason() == "control_not_run" diff --git a/tests/harness/test_nudge_turn.py b/tests/harness/test_nudge_turn.py index 2aed04058..34dc438fa 100644 --- a/tests/harness/test_nudge_turn.py +++ b/tests/harness/test_nudge_turn.py @@ -95,6 +95,17 @@ def _scenario(name: str = "adds-a-big-mac", **overrides) -> dict: {"tool": "add", "arguments": {"item_id": "big_mac"}}, ], "sub_goals": ["item-added", "right-item"], + "negative_controls": [ + { + "name": "adds_twice", + "why": "adds the item twice", + "steps": [ + {"tool": "lst", "arguments": {}}, + {"tool": "add", "arguments": {"item_id": "big_mac"}}, + {"tool": "add", "arguments": {"item_id": "big_mac"}}, + ], + } + ], } payload.update(overrides) return payload diff --git a/tests/harness/test_proofs.py b/tests/harness/test_proofs.py index 3b10e98e0..426853c2e 100644 --- a/tests/harness/test_proofs.py +++ b/tests/harness/test_proofs.py @@ -6,16 +6,21 @@ from fi.alk.harness.proofs import load_proofs, proof_counts, record_proof, record_unprovable from fi.alk.harness.prove import Proof -from proof_world import ONE_REFUND, catalogue, install, scenario +from proof_world import ONE_REFUND, REFUND_CATALOGUE, catalogue, check, control, install, refund_case, scenario + +RAN = [{"name": "wrong", "why": "does it wrong", "failed": True, "held": [], "assumed": []}] +NOT_RUN = [{"name": "wrong", "why": "does it wrong", "failed": True, "held": [], "assumed": ["place_order"]}] REFUND = [{"tool": "refund_order", "arguments": {"order_id": "O-10"}, "fixed": {"order_id": "the caller"}}] def test_assumed_steps_are_saved_and_not_proved(monkeypatch, tmp_path): install(monkeypatch, runtime_tools={"place_order"}) + place = {"tool": "place_order", "arguments": {"order_id": "O-10"}, "fixed": {"order_id": "the caller"}} case = scenario( - [{"tool": "place_order", "arguments": {"order_id": "O-10"}, "fixed": {"order_id": "the caller"}}], + [place], sub_goals=["placed"], + negative_controls=[control("places_twice", "places the order twice", [place, place])], ) proof = module.prove(case, catalogue(placed="def check(world, calls):\n return None if any(c.ok and c.name == 'place_order' and c.arguments.get('order_id') == 'O-10' for c in calls) else 'not placed'\n"), tmp_path) @@ -31,10 +36,10 @@ def test_assumed_steps_are_saved_and_not_proved(monkeypatch, tmp_path): def test_proof_counts_split_kept_as_written(tmp_path): - proved = Proof(ready=True, solvable=True, vacuous=False) - assumed = Proof(ready=True, solvable=True, vacuous=False, assumed=["place_order"]) + proved = Proof(ready=True, solvable=True, vacuous=False, controls=RAN) + assumed = Proof(ready=True, solvable=True, vacuous=False, assumed=["place_order"], controls=NOT_RUN) deferred = Proof(deferred=True) - failing = Proof(ready=True, solvable=False, vacuous=False) + failing = Proof(ready=True, solvable=False, vacuous=False, controls=RAN) for key in ("a", "b", "c"): record_proof(tmp_path, key, proved) record_proof(tmp_path, "d", assumed) @@ -66,10 +71,141 @@ def test_deferred_reads_proof_deferred_not_not_ready(monkeypatch, tmp_path): def test_holds_failures_map_to_codes(): - base = dict(ready=True, solvable=True, vacuous=False) + base = dict(ready=True, solvable=True, vacuous=False, controls=RAN) assert Proof(**base, weak=["no_refund"]).blocked_reason() == "weak_check" assert Proof(**base, failed_attempts=["tried"]).blocked_reason() == "credits_failed_attempt" assert Proof(**base, too_strict=["refund_once (alternative 1)"]).blocked_reason() == "too_strict" assert Proof(**base, unchecked=["refund_order.order_id"]).blocked_reason() == "unchecked" for proof in (Proof(**base, weak=["x"]), Proof(**base, unchecked=["y"])): assert proof.kept_as_written is False + + +CALLED_ONCE = check( + "n = sum(1 for c in calls if c.ok and c.name == 'refund_order' and c.arguments.get('order_id') == 'O-10')\n" + "return None if n == 1 else f'{n} refund calls'" +) + + +def test_lane_codes_never_hide_a_fault(monkeypatch, tmp_path): + # The refund tool has no endpoint, so its steps are recorded unrun and the check reads the calls. + install(monkeypatch, runtime_tools={"refund_order"}) + leaked = refund_case(instruction="Make sure the agent will refund once for order O-10.") + goals = catalogue(refund_once=CALLED_ONCE, looked_up=REFUND_CATALOGUE["looked_up"]) + proof = module.prove(leaked, goals, tmp_path, contract=None, actors=[]) + assert proof.assumed == ["refund_order"] + assert proof.blocked_reason() == "answer_leaked" and proof.kept_as_written is False + + clean = module.prove(refund_case(), goals, tmp_path, contract=None, actors=[]) + assert clean.blocked_reason() == "solution_assumed" and clean.kept_as_written is True + + +def test_why_names_the_block_not_the_lane(tmp_path): + record_proof(tmp_path, "k", Proof(ready=True, solvable=True, vacuous=False, assumed=["place_order"], controls=[])) + saved = load_proofs(tmp_path)["k"] + assert saved["blocked_reason"] == "no_negative_control" + assert saved["why"].startswith("no negative control") + + +def _judged_case(**extra): + from fi.alk.harness.catalogue import Catalogue, SubGoal + + goals = Catalogue(sub_goals=[SubGoal(name="explains_the_limit", what="The agent explains the limit.", judged="read it")]) + return scenario([], sub_goals=["explains_the_limit"], **extra), goals + + +def _said_control(name="promises_a_refund"): + from fi.alk.harness.scenario import NegativeControl + + return NegativeControl( + name=name, + why="promises a refund outside the window", + transcript=[ + {"role": "user", "content": "Can I get my money back for the 20-day order?"}, + {"role": "assistant", "content": "Of course, I have refunded it."}, + ], + ) + + +def test_judged_only_is_kept_as_written_not_proved(monkeypatch, tmp_path): + install(monkeypatch) + case, goals = _judged_case() + asked: list[str] = [] + + proof = module.prove(case, goals, tmp_path, contract=None, actors=[], judge=lambda *a: asked.append("x")) + record_proof(tmp_path, "judged", proof) + saved = load_proofs(tmp_path)["judged"] + + assert proof.judged_only is True and proof.holds is True + assert proof.proved is False + assert proof.blocked_reason() == "judged_only" and proof.kept_as_written is True + assert saved["why"] == module.JUDGED_ONLY_TEXT + assert asked == [] + + +def test_judged_only_control_with_no_transcript_is_not_played(monkeypatch, tmp_path): + install(monkeypatch) + from proof_world import LOOKUP_O10 + + case, goals = _judged_case(negative_controls=[control("looks_up", "looks the order up", [LOOKUP_O10])]) + asked: list[str] = [] + + proof = module.prove(case, goals, tmp_path, contract=None, actors=[], judge=lambda *a: asked.append("x")) + + assert asked == [] + assert proof.controls[0]["failed"] is None and proof.controls[0]["judged"] is True + assert proof.blocked_reason() == "judged_only" + + +def test_judged_control_the_judge_fails_proves_the_case(monkeypatch, tmp_path): + install(monkeypatch) + case, goals = _judged_case(negative_controls=[_said_control()]) + seen: list[list[dict]] = [] + + def judge(goal, world, calls, messages, scenario): + seen.append(messages) + return False, "it promised a refund the rule forbids" + + proof = module.prove(case, goals, tmp_path, contract=None, actors=[], judge=judge) + + assert seen and seen[0][1]["content"] == "Of course, I have refunded it." + assert proof.controls[0]["failed"] is True + assert proof.controls[0]["verdicts"]["explains_the_limit"]["verdict"] == "fail" + assert proof.proved is True and proof.blocked_reason() is None + + +def test_judged_control_the_judge_passes_is_a_passed_control(monkeypatch, tmp_path): + install(monkeypatch) + case, goals = _judged_case(negative_controls=[_said_control()]) + + proof = module.prove(case, goals, tmp_path, contract=None, actors=[], judge=lambda *a: (True, "looks fine")) + + assert proof.proved is False + assert proof.blocked_reason() == "negative_control_passed" and proof.kept_as_written is False + + +def test_inconclusive_judge_keeps_the_case_as_written(monkeypatch, tmp_path): + install(monkeypatch) + case, goals = _judged_case(negative_controls=[_said_control()]) + + proof = module.prove(case, goals, tmp_path, contract=None, actors=[], judge=lambda *a: (None, "")) + + assert proof.controls[0]["failed"] is None + assert proof.blocked_reason() == "judged_only" and proof.kept_as_written is True + + +def test_check_reading_calls_records_tool_calls_evidence(monkeypatch, tmp_path): + from proof_world import LOOKED_UP_O10 + + assert module.check_evidence(LOOKED_UP_O10) == ["tool_calls"] + assert module.check_evidence(ONE_REFUND) == ["world_state"] + both = check("return None if calls and world.data['refunds'] else 'no'") + assert module.check_evidence(both) == ["tool_calls", "world_state"] + assert module.check_evidence("def check(:\n") == [] + + install(monkeypatch) + proof = module.prove(refund_case(), catalogue(**REFUND_CATALOGUE), tmp_path, contract=None, actors=[]) + record_proof(tmp_path, "refund_case", proof) + assert load_proofs(tmp_path)["refund_case"]["evidence"] == { + "refund_once": ["world_state"], + "looked_up": ["tool_calls"], + } diff --git a/tests/harness/test_prove_suite.py b/tests/harness/test_prove_suite.py new file mode 100644 index 000000000..79a5ab92c --- /dev/null +++ b/tests/harness/test_prove_suite.py @@ -0,0 +1,79 @@ +"""Every certification proves every case again, writing each proof as it goes.""" + +from __future__ import annotations + +import json +import time + +from fi.alk.harness import catalogue as catalogue_module +from fi.alk.harness import prove as module +from fi.alk.harness import scenario_tools +from fi.alk.harness.prove_suite import main, prove_suite + +from proof_world import REFUND_CATALOGUE, catalogue, install, refund_case + + +def _suite(monkeypatch, cases): + monkeypatch.setattr(scenario_tools, "load_scenarios", lambda path: cases) + monkeypatch.setattr(catalogue_module, "load_catalogue", lambda path: catalogue(**REFUND_CATALOGUE)) + monkeypatch.setattr(module, "_load_contract", lambda root: None) + monkeypatch.setattr(module, "_load_actors", lambda root: []) + + +def test_writes_every_key_and_survives_one_crash(monkeypatch, tmp_path): + install(monkeypatch) + cases = [refund_case(name="one", scenario_key="one"), refund_case(name="two", scenario_key="two")] + _suite(monkeypatch, cases) + real = module.prove + + def flaky(scenario, *args, **kwargs): + if scenario.name == "two": + raise ValueError("the world exploded\nsecond line") + return real(scenario, *args, **kwargs) + + monkeypatch.setattr(module, "prove", flaky) + output = tmp_path / "out" / "scenario-proofs.json" + assert main(["--authoring", str(tmp_path), "--output", str(output)]) == 0 + proofs = json.loads(output.read_text(encoding="utf-8")) + assert proofs["one"]["proved"] is True + assert proofs["two"]["blocked_reason"] == "broken_check" + assert proofs["two"]["why"] == "ValueError: the world exploded" + + +def test_missing_record_reads_not_ready(monkeypatch, tmp_path): + install(monkeypatch, ready=False, ready_said="order O-999 is not in the world") + _suite(monkeypatch, [refund_case(name="o999", scenario_key="o999")]) + proofs = prove_suite(tmp_path, tmp_path / "proofs.json") + assert proofs["o999"]["blocked_reason"] == "not_ready" + assert "O-999" in proofs["o999"]["why"] + + +def test_budget_writes_remaining_as_incomplete(monkeypatch, tmp_path): + install(monkeypatch) + _suite(monkeypatch, [refund_case(name="a", scenario_key="a"), refund_case(name="b", scenario_key="b")]) + output = tmp_path / "proofs.json" + assert main(["--authoring", str(tmp_path), "--output", str(output), "--budget-seconds", "0"]) == 0 + proofs = json.loads(output.read_text(encoding="utf-8")) + assert set(proofs) == {"a", "b"} + assert all(one["blocked_reason"] == "validation_incomplete" for one in proofs.values()) + assert all("budget" in one["why"] for one in proofs.values()) + past = str(time.time() - 10) + assert main(["--authoring", str(tmp_path), "--output", str(output), "--deadline-at", past]) == 0 + assert all(one["blocked_reason"] == "validation_incomplete" for one in json.loads(output.read_text()).values()) + + +def test_keys_prove_only_the_named_cases(monkeypatch, tmp_path): + install(monkeypatch) + _suite(monkeypatch, [refund_case(name=key, scenario_key=key) for key in ("a", "b", "c")]) + proved: list[str] = [] + real = module.prove + + def counting(scenario, *args, **kwargs): + proved.append(scenario.scenario_key) + return real(scenario, *args, **kwargs) + + monkeypatch.setattr(module, "prove", counting) + output = tmp_path / "proofs.json" + assert main(["--authoring", str(tmp_path), "--output", str(output), "--keys", "b, c"]) == 0 + assert proved == ["b", "c"] + assert set(json.loads(output.read_text(encoding="utf-8"))) == {"b", "c"} diff --git a/tests/harness/test_world_handle.py b/tests/harness/test_world_handle.py index f59c9e9e1..c2cb4e73d 100644 --- a/tests/harness/test_world_handle.py +++ b/tests/harness/test_world_handle.py @@ -663,7 +663,7 @@ def test_read_only_dunder_lookups_stay_plain_attribute_errors() -> None: view.__wrapped__ -# --- call(): not implemented, the http_tool shim's wire format is unpinned ------------------ +# --- call(): a tool with no bound service endpoint is unavailable ------------------------ def test_call_is_unavailable() -> None: diff --git a/tests/test_harness.py b/tests/test_harness.py index 122defdc6..9c8214b3c 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -2673,6 +2673,19 @@ def _delta(**overrides): {"tool": "add", "arguments": {"item_id": "big_mac"}}, ], "sub_goals": ["item-added", "right-item"], + # The builder cites where each new case comes from; an assumption source when nothing does. + "sources": [{"kind": "assumption", "quote": "Customers order one item at a time.", "supports": "one Big Mac"}], + "negative_controls": [ + { + "name": "adds_twice", + "why": "adds the item twice", + "steps": [ + {"tool": "lst", "arguments": {}}, + {"tool": "add", "arguments": {"item_id": "big_mac"}}, + {"tool": "add", "arguments": {"item_id": "big_mac"}}, + ], + } + ], } payload.update(overrides) return payload @@ -9311,3 +9324,60 @@ def test_an_old_contract_gains_the_new_tool_fields_through_fix_tool(tmp_path): tool.effects == "clears the basket total" and tool.refuses == "returns ok false for an unknown item" ) + + +def test_new_case_without_sources_is_refused(tmp_path): + from fi.alk.harness.scenario_tools import NO_SOURCES, accept_scenario + + root, _contract, catalogue = _built_environment(tmp_path) + kept = [] + said = accept_scenario(_delta(sources=[]), world_root=root, catalogue=catalogue, kept=kept, require_sources=True) + assert said.get("is_error") and NO_SOURCES in said["content"][0]["text"] + assert kept == [] + blank = [{"kind": "assumption", "quote": " ", "supports": ""}] + said = accept_scenario(_delta(sources=blank), world_root=root, catalogue=catalogue, kept=kept, require_sources=True) + assert said.get("is_error") and NO_SOURCES in said["content"][0]["text"] + + assumed = [{"kind": "assumption", "quote": "Customers order one item at a time.", "supports": "one Big Mac"}] + said = accept_scenario( + _delta(sources=assumed), world_root=root, catalogue=catalogue, kept=kept, require_sources=True + ) + assert not said.get("is_error"), said + + +def test_unchanged_legacy_case_without_sources_saves(tmp_path): + from fi.alk.harness.scenario_tools import NO_SOURCES, accept_scenario + + root, _contract, catalogue = _built_environment(tmp_path) + kept = [] + # Written before sources were required. + assert not accept_scenario(_delta(sources=[]), world_root=root, catalogue=catalogue, kept=kept).get("is_error") + key = kept[0].scenario_key + + same = accept_scenario( + _delta(scenario_key=key, sources=[]), world_root=root, catalogue=catalogue, kept=kept, require_sources=True + ) + assert not same.get("is_error"), same + edited = accept_scenario( + _delta(scenario_key=key, instruction="Order one Big Mac, please.", sources=[]), + world_root=root, + catalogue=catalogue, + kept=kept, + require_sources=True, + ) + assert edited.get("is_error") and NO_SOURCES in edited["content"][0]["text"] + + +def test_must_pass_round_trips(tmp_path): + from fi.alk.harness.folder import document_for + from fi.alk.harness.scenario_tools import accept_scenario, load_scenarios + + assert Scenario(name="plain").must_pass is False + root, _contract, catalogue = _built_environment(tmp_path) + kept = [] + said = accept_scenario(_delta(must_pass=True), world_root=root, catalogue=catalogue, kept=kept) + assert not said.get("is_error"), said + assert document_for(kept[0])["must_pass"] is True + assert [one.must_pass for one in load_scenarios(root)] == [True] + # An unflagged case writes no must_pass, so an older suite rewritten as it was is unchanged. + assert "must_pass" not in document_for(Scenario(name="plain"))