From 27a5cd69cdc263bca4cdf1bc7667463966bd9deb Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Sat, 22 Aug 2026 11:01:50 +0530 Subject: [PATCH 1/2] feat(harness): generate a suite with one writer per use case, counted by use case and branch --- harness-ui/server.py | 14 ++ src/fi/alk/harness/persona_guides.py | 199 +++++++++++++++ src/fi/alk/harness/progress.py | 113 +++++++++ src/fi/alk/harness/scenario.py | 12 + src/fi/alk/harness/scenario_tools.py | 144 +++++++++-- src/fi/alk/harness/scenarios.py | 238 ++++++++++++++++++ .../harness/skills/write-scenarios/SKILL.md | 23 +- 7 files changed, 717 insertions(+), 26 deletions(-) create mode 100644 src/fi/alk/harness/persona_guides.py create mode 100644 src/fi/alk/harness/progress.py diff --git a/harness-ui/server.py b/harness-ui/server.py index 153fbe66..f084f1e9 100644 --- a/harness-ui/server.py +++ b/harness-ui/server.py @@ -460,6 +460,20 @@ async def world(): db.close() +@app.get("/api/generation") +async def generation(session: str = ""): + """What the suite generation is doing right now, so the page can draw it while it runs. + + Polled rather than streamed: the page may be opened halfway through a suite, refreshed, or + opened somewhere else entirely, and each of those has to show the same thing. Empty when + nothing has been generated here, which the page reads as "no fan-out to show". + """ + from fi.alk.harness import progress + + out = _folder(session) + return progress.read(out) if out else {} + + @app.get("/api/scenarios") async def scenarios(): """Every scenario, with its files and its three gates re-run. diff --git a/src/fi/alk/harness/persona_guides.py b/src/fi/alk/harness/persona_guides.py new file mode 100644 index 00000000..01231a6d --- /dev/null +++ b/src/fi/alk/harness/persona_guides.py @@ -0,0 +1,199 @@ +"""The behaviour guidance the platform already uses for a simulated caller. + +A persona profile names what somebody is like: impatient and direct, cautious and skeptical. It +does not say how that should sound turn by turn, and a model handed only the label improvises +one, which is how "in a hurry" became a caller who says it every turn instead of a caller who +cuts in once and accepts the first workable answer. + +The platform solved that with lookup tables mapping each value to a sentence of guidance, and +voice simulation has run on them for months. They are read from there rather than restated here, +because two copies of the same wording drift and then a caller behaves one way on the platform +and another way through the harness, for reasons nobody can see. + +Read, not imported: the tables live inside a Django app this package cannot import, but they are +plain literals, so they are parsed out of the file. Absent, every lookup answers with nothing and +a persona still renders — one without guidance, never a crash. +""" + +from __future__ import annotations + +import ast +import os +from functools import lru_cache +from pathlib import Path + +# Where the platform's tables are mounted. Colon-separated so voice and chat guides can both be +# offered; the first file defining a table wins, so voice takes precedence when both are present. +GUIDES_ENV = "HARNESS_PERSONA_GUIDES" + +WANTED = ( + "VOICE_PERSONALITY_GUIDES", + "VOICE_COMMUNICATION_STYLE_GUIDES", + "CHAT_PERSONALITY_GUIDES", + "CHAT_COMMUNICATION_STYLE_GUIDES", + "CHAT_TONE_GUIDES", + "CHAT_VERBOSITY_GUIDES", +) + + +def _tables_in(path: Path) -> dict[str, dict[str, str]]: + """Every guidance table defined in one file, by name. + + Parsed rather than executed. The file sits in an app with imports this process cannot + satisfy, and running it to read a dictionary would fail for reasons that have nothing to do + with the dictionary. + """ + found: dict[str, dict[str, str]] = {} + try: + tree = ast.parse(path.read_text(encoding="utf-8")) + except (OSError, SyntaxError): + return found + for node in tree.body: + targets = ( + [node.target] if isinstance(node, ast.AnnAssign) else getattr(node, "targets", []) + ) + for target in targets: + name = getattr(target, "id", "") + if name not in WANTED or node.value is None: + continue + try: + value = ast.literal_eval(node.value) + except ValueError: + continue + if isinstance(value, dict) and value: + found[name] = {str(k).lower(): str(v) for k, v in value.items()} + return found + + +@lru_cache(maxsize=1) +def guides() -> dict[str, dict[str, str]]: + """Every table the platform offers this harness, merged.""" + merged: dict[str, dict[str, str]] = {} + for raw in (os.environ.get(GUIDES_ENV) or "").split(":"): + if not raw.strip(): + continue + for name, table in _tables_in(Path(raw.strip())).items(): + merged.setdefault(name, table) + return merged + + +def guidance_for(kind: str, value: str, *, voice: bool = True) -> str: + """The platform's sentence for one persona value, or nothing. + + ``kind`` is ``personality``, ``communication_style``, ``tone`` or ``verbosity``. Voice tables + are preferred for a spoken call and the chat table is the fallback, because the two describe + the same disposition and only one of them is written for speech. + """ + if not value.strip(): + return "" + tables = guides() + order = ("VOICE", "CHAT") if voice else ("CHAT", "VOICE") + for prefix in order: + table = tables.get(f"{prefix}_{kind.upper()}_GUIDES") or {} + found = table.get(value.strip().lower()) + if found: + return found + return "" + + +def available() -> bool: + """Whether any guidance was found, so a build can say so rather than silently omitting it.""" + return bool(guides()) + + +# Where the platform's persona model is mounted, for the values it accepts. +VOCABULARY_ENV = "HARNESS_PERSONA_VOCABULARY" + +# The persona fields worth constraining, and the choice class each is drawn from. Only the ones +# that change behaviour or routing: a free-text occupation harms nothing, an accent nobody +# recognises silently loses the voice it was supposed to select. +FIELDS = { + "gender": "GenderChoices", + "age_group": "AgeGroupChoices", + "occupation": "ProfessionChoices", + "location": "LocationChoices", + "personality": "PersonalityChoices", + "communication_style": "CommunicationStyleChoices", + "accent": "AccentChoices", + "languages": "LanguageChoices", +} + +# Constrained because something downstream reads them. The rest are offered as vocabulary but a +# writer who needs a value outside them is not stopped: an unknown occupation costs nothing, +# an unknown accent costs the voice. +ENFORCED = ("personality", "communication_style", "accent", "languages") + + +@lru_cache(maxsize=1) +def vocabulary() -> dict[str, list[str]]: + """What the platform accepts for each persona field. + + Parsed out of the model's ``TextChoices`` classes for the same reason the guidance is read + rather than restated: the platform is the one that has to understand these values, so it is + the one that decides what they are. A persona written in words of its own renders fine, gets + no behaviour guidance, and cannot be grouped with anything on the platform afterwards. + """ + path = os.environ.get(VOCABULARY_ENV) or "" + if not path or not Path(path).exists(): + return {} + try: + tree = ast.parse(Path(path).read_text(encoding="utf-8")) + except (OSError, SyntaxError): + return {} + + by_class: dict[str, list[str]] = {} + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + values: list[str] = [] + for item in node.body: + if not isinstance(item, ast.Assign): + continue + try: + held = ast.literal_eval(item.value) + except ValueError: + continue + # ``NAME = "value", "Label"`` is the choices shape; a bare string is also accepted. + if isinstance(held, tuple) and held and isinstance(held[0], str): + values.append(held[0]) + elif isinstance(held, str): + values.append(held) + if values: + by_class[node.name] = values + + return { + field: by_class[cls] for field, cls in FIELDS.items() if by_class.get(cls) + } + + +def offered(field: str) -> list[str]: + """The values this field accepts, or nothing if the platform's model was not readable.""" + return list(vocabulary().get(field, [])) + + +def unrecognised(persona: dict[str, object]) -> list[str]: + """Persona values the platform would not recognise, as sentences saying what to use instead. + + Only the fields something downstream actually reads, and only when the vocabulary was found: + a harness that cannot see the platform's model must not start refusing personas over it. + """ + known = vocabulary() + if not known: + return [] + problems: list[str] = [] + for field in ENFORCED: + allowed = known.get(field) or [] + if not allowed: + continue + held = persona.get(field) + values = held if isinstance(held, list) else ([held] if held else []) + lowered = {str(one).strip().lower() for one in allowed} + for one in values: + text = str(one).strip() + if text and text.lower() not in lowered: + problems.append( + f"persona {field} {text!r} is not one the platform knows, so it will not " + f"reach the call. Use one of: {', '.join(allowed)}. Anything else this " + "person is like belongs in persona.metadata." + ) + return problems diff --git a/src/fi/alk/harness/progress.py b/src/fi/alk/harness/progress.py new file mode 100644 index 00000000..eb765917 --- /dev/null +++ b/src/fi/alk/harness/progress.py @@ -0,0 +1,113 @@ +"""What the fan-out is doing right now, written where a UI can read it. + +Generating a suite in parallel is the one thing this harness does where nothing appears for +several minutes and then everything appears at once. Told nothing, a person cannot tell a +working run from a hung one, and the honest answer to "is it stuck" is the only thing they want. + +So the fan-out writes its own state as it goes: which use cases it split the work into, which +are running, how many scenarios each has proved, and which have finished. A file rather than a +stream, because the reader is a page that may be opened halfway through, refreshed, or opened on +another machine, and each of those has to show the same thing. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from pathlib import Path +from typing import Any + +PROGRESS = "generation.json" + +WAITING = "waiting" +RUNNING = "running" +DONE = "done" +FAILED = "failed" + + +def _path(destination: Path) -> Path: + return Path(destination) / PROGRESS + + +def _write(destination: Path, state: dict[str, Any]) -> None: + """Replace the file atomically. + + A reader polling this will otherwise catch a half-written file and show nothing, which looks + exactly like the failure it is meant to rule out. + """ + path = _path(destination) + path.parent.mkdir(parents=True, exist_ok=True) + handle, temporary = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp") + try: + with os.fdopen(handle, "w", encoding="utf-8") as writing: + json.dump(state, writing, indent=2) + os.replace(temporary, path) + except BaseException: + Path(temporary).unlink(missing_ok=True) + raise + + +def read(destination: Path) -> dict[str, Any]: + """The current state, or nothing if no suite has been generated here.""" + path = _path(destination) + if not path.exists(): + return {} + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + + +def planned( + destination: Path, allocation: list[tuple[str, int]], *, at_once: int, asked: int +) -> None: + """The split, before any of it starts. Written first so the tree appears immediately.""" + _write( + destination, + { + "state": RUNNING, + "asked": asked, + "at_once": at_once, + "kept": 0, + "slices": [ + {"use_case": case, "wanted": count, "kept": 0, "state": WAITING} + for case, count in allocation + ], + }, + ) + + +def _change(destination: Path, use_case: str, **fields: Any) -> None: + state = read(destination) + for slice_ in state.get("slices", []): + if slice_.get("use_case") == use_case: + slice_.update(fields) + break + state["kept"] = sum(one.get("kept", 0) for one in state.get("slices", [])) + _write(destination, state) + + +def started(destination: Path, use_case: str) -> None: + _change(destination, use_case, state=RUNNING) + + +def kept(destination: Path, use_case: str, count: int) -> None: + """How many this slice has proved so far. Called as they land, not at the end.""" + _change(destination, use_case, kept=count) + + +def finished(destination: Path, use_case: str, count: int) -> None: + _change(destination, use_case, state=DONE, kept=count) + + +def failed(destination: Path, use_case: str, why: str) -> None: + _change(destination, use_case, state=FAILED, why=why[:300]) + + +def settled(destination: Path, *, kept_total: int) -> None: + """The whole fan-out is over and the suite is written.""" + state = read(destination) + state["state"] = DONE + state["kept"] = kept_total + _write(destination, state) diff --git a/src/fi/alk/harness/scenario.py b/src/fi/alk/harness/scenario.py index 90428c6c..1c62c5f1 100644 --- a/src/fi/alk/harness/scenario.py +++ b/src/fi/alk/harness/scenario.py @@ -154,6 +154,11 @@ class Scenario(BaseModel): name: str use_case: str = "" + # Which branch of that use case this is: the condition that makes this row different from + # its siblings. A use case fans out into several — the ordinary path, the one that cannot be + # completed, the rule under pressure — and each is its own test. Coverage is counted on the + # pair, so a use case can carry many scenarios without any of them reading as a duplicate. + branch: str = "" tests: str = "" # What this scenario changes about the world after it is reset, as code: a file defining @@ -232,6 +237,13 @@ def validate_scenario( missing := scenario.persona.missing_profile_fields() ): problems.append("persona is incomplete: " + ", ".join(missing)) + elif scenario.persona is not None: + # A persona written in words of its own renders fine and then does nothing: no behaviour + # guidance attaches to it, and the accent it names selects no voice. Caught here, where + # the writer is still holding the scenario and can fix it in one turn. + from .persona_guides import unrecognised + + problems.extend(unrecognised(scenario.persona.model_dump())) if not scenario.sub_goals: problems.append( "no sub_goals: nothing would be graded. Name the entries of the catalogue this " diff --git a/src/fi/alk/harness/scenario_tools.py b/src/fi/alk/harness/scenario_tools.py index e6249837..6bc78c10 100644 --- a/src/fi/alk/harness/scenario_tools.py +++ b/src/fi/alk/harness/scenario_tools.py @@ -50,6 +50,31 @@ def _err(text: str) -> dict[str, Any]: return {"content": [{"type": "text", "text": text}], "is_error": True} +def persona_field(name: str) -> dict[str, Any]: + """The schema for one persona field, carrying the platform's own values where it has them. + + Offered as an enum so the values arrive right the first time. Without the platform's model + to read, it stays a plain string rather than an enum of nothing. + """ + from .persona_guides import offered + + allowed = offered(name) + return {"type": "string", "enum": allowed} if allowed else {"type": "string"} + + +def persona_vocabulary_note() -> str: + """A sentence about why the persona fields are constrained, when they are.""" + from .persona_guides import vocabulary + + if not vocabulary(): + return "" + return ( + " The listed values are the ones the platform understands: they carry behaviour " + "guidance into the call and select the caller's voice. Anything else about this person " + "goes in metadata, where it is free text." + ) + + def write_scenarios( scenarios: list[Scenario], destination: Path, catalogue: Catalogue | None = None ) -> Path: @@ -168,17 +193,22 @@ def not_ready(kept: list[Scenario], wanted: int, catalogue: Catalogue) -> list[s # "cancel a pending order", which is neither what it tests nor distinguishable afterwards # from the scenario that really does test that. A use case is how coverage is counted, so a # duplicate quietly overstates it. - claimed: dict[str, list[str]] = {} + # Keyed on the pair, not the use case alone. A use case fans out into several branches and + # each is a separate test, so keying on the use case alone caps a suite at one scenario per + # use case — which is how a request for forty against fourteen use cases became unsaveable. + claimed: dict[tuple[str, str], list[str]] = {} for one in kept: case = (one.use_case or "").strip().lower() + branch = (one.branch or "").strip().lower() if case: - claimed.setdefault(case, []).append(one.name) - for case, names in claimed.items(): + claimed.setdefault((case, branch), []).append(one.name) + for (case, branch), names in claimed.items(): if len(names) > 1: + where = f"{case!r}" if not branch else f"{case!r} / {branch!r}" problems.append( - f"{' and '.join(names)} both claim the use case {case!r}. Give each the use case " - "it actually exercises, or drop the one that duplicates the other. Coverage is " - "counted by use case, so two scenarios sharing one hides a gap." + f"{' and '.join(names)} both claim {where}. Give each the branch it actually " + "exercises, or drop the one that duplicates the other. Coverage is counted by " + "use case and branch, so two scenarios sharing both hides a gap." ) # Sub-goals are shared so results roll up. A suite where every scenario invents its own is a @@ -193,10 +223,27 @@ def not_ready(kept: list[Scenario], wanted: int, catalogue: Catalogue) -> list[s def scenario_tools( - contract: AgentContract, world_root: Path, destination: Path, *, wanted: int + contract: AgentContract, + world_root: Path, + destination: Path, + *, + wanted: int, + can_save: bool = True, + start_from: list[Scenario] | None = None, ) -> tuple[Any, list[Scenario]]: - """A server for writing scenarios against one built environment.""" - kept: list[Scenario] = load_scenarios(destination) + """A server for writing scenarios against one built environment. + + ``can_save`` is what makes several writers safe at once. Saving rewrites the index and + removes any folder not in the saver's own list, so two writers saving concurrently delete + each other's work. A writer that only submits keeps its scenarios in ``kept``, and whoever + spawned it merges the lists and writes once. + + ``start_from`` seeds that list. A parallel writer starts empty rather than from disk, so it + is never counted as already having what a sibling wrote. + """ + kept: list[Scenario] = ( + list(start_from) if start_from is not None else load_scenarios(destination) + ) catalogue = load_catalogue(destination) simulator_prompt = load_simulator_prompt(destination) target = {"count": wanted} @@ -373,6 +420,12 @@ async def add_sub_goal(args: dict[str, Any]) -> dict[str, Any]: "type": "string", "description": "Which of the agent's use cases this belongs to.", }, + "branch": { + "type": "string", + "description": "The condition that makes this scenario different from the " + "others in the same use case, in one line: what is true here that is not " + "true of its siblings.", + }, "tests": { "type": "string", "description": "One line: what this scenario is trying to find out.", @@ -386,23 +439,27 @@ async def add_sub_goal(args: dict[str, Any]) -> dict[str, Any]: "type": "object", "description": "Who the simulated person is, separate from the task. Use " "the established voice-scenario shape and only grounded, test-relevant " - "details. This fills the simulator prompt's persona slot.", + "details. This fills the simulator prompt's persona slot." + + persona_vocabulary_note(), "properties": { "name": {"type": "string"}, - "gender": {"type": "string"}, - "age_group": {"type": "string"}, - "occupation": {"type": "string"}, - "location": {"type": "string"}, - "personality": {"type": "string"}, - "communication_style": {"type": "string"}, + "gender": persona_field("gender"), + "age_group": persona_field("age_group"), + "occupation": persona_field("occupation"), + "location": persona_field("location"), + "personality": persona_field("personality"), + "communication_style": persona_field("communication_style"), "initial_message": { "type": "string", "description": "The caller's natural opening request, specific to " "this scenario. Do not use a generic greeting.", }, "keywords": {"type": "array", "items": {"type": "string"}}, - "languages": {"type": "array", "items": {"type": "string"}}, - "accent": {"type": "string"}, + "languages": { + "type": "array", + "items": persona_field("languages"), + }, + "accent": persona_field("accent"), "multilingual": {"type": "boolean"}, "metadata": {"type": "object"}, }, @@ -612,6 +669,50 @@ async def drop_scenario(args: dict[str, Any]) -> dict[str, Any]: write_scenarios(kept, destination, catalogue) return _ok(f"{name} dropped. {len(kept)} left") + @tool( + "generate_suite", + "Write a whole suite at once by splitting it across the agent's use cases, one writer " + "per use case, several running at the same time. Use this when somebody asks for a " + "suite rather than a particular scenario: writing twenty or fifty one at a time runs " + "out of turns long before it finishes. Everything it produces has cleared the same " + "three gates. Saved when it completes.", + schema({"count": int, "at_once": int}, ["count"]), + ) + async def generate_suite(args: dict[str, Any]) -> dict[str, Any]: + from .scenarios import write_in_parallel + + count = int(args.get("count") or 0) + if count < 1: + return _err("say how many scenarios the suite should have") + at_once = int(args.get("at_once") or 0) or 4 + cases = [one for one in contract.real_use_cases if one.strip()] + if not cases: + return _err( + "this contract names no use cases, so there is nothing to split the work " + "across. Write them one at a time with submit_scenario, or fix the contract." + ) + produced = await write_in_parallel( + contract, + out=destination, + wanted=count, + use_cases=cases, + at_once=at_once, + ) + # The suite is already on disk. The open session's own list has to be brought level with + # it, or a later save_scenarios here would write out the stale list and delete every + # folder the fan-out just produced. + kept[:] = produced + target["count"] = len(produced) + by_case: dict[str, int] = {} + for one in produced: + name = one.use_case or "unassigned" + by_case[name] = by_case.get(name, 0) + 1 + lines = "\n".join(f" {n} x {case[:70]}" for case, n in sorted(by_case.items())) + return _ok( + f"{len(produced)} scenarios across {len(by_case)} use cases, {at_once} writers at a " + f"time. Each cleared all three gates and the suite is saved.\n{lines}" + ) + @tool( "save_scenarios", "Write the kept scenarios out. Every one has already been proved by submit_scenario, so " @@ -670,8 +771,10 @@ async def save_scenarios(_args: dict[str, Any]) -> dict[str, Any]: fix_tool_tool, aim_for, drop_scenario, - save_scenarios, - ], + ] + # Only the session a person is talking to may fan out. A writer that is itself one slice + # of a fan-out calling this would split its own slice again, and so on. + + ([generate_suite, save_scenarios] if can_save else []), ) return server, kept @@ -688,6 +791,7 @@ async def save_scenarios(_args: dict[str, Any]) -> dict[str, Any]: "fix_tool", "aim_for", "drop_scenario", + "generate_suite", "save_scenarios", ) diff --git a/src/fi/alk/harness/scenarios.py b/src/fi/alk/harness/scenarios.py index 57a4a581..dc32223a 100644 --- a/src/fi/alk/harness/scenarios.py +++ b/src/fi/alk/harness/scenarios.py @@ -10,6 +10,7 @@ from __future__ import annotations +import asyncio from collections.abc import Callable from pathlib import Path from typing import Any @@ -25,6 +26,8 @@ permission_gate, provider_env, ) +from . import progress +from .catalogue import load_catalogue from .contract import AgentContract from .scenario import Scenario from .scenario_tools import ( @@ -33,6 +36,7 @@ load_scenarios, scenario_tools, world_summary, + write_scenarios, ) from .session import Stage from .tools import qualified @@ -131,6 +135,10 @@ def opening(contract: AgentContract, wanted: int = 10, existing: int = 0) -> str "across several turns. If a proof says an intended check is vacuous or broken, repair " "that named sub-goal with add_sub_goal and resubmit. Never evade a gate by deleting a " "check for behavior the scenario still claims to test. Then save_scenarios." + "\n\nFor a suite rather than one scenario, say briefly how you are splitting it " + "across the agent's use cases and then write it with generate_suite in the same turn: " + "it runs a writer per use case at the same time and saves what they prove, where " + "writing this many one at a time would run out of turns before finishing." ) @@ -139,6 +147,236 @@ def load(destination: Path) -> list[Scenario]: return load_scenarios(Path(destination)) +# How many writers run at once. Each is a model session with its own subprocess, and each gate +# restores its own copy of the world, so this is bounded by the machine rather than by the API. +AT_ONCE = 4 + + +def shares(wanted: int, use_cases: list[str]) -> list[tuple[str, int]]: + """How many scenarios each use case is asked to produce. + + Evenly, with the remainder going to the ones named first, because a contract lists its + primary use cases before its marginal ones. A use case that turns out to have less in it + than its share says returns fewer; nothing forces it to pad. + """ + if not use_cases: + return [] + if wanted <= len(use_cases): + return [(case, 1) for case in use_cases[:wanted]] + each, extra = divmod(wanted, len(use_cases)) + return [(case, each + (1 if i < extra else 0)) for i, case in enumerate(use_cases)] + + +def callers_for(index: int, wanted: int) -> str: + """Which callers this slice should write, so the suite varies across slices as well as within. + + Instruction alone cannot do this. Each writer is blind to the others, so each independently + picks the safest value and the suite converges on it: measured across three suites, more + than half the callers came out "Professional and formal" and over three quarters American, + with nobody doing anything wrong. Worse, a slice writing a single scenario has nothing to + vary at all. + + So the spread is dealt out here, the same way the work is. Each slice is handed a different + starting point in the platform's own vocabularies and told to begin there. It is a + suggestion rather than a rule, because the caller still has to suit the scenario: a stolen + phone is not a cheerful call whatever this hands out. + """ + from .persona_guides import offered + + people = offered("personality") + accents = offered("accent") + if not people: + return "" + picks = [people[(index + step) % len(people)] for step in range(max(1, wanted))] + accent = accents[index % len(accents)] if accents else "" + said = ( + "\n\nStart from these callers, and move off them only where the scenario calls for " + f"somebody else: {', '.join(picks)}." + ) + if accent: + said += ( + f" At least one of your callers has a {accent} accent. Other writers are covering " + "other use cases with other callers, so a suite where everyone sounds the same is " + "what happens when each of us picks the safest option." + ) + return said + + +def branch_opening( + contract: AgentContract, use_case: str, wanted: int, callers: str = "" +) -> str: + return ( + f"Write {wanted} scenarios for {contract.agent!r}, all of them within this one " + "use case:\n\n" + f" {use_case}\n\n" + "Write nothing outside it. Somebody else is covering the other use cases at the same " + "time, so a scenario that strays is either a duplicate of theirs or a gap in yours.\n\n" + "Every scenario carries this use case verbatim in `use_case`, and its own one-line " + "`branch` saying what makes it different from the others you write here. Branches are " + "where the variety lives: the ordinary path, the branch that cannot be completed, the " + "rule under pressure, state that has to carry across turns, the same request against a " + "differently seeded world.\n\n" + "Look at the world first with inspect_world so every scenario names real records, and " + "read the sub-goals already defined. Work out each solution with try_calls before you " + "submit it. Submit each one with submit_scenario and then stop: do not save, and do not " + "ask what to do next. Whoever asked for this collects the suite and writes it.\n\n" + "Vary the caller across the scenarios you write. Everyone else is writing their own use " + "case and cannot see yours, so a suite where every caller is professional and formal is " + "what happens when each writer picks the safest value. Give different scenarios " + "different personalities, communication styles and accents from the values offered, and " + "let the caller suit the situation: somebody whose card was declined is not in the same " + "mood as somebody booking a routine morning ride." + callers + ) + + +async def _write_one_use_case( + contract: AgentContract, + use_case: str, + count: int, + *, + index: int = 0, + destination: Path, + on_event: Callable[..., Any] | None, + ask: Callable[..., Any] | None, +) -> list[Scenario]: + """One use case's share, written by its own session. Returns what it proved, unsaved.""" + server, kept = scenario_tools( + contract, + destination, + destination, + wanted=count, + can_save=False, + start_from=[], + ) + progress.started(destination, use_case) + + def watch(event: Any) -> None: + # Report as they land rather than at the end. A slice that proves its first scenario + # four minutes in is the difference between a run that looks alive and one that does not. + progress.kept(destination, use_case, len(kept)) + if on_event: + on_event(event) + allowed = [ + qualified(SCENARIO_SERVER, name) for name in TOOL_NAMES if name != "save_scenarios" + ] + options = ClaudeAgentOptions( + system_prompt=( + f"{load_skill(SKILL)}\n\n## This agent\n\n{contract.brief(with_data=True)}" + f"\n\n## Its world\n\n{world_summary(destination)}" + f"\n\n## Your slice\n\nYou are writing only the scenarios for: {use_case}" + ), + allowed_tools=allowed, + mcp_servers={SCENARIO_SERVER: server}, + permission_mode="default", + cwd=str(destination.parent if destination.parent.exists() else Path.cwd()), + setting_sources=[], + max_turns=turns_for(count), + model=chosen_model(), + env=provider_env(), + ) + options.disallowed_tools = list(UNWANTED) + options.hooks = gate_hooks(allowed) + options.can_use_tool = permission_gate(ask, allowed) + stage = Stage(options, name=f"{SKILL}:{use_case[:40]}") + try: + async with stage: + await stage.say( + branch_opening(contract, use_case, count, callers_for(index, count)), + on_event=watch, + ) + except Exception as broke: # noqa: BLE001 - one slice failing must not lose the others + progress.failed(destination, use_case, str(broke)) + if on_event: + on_event({"type": "slice_failed", "use_case": use_case, "why": str(broke)[:300]}) + return list(kept) + progress.finished(destination, use_case, len(kept)) + return list(kept) + + +def merged(written: list[list[Scenario]]) -> list[Scenario]: + """One suite out of several writers, with the collisions they could not see removed. + + The writers run blind to each other, so two can land on the same folder name or on the same + use case and branch. Both are dropped here rather than at save time, where the loser would + silently overwrite the winner's folder. + """ + suite: list[Scenario] = [] + names: set[str] = set() + pairs: set[tuple[str, str]] = set() + for batch in written: + for one in batch: + pair = ((one.use_case or "").strip().lower(), (one.branch or "").strip().lower()) + if one.name in names or (pair[0] and pair in pairs): + continue + names.add(one.name) + if pair[0]: + pairs.add(pair) + suite.append(one) + return suite + + +async def write_in_parallel( + contract: AgentContract, + *, + out: Path | None = None, + wanted: int = 10, + use_cases: list[str] | None = None, + at_once: int = AT_ONCE, + on_event: Callable[..., Any] | None = None, + ask: Callable[..., Any] | None = None, +) -> list[Scenario]: + """Write a suite with one session per use case, then save it once. + + Sequentially, a suite costs roughly three turns a scenario against one budget, which is why + asking for forty stopped around twenty-five. Here each use case is written by its own + session, so the wall clock is the slowest use case rather than the sum of all of them, and + the turn budget is per slice rather than shared. + + Saving stays here, once, for a reason: ``save_scenarios`` regenerates the index and deletes + any folder it does not know about, so letting the writers save would have each of them + remove the others' work. + """ + destination = out or artifact_dir(contract.agent) + cases = [case for case in (use_cases or contract.real_use_cases) if case.strip()] + if not cases: + # Nothing to partition on. One writer, the ordinary path, rather than no scenarios. + return await write(contract, out=destination, wanted=wanted, on_event=on_event, ask=ask) + + allocation = shares(wanted, cases) + progress.planned(destination, allocation, at_once=at_once, asked=wanted) + if on_event: + on_event({"type": "planned", "slices": allocation, "at_once": at_once}) + + limit = asyncio.Semaphore(max(1, at_once)) + + async def guarded(use_case: str, count: int, index: int) -> list[Scenario]: + async with limit: + return await _write_one_use_case( + contract, + use_case, + count, + index=index, + destination=destination, + on_event=on_event, + ask=ask, + ) + + written = await asyncio.gather( + *( + guarded(case, count, index) + for index, (case, count) in enumerate(allocation) + ), + return_exceptions=False, + ) + + suite = merged([load_scenarios(destination), *written]) + write_scenarios(suite, destination, load_catalogue(destination)) + progress.settled(destination, kept_total=len(suite)) + if on_event: + on_event({"type": "saved", "kept": len(suite), "asked": wanted}) + return load(destination) + + async def write( contract: AgentContract, *, diff --git a/src/fi/alk/harness/skills/write-scenarios/SKILL.md b/src/fi/alk/harness/skills/write-scenarios/SKILL.md index 86ac6c86..474f8ce7 100644 --- a/src/fi/alk/harness/skills/write-scenarios/SKILL.md +++ b/src/fi/alk/harness/skills/write-scenarios/SKILL.md @@ -20,6 +20,7 @@ afterwards. ``` name short identifier; it becomes this scenario's folder use_case which of the agent's use cases this belongs to +branch what makes this one different from its siblings in that use case tests one line: what this scenario is trying to find out instruction the task, written to the person the agent is serving persona who that person is: identity, communication style, languages/accent and characteristics @@ -157,6 +158,16 @@ rule under pressure, the state that has to carry, the same request against a dif world. Keep that plan concise and continue immediately unless the person explicitly asked to review it. + +**Then write the suite with `generate_suite`, not one scenario at a time.** It splits the work +across the agent's use cases and runs several writers at once, each proving its own scenarios +through the same three gates. Writing a suite yourself with `submit_scenario` costs about three +turns per scenario against one budget, so a request for twenty or fifty runs out long before it +finishes, and what does get written is lost because nothing was saved. + +Use `submit_scenario` for what it is good at: one scenario somebody asked for by name, a +replacement for one that came back wrong, or filling a specific gap in a suite that already +exists. Anything described as a number of scenarios is a suite. After inspecting the world, submit the first scenario in the same response. Then prove and save one scenario at a time. Never silently compose the whole suite before the next tool call: the UI must show progress, and already-proved work must survive a stopped or timed-out model turn. @@ -212,9 +223,7 @@ Every stance still obeys the bar above: a real person could bring it, a competen fail it, and the values are real. A stance chooses *what to look at*, never whether the scenario has to be honest. -Two rules keep this from turning into noise. **Each scenario carries one use case, and no two -scenarios carry the same one** — a duplicate is either the same test twice or one of them is -mislabelled, and it hides a gap while appearing to fill it. And a stance that produces nothing new +Two rules keep this from turning into noise. **Each scenario carries one use case and one branch, and no two scenarios carry the same pair** — a duplicate is either the same test twice or one of them is mislabelled, and it hides a gap while appearing to fill it. Several scenarios sharing a use case is normal and expected; that is what branches are for. What is not allowed is two rows that agree on both. And a stance that produces nothing new for a given agent produces nothing: an agent with no rules to bend does not need an adversarial scenario invented for it. @@ -376,10 +385,12 @@ hides the problem and everything built afterwards inherits it. 1. `inspect_world` with no table, then look at the ones that matter. Read the sub-goals already defined. 2. Read the agent's hard rules. Each one is a branch waiting to be written. -3. For each scenario: work out the solution, `try_calls` it with your `setup_code`, then +3. For a suite, say how you are splitting it and then `generate_suite` with the count. It + writes the whole thing and saves it, and you report what came back. +4. For a single scenario: work out the solution, `try_calls` it with your `setup_code`, then `submit_scenario`. -4. Read what comes back. A refusal names which gate failed and why. -5. `save_scenarios` when you have the number that was asked for. +5. Read what comes back. A refusal names which gate failed and why. +6. `save_scenarios` when you have the number that was asked for. ## Finishing From efdb0253e2c8302cf03665408e427291b24b88ed Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Sat, 22 Aug 2026 12:10:19 +0530 Subject: [PATCH 2/2] feat(harness): write a suite to a plan, review what came back, and cap a large ask --- src/fi/alk/harness/scenario_tools.py | 81 +++- src/fi/alk/harness/scenarios.py | 393 +++++++++++++++--- .../harness/skills/write-scenarios/SKILL.md | 11 + 3 files changed, 413 insertions(+), 72 deletions(-) diff --git a/src/fi/alk/harness/scenario_tools.py b/src/fi/alk/harness/scenario_tools.py index 6bc78c10..fe373417 100644 --- a/src/fi/alk/harness/scenario_tools.py +++ b/src/fi/alk/harness/scenario_tools.py @@ -672,30 +672,79 @@ async def drop_scenario(args: dict[str, Any]) -> dict[str, Any]: @tool( "generate_suite", "Write a whole suite at once by splitting it across the agent's use cases, one writer " - "per use case, several running at the same time. Use this when somebody asks for a " - "suite rather than a particular scenario: writing twenty or fifty one at a time runs " - "out of turns long before it finishes. Everything it produces has cleared the same " - "three gates. Saved when it completes.", - schema({"count": int, "at_once": int}, ["count"]), + "per slice, several running at the same time, then reviewing what came back and " + "filling what it missed. Use this whenever somebody asks for a number of scenarios " + "rather than one in particular: writing twenty or fifty one at a time runs out of " + "turns long before it finishes.\n\n" + "Pass `slices` when you know how the suite should be divided, which you do once you " + "have looked at the world: give each use case a share in proportion to how much can " + "genuinely go wrong in it, and name the angle each slice should take. Without it the " + "work is divided evenly, which pads the thin use cases and under-covers the rich ones. " + "Everything produced clears the same three gates, and the suite is saved.", + schema( + { + "count": int, + "at_once": int, + "slices": { + "type": ["array", "null"], + "description": "How to divide the suite. One entry per writer.", + "items": { + "type": "object", + "properties": { + "use_case": { + "type": "string", + "description": "One of the agent's use cases, worded as the " + "contract words it.", + }, + "angle": { + "type": "string", + "description": "What this slice should look for: the ordinary " + "path, the branch that cannot be completed, the rule under " + "pressure, state that has to carry.", + }, + "count": { + "type": "integer", + "description": "How many scenarios this slice is worth, in " + "proportion to how much can genuinely go wrong in it.", + }, + "why": { + "type": "string", + "description": "Why it earns that share.", + }, + }, + "required": ["use_case", "count"], + }, + }, + }, + ["count"], + ), ) async def generate_suite(args: dict[str, Any]) -> dict[str, Any]: - from .scenarios import write_in_parallel + from .scenarios import MOST_AT_ONCE, MOST_IN_ONE_GO, write_in_parallel - count = int(args.get("count") or 0) - if count < 1: + asked = int(args.get("count") or 0) + if asked < 1: return _err("say how many scenarios the suite should have") - at_once = int(args.get("at_once") or 0) or 4 cases = [one for one in contract.real_use_cases if one.strip()] - if not cases: + given = args.get("slices") or None + if not cases and not given: return _err( "this contract names no use cases, so there is nothing to split the work " "across. Write them one at a time with submit_scenario, or fix the contract." ) + + # A large ask is served a batch at a time. Spinning up a writer per scenario would put + # hundreds of model sessions on one machine, and the person waiting would see nothing + # for an hour. A batch they can read, and an offer of the rest, is the better trade. + count = min(asked, MOST_IN_ONE_GO) + at_once = max(1, min(int(args.get("at_once") or 0) or 4, MOST_AT_ONCE)) + produced = await write_in_parallel( contract, out=destination, wanted=count, use_cases=cases, + slices=given, at_once=at_once, ) # The suite is already on disk. The open session's own list has to be brought level with @@ -703,15 +752,25 @@ async def generate_suite(args: dict[str, Any]) -> dict[str, Any]: # folder the fan-out just produced. kept[:] = produced target["count"] = len(produced) + by_case: dict[str, int] = {} for one in produced: name = one.use_case or "unassigned" by_case[name] = by_case.get(name, 0) + 1 lines = "\n".join(f" {n} x {case[:70]}" for case, n in sorted(by_case.items())) - return _ok( + said = ( f"{len(produced)} scenarios across {len(by_case)} use cases, {at_once} writers at a " f"time. Each cleared all three gates and the suite is saved.\n{lines}" ) + if asked > count: + said += ( + f"\n\n{asked - count} of the {asked} asked for are still to write. " + f"{MOST_IN_ONE_GO} is as many as one pass does, so that the suite can be looked " + "at before more is spent on it. Show what came back, then ask whether to carry " + "on with the rest, change direction first, or stop here. Call generate_suite " + "again for the next batch once they have said." + ) + return _ok(said) @tool( "save_scenarios", diff --git a/src/fi/alk/harness/scenarios.py b/src/fi/alk/harness/scenarios.py index dc32223a..e877ce36 100644 --- a/src/fi/alk/harness/scenarios.py +++ b/src/fi/alk/harness/scenarios.py @@ -11,11 +11,13 @@ from __future__ import annotations import asyncio +import os +from dataclasses import dataclass from collections.abc import Callable from pathlib import Path from typing import Any -from claude_agent_sdk import ClaudeAgentOptions +from claude_agent_sdk import ClaudeAgentOptions, create_sdk_mcp_server, tool from .config import ( UNWANTED, @@ -39,7 +41,7 @@ write_scenarios, ) from .session import Stage -from .tools import qualified +from .tools import qualified, schema SKILL = "write-scenarios" @@ -147,24 +149,107 @@ def load(destination: Path) -> list[Scenario]: return load_scenarios(Path(destination)) -# How many writers run at once. Each is a model session with its own subprocess, and each gate -# restores its own copy of the world, so this is bounded by the machine rather than by the API. +# What a suite costs, and what it is allowed to cost. +# +# Writers run as separate model sessions, so wall clock is roughly the number of scenarios +# divided by how many run at once. The two ceilings below exist for different reasons: one +# protects the machine, the other protects the person waiting. Asking for a thousand scenarios +# is a reasonable thing to want and an unreasonable thing to do in one go, so a large ask is +# served a batch at a time with the rest offered back. AT_ONCE = 4 +MOST_AT_ONCE = int(os.environ.get("HARNESS_WRITERS_AT_ONCE") or 8) +MOST_IN_ONE_GO = int(os.environ.get("HARNESS_SUITE_BATCH") or 50) +# How many times the suite is reviewed and topped up after the first pass. One is enough to +# catch a slice that came back short or a use case nobody covered; more turns it into a loop +# that keeps finding smaller things to say. +TOP_UP_ROUNDS = 1 -def shares(wanted: int, use_cases: list[str]) -> list[tuple[str, int]]: - """How many scenarios each use case is asked to produce. + +@dataclass(frozen=True) +class Slice: + """One writer's share of a suite: what to write, how much, and why it is worth writing.""" + + use_case: str + angle: str = "" + count: int = 1 + why: str = "" + + def named(self) -> str: + return f"{self.use_case} — {self.angle}" if self.angle else self.use_case + + +def even_slices(wanted: int, use_cases: list[str]) -> list[Slice]: + """The fallback split, when nobody said how the work should be divided. Evenly, with the remainder going to the ones named first, because a contract lists its - primary use cases before its marginal ones. A use case that turns out to have less in it - than its share says returns fewer; nothing forces it to pad. + primary use cases before its marginal ones. It is a poor plan and it is meant to be: a use + case with one real branch gets the same share as one with six, so the first pads and the + second under-covers. It exists so a caller that supplies no plan still gets a suite. """ if not use_cases: return [] if wanted <= len(use_cases): - return [(case, 1) for case in use_cases[:wanted]] + return [Slice(use_case=case, count=1) for case in use_cases[:wanted]] each, extra = divmod(wanted, len(use_cases)) - return [(case, each + (1 if i < extra else 0)) for i, case in enumerate(use_cases)] + return [ + Slice(use_case=case, count=each + (1 if i < extra else 0)) + for i, case in enumerate(use_cases) + ] + + +def planned(wanted: int, use_cases: list[str], given: list[dict] | None) -> list[Slice]: + """The split this suite will actually be written to. + + A plan supplied by the caller wins, because whoever is talking to the person has just read + the contract and the world and knows which use cases have something in them. Sizing every + use case identically is the thing that made suites pad in one place and under-cover in + another, and the plan is the only part of the process that knows the difference. + + Anything the plan leaves out is filled in evenly, and anything it over-asks for is trimmed, + so a plan can be rough without producing a suite nobody asked for. + """ + if not given: + return even_slices(wanted, use_cases) + + known = {case.strip().lower(): case for case in use_cases} + slices: list[Slice] = [] + for one in given: + if not isinstance(one, dict): + continue + case = str(one.get("use_case") or "").strip() + if not case: + continue + # Match the contract's own wording where the plan paraphrased it, so a slice is filed + # under a use case the coverage count recognises rather than a near-miss of one. + case = known.get(case.lower(), case) + try: + count = max(1, int(one.get("count") or 1)) + except (TypeError, ValueError): + count = 1 + slices.append( + Slice( + use_case=case, + angle=str(one.get("angle") or "").strip(), + count=count, + why=str(one.get("why") or "").strip(), + ) + ) + if not slices: + return even_slices(wanted, use_cases) + + # Trim from the end rather than scaling everything down: the plan put its most valuable + # slices first, and shaving one scenario off each is how a deliberate plan becomes an even + # one again. + total = sum(one.count for one in slices) + while total > wanted and slices: + last = slices[-1] + if last.count > 1: + slices[-1] = Slice(last.use_case, last.angle, last.count - 1, last.why) + else: + slices.pop() + total = sum(one.count for one in slices) + return slices def callers_for(index: int, wanted: int) -> str: @@ -202,60 +287,84 @@ def callers_for(index: int, wanted: int) -> str: return said -def branch_opening( - contract: AgentContract, use_case: str, wanted: int, callers: str = "" +def brief_for( + contract: AgentContract, mine: Slice, siblings: list[Slice], callers: str ) -> str: + """What one writer is told: its share, what everyone else holds, and the bar. + + Written as a brief rather than a template because a writer that cannot see its siblings + will otherwise write what they are writing. Naming their angles is cheaper than discovering + the overlap at the merge and throwing the loser away. + """ + others = "\n".join(f" - {one.named()}" for one in siblings if one is not mine) + aim = f" {mine.use_case}" + if mine.angle: + aim += f"\n Angle: {mine.angle}" + if mine.why: + aim += f"\n Worth testing because: {mine.why}" + return ( - f"Write {wanted} scenarios for {contract.agent!r}, all of them within this one " - "use case:\n\n" - f" {use_case}\n\n" - "Write nothing outside it. Somebody else is covering the other use cases at the same " - "time, so a scenario that strays is either a duplicate of theirs or a gap in yours.\n\n" - "Every scenario carries this use case verbatim in `use_case`, and its own one-line " + f"Write {mine.count} scenario{'s' if mine.count != 1 else ''} for {contract.agent!r}, " + "all of them within this one slice:\n\n" + f"{aim}\n\n" + + ( + "The rest of the suite is being written at the same time by others, covering:\n" + f"{others}\n\nStay out of theirs. A scenario that strays is either a duplicate of " + "somebody else's or a gap in yours.\n\n" + if others + else "" + ) + + "Every scenario carries this use case verbatim in `use_case`, and its own one-line " "`branch` saying what makes it different from the others you write here. Branches are " "where the variety lives: the ordinary path, the branch that cannot be completed, the " "rule under pressure, state that has to carry across turns, the same request against a " "differently seeded world.\n\n" - "Look at the world first with inspect_world so every scenario names real records, and " - "read the sub-goals already defined. Work out each solution with try_calls before you " - "submit it. Submit each one with submit_scenario and then stop: do not save, and do not " - "ask what to do next. Whoever asked for this collects the suite and writes it.\n\n" - "Vary the caller across the scenarios you write. Everyone else is writing their own use " - "case and cannot see yours, so a suite where every caller is professional and formal is " - "what happens when each writer picks the safest value. Give different scenarios " - "different personalities, communication styles and accents from the values offered, and " - "let the caller suit the situation: somebody whose card was declined is not in the same " - "mood as somebody booking a routine morning ride." + callers + "What each one has to be, before you submit it:\n" + " - every value real, read out of the world with inspect_world, never invented\n" + " - an instruction that is a circumstance the person is living through, not a script " + "of lines to say\n" + " - a setup that makes true whatever the instruction presumes, and a ready check that " + "proves it\n" + " - a solution worked out with try_calls first, so the gates are not where you find " + "out it cannot be passed\n" + " - sub-goals named from the shared catalogue, and checks that assert the right call " + "with the right arguments or the right end state, never that something merely happened\n" + " - a scenario a competent agent could plausibly fail. If any correct implementation " + "passes it for free, it teaches nothing and is not worth the run\n\n" + "Look at the world first, and read the sub-goals already defined. Submit each scenario " + "with submit_scenario and then stop: do not save, and do not ask what to do next. " + "Whoever asked for this collects the suite and writes it." + callers ) -async def _write_one_use_case( +async def _write_slice( contract: AgentContract, - use_case: str, - count: int, + mine: Slice, + siblings: list[Slice], *, - index: int = 0, + index: int, destination: Path, on_event: Callable[..., Any] | None, ask: Callable[..., Any] | None, ) -> list[Scenario]: - """One use case's share, written by its own session. Returns what it proved, unsaved.""" + """One slice, written by its own session. Returns what it proved, unsaved.""" server, kept = scenario_tools( contract, destination, destination, - wanted=count, + wanted=mine.count, can_save=False, start_from=[], ) - progress.started(destination, use_case) + progress.started(destination, mine.named()) def watch(event: Any) -> None: # Report as they land rather than at the end. A slice that proves its first scenario # four minutes in is the difference between a run that looks alive and one that does not. - progress.kept(destination, use_case, len(kept)) + progress.kept(destination, mine.named(), len(kept)) if on_event: on_event(event) + allowed = [ qualified(SCENARIO_SERVER, name) for name in TOOL_NAMES if name != "save_scenarios" ] @@ -263,33 +372,33 @@ def watch(event: Any) -> None: system_prompt=( f"{load_skill(SKILL)}\n\n## This agent\n\n{contract.brief(with_data=True)}" f"\n\n## Its world\n\n{world_summary(destination)}" - f"\n\n## Your slice\n\nYou are writing only the scenarios for: {use_case}" + f"\n\n## Your slice\n\nYou are writing only: {mine.named()}" ), allowed_tools=allowed, mcp_servers={SCENARIO_SERVER: server}, permission_mode="default", cwd=str(destination.parent if destination.parent.exists() else Path.cwd()), setting_sources=[], - max_turns=turns_for(count), + max_turns=turns_for(mine.count), model=chosen_model(), env=provider_env(), ) options.disallowed_tools = list(UNWANTED) options.hooks = gate_hooks(allowed) options.can_use_tool = permission_gate(ask, allowed) - stage = Stage(options, name=f"{SKILL}:{use_case[:40]}") + stage = Stage(options, name=f"{SKILL}:{mine.named()[:40]}") try: async with stage: await stage.say( - branch_opening(contract, use_case, count, callers_for(index, count)), + brief_for(contract, mine, siblings, callers_for(index, mine.count)), on_event=watch, ) except Exception as broke: # noqa: BLE001 - one slice failing must not lose the others - progress.failed(destination, use_case, str(broke)) + progress.failed(destination, mine.named(), str(broke)) if on_event: - on_event({"type": "slice_failed", "use_case": use_case, "why": str(broke)[:300]}) + on_event({"type": "slice_failed", "slice": mine.named(), "why": str(broke)[:300]}) return list(kept) - progress.finished(destination, use_case, len(kept)) + progress.finished(destination, mine.named(), len(kept)) return list(kept) @@ -315,22 +424,140 @@ def merged(written: list[list[Scenario]]) -> list[Scenario]: return suite +def _suite_summary(suite: list[Scenario]) -> str: + """The whole suite as a reviewer needs to see it: what each row claims to test.""" + return "\n".join( + f" {one.name} | use case: {one.use_case} | branch: {one.branch} | tests: {one.tests}" + for one in suite + ) + + +async def gaps_in( + contract: AgentContract, + suite: list[Scenario], + *, + destination: Path, + wanted: int, + ask: Callable[..., Any] | None = None, +) -> list[Slice]: + """What the finished suite is missing, as slices that would fill it. + + Nobody looks at a suite written in parallel. Each writer sees its own slice and the merge + only removes collisions, so a use case that came back one short, or an obvious branch that + every writer assumed somebody else had, survives to the end and nobody notices. This is the + one pass that reads the suite as a whole. + """ + if not suite: + return [] + found: list[Slice] = [] + + @tool( + "submit_gaps", + "The gaps worth filling in this suite, as the slices that would fill them. Return " + "nothing when the suite covers what it should: a suite that is finished is a real " + "answer, and inventing work to report is worse than saying so.", + schema( + { + "gaps": { + "type": "array", + "description": "One entry per gap. Empty when the suite is covering what " + "it should.", + "items": { + "type": "object", + "properties": { + "use_case": {"type": "string"}, + "angle": { + "type": "string", + "description": "The scenario that is missing, in one line.", + }, + "why": {"type": "string"}, + }, + "required": ["use_case", "angle"], + }, + } + }, + ["gaps"], + ), + ) + async def submit_gaps(args: dict[str, Any]) -> dict[str, Any]: + for one in args.get("gaps") or []: + if not isinstance(one, dict): + continue + case = str(one.get("use_case") or "").strip() + if case: + found.append( + Slice( + use_case=case, + angle=str(one.get("angle") or "").strip(), + count=1, + why=str(one.get("why") or "").strip(), + ) + ) + return { + "content": [ + {"type": "text", "text": f"{len(found)} gap(s) recorded. Nothing else to do."} + ] + } + + server = create_sdk_mcp_server(name=REVIEW_SERVER, version="0.1.0", tools=[submit_gaps]) + allowed = [qualified(REVIEW_SERVER, "submit_gaps")] + options = ClaudeAgentOptions( + system_prompt=( + "You are reviewing a suite of tests somebody else wrote for an AI agent, in " + "parallel, each writer blind to the others. Your only job is to say what is " + "missing.\n\n" + "Look for: a use case of this agent that nothing covers; a use case covered only " + "on its ordinary path, where the branch that cannot be completed or the rule under " + "pressure is the interesting one; two rows that are the same test under different " + "names, leaving the branch one of them claimed uncovered.\n\n" + "Judge coverage of the agent, not of the plan. Do not ask for more of what is " + "already well covered, and do not report a gap you cannot name a scenario for. " + "A suite of the right size that covers what matters is finished, and saying so is " + f"the useful answer.\n\n## This agent\n\n{contract.brief()}" + ), + allowed_tools=allowed, + mcp_servers={REVIEW_SERVER: server}, + permission_mode="default", + cwd=str(destination.parent if destination.parent.exists() else Path.cwd()), + setting_sources=[], + max_turns=8, + model=chosen_model(), + env=provider_env(), + ) + options.disallowed_tools = list(UNWANTED) + options.hooks = gate_hooks(allowed) + options.can_use_tool = permission_gate(ask, allowed) + stage = Stage(options, name=f"{SKILL}:review") + try: + async with stage: + await stage.say( + f"This suite has {len(suite)} scenarios against a target of {wanted}:\n\n" + f"{_suite_summary(suite)}\n\n" + "Say what it is missing, then submit_gaps. Submit an empty list if it is " + "covering what it should." + ) + except Exception: # noqa: BLE001 - a review that fails leaves the suite as written + return [] + return found + + async def write_in_parallel( contract: AgentContract, *, out: Path | None = None, wanted: int = 10, use_cases: list[str] | None = None, + slices: list[dict] | None = None, at_once: int = AT_ONCE, + rounds: int = TOP_UP_ROUNDS, on_event: Callable[..., Any] | None = None, ask: Callable[..., Any] | None = None, ) -> list[Scenario]: - """Write a suite with one session per use case, then save it once. + """Write a suite with one session per slice, review it, fill what it missed, and save once. Sequentially, a suite costs roughly three turns a scenario against one budget, which is why - asking for forty stopped around twenty-five. Here each use case is written by its own - session, so the wall clock is the slowest use case rather than the sum of all of them, and - the turn budget is per slice rather than shared. + asking for forty stopped around twenty-five. Here the work is split into slices that run at + the same time, so the wall clock is the slowest slice rather than the sum of all of them. Saving stays here, once, for a reason: ``save_scenarios`` regenerates the index and deletes any folder it does not know about, so letting the writers save would have each of them @@ -338,23 +565,35 @@ async def write_in_parallel( """ destination = out or artifact_dir(contract.agent) cases = [case for case in (use_cases or contract.real_use_cases) if case.strip()] - if not cases: + if not cases and not slices: # Nothing to partition on. One writer, the ordinary path, rather than no scenarios. return await write(contract, out=destination, wanted=wanted, on_event=on_event, ask=ask) - allocation = shares(wanted, cases) - progress.planned(destination, allocation, at_once=at_once, asked=wanted) + at_once = max(1, min(at_once or AT_ONCE, MOST_AT_ONCE)) + allocation = planned(wanted, cases, slices) + progress.planned( + destination, + [(one.named(), one.count) for one in allocation], + at_once=at_once, + asked=wanted, + ) if on_event: - on_event({"type": "planned", "slices": allocation, "at_once": at_once}) + on_event( + { + "type": "planned", + "slices": [(one.named(), one.count) for one in allocation], + "at_once": at_once, + } + ) - limit = asyncio.Semaphore(max(1, at_once)) + limit = asyncio.Semaphore(at_once) - async def guarded(use_case: str, count: int, index: int) -> list[Scenario]: + async def guarded(mine: Slice, siblings: list[Slice], index: int) -> list[Scenario]: async with limit: - return await _write_one_use_case( + return await _write_slice( contract, - use_case, - count, + mine, + siblings, index=index, destination=destination, on_event=on_event, @@ -362,14 +601,46 @@ async def guarded(use_case: str, count: int, index: int) -> list[Scenario]: ) written = await asyncio.gather( - *( - guarded(case, count, index) - for index, (case, count) in enumerate(allocation) - ), + *(guarded(one, allocation, index) for index, one in enumerate(allocation)), return_exceptions=False, ) - suite = merged([load_scenarios(destination), *written]) + + # Read the whole thing and fill what nobody covered. Bounded, because a reviewer asked + # twice will always find something smaller to say. + for _ in range(max(0, rounds)): + if len(suite) >= wanted: + break + missing = await gaps_in( + contract, suite, destination=destination, wanted=wanted, ask=ask + ) + missing = missing[: max(0, wanted - len(suite))] + if not missing: + break + if on_event: + on_event({"type": "topping_up", "slices": [one.named() for one in missing]}) + progress.planned( + destination, + [(one.named(), one.count) for one in [*allocation, *missing]], + at_once=at_once, + asked=wanted, + ) + for one in [*allocation, *missing]: + if one in allocation: + progress.finished(destination, one.named(), one.count) + more = await asyncio.gather( + *( + guarded(one, missing, len(allocation) + index) + for index, one in enumerate(missing) + ), + return_exceptions=False, + ) + before = len(suite) + suite = merged([suite, *more]) + allocation = [*allocation, *missing] + if len(suite) == before: + break + write_scenarios(suite, destination, load_catalogue(destination)) progress.settled(destination, kept_total=len(suite)) if on_event: diff --git a/src/fi/alk/harness/skills/write-scenarios/SKILL.md b/src/fi/alk/harness/skills/write-scenarios/SKILL.md index 474f8ce7..67405d97 100644 --- a/src/fi/alk/harness/skills/write-scenarios/SKILL.md +++ b/src/fi/alk/harness/skills/write-scenarios/SKILL.md @@ -165,6 +165,17 @@ through the same three gates. Writing a suite yourself with `submit_scenario` co turns per scenario against one budget, so a request for twenty or fifty runs out long before it finishes, and what does get written is lost because nothing was saved. +**Pass your plan to it.** The tool takes the split as an argument, and you have just read the +world and know which use cases have +something in them; it is the part of this only you can do. Each slice names its use case, +the angle it should take, how many scenarios it is worth, and why. Left to itself the work is +divided evenly, which is how a use case with one real branch pads to three and one with six gets +three. + +A large request comes back a batch at a time rather than all at once, with the rest offered. When +that happens, show what came back and ask whether to carry on, change direction first, or stop. +Do not silently loop until the number is reached. + Use `submit_scenario` for what it is good at: one scenario somebody asked for by name, a replacement for one that came back wrong, or filling a specific gap in a suite that already exists. Anything described as a number of scenarios is a suite.