diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py index 661190ef4..f6cd5045a 100644 --- a/src/agent_env/task/episode/rules.py +++ b/src/agent_env/task/episode/rules.py @@ -158,7 +158,7 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s "browser_release_context", "browser_set_cookies", "browser_sign_in", "browser_set_local_storage", "build_mcp_cli", "combine_universe_verdicts", "configure_scheduled_checks", "deploy_env", "deploy_sandbox", "end_browser_recording", "end_computer_recording", "start_browser_recording", "start_computer_recording", "env_outcome_verifier", - "modify_env_tool_access", "reset_env", "review", + "modify_env_tool_access", "reset_env", "review", "user_sim_judge", "run_container_unit_tests_verifier", "run_docker_container", "set_env_clock_rate", "snapshot_env", "snapshot_computer_env", "sync_env_clock", "validation_gate_aggregator", "verify_env_card", "verify_env_core_protocol", "verify_mcp_tool_schema", diff --git a/src/agent_env/task_step/registry.py b/src/agent_env/task_step/registry.py index 94df3ede2..826c94255 100644 --- a/src/agent_env/task_step/registry.py +++ b/src/agent_env/task_step/registry.py @@ -73,6 +73,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: from agent_env.task_step.task_steps.configure_scheduled_checks import ConfigureScheduledChecksTaskStep from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep + from agent_env.task_step.task_steps.verifiers.user_sim_judge import UserSimJudgeTaskStep from agent_env.task_step.task_steps.verifiers.always_on_agent_verifier import AlwaysOnAgentVerifier from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import RunContainerUnitTestsVerifierTaskStep from agent_env.task_step.task_steps.run_docker_container import RunDockerContainerTaskStep @@ -148,6 +149,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: _get_type(ConfigureScheduledChecksTaskStep): ConfigureScheduledChecksTaskStep, _get_type(InstallAgentTaskStep): InstallAgentTaskStep, _get_type(RubricsVerifierTaskStep): RubricsVerifierTaskStep, + _get_type(UserSimJudgeTaskStep): UserSimJudgeTaskStep, _get_type(AlwaysOnAgentVerifier): AlwaysOnAgentVerifier, _get_type(RunContainerUnitTestsVerifierTaskStep): RunContainerUnitTestsVerifierTaskStep, _get_type(RunDockerContainerTaskStep): RunDockerContainerTaskStep, diff --git a/src/agent_env/task_step/task_steps/agent/deploy_user_sim_agent.py b/src/agent_env/task_step/task_steps/agent/deploy_user_sim_agent.py index 3224d2fe6..0d6b84f29 100644 --- a/src/agent_env/task_step/task_steps/agent/deploy_user_sim_agent.py +++ b/src/agent_env/task_step/task_steps/agent/deploy_user_sim_agent.py @@ -33,6 +33,8 @@ - ``prompt_set``: the system prompt and model are on the agent, so always_on sets neither; ``multi_day``: the done line the prompt was written with, which always_on refuses to run against a run of the other kind. - ``model``, ``model_source``, ``prompt_sha256``, ``user_version``, ``setup``: for the run's record. +- ``persona_sha256``: ``user_sim_common.persona_fingerprint`` of the persona and ``opening_prompt`` as written, which + user_sim_judge compares a rubric's ``rubric_persona_sha256`` with. - ``tagged_lines``: each tag the system prompt shows with its line as written there (``{"W1": "...", "D9": "..."}``), so an answer's ``used`` resolves against the lines this run's user sim saw, whatever later edits to the persona move. @@ -102,7 +104,8 @@ from agent_env.entity.scopes import FieldScope from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps.agent import user_sim_versions -from agent_env.task_step.task_steps.agent.user_sim_common import attached, opening_parts, opening_problems, opening_text +from agent_env.task_step.task_steps.agent.user_sim_common import (attached, opening_parts, opening_problems, opening_text, + persona_fingerprint, recorded_opening) from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep logger = logging.getLogger(__name__) @@ -188,6 +191,31 @@ def persona_problems(persona: Any) -> list[str]: return out +def person(key: str, persona: dict, opening: list[dict], seed: dict, multi_day: Optional[bool]) -> dict: + """The user as the built-in user sim's ``setup`` takes them: ``persona``'s fields, the ``opening`` parts with the + run's ```` placeholders filled in (as always_on fills them), and the done line ``multi_day`` picks. The step + writes its prompt from this, and user_sim_judge rebuilds that prompt from the step's record (``recorded_person``).""" + from agent_env.task_step.task_steps.prompt_agent import apply_seed + + name = persona["display_name"] + parts = apply_seed(opening, seed) + text, files = opening_text(parts), attached(parts) + return {"key": key, "name": name, "first": persona.get("first_name") or name.split()[0], + "description": persona.get("description"), "permissions": persona.get("permissions"), + "keep_private": persona.get("keep_private"), + "opening": {"text": text, **({"files": files} if files else {})} if text else None, + "done_ends_run": not multi_day} + + +def recorded_person(us: dict, seed: dict, multi_day: Optional[bool] = None) -> dict: + """``person`` from the step's ``metadata["user_sims"]`` record, with its ``opening_parts`` when it has them, so + ``setup`` gives the prompt the step wrote; ``multi_day`` is the record's, else the one given.""" + persona = {**{k: us.get(k) for k in PERSONA_KEYS}, + "display_name": us.get("display_name") or us.get("name") or "the user"} + return person(us.get("name"), persona, recorded_opening(us), seed, + us["multi_day"] if us.get("multi_day") is not None else multi_day) + + class DeployUserSimAgentTaskStep(DeployAgentTaskStep): """Deploys the agent that plays the task's user, its persona set at deploy, and records the user under ``agent_name`` in ``metadata["user_sims"]``.""" @@ -250,20 +278,8 @@ def _resolve_agent_ref(self, context: TaskStepContext) -> tuple[str, Optional[in def person(self, context: TaskStepContext) -> dict: """The user as the built-in user sim's ``setup`` takes them, as always_on hands them over.""" - name = self.persona["display_name"] - parts = self._opening(context) - text, files = opening_text(parts), attached(parts) - return {"key": self.agent_name, "name": name, "first": self.persona.get("first_name") or name.split()[0], - "description": self.persona.get("description"), "permissions": self.persona.get("permissions"), - "keep_private": self.persona.get("keep_private"), - "opening": {"text": text, **({"files": files} if files else {})} if text else None, - "done_ends_run": not self.multi_day} - - def _opening(self, context: TaskStepContext) -> list[dict]: - """``opening_prompt`` as parts with the run's ```` placeholders filled in, as always_on fills them.""" - from agent_env.task_step.task_steps.prompt_agent import apply_seed - - return apply_seed(opening_parts(self.opening_prompt), context.metadata.get("seed") or {}) + return person(self.agent_name, self.persona, opening_parts(self.opening_prompt), + context.metadata.get("seed") or {}, self.multi_day) def _setup(self, context: TaskStepContext) -> dict: return user_sim_versions.setup(user_sim_versions.VERSION, self.person(context)) @@ -319,6 +335,8 @@ async def _set_user(self, context: TaskStepContext, agent, setup: dict) -> TaskS **({"opening_parts": opening_parts(self.opening_prompt)} if isinstance(self.opening_prompt, list) else {}), "user_version": user_sim_versions.VERSION, "model": model, "model_source": source, "prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest(), "prompt_set": True, + # Who the user is, as written: what user_sim_judge's rubric_persona_sha256 is compared with. + "persona_sha256": persona_fingerprint(persona, self.opening_prompt), # Each tag the prompt shows with its line, so an answer's ``used`` ("D9") names what this run saw. "tagged_lines": user_sim_versions.tagged_lines(self.person(context)), "multi_day": self.multi_day, diff --git a/src/agent_env/task_step/task_steps/agent/user_sim_common.py b/src/agent_env/task_step/task_steps/agent/user_sim_common.py index 7cd09efa2..28581486d 100644 --- a/src/agent_env/task_step/task_steps/agent/user_sim_common.py +++ b/src/agent_env/task_step/task_steps/agent/user_sim_common.py @@ -2,6 +2,8 @@ from __future__ import annotations import copy +import hashlib +import json from datetime import datetime, timezone from typing import Any, Optional @@ -20,6 +22,22 @@ def person_of(us: dict) -> dict: "permissions": us.get("permissions"), "keep_private": us.get("keep_private")} +def persona_fingerprint(persona: dict, opening_prompt: Any) -> str: + """Which user sim a run had: the hex sha256 of who the user is and what they first sent, as written (the seed not + filled in). deploy_user_sim_agent records it as ``persona_sha256``, and user_sim_judge grades a rubric written for + a fingerprint only on a run that has it. psi-benchmark-tasks's converter computes it too, so the recipe is fixed:: + + sha256(json.dumps({"description": persona.get("description"), "permissions": persona.get("permissions"), + "keep_private": persona.get("keep_private"), "opening_prompt": opening_prompt}, + sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")).hexdigest() + + ``opening_prompt`` is the step's as written: its text, or its list of parts.""" + body = json.dumps({"description": persona.get("description"), "permissions": persona.get("permissions"), + "keep_private": persona.get("keep_private"), "opening_prompt": opening_prompt}, + sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return hashlib.sha256(body.encode("utf-8")).hexdigest() + + def task_zone(us: dict, clocks: list[dict], where: str) -> str: """The user's time zone, else the one sync_env_clock gave the world; refused when the two disagree. ``where`` names the step in the error.""" diff --git a/src/agent_env/task_step/task_steps/verifiers/user_sim_judge.py b/src/agent_env/task_step/task_steps/verifiers/user_sim_judge.py new file mode 100644 index 000000000..e2ca71be2 --- /dev/null +++ b/src/agent_env/task_step/task_steps/verifiers/user_sim_judge.py @@ -0,0 +1,505 @@ +"""The user_sim_judge step: after a run with the built-in user sim, grade the user sim (the AI playing the user), not +the assistant. A task writer adds it after the run's step; nothing turns it on by default. + +It judges either kind of run the built-in user sim plays (``deploy_user_sim_agent``): + +- an ``always_on`` run: every ``user_sim_read`` in the timeline stored at ``metadata["always_on"][step]["object_url"]`` + (else the record's ``timeline_tail``, which keeps answers but not what the user sim was sent); +- a ``prompt_agent`` chat run: each short entry in ``metadata["user_sim_reads"][step]``, read whole from its + ``object_url`` (else the entry, which keeps the answer only). + +``run_step_id`` names that step; unset, the run must have exactly one of them. The user sim's instructions are rebuilt +from its ``metadata["user_sims"]`` record by ``deploy_user_sim_agent``'s own path (``recorded_person``, then +``user_sim_versions.setup``), its opening's files included, and checked against the record's ``prompt_sha256``. ``used`` is never shown to the judge: it is the user sim's +own pulse check, not evidence. + +Each check (``user_sim_judge_core.CHECKS``) is one narrow question to ``judge_model``, in its own call; a task's +``rubric_items`` ride along with the check that owns their family, at most ``MAX_ITEMS_PER_CALL`` to a call. + +A rubric item is the shape of a criterion in a task's ``usersim.yaml`` (psi-benchmark-tasks), which that repo turns +into this step's settings; this step reads no file: + +- ``id`` and ``family`` (A1-A5, F1-F4, G1-G5), both required; ids are unique. +- ``kind``: ``hard`` (an exact rule from the user sim's instructions: the judge fails it on any breach), ``general`` + (a judgement: it fails only on a clear instance; the default) or ``auto`` (never sent to the model: ``metric`` + ``max_words`` or ``max_sentences`` is counted on each graded answer, and any answer over ``max`` fails it, naming + the turn; see ``user_sim_judge_core.count_sentences``). If none is over but the record kept only the start of one + (always_on's tail, when its stored timeline can't be read), the item is ``error``: that length can't be counted. +- ``check``: the text the judge applies, required unless auto. ``text`` is its old name and still loads. +- ``name`` and ``short``: a title and one line for reviewers, recorded but not shown to the judge. +- ``na_when``: the situation in which it never came up, told to the judge. ``always`` makes it n/a with no call, + recording ``static_na_reason`` (required with it); ``never`` means it always applies, and an n/a is asked again. + An auto item is counted, not judged, so it takes only ``always``; any other ``na_when`` is refused when the task loads. +- ``can_error``: whether its fail makes the run a user sim error; unset, as the check that owns its family. + +For example:: + + rubric_persona_sha256: 3f1c... # the run's persona_sha256, all 64 hex digits + rubric_items: + - {id: a1_call_rae, family: A1, kind: hard, name: "Rae: yes if it fits", + check: "If the assistant asks about Rae's order, Tess says yes only if it fits, gives both floors ...", + na_when: "The assistant didn't raise Rae's order."} + - {id: f2_her_length, family: F2, kind: auto, check: "Every reply is at most 25 words.", metric: max_words, max: 25} + +``rubric_persona_sha256`` is the fingerprint of the user sim the items were written for: +``user_sim_common.persona_fingerprint`` of its persona and ``opening_prompt``, which deploy_user_sim_agent records as +the run's ``persona_sha256``. When it's set and differs from the run's, the task items aren't graded and the entry's +``rubric_persona`` (``status``, ``rubric``, ``run``, ``note``) says so; the general checks still run. A run that +recorded no ``persona_sha256`` is ``unknown``, and the items are graded. Edits to the task that leave the persona and +the opening as they were keep the items graded. + +What it writes: ``metadata["user_sim_judge"][verifier_id]``, with every verdict, its reason and cited turns, the task +items, ``user_sim_error`` (``value`` yes / no / unknown, ``because``, ``marked``), the instructions check, token use +and cost. It writes no ``verifications`` entry, so its verdicts never mix into the assistant's scores. + +Record only, by default: with ``mark_user_sim_error`` off it changes nothing a grader reads. Turned on, a run whose +``user_sim_error`` is yes, judged against instructions that match the recorded hash, has its prompt response marked +``error_type = "user_sim_error"``, as a user sim that fails mid-run marks it, so graders that run after this step skip +it. A failure of this step doesn't fail the task unless ``fail_task_on_error`` says so. + +A run that ended in an error (the always_on record's or the prompt response's ``error_type``) is not judged: every +verifier skips its prompt anyway. +""" +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +import re +import uuid +from typing import Any, ClassVar, Optional + +from agent_env.task_step.context import PromptResponse, TaskStepContext +from agent_env.task_step.task_step import TaskStep, TaskStepDependency +from agent_env.task_step.task_steps.verifiers import user_sim_judge_core as core + +logger = logging.getLogger(__name__) + +JUDGE_KEY = "user_sim_judge" +# Not imported: always_on and prompt_agent import the verifiers' peers. Each is the key its module names. +ALWAYS_ON_KEY = "always_on" # always_on.ALWAYS_ON_KEY +READS_KEY = "user_sim_reads" # prompt_agent's _BuiltInUserSim +USER_SIMS_KEY = "user_sims" # deploy_user_sim_agent.USER_SIMS_KEY +ALWAYS_ON, CHAT = "always_on", "prompt_agent" +RUN_STEP_TYPES = (ALWAYS_ON, CHAT) +FALLBACK_JUDGE_MODEL = "claude-sonnet-4-6" # rubrics_verifier's, when the config names no judge +USER_SIM_ERROR = "user_sim_error" +FLAG_ASSISTANT = "this answer looks like the assistant's (very long, or markdown)" +FLAG_LAST = "after the run's last wake: this answer never reached the assistant" +# always_on.ASSISTANT_LIKE_MESSAGE_CHARS: a chat run's reads carry no flag of their own, so it's computed the same way. +ASSISTANT_LIKE_MESSAGE_CHARS = 1200 +SCHEMA_ERRORS_SHOWN_CHARS = 300 # of a malformed answer's schema errors, in the judge's turn note +USER_SIM_ERROR_SHOWN_CHARS = 200 # of a user sim agent error, in the judge's turn note + + +def default_judge_model() -> str: + """The model rubrics_verifier defaults to: the config's ``judge`` role, else claude-sonnet-4-6.""" + from agent_env.config import get_config + + return get_config().get_model_for_role("judge") or FALLBACK_JUDGE_MODEL + + +class UserSimJudgeTaskStep(TaskStep): + type: ClassVar[str] = "user_sim_judge" + entity_refs = () + DEFAULT_TIMEOUT_SECONDS: ClassVar[int] = 600 + DEFAULT_MAX_RETRIES: ClassVar[int] = 3 + + def __init__( + self, + id: str, + version: Optional[int], + run_step_id: Optional[str] = None, + judge_model: Optional[str] = None, + checks: Optional[list[str]] = None, + error_checks: Optional[list[str]] = None, + rubric_items: Optional[list[dict]] = None, + rubric_persona_sha256: Optional[str] = None, + mark_user_sim_error: bool = False, + verifier_id: Optional[str] = None, + judge_timeout_seconds: Optional[int] = None, + depends_on: Optional[list[TaskStepDependency]] = None, + fail_task_on_error: bool = False, + ): + super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + self.run_step_id = run_step_id + self.judge_model = judge_model # None: default_judge_model() when the step runs + self.checks = list(checks) if checks is not None else list(core.CHECK_NAMES) + self.error_checks = (list(error_checks) if error_checks is not None + else [c for c in core.DEFAULT_ERROR_CHECKS if c in self.checks]) + self.rubric_items = list(rubric_items or []) + self.rubric_persona_sha256 = rubric_persona_sha256 + self.mark_user_sim_error = mark_user_sim_error + self.verifier_id = verifier_id or uuid.uuid4().hex + self.judge_timeout_seconds = judge_timeout_seconds or self.DEFAULT_TIMEOUT_SECONDS + + # ---- settings ---------------------------------------------------------------------------------------------- + def config_problems(self) -> list[str]: + return _problems(self.judge_model, self.checks, self.error_checks, self.rubric_items, self.mark_user_sim_error, + self.rubric_persona_sha256) + + def task_problems(self, steps: list[TaskStep]) -> list[str]: + """The run it judges must be an always_on or prompt_agent step's; unset, the task must have exactly one.""" + runs = [s for s in steps if s.type in RUN_STEP_TYPES] + if self.run_step_id is not None: + named = next((s for s in steps if s.id == self.run_step_id), None) + if named is not None and named.type not in RUN_STEP_TYPES: + return [f"step '{self.id}' ({self.type}) names the user sim of step '{named.id}' ({named.type}); " + "it judges an always_on or prompt_agent step's"] + return [] + if len(runs) > 1: + return [f"step '{self.id}' ({self.type}): set run_step_id; this task has {len(runs)} steps it could judge " + f"({sorted(s.id for s in runs)})"] + return [] + + def to_dict(self) -> dict: + base = super().to_dict() + base.update(run_step_id=self.run_step_id, judge_model=self.judge_model, checks=self.checks, + error_checks=self.error_checks, rubric_items=self.rubric_items, + rubric_persona_sha256=self.rubric_persona_sha256, mark_user_sim_error=self.mark_user_sim_error, + verifier_id=self.verifier_id, + judge_timeout_seconds=self.judge_timeout_seconds) + return base + + @classmethod + def from_dict(cls, data: dict) -> "UserSimJudgeTaskStep": + base = {**cls._base_from_dict(data), "fail_task_on_error": data.get("fail_task_on_error", False)} + return cls(**base, run_step_id=data.get("run_step_id"), judge_model=data.get("judge_model"), + checks=data.get("checks"), error_checks=data.get("error_checks"), + rubric_items=data.get("rubric_items"), rubric_persona_sha256=data.get("rubric_persona_sha256"), + mark_user_sim_error=data.get("mark_user_sim_error", False), + verifier_id=data.get("verifier_id"), judge_timeout_seconds=data.get("judge_timeout_seconds")) + + def _settings(self, context: TaskStepContext) -> dict: + """This run's settings: a run's ``step_params`` for this step override the stored ones.""" + over = self.step_param_overrides(context) + model = over.get("judge_model", self.judge_model) + checks = list(over.get("checks", self.checks)) + error_checks = list(over["error_checks"]) if "error_checks" in over else ( + [c for c in self.error_checks if c in checks] if "checks" in over else list(self.error_checks)) + items = over.get("rubric_items", self.rubric_items) + persona = over.get("rubric_persona_sha256", self.rubric_persona_sha256) + mark = over.get("mark_user_sim_error", self.mark_user_sim_error) + problems = _problems(model, checks, error_checks, items, mark, persona) + if problems: + raise ValueError(f"user_sim_judge '{self.id}': " + "; ".join(problems)) + return {"model": model or default_judge_model(), "checks": checks, "error_checks": error_checks, + "mark": mark, "items": [core.RubricItem.from_dict(i) for i in items], "rubric_persona_sha256": persona} + + # ---- the run ----------------------------------------------------------------------------------------------- + async def execute(self, context: TaskStepContext) -> TaskStepContext: + settings = self._settings(context) + kind, step_id = self._find_run(context) + response = _prompt_response(context, step_id) + entry: dict[str, Any] = {"run_step_id": step_id, "run_kind": kind, "judge_model": settings["model"], + "checks_run": settings["checks"], "error_checks": settings["error_checks"], + "rubric_items": [i.to_dict() for i in settings["items"]], + "mark_user_sim_error": settings["mark"]} + items = settings["items"] + record = (context.metadata.get(ALWAYS_ON_KEY) or {}).get(step_id) if kind == ALWAYS_ON else None + error_type = (record or {}).get("error_type") or (response.error_type if response else None) + if error_type: + entry.update(skipped=f"the run ended with error_type={error_type}") + elif kind == CHAT and not (context.metadata.get(READS_KEY) or {}).get(step_id): + entry.update(skipped="the user sim never read: the run has no user_sim_reads") + else: + run, about = await asyncio.to_thread(_read_run, kind, step_id, record, context) + entry["user_sim"], entry["instructions"] = about["user_sim"], about["instructions"] + if settings["rubric_persona_sha256"] is not None and items: + us = (context.metadata.get(USER_SIMS_KEY) or {}).get(about["user_sim"]) or {} + entry["rubric_persona"] = _rubric_persona(settings["rubric_persona_sha256"], us.get("persona_sha256")) + if entry["rubric_persona"]["status"] == "mismatch": + items = [] + context.record_model("user sim judge", self.id, settings["model"]) + result = await core.judge_run( + run, model=settings["model"], checks=settings["checks"], items=items, + error_checks=settings["error_checks"], complete=self._complete(context)) + result.pop("model", None) # the entry's judge_model + entry.update(result) + if "skipped" in entry: + logger.warning(f"user_sim_judge '{self.id}': skipped, {entry['skipped']}") + entry["user_sim_error"] = dict(core.NOT_JUDGED) + entry["user_sim_error"].update(self._mark(entry, settings["mark"], response)) + context.metadata.setdefault(JUDGE_KEY, {})[self.verifier_id] = entry + error = entry["user_sim_error"] + logger.info(f"user_sim_judge '{self.id}': {kind} '{step_id}' user_sim_error={error['value']} " + f"({', '.join(error['because'] + error['unknown']) or 'no check'}), marked={error['marked']}; " + + ", ".join(f"{k}={v['verdict']}" for k, v in (entry.get("checks") or {}).items())) + return context + + def _find_run(self, context: TaskStepContext) -> tuple[str, str]: + """The kind of run and its step id: ``run_step_id``'s, else the run's only always_on or chat run.""" + records = context.metadata.get(ALWAYS_ON_KEY) or {} + reads = context.metadata.get(READS_KEY) or {} + if self.run_step_id: + if self.run_step_id in records: + return ALWAYS_ON, self.run_step_id + if self.run_step_id in reads or _prompt_response(context, self.run_step_id) is not None: + return CHAT, self.run_step_id # a chat run whose user sim never read leaves a response, no reads + raise RuntimeError(f"user_sim_judge '{self.id}': step '{self.run_step_id}' left no run to judge (always_on " + f"records: {sorted(records)}; prompt_agent user sim reads: {sorted(reads)})") + runs = [(ALWAYS_ON, s) for s in records] + [(CHAT, s) for s in reads if s not in records] + if len(runs) != 1: + raise RuntimeError(f"user_sim_judge '{self.id}': set run_step_id; this run has {len(runs)} runs with a " + f"built-in user sim ({sorted(s for _, s in runs)})") + return runs[0] + + def _mark(self, entry: dict, mark: bool, response: Optional[PromptResponse]) -> dict: + """With ``mark`` on, a user sim error marks the run's prompt response, so graders after this step skip it.""" + error = entry["user_sim_error"] + if error["value"] != "yes": + return {"marked": False} + if not mark: + return {"marked": False, "not_marked": "mark_user_sim_error is off: verdicts are recorded only"} + if not (entry.get("instructions") or {}).get("matches"): + return {"marked": False, "not_marked": "the judge's copy of the user sim's instructions isn't confirmed " + "to be the one it ran with"} + if response is None: + return {"marked": False, "not_marked": f"step '{entry['run_step_id']}' left no prompt response"} + response.error_type = USER_SIM_ERROR + response.error_message = response.error_message or ( + "The user sim judge found a user sim error: " + ", ".join(error["because"])) + return {"marked": True} + + def _complete(self, context: TaskStepContext) -> core.Complete: + """One judge call through LiteLLM, with the key and base the run's overrides name (as rubrics_verifier's + judge), the run's two shared parts marked for Anthropic's prompt cache.""" + overrides = context.metadata.get("user_overrides", {}) + timeout, retries = self.judge_timeout_seconds, self.DEFAULT_MAX_RETRIES + + async def complete(model: str, system: str, parts: list[str], schema: dict) -> dict: + import litellm + + from agent_env.config import get_config + from agent_env.config.model import ModelParam + + config = get_config() + call = config.resolve_model_call( + model, + default_api_key=config.resolve_reference(overrides.get("judge_litellm_api_key") + or overrides.get("litellm_api_key")), + base_override=overrides.get("judge_litellm_base_url") or overrides.get("litellm_base_url")) + cache = _anthropic(model) + content = [{"type": "text", "text": p, **({"cache_control": {"type": "ephemeral"}} if cache and i < 2 + else {})} + for i, p in enumerate(parts) if p] + kwargs = dict(model=model, messages=[{"role": "system", "content": system}, + {"role": "user", "content": content}], + timeout=timeout, **call.client_kwargs()) + kwargs[ModelParam.RESPONSE_FORMAT] = {"type": "json_schema", "json_schema": { + "name": "user_sim_check", "schema": schema, "strict": True}} + last: Optional[Exception] = None + for attempt in range(1, retries + 1): + try: + response = await litellm.acompletion(**kwargs) + usage = getattr(response, "usage", None) + hidden = getattr(response, "_hidden_params", None) or {} + return {"text": response.choices[0].message.content or "", + "usage": usage.model_dump() if hasattr(usage, "model_dump") else dict(usage or {}), + "cost": hidden.get("response_cost")} + except Exception as e: + last = e + logger.warning(f"user_sim_judge: {model} call {attempt}/{retries} failed: {type(e).__name__}: {e}") + raise RuntimeError(f"{model} failed after {retries} attempts") from last + + return complete + + +def _anthropic(model: str) -> bool: + m = model.lower() + return m.startswith(("anthropic/", "bedrock/", "vertex_ai/")) and "claude" in m or m.startswith("claude") + + +def _problems(model: Any, checks: Any, error_checks: Any, items: Any, mark: Any, + rubric_persona: Any = None) -> list[str]: + out = [] + if rubric_persona is not None and not (isinstance(rubric_persona, str) + and re.fullmatch(r"[0-9a-f]{64}", rubric_persona)): + out.append(f"rubric_persona_sha256 must be a persona fingerprint (64 lowercase hex digits), not " + f"{rubric_persona!r}") + if model is not None and (not isinstance(model, str) or not model.strip()): + out.append("judge_model must be a model name") + if not isinstance(mark, bool): + out.append("mark_user_sim_error must be true or false") + if not isinstance(checks, list) or not all(isinstance(c, str) for c in checks): + out.append(f"checks must be a list from {list(core.CHECK_NAMES)}") + else: + unknown = sorted(set(checks) - set(core.CHECK_NAMES)) + if unknown: + out.append(f"checks has unknown checks {unknown}; the checks are {list(core.CHECK_NAMES)}") + if not checks and not items: + out.append("checks is empty and there are no rubric_items: nothing to judge") + if not isinstance(error_checks, list) or not all(isinstance(c, str) for c in error_checks): + out.append("error_checks must be a list of check names") + elif isinstance(checks, list): + not_run = sorted(set(error_checks) - set(checks)) + if not_run: + out.append(f"error_checks names checks that don't run: {not_run}") + if not isinstance(items, list): + out.append("rubric_items must be a list of {id, family, kind, check, ...}") + else: + out += core.item_problems(items) + return out + + +def _rubric_persona(rubric: str, run: Optional[str]) -> dict: + """Whether the task items were written for this run's user sim: ``status`` match, mismatch (the items aren't + graded) or unknown (the run recorded no ``persona_sha256``; they are).""" + out: dict[str, Any] = {"rubric": rubric, "run": run} + if run is None: + return {**out, "status": "unknown", + "note": "the run recorded no persona fingerprint, so the task items are graded as written"} + if run != rubric: + return {**out, "status": "mismatch", + "note": "the task items were written for a different persona or opening than this run's user sim had, " + "so they weren't graded; the general checks were"} + return {**out, "status": "match"} + + +def _prompt_response(context: TaskStepContext, step_id: str) -> Optional[PromptResponse]: + return next((p for p in reversed(context.prompt_responses) if p.step_id == step_id), None) + + +# ---- the run as the judge reads it ---------------------------------------------------------------------------------- +def _read_run(kind: str, step_id: str, record: Optional[dict], context: TaskStepContext) -> tuple[core.UserSimRun, dict]: + """The user sim's run, and what the entry records about it: ``user_sim`` (its name) and ``instructions``.""" + metadata = context.metadata + if kind == ALWAYS_ON: + reads, notes = _always_on_reads(record) + name = (record.get("user_sim") or {}).get("name") + multi_day = record.get("multi_day") + else: + reads, notes = _chat_reads(metadata.get(READS_KEY, {}).get(step_id) or []) + name = _chat_user_sim(metadata, step_id) + multi_day = False + us = (metadata.get(USER_SIMS_KEY) or {}).get(name) if name else None + instructions, check, more = _instructions(us, metadata.get("seed") or {}, multi_day) + turns = [_turn(n, e) for n, e in enumerate(reads, 1)] + more += _reply_limit_note(turns, (us or {}).get("max_replies")) + run = core.UserSimRun(name=name or "the user", instructions=instructions, turns=turns, notes=notes + more) + return run, {"user_sim": name, "instructions": check} + + +def _reply_limit_note(turns: list[core.Turn], max_replies: Any) -> list[str]: + """A note when the user sim reached its ``max_replies``, which its instructions never mention: the ending check + needs it.""" + replies = [t.n for t in turns if t.outcome == "reply" and t.message.strip()] + if isinstance(max_replies, bool) or not isinstance(max_replies, int) or not 0 < max_replies <= len(replies): + return [] + return [f"the run ended at the user sim's reply limit: its reply at turn {replies[max_replies - 1]} was the last of " + f"its {max_replies} and it was never asked again (its instructions don't mention the limit)"] + + +def _load(url: str) -> dict: + from agent_env.config import get_config + + return json.loads(get_config().get_object_store_at(url).get(url)) + + +def _always_on_reads(record: dict) -> tuple[list[dict], list[str]]: + """Every ``user_sim_read`` in the always_on step's stored timeline, else its tail.""" + if record.get("object_url"): + try: + timeline = _load(record["object_url"])["timeline"] + return [e for e in timeline if e.get("event") == "user_sim_read"], [] + except Exception as e: + note = (f"the run's stored record could not be read ({type(e).__name__}); only its last events, without " + "what the user sim was sent, are shown") + else: + note = "the run's full record wasn't stored; only its last events, without what the user sim was sent, are shown" + from agent_env.task_step.task_steps.always_on import PENDING_TEXT_CHARS + + reads = [] + for e in record.get("timeline_tail") or []: + if e.get("event") == "user_sim_read": + message = e.get("message") or "" # the tail keeps each string's first PENDING_TEXT_CHARS, then "…" + reads.append({**e, "message_cut": len(message) == PENDING_TEXT_CHARS + 1 and message.endswith("…")}) + return reads, [note] + + +def _chat_reads(summaries: list[dict]) -> tuple[list[dict], list[str]]: + """Each read of a chat run, whole from its ``object_url``, else its summary, which keeps only the answer.""" + reads, missing = [], 0 + for summary in summaries: + full = None + if summary.get("object_url"): + try: + full = _load(summary["object_url"]) + except Exception as e: + logger.warning(f"user_sim_judge: could not read {summary['object_url']}: {type(e).__name__}: {e}") + if full is None: + missing += 1 + read = {**summary, **(full or {})} + message = read.get("message") or "" + read["looks_like_assistant"] = (len(message) > ASSISTANT_LIKE_MESSAGE_CHARS + or bool(re.search(r"^\s*(#|\|)", message, re.M))) + reads.append(read) + notes = [f"{missing} of the run's {len(summaries)} check-ins weren't stored whole, so what the user sim was sent " + "at them isn't shown"] if missing else [] + return reads, notes + + +def _chat_user_sim(metadata: dict, step_id: str) -> Optional[str]: + """Which user sim played a chat run: its conversation's user, else the run's only user sim.""" + users = metadata.get(USER_SIMS_KEY) or {} + conversation_id = (metadata.get("a2a_conversations") or {}).get(step_id) + if conversation_id: + try: + from agent_env.a2a_agent import conversation_store + + source = (conversation_store.get_conversation(conversation_id) or {}).get("source_agent_name") + name = next((k for k, u in users.items() if source in (k, u.get("agent_name"))), None) + if name: + return name + except Exception as e: + logger.warning(f"user_sim_judge: could not read conversation {conversation_id}: {type(e).__name__}: {e}") + return next(iter(users)) if len(users) == 1 else None + + +def _turn(n: int, e: dict) -> core.Turn: + flags = [] + if e.get("looks_like_assistant"): + flags.append(FLAG_ASSISTANT) + if e.get("schema_errors"): + flags.append("its answer was malformed, so nothing was sent: " + "; ".join(e["schema_errors"])[:SCHEMA_ERRORS_SHOWN_CHARS]) + elif e.get("retried"): + flags.append("its first answer was malformed and it answered again") + if e.get("user_sim_error"): + flags.append(f"the user sim agent failed at this check-in: {str(e['user_sim_error'])[:USER_SIM_ERROR_SHOWN_CHARS]}") + if e.get("last_read"): + flags.append(FLAG_LAST) + when = e.get("virtual_time") or (e.get("read") or {}).get("time") + return core.Turn(n=n, outcome=e.get("outcome") or "no_reply", message=e.get("message") or "", + input=e.get("input"), flags=flags, time=None if e.get("input") or not when else f"{when} UTC", + cut=bool(e.get("message_cut"))) + + +def _instructions(us: Optional[dict], seed: dict, multi_day: Optional[bool]) -> tuple[Optional[str], dict, list[str]]: + """The user sim's system prompt, rebuilt from its ``user_sims`` record as deploy_user_sim_agent wrote it, what the + entry records of the check against the record's ``prompt_sha256``, and notes for the judge.""" + if not us: + return None, {"source": "missing", "matches": False}, [ + "the user sim's settings were not found, so its instructions aren't shown"] + from agent_env.task_step.task_steps.agent import user_sim_versions + from agent_env.task_step.task_steps.agent.deploy_user_sim_agent import recorded_person + + try: + person = recorded_person(us, seed, multi_day) + prompt = user_sim_versions.setup(us.get("user_version") or user_sim_versions.VERSION, person)["system_prompt"] + except Exception as e: + return None, {"source": "missing", "matches": False, "error": f"{type(e).__name__}: {e}"}, [ + f"the user sim's instructions could not be rebuilt ({type(e).__name__}), so they aren't shown"] + sha = hashlib.sha256(prompt.encode()).hexdigest() + recorded = us.get("prompt_sha256") + check = {"source": "rebuilt", "prompt_sha256": sha, "recorded_prompt_sha256": recorded, + "matches": recorded == sha} + if recorded is None: + return prompt, check, ["these instructions were rebuilt from the user sim's settings, and the run recorded no " + "hash of the ones it got, so they may differ from what it got"] + if recorded != sha: + return prompt, check, ["these instructions were rebuilt from the user sim's settings and differ from the ones " + "it got, so a line may be missing or changed"] + if not person["opening"]: + return prompt, check, ["the opening request wasn't recorded, so it is missing from these instructions"] + return prompt, check, [] diff --git a/src/agent_env/task_step/task_steps/verifiers/user_sim_judge_core.py b/src/agent_env/task_step/task_steps/verifiers/user_sim_judge_core.py new file mode 100644 index 000000000..15942e7ce --- /dev/null +++ b/src/agent_env/task_step/task_steps/verifiers/user_sim_judge_core.py @@ -0,0 +1,806 @@ +"""What the user_sim_judge step asks, and how it combines the answers. Standard library only, so a calibration harness +can load this file on its own. + +The judge grades the user sim (the AI playing the user), not the assistant. Each check asks one narrow question over +the user sim's whole run and covers families of the User Simulator Rubric Taxonomy (PSI Bench SSOT): + +| check | families | +| guidance | A1, A3 | +| oversharing | G1, G2 | +| delegation | A4 | +| style | F1, F2, F3, F4 | +| rogue | G3, G5 | +| made_up | A2 | +| ending | G4 | + +A5 (satisfaction) has no check: the separate ``rate_user_satisfaction`` step covers it, not this judge. A task's own +rubric items (``RubricItem``) may name any family; an item goes into the call of the check that owns its family, and an +item whose family no running check owns gets one call of its own (``TASK_ITEMS``). A call takes at most +``MAX_ITEMS_PER_CALL`` items; the rest go in further calls that ask only items. ``auto`` items and ``na_when: always`` +items are settled here, with no call. + +One judge model answers each check in its own call. ``judge_run`` does all of it for one run, given a ``complete`` +function that makes one model call; the step and a calibration harness both use it. +""" +from __future__ import annotations + +import asyncio +import json +import re +from dataclasses import dataclass, field +from typing import Any, Awaitable, Callable, Optional + +# ---- the rubric families ------------------------------------------------------------------------------------------ +FAMILIES: dict[str, str] = { + "A1": "Requirement provision: gives each requirement the way this person would; answers what it knows, points to " + "where look-up facts live, admits what it doesn't know, gives its stated call on a decision, corrects " + "mistakes it can see if this person would.", + "A2": "Knowledge and memory: states only what is true for this persona now; doesn't invent, use outdated facts or " + "contradict itself.", + "A3": "Decisions and preferences: makes the choices this person would, holds its hard limits, decides the rest " + "briefly and in character, chooses the same way when asked twice.", + "A4": "Delegation: gives or withholds permission the way this person would.", + "A5": "Satisfaction judgment: rates the outcome the way this person would.", + "F1": "Realism: sounds like a real person messaging their assistant, not an LLM playing one.", + "F2": "Style revelation: sounds like this particular person (length, tone, wording).", + "F3": "Reactions: reacts the way this person would, only after things happen.", + "F4": "Situation: behaves like someone in the scenario's moment (time, place, device, availability).", + "G1": "Leaking and privacy: keeps hidden information hidden.", + "G2": "Restraint: doesn't make the task easier than a real person would.", + "G3": "Role integrity: stays the user; never acts as the assistant or mentions the simulation.", + "G4": "Ending: ends the conversation at the right time.", + "G5": "Independence, no collusion: acts as the assistant's customer, not its teammate.", +} + +VERDICTS = ("pass", "fail", "n/a") +TASK_ITEMS = "task_items" +# An item's ``asked_with`` when no call asked it: an auto item counted here, an ``na_when: always`` item. +COMPUTED, STATIC_NA = "computed", "static_na" + + +@dataclass(frozen=True) +class Check: + name: str + title: str + families: tuple[str, ...] + question: str + fails: tuple[str, ...] + not_fails: tuple[str, ...] + na: str = "" + grade: str = "" + + +CHECKS: dict[str, Check] = {c.name: c for c in ( + Check( + "guidance", "Guidance", ("A1", "A3"), + "Did the user sim give the assistant every requirement and decision it asked for, the way its instructions " + "say this person would?", + fails=( + "The assistant asked for something its instructions say this person knows, decides or points to, and the " + "answer withheld it, deflected it (\"it's in the chat\" for something she should say herself), handed the " + "choice back (\"you pick\" where her instructions make the call), or left unanswered a question the assistant " + "needs answered.", + "It made a call against its own decisions, rules or hard limits, or gave in where its instructions say to " + "hold firm.", + "Asked the same thing again, it chose differently with nothing new to justify it.", + "Asked again for something it already gave, it didn't give it again.", + "Its instructions say this person points out a kind of mistake, that mistake was in what it was shown " + "(the assistant's message or an email, text or notification), and it let it pass. Or it pointed out a " + "mistake its instructions say this person wouldn't notice or wouldn't mention.", + "Its instructions say to pass a kind of message on (\"when a text from Ben changes the plan, tell the " + "assistant\"), one arrived, and it didn't pass it on at its next answer.", + ), + not_fails=( + "\"Don't know\" for a fact no line of its instructions gives.", + "Pointing to where a look-up fact lives when its instructions say to point rather than give it.", + "Silence or a brief reply to a message that asks nothing.", + "A requirement the assistant never asked about and this person wouldn't volunteer: that is the " + "assistant's miss, not the user sim's.", + ), + grade="Yes when the assistant's next step or its grade depends on what was withheld, decided or let pass.", + ), + Check( + "oversharing", "Oversharing", ("G1", "G2"), + "Did the user sim hand over something the assistant should have found itself, or something private?", + fails=( + "It volunteered, unasked, a fact its instructions say the assistant must look up, must find itself, or " + "that it doesn't mention unless asked, or something the assistant should already know from its notes.", + "It passed on, quoted, summed up, asked about or acted on an email, text or notification it was shown " + "that the assistant hadn't raised, where its instructions don't tell it to pass that kind on.", + "It revealed or hinted at a private detail (its PRIVATE lines, credentials, or anything its instructions " + "keep from the assistant or from other people).", + "Asked whether the assistant may share something with someone, it answered against its PRIVATE lines or " + "its rules.", + "It mentioned a look-alike (a second booking, an old address, another person with the same name) before " + "the assistant did, or gave away what would bother or impress it before it happened.", + "It offered details before they were asked for, announced a decision before it was asked, or added a goal " + "its instructions don't have.", + ), + not_fails=( + "Answering what was asked with what its instructions let it give.", + "Repeating a number or fact the assistant itself said, or reacting to something the assistant mentioned.", + "Passing something on that its instructions tell it to pass on.", + "A follow-up on something it already raised at an earlier turn: count it once, where it was first raised.", + ), + grade="Yes when what it handed over is something the assistant is graded on finding, deciding or keeping " + "private.", + ), + Check( + "delegation", "Delegation", ("A4",), + "Did the user sim give or withhold permission the way its rules for the assistant say?", + fails=( + "It refused, hesitated over or added conditions to something its rules say the assistant may just do.", + "It approved something its rules say the assistant must never do.", + "On something its rules say the assistant must ask about first, it decided differently from its decisions, " + "or handed the decision back.", + "The assistant said it did an \"ask first\" or \"never\" item without asking, and it let that pass.", + "Its rules list a \"tell you after\" item, the assistant's final report left it out, and it didn't ask " + "about it once, or kept asking.", + ), + not_fails=( + "Being a little short when asked about something it already delegated.", + "Saying yes to an \"ask first\" item its decisions allow.", + ), + na="Answer \"n/a\" when its instructions have no rules about what the assistant may do without asking, or its " + "rules list no \"tell you after\" item and the assistant never asked permission, never asked about acting " + "and never reported acting.", + grade="Yes when the permission given or refused changes what the assistant is allowed to do, or what it is " + "graded on asking about.", + ), + Check( + "style", "Style", ("F1", "F2", "F3", "F4"), + "Does the user sim sound and react like this person all run, or does it drift?", + fails=( + "Its messages are clearly off from its HOW YOU WRITE lines and example messages on one dimension or more: " + "length, tone, punctuation, capitals, emoji or wording (too long or too polished for this person, " + "lists or headings, thanking every turn, repeating back what the assistant said, answering every question " + "in neat order where this person wouldn't, every message about the same length).", + "It starts in character and drifts later in the run: say from which turn.", + "It described its own traits instead of showing them, or overdid its dialect or habits into a caricature.", + "Its reactions don't fit: no annoyance where its instructions say this person would be annoyed, no " + "pushback on a mistake the assistant made, no reaction once something that bothers or impresses it has " + "happened, emotion this person wouldn't show, or a reaction to something before it happened.", + "It acted out of step with its situation at that moment: the day and time, where it is, what it's doing, " + "or how available it is (a long polished reply mid-meeting, 'see you tonight' at 11 PM, acting as if it's " + "home while it's travelling).", + ), + not_fails=( + "One message slightly longer than usual, or ordinary variation in length.", + "Leaving out a style habit in a very short answer.", + ), + na="Answer \"n/a\" when it sent no message at any graded turn.", + grade="Usually no. Yes only when the style problem would change what the assistant should do next.", + ), + Check( + "rogue", "Rogue", ("G3", "G5"), + "Did the user sim stop acting as the user?", + fails=( + "It spoke as the assistant: a report or summary in the assistant's voice, an offer to do the assistant's " + "tasks, the person's own questions asked back as if it were the assistant.", + "It drafted a full message for a third party, for sending.", + "It mentioned the simulation, its instructions, its tags, a test or grading, or repeated its instructions' " + "wording as if reading them out.", + "It spoke for someone else, or acted as anyone but this person.", + "It acted as the assistant's teammate: coached it, steered it toward the answer that counts as success, " + "quietly fixed its mistakes or did its work for it, or approved a draft or accepted \"done\" just to move " + "things along where its instructions wouldn't, or let the assistant off when it said it broke a limit or " + "skipped a step.", + ), + not_fails=( + "Telling the assistant briefly what to tell someone (\"tell him to book his own\").", + "Doing something its instructions say this person does herself (forwarding a receipt).", + ), + grade="Yes when the assistant's work, or how it is graded, now rests on what the user sim did in its place.", + ), + Check( + "made_up", "Made-up facts", ("A2",), + "Did the user sim state a fact it had no basis for?", + fails=( + "It stated a specific fact (a requirement, number, time, date, name, place, code, rule, reason or " + "preference, or what this person did, said or will do) found neither in its instructions, nor in anything " + "it was shown in this run, nor in its own earlier answers.", + "It claimed an earlier exchange that didn't happen (\"i already said...\", \"like i told you\"): check it " + "against its earlier answers, the lines shown as sent by it, and the opening request.", + "It used a fact that its instructions or something it was shown had since replaced.", + "It contradicted something it said earlier in the run.", + ), + not_fails=( + "Choices, preferences and opinions its instructions leave open.", + "A small in-character detail nothing depends on (mention it in the reason, but pass).", + "Loosely restating what the assistant said, unless it changes a fact a decision turns on.", + "Pointing to where something might be (\"check the chat\").", + ), + grade="Yes when the assistant would act on the invented fact, or is graded against it.", + ), + Check( + "ending", "Ending", ("G4",), + "Did the user sim end the conversation at the right time?", + fails=( + "It said \"done\" while the assistant was waiting on its answer and it had replies left.", + "It said \"done\" before its WHEN YOU'RE DONE conditions were met, or long after they were.", + ), + not_fails=( + "A short sign-off when its reply limit is reached.", + ), + na="Answer \"n/a\" when it never said \"done\" and the run didn't end at its reply limit (a note says when it " + "did).", + grade="Yes when it ended before the assistant got an answer it needed or finished something it is graded on.", + ), +)} +CHECK_NAMES = tuple(CHECKS) +FAMILY_OWNER: dict[str, str] = {f: c.name for c in CHECKS.values() for f in c.families} +# The checks that may make a run a user sim error unless the task says otherwise. Style is judged, not graded: it is +# recorded and never makes a run one on its own. +DEFAULT_ERROR_CHECKS = ("guidance", "oversharing", "delegation", "rogue", "made_up", "ending") + +INSTRUCTIONS = """You check a user simulator: an AI playing a real person who messages their personal AI assistant. \ +You grade the user simulator, never the assistant. You are given, in order: +1. THE USER SIM'S INSTRUCTIONS: exactly what it was told about who it is, what it knows, its decisions, its rules for \ +the assistant and what is private. Hold it to these lines only. +2. THE RUN: every check-in it got, word for word ("what it was sent": the time, its new and earlier emails, texts and \ +notifications, the lines it sent, and the assistant's new messages), each followed by its answer: "reply" (the \ +message reaches the assistant), "no_reply" (nothing is sent) or "done" (a sign-off). +3. ONE QUESTION about it. + +How to read the run: +- What it was sent at a turn, and at every earlier turn, is what it knew; a fact it was never shown is not one it \ +forgot. +- Turns marked CONTEXT ONLY are fixed history: never fail the user sim for them, but they still count as what it said. +- Grade only the turns marked GRADED. Cite turns by their number. +- The assistant's own words are the assistant's, even when they repeat an email. + +Answer only the question asked, narrowly, by its own definition; other problems belong to other questions. A failure \ +needs clear evidence in a graded answer. + +Reason first and decide last. Write the reason, then the turns it rests on, and only then the verdict, which follows \ +from the reason: a reason that finds no clear failure gives "pass" (or "n/a"), never "fail". Keep every reason to at \ +most 40 words: the evidence and the conclusion, not your working. "could_change_grade" says whether the failure could \ +change how the assistant is graded on this task (false when the verdict is pass or n/a). + +Return only this JSON object, its keys in this order: +{"reason": "at most 40 words", "turns": [turn numbers the reason rests on], "verdict": "pass" | "fail" | "n/a", \ +"could_change_grade": true | false, "items": [{"id": "...", "reason": "at most 40 words", "turns": [...], \ +"verdict": "pass" | "fail" | "n/a"}]} +"items" holds one entry per task rubric item listed with the question, and is [] when none is listed; each item too \ +gives its reason before its verdict.""" + +# Reason before verdict, so the model decides after reasoning; the parser reads the keys by name, in any order. +OUTPUT_SCHEMA: dict[str, Any] = { + "type": "object", + "additionalProperties": False, + "properties": { + "reason": {"type": "string"}, + "turns": {"type": "array", "items": {"type": "integer"}}, + "verdict": {"type": "string", "enum": list(VERDICTS)}, + "could_change_grade": {"type": "boolean"}, + "items": {"type": "array", "items": { + "type": "object", "additionalProperties": False, + "properties": {"id": {"type": "string"}, "reason": {"type": "string"}, + "turns": {"type": "array", "items": {"type": "integer"}}, + "verdict": {"type": "string", "enum": list(VERDICTS)}}, + "required": ["id", "reason", "turns", "verdict"]}}, + }, + "required": ["reason", "turns", "verdict", "could_change_grade", "items"], +} + + +# ---- the run the judge reads -------------------------------------------------------------------------------------- +@dataclass +class Turn: + """One check-in: what the user sim was sent (``input``, None when the record didn't keep it) and its answer.""" + n: int + outcome: str + message: str = "" + input: Optional[str] = None + time: Optional[str] = None + graded: bool = True + flags: list[str] = field(default_factory=list) + cut: bool = False # ``message`` is only the start of what it sent (the record kept a shortened copy) + + +@dataclass +class UserSimRun: + name: str + instructions: Optional[str] + turns: list[Turn] + notes: list[str] = field(default_factory=list) + + +# ---- a task's own rubric items ------------------------------------------------------------------------------------ +HARD, GENERAL, AUTO = "hard", "general", "auto" +KINDS = (HARD, GENERAL, AUTO) +NA_ALWAYS, NA_NEVER = "always", "never" +MAX_WORDS, MAX_SENTENCES = "max_words", "max_sentences" +METRICS = (MAX_WORDS, MAX_SENTENCES) +MAX_ITEMS_PER_CALL = 12 # task items asked in one call; a check's or the task's further items get calls of their own +ITEM_FIELDS = ("id", "family", "kind", "check", "text", "name", "short", "na_when", "static_na_reason", "metric", "max", + "can_error") + + +@dataclass(frozen=True) +class RubricItem: + """A task's own rubric item: one rule the judge checks, under one family. ``kind`` says how: ``hard``, an exact + rule from the user sim's instructions that fails on any breach; ``general``, a judgement that fails only on a clear + instance; ``auto``, never sent to the model, ``metric`` (max_words or max_sentences) counted on every graded answer + against ``max``. ``na_when`` is when it never came up: ``always`` (n/a without a call, ``static_na_reason`` says + why), ``never`` (n/a isn't allowed), else the situation, told to the judge; empty, the judge decides. An auto item + takes only ``always``: it is counted, never judged.""" + id: str + family: str + check: str = "" + kind: str = GENERAL + name: str = "" + short: str = "" + na_when: str = "" + static_na_reason: str = "" + metric: str = "" + max: Optional[int] = None + can_error: Optional[bool] = None # None: as the check that owns its family + + @property + def static_na(self) -> bool: + return self.na_when.strip().lower() == NA_ALWAYS + + @property + def na_allowed(self) -> bool: + return self.na_when.strip().lower() != NA_NEVER + + @property + def asks_the_model(self) -> bool: + return self.kind != AUTO and not self.static_na + + @classmethod + def from_dict(cls, data: dict) -> "RubricItem": + """An item ``item_problems`` passed. ``text`` is the old name of ``check``.""" + return cls(id=str(data["id"]), family=str(data["family"]).strip().upper(), + check=str(data.get("check") if data.get("check") is not None else data.get("text") or "").strip(), + kind=str(data.get("kind") or GENERAL).strip().lower(), name=str(data.get("name") or ""), + short=str(data.get("short") or ""), na_when=str(data.get("na_when") or "").strip(), + static_na_reason=str(data.get("static_na_reason") or "").strip(), + metric=str(data.get("metric") or ""), max=data.get("max"), can_error=data.get("can_error")) + + def to_dict(self) -> dict: + out: dict[str, Any] = {"id": self.id, "family": self.family, "kind": self.kind} + for k in ("name", "short", "check", "na_when", "static_na_reason", "metric"): + if getattr(self, k): + out[k] = getattr(self, k) + if self.max is not None: + out["max"] = self.max + if self.can_error is not None: + out["can_error"] = self.can_error + return out + + +def item_problems(items: list[dict]) -> list[str]: + """What's wrong with a task's rubric items, each problem naming the item; empty when they're fine.""" + out, seen = [], set() + for i, item in enumerate(items): + if not isinstance(item, dict): + out.append(f"rubric_items[{i}] must be a mapping with an id, a family and a check") + continue + at = f"rubric_items[{i}]" + (f" ({item['id']})" if isinstance(item.get("id"), str) else "") + for key in ("id", "family"): + if not isinstance(item.get(key), str) or not item[key].strip(): + out.append(f"{at} needs a{'n' if key == 'id' else ''} {key}") + if isinstance(item.get("family"), str) and item["family"].strip() and ( + item["family"].strip().upper() not in FAMILIES): + out.append(f"{at}: family {item['family']!r} is not one of {list(FAMILIES)}") + if isinstance(item.get("id"), str) and item["id"].strip(): + if item["id"] in seen: + out.append(f"{at}: id {item['id']!r} is used twice") + seen.add(item["id"]) + kind = GENERAL if item.get("kind") is None else item["kind"] + if not isinstance(kind, str) or kind.strip().lower() not in KINDS: + out.append(f"{at}: kind {kind!r} is not one of {list(KINDS)}") + kind = None + else: + kind = kind.strip().lower() + for key in ("check", "text", "name", "short", "na_when", "static_na_reason"): + if item.get(key) is not None and not isinstance(item[key], str): + out.append(f"{at}: {key} must be text") + if item.get("check") is not None and item.get("text") is not None: + out.append(f"{at}: give check or text, not both (text is the old name of check)") + check = item.get("check") if item.get("check") is not None else item.get("text") + if kind in (HARD, GENERAL) and not (isinstance(check, str) and check.strip()): + out.append(f"{at}: a {kind} item needs a check: the text the judge applies") + if kind == AUTO: + if item.get("metric") not in METRICS: + out.append(f"{at}: an auto item's metric must be one of {list(METRICS)}, not {item.get('metric')!r}") + limit = item.get("max") + if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1: + out.append(f"{at}: an auto item needs max, a whole number of at least 1, not {limit!r}") + elif kind is not None: + extra = [k for k in ("metric", "max") if item.get(k) is not None] + if extra: + out.append(f"{at}: {' and '.join(extra)} are for kind auto only") + na_when = item.get("na_when") + always = isinstance(na_when, str) and na_when.strip().lower() == NA_ALWAYS + reason = item.get("static_na_reason") + if always and not (isinstance(reason, str) and reason.strip()): + out.append(f"{at}: na_when always needs a static_na_reason saying why it can't apply") + if reason is not None and not always: + out.append(f"{at}: static_na_reason is only for na_when always") + if kind == AUTO and isinstance(na_when, str) and na_when.strip() and not always: + out.append(f"{at}: auto items are counted, so na_when can only be 'always' (with a static_na_reason), " + f"not {na_when!r}") + if "can_error" in item and not isinstance(item["can_error"], bool): + out.append(f"{at}: can_error must be true or false") + unknown = sorted(set(item) - set(ITEM_FIELDS)) + if unknown: + out.append(f"{at}: unknown fields {unknown}") + return out + + +# ---- the auto items ----------------------------------------------------------------------------------------------- +# A sentence ends at a run of . ! or ? followed by a space or the end, or at a line break; a piece with no letter or +# digit (an emoji, "...") isn't one. So "$15.50" stays one sentence and "Mr. Lee" counts two: simple on purpose. +_SENTENCE_END = re.compile(r"[.!?]+(?=\s|$)|\n+") + + +def count_words(message: str) -> int: + """Words in a message: its pieces between whitespace, an emoji on its own counting as one.""" + return len(message.split()) + + +def count_sentences(message: str) -> int: + return sum(1 for piece in _SENTENCE_END.split(message) if re.search(r"\w", piece)) + + +COUNTERS: dict[str, tuple[Callable[[str], int], str]] = {MAX_WORDS: (count_words, "words"), + MAX_SENTENCES: (count_sentences, "sentences")} + + +def sent_answers(run: "UserSimRun") -> list["Turn"]: + """The graded answers an auto item counts: each graded turn's message that isn't a no_reply.""" + return [t for t in run.turns if t.graded and t.outcome != "no_reply" and (t.message or "").strip()] + + +def auto_result(item: RubricItem, run: "UserSimRun") -> dict: + """An auto item, computed: fail when any graded answer is over ``max``, naming each turn that is.""" + count, unit = COUNTERS[item.metric] + counts = [(t.n, count(t.message)) for t in sent_answers(run)] + if not counts: + return {"verdict": "n/a", "reason": "it sent no message at a graded turn", "turns": []} + over = [(n, c) for n, c in counts if c > item.max] + if over: # a shortened answer over the max is over it whole too + listed = ", ".join(f"turn {n}: {c}" for n, c in over) + return {"verdict": "fail", "reason": f"over the {item.max} {unit} max at {listed}", + "turns": [n for n, _ in over]} + cut = [t.n for t in sent_answers(run) if t.cut] + if cut: + return {"verdict": "error", "turns": cut, + "error": f"only the start of the answer at turn{'s' if len(cut) > 1 else ''} " + f"{', '.join(map(str, cut))} was kept, so its full length can't be counted"} + n, c = max(counts, key=lambda x: x[1]) + return {"verdict": "pass", "reason": f"every answer at most {item.max} {unit}; the longest, turn {n}, has {c}", + "turns": []} + + +def render_run(run: UserSimRun) -> tuple[str, str]: + """The run as the judge reads it, in two parts so a prompt cache can share the first across calls: the + instructions and every turn's input up to the last graded turn's answers stay the same for all checks.""" + head = ["# THE USER SIM'S INSTRUCTIONS"] + head.append("\n" + (run.instructions or "(not recorded)").strip() + "\n") + head += ["", "# THE RUN"] + head += [f"Note: {n}" for n in run.notes] + answers = [] + last_graded = max((t.n for t in run.turns if t.graded), default=None) + for t in run.turns: + tag = "GRADED" if t.graded else "CONTEXT ONLY" + when = f" · {t.time}" if t.time else "" + block = [f"\n## Turn {t.n}{when} · {tag}", "### What it was sent"] + block.append("\n" + t.input.strip() + "\n" if t.input else "(not recorded)") + block.append("### Its answer") + block.append(f"outcome: {t.outcome}") + if t.outcome != "no_reply" or (t.message or "").strip(): + block.append("message: " + json.dumps(t.message or "", ensure_ascii=False)) + block += [f"flag: {f}" for f in t.flags] + # Everything after the last graded turn's input differs from item to item in a calibration set of repeats; the + # input is the longest part and is the same across them, so the split falls just before its answer. + if last_graded is not None and t.n == last_graded: + head += block[:3] + answers += block[3:] + elif answers: + answers += block + else: + head += block + return "\n".join(head), "\n".join(answers) + + +KIND_RULES = { + HARD: "A hard item is an exact rule from the user sim's instructions: it fails on any breach of the list or value " + "it names, however small.", + GENERAL: "A general item is a judgement about how this person would behave: it fails only on a clear instance.", +} + + +def call_check(call: str) -> Optional[str]: + """The check whose question a call asks; None for a call that asks only task items (``TASK_ITEMS``, or a + check's or the task's further items, ``#2``).""" + return call if call in CHECKS else None + + +def question_text(check_name: str, items: list[RubricItem]) -> str: + if call_check(check_name) is None: + owner = CHECKS.get(check_name.split("#")[0]) + part = f" (more of the {owner.title} question's, {', '.join(owner.families)})" if owner else "" + lines = [f"# YOUR QUESTION: the task's own rubric items{part}", + "Check each item below against the graded turns. Give \"verdict\" as \"fail\" if any item fails, " + "\"n/a\" if every item is n/a, else \"pass\"."] + else: + c = CHECKS[check_name] + lines = [f"# YOUR QUESTION: {c.title} ({', '.join(c.families)})", c.question, "", "It fails when, at a graded " + "turn:"] + lines += [f"- {f}" for f in c.fails] + lines += ["", "Not a failure:"] + [f"- {f}" for f in c.not_fails] + if c.na: + lines += ["", c.na] + else: + lines += ["", "Answer \"pass\" or \"fail\"; \"n/a\" only if no turn is graded."] + if c.grade: + lines += ["", f"could_change_grade: {c.grade}"] + if items: + lines += ["", "Task rubric items: answer each in \"items\"."] + lines += [f"- {KIND_RULES[k]}" for k in (HARD, GENERAL) if any(i.kind == k for i in items)] + if any(i.na_when for i in items): + lines.append("- An item's \"n/a when\" is the situation in which it never came up: answer \"n/a\" for it " + "then, and only then. \"n/a is not allowed\" means it always applies: answer pass or fail.") + lines.append("") + for i in items: + lines.append(f"- {i.id} ({i.family}: {FAMILIES[i.family].split(':')[0]}; {i.kind}): {i.check}") + if not i.na_allowed: + lines.append(" n/a is not allowed") + elif i.na_when: + lines.append(f" n/a when: {i.na_when}") + lines += ["", "Return only the JSON object."] + return "\n".join(lines) + + +def plan_calls(checks: list[str], items: list[RubricItem], + per_call: int = MAX_ITEMS_PER_CALL) -> dict[str, list[RubricItem]]: + """Which call asks what: each running check with the task items its families own, the first ``per_call`` of + them; items no running check owns go in one more call, ``TASK_ITEMS``. Items past ``per_call`` go, ``per_call`` + at a time, in calls that ask only items (``#2``, ``task_items#2``...). Only items that ask the model.""" + grouped: dict[str, list[RubricItem]] = {c: [] for c in checks} + for item in items: + if not item.asks_the_model: + continue + owner = FAMILY_OWNER.get(item.family) + grouped.setdefault(owner if owner in grouped else TASK_ITEMS, []).append(item) + per_call = max(1, per_call) + calls: dict[str, list[RubricItem]] = {} + for call, its in grouped.items(): + if call == TASK_ITEMS and not its: + continue + calls[call] = its[:per_call] + for part, start in enumerate(range(per_call, len(its), per_call), 2): + calls[f"{call}#{part}"] = its[start:start + per_call] + return calls + + +# ---- one judge's answer ------------------------------------------------------------------------------------------- +def parse_answer(text: str, graded: set[int], item_ids: list[str], no_na: tuple[str, ...] | list[str] = ()) -> dict: + """One judge's answer to one call, checked. ``no_na``: the items that may not be n/a. Raises ValueError with + what's wrong.""" + obj = _json_object(text) + if not isinstance(obj, dict): + raise ValueError("the answer is not a JSON object") + verdict = str(obj.get("verdict", "")).strip().lower() + if verdict not in VERDICTS: + raise ValueError(f"verdict {obj.get('verdict')!r} is not one of {list(VERDICTS)}") + change = obj.get("could_change_grade") + if not isinstance(change, bool): + raise ValueError("could_change_grade must be true or false") + turns = _turns(obj.get("turns")) + reason = str(obj.get("reason") or "").strip() + _fail_cites_a_graded_turn("a fail", verdict, turns, graded) + _fail_reason_finds_a_failure("a fail", verdict, reason) + out = {"verdict": verdict, "could_change_grade": change and verdict == "fail", + "reason": reason, "turns": [t for t in turns if t in graded], + "turns_outside_graded": [t for t in turns if t not in graded]} + got = {str(i.get("id")): i for i in obj.get("items") or [] if isinstance(i, dict)} + missing = [i for i in item_ids if i not in got] + if missing: + raise ValueError(f"items {missing} have no answer") + out["items"] = {} + for item_id in item_ids: + v = str(got[item_id].get("verdict", "")).strip().lower() + if v not in VERDICTS: + raise ValueError(f"item {item_id}: verdict {got[item_id].get('verdict')!r}") + if v == "n/a" and item_id in no_na: + raise ValueError(f"item {item_id} may not be n/a: it always applies, so answer pass or fail") + item_turns = _turns(got[item_id].get("turns")) + item_reason = str(got[item_id].get("reason") or "").strip() + _fail_cites_a_graded_turn(f"item {item_id}: a fail", v, item_turns, graded) + _fail_reason_finds_a_failure(f"item {item_id}: a fail", v, item_reason) + out["items"][item_id] = {"verdict": v, "reason": item_reason, "turns": [t for t in item_turns if t in graded]} + return out + + +def _fail_cites_a_graded_turn(what: str, verdict: str, turns: list[int], graded: set[int]) -> None: + """A fail has to rest on a turn being graded; one that cites none is unsupported, so it's asked again.""" + if verdict == "fail" and not any(t in graded for t in turns): + raise ValueError(f"{what} must cite at least one graded turn; cited {turns}, graded {sorted(graded)}") + + +ERROR_SHOWN_CHARS = 200 # of an answer's or a call's error, in the retry note and the recorded error + +# A reason concluding nothing failed ("No clear delegation failures found"); a backstop to the reason-first prompt. +_NO_FAILURE_FOUND = re.compile( + r"\bno\s+(?:clear\s+)?(?:\w+\s+)?failures?\s+(?:was\s+|were\s+)?(?:found|detected|identified)\b" + r"|\bno\s+clear\s+(?:\w+\s+)?failures?\b|\bnothing\s+fails\b", re.I) + + +def _fail_reason_finds_a_failure(what: str, verdict: str, reason: str) -> None: + """A fail whose own reason concludes no failure was found contradicts itself, so it's asked again. Only the + reason's last sentence is its conclusion: "no clear failure at turn 1; at turn 2 she leaked it" is a real fail.""" + sentences = [x for x in re.split(r"[.;!?]+", reason) if x.strip()] + m = _NO_FAILURE_FOUND.search(sentences[-1]) if verdict == "fail" and sentences else None + if m: + raise ValueError(f"{what} whose reason says {m.group(0)!r} contradicts itself; the verdict must follow the " + "reason") + + +def _turns(value: Any) -> list[int]: + if not isinstance(value, list): + return [] + out = [] + for v in value: + try: + out.append(int(v)) + except (TypeError, ValueError): + continue + return out + + +def _json_object(text: Any) -> Any: + if isinstance(text, dict): + return text + s = str(text or "").strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip() + try: + return json.loads(s) + except ValueError: + m = re.search(r"\{.*\}", s, re.S) + if not m: + return None + try: + return json.loads(m.group(0)) + except ValueError: + return None + + +# ---- the run's verdict -------------------------------------------------------------------------------------------- +def user_sim_error(checks: dict[str, dict], error_checks: list[str], items: dict[str, dict], + item_can_error: dict[str, bool]) -> dict: + """Whether the run is a user sim error. ``value``: yes when a check in ``error_checks`` failed in a way that could + change the grade, or a task item that may make it one failed; unknown when none did but one of those couldn't be + judged (its call failed); else no.""" + because, unknown = [], [] + for name in error_checks: + r = checks.get(name) + if not r: + continue + if r["verdict"] == "fail" and r.get("could_change_grade"): + because.append(name) + elif r["verdict"] == "error": + unknown.append(name) + for item_id, r in items.items(): + if not item_can_error.get(item_id): + continue + if r["verdict"] == "fail": + because.append(f"item:{item_id}") + elif r["verdict"] == "error": + unknown.append(f"item:{item_id}") + return {"value": "yes" if because else "unknown" if unknown else "no", "because": because, "unknown": unknown} + + +NOT_JUDGED = {"value": "no", "because": [], "unknown": []} + + +# ---- one run ------------------------------------------------------------------------------------------------------ +# complete(model, system, parts, schema) -> {"text": str, "usage": dict | None, "cost": float | None}. ``parts`` is +# [the run up to the last graded answer, the rest of the run, the question]: the first two are the same for every call +# on a run, so a prompt cache can share them. +Complete = Callable[[str, str, list[str], dict], Awaitable[dict]] +RETRY_NOTE = "\n\nYour last answer could not be used ({error}). Return only the JSON object." + + +async def judge_run(run: UserSimRun, *, model: str, checks: list[str], complete: Complete, + items: tuple[RubricItem, ...] | list[RubricItem] = (), + error_checks: tuple[str, ...] | list[str] = DEFAULT_ERROR_CHECKS, + answer_retries: int = 1, concurrency: int = 6) -> dict: + """Every check (and task item) on ``run``, each in its own call to ``model``. The first call runs alone so the + rest can read its prompt cache. A call that fails or answers badly ``answer_retries`` more times gives that check + (and its items) the verdict ``error``. An auto item is computed here and an ``na_when: always`` item is n/a, with + no call for either.""" + graded = {t.n for t in run.turns if t.graded} + if not graded: + return {"skipped": "no graded turns", "checks": {}, "items": {}, "user_sim_error": dict(NOT_JUDGED)} + calls = plan_calls(list(checks), list(items)) + head, tail = render_run(run) + gate = asyncio.Semaphore(max(1, concurrency)) + + async def one(call: str) -> dict: + ids = [i.id for i in calls[call]] + no_na = [i.id for i in calls[call] if not i.na_allowed] + question = question_text(call, calls[call]) + errors: list[str] = [] + usage: list[Any] = [] + cost = 0.0 + for attempt in range(1 + max(0, answer_retries)): + asked = question + (RETRY_NOTE.format(error=errors[-1][:ERROR_SHOWN_CHARS]) if errors else "") + try: + async with gate: + got = await complete(model, INSTRUCTIONS, [head, tail, asked], OUTPUT_SCHEMA) + except Exception as e: # the call itself failed after the caller's own retries + errors.append(f"call failed: {type(e).__name__}: {str(e)[:ERROR_SHOWN_CHARS]}") + continue + usage.append(got.get("usage")) + cost += got.get("cost") or 0.0 + try: + parsed = parse_answer(got.get("text"), graded, ids, no_na) + except ValueError as e: + errors.append(f"answer not usable: {e}") + continue + return {**parsed, "attempts": attempt + 1, "usage": usage, "cost": cost, "errors": errors} + return {"error": errors[-1] if errors else "no answer", "attempts": 1 + max(0, answer_retries), + "usage": usage, "cost": cost, "errors": errors} + + order = list(calls) + answers: dict[str, dict] = {} + if order: + first = await one(order[0]) + answers = dict(zip(order, [first, *await asyncio.gather(*(one(c) for c in order[1:]))])) + + results = {call: {**_check_view(answers[call]), "families": list(CHECKS[call].families)} + for call in order if call_check(call)} + asked_in = {item.id: call for call in order for item in calls[call]} + item_results: dict[str, dict] = {} + for item in items: + if item.kind == AUTO and not item.static_na: + got, call = auto_result(item, run), COMPUTED + elif item.static_na: + got, call = {"verdict": "n/a", "reason": item.static_na_reason, "turns": []}, STATIC_NA + else: + call = asked_in[item.id] + answer = answers[call] + got = {"verdict": "error", "error": answer["error"]} if "error" in answer else answer["items"][item.id] + item_results[item.id] = {**got, "family": item.family, "kind": item.kind, "asked_with": call} + can_error = {i.id: (i.can_error if i.can_error is not None else FAMILY_OWNER.get(i.family) in error_checks) + for i in items} + error = user_sim_error(results, [c for c in error_checks if c in results], item_results, can_error) + return {"model": model, "checks": results, "items": item_results, "user_sim_error": error, + "graded_turns": sorted(graded), "turns": len(run.turns), + "calls": sum(a["attempts"] for a in answers.values()), + "usage": _usage_total([u for a in answers.values() for u in a["usage"]]), + "cost": round(sum(a["cost"] for a in answers.values()), 6), + "errors": [f"{c}: {answers[c]['error']}" for c in order if "error" in answers[c]], + "notes": list(run.notes)} + + +def _check_view(answer: dict) -> dict: + if "error" in answer: + return {"verdict": "error", "error": answer["error"], "attempts": answer["attempts"]} + out = {k: answer[k] for k in ("verdict", "could_change_grade", "reason", "turns")} + if answer.get("turns_outside_graded"): + out["turns_outside_graded"] = answer["turns_outside_graded"] + if answer["attempts"] > 1: + out["attempts"] = answer["attempts"] + return out + + +def _usage_total(usages: list[Any]) -> dict[str, int]: + keys = ("prompt_tokens", "completion_tokens", "cache_read_input_tokens", "cache_creation_input_tokens") + total = dict.fromkeys(keys, 0) + for u in usages: + if not isinstance(u, dict): + continue + for k in keys: + v = u.get(k) + if isinstance(v, (int, float)): + total[k] += int(v) + cached = (u.get("prompt_tokens_details") or {}).get("cached_tokens") if isinstance( + u.get("prompt_tokens_details"), dict) else None + if isinstance(cached, int) and not u.get("cache_read_input_tokens"): + total["cache_read_input_tokens"] += cached + return total diff --git a/tst/unit/task_step/test_deploy_user_sim_agent.py b/tst/unit/task_step/test_deploy_user_sim_agent.py index c92557584..c5dcf0fc2 100644 --- a/tst/unit/task_step/test_deploy_user_sim_agent.py +++ b/tst/unit/task_step/test_deploy_user_sim_agent.py @@ -21,6 +21,7 @@ from agent_env.task_step.task_steps.agent import user_sim_versions from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.agent.deploy_user_sim_agent import USER_SIMS_KEY, DeployUserSimAgentTaskStep +from agent_env.task_step.task_steps.agent.user_sim_common import persona_fingerprint from tst.unit.task_step.always_on_harness import USER_SIM, Rig DATA = Path(__file__).parents[2] / "data" / "user_sim" @@ -121,6 +122,38 @@ async def test_the_record_keeps_each_tag_s_line_so_an_old_used_resolves_after_th assert record["prompt_sha256"] == hashlib.sha256(PROMPT.encode()).hexdigest() +# The persona fingerprint, pinned: psi-benchmark-tasks's converter computes it too, and its test uses this sample. +FINGERPRINT_PERSONA = { + "display_name": "Hannah Reyes", + "description": {"who_you_are": ["Hannah Reyes, 32, a project manager in San Francisco."], + "decisions": ["A place at $180 a night or less: yes."]}, + "permissions": {"just_do": ["book flights"]}, + "keep_private": ["Your card number."], +} +FINGERPRINT_OPENING = "ok mom and dad are coming thu\u2013mon (nov 19\u201323), can you handle it?" +FINGERPRINT = "2d287ae4fd69f1f2398e223d7a3ed94976c912c825c606378859656b2ec2fdcd" + + +def test_the_persona_fingerprint_is_pinned(): + assert persona_fingerprint(FINGERPRINT_PERSONA, FINGERPRINT_OPENING) == FINGERPRINT + # The name and the reading settings aren't in it; the opening as written is. + assert persona_fingerprint({**FINGERPRINT_PERSONA, "display_name": "H. Reyes", "max_replies": 3}, + FINGERPRINT_OPENING) == FINGERPRINT + assert persona_fingerprint(FINGERPRINT_PERSONA, [{"kind": "text", "text": FINGERPRINT_OPENING}]) != FINGERPRINT + + +@pytest.mark.asyncio +async def test_the_record_has_the_persona_fingerprint_of_the_step_as_written(monkeypatch): + opening = [{"kind": "text", "text": "the trip, can you handle it?"}, + {"kind": "file", "file": {"uri": "s3://reyes-family//itinerary.pdf", "name": "itinerary.pdf"}}] + context = TaskStepContext() + context.metadata["seed"] = {"trip": "nov-19"} + record = await _published(monkeypatch, _step(opening_prompt=opening), context) + assert record["persona_sha256"] == persona_fingerprint(HANNAH, opening) # unseeded + assert (await _published(monkeypatch, _step()))["persona_sha256"] == persona_fingerprint( + HANNAH, PERSON["opening"]["text"]) + + @pytest.mark.asyncio async def test_a_one_sitting_user_gets_the_done_is_a_sign_off_line(monkeypatch): deploy = _Deploy(monkeypatch) diff --git a/tst/unit/task_step/verifiers/test_user_sim_judge.py b/tst/unit/task_step/verifiers/test_user_sim_judge.py new file mode 100644 index 000000000..4066699e3 --- /dev/null +++ b/tst/unit/task_step/verifiers/test_user_sim_judge.py @@ -0,0 +1,951 @@ +"""user_sim_judge: grades the user sim of an always_on run or a prompt_agent chat run, one narrow question per check, +by one judge model, with a fake model standing in for it. It records only, unless marking is turned on.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import replace + +import pytest + +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.registry import _builtin_registry +from agent_env.task_step.task_steps import always_on as ao +from agent_env.task_step.task_steps.agent import user_sim_versions +from agent_env.task_step.task_steps.agent.deploy_user_sim_agent import USER_SIMS_KEY +from agent_env.task_step.task_steps.verifiers import user_sim_judge as usj +from agent_env.task_step.task_steps.verifiers import user_sim_judge_core as core +from agent_env.task_step.task_steps.verifiers.agent_prompt_response_verifier import AgentPromptResponseVerifierTaskStep +from tst.unit.task_step.always_on_harness import Rig +from tst.unit.task_step.test_deploy_user_sim_agent import HANNAH, _Deploy +from tst.unit.task_step.test_deploy_user_sim_agent import _step as _deploy_step +from tst.unit.task_step.test_prompt_agent_built_in_user_sim import _context as _chat_context +from tst.unit.task_step.test_prompt_agent_built_in_user_sim import _step as _chat_step +from tst.unit.task_step.test_prompt_agent_built_in_user_sim import said +from tst.unit.task_step.test_prompt_agent_user_side import AGENT, USER, Peers, text + +JUDGE = "fake/judge" +OTHER_PERSONA = "0" * 64 # a persona fingerprint no run here has +TASK_ITEMS_Q = "task's own rubric items" + + +def _pass(**kw) -> str: + return json.dumps({"verdict": "pass", "could_change_grade": False, "reason": "fine", "turns": [], "items": [], + **kw}) + + +def _fail(turns=(1,), change=True, **kw) -> str: + return json.dumps({"verdict": "fail", "could_change_grade": change, "reason": "it broke the rule", + "turns": list(turns), "items": [], **kw}) + + +class FakeModel: + """Answers each call from ``script[check]``, a text, an exception, or a list of them used in turn; anything + unscripted passes. Records every call.""" + + def __init__(self, script=None): + self.script = dict(script or {}) + self.calls: list[dict] = [] + + async def __call__(self, model, system, parts, schema): + check = next(c for c in [*core.CHECK_NAMES, TASK_ITEMS_Q] if self._asks(parts[-1], c)) + self.calls.append({"model": model, "check": check, "system": system, "parts": parts, "schema": schema}) + got = self.script.get(check, _pass()) + if isinstance(got, list): + got = got.pop(0) + if isinstance(got, Exception): + raise got + return {"text": got, "usage": {"prompt_tokens": 100, "completion_tokens": 20}, "cost": 0.001} + + @staticmethod + def _asks(question: str, check: str) -> bool: + if check in core.CHECKS: + return question.startswith(f"# YOUR QUESTION: {core.CHECKS[check].title} (") + return question.startswith(f"# YOUR QUESTION: the {check}") + + def run_text(self) -> str: + """The run as the judge's first call read it.""" + return self.calls[0]["parts"][0] + "\n" + self.calls[0]["parts"][1] + + +def _run(*answers: tuple[str, str], graded=None) -> core.UserSimRun: + turns = [core.Turn(n=i, outcome=o, message=m, input=f"It's Mon. NEW FROM YOUR AI ASSISTANT: question {i}?", + graded=graded is None or i in graded) for i, (o, m) in enumerate(answers, 1)] + return core.UserSimRun(name="hannah", instructions="You are Hannah Reyes. K1 Dad is allergic to shellfish.", + turns=turns) + + +async def _judge(run, fake, **kw): + kw.setdefault("model", JUDGE) + kw.setdefault("checks", list(core.CHECK_NAMES)) + return await core.judge_run(run, complete=fake, **kw) + + +# ---- what the judge is asked -------------------------------------------------------------------------------------- +def test_each_check_gets_one_narrow_question_naming_its_families(): + for name, check in core.CHECKS.items(): + q = core.question_text(name, []) + assert q.startswith(f"# YOUR QUESTION: {check.title} ({', '.join(check.families)})\n{check.question}") + assert all(f in q for f in check.fails) and "Return only the JSON object." in q + assert sorted(core.FAMILY_OWNER) == ["A1", "A2", "A3", "A4", "F1", "F2", "F3", "F4", "G1", "G2", "G3", "G4", "G5"] + + +def test_self_contradiction_is_asked_under_made_up_not_guidance(): + line = "It contradicted something it said earlier in the run." + assert line in core.question_text("made_up", []) and "contradict" not in core.question_text("guidance", []) + + +def test_the_run_is_split_before_the_last_graded_answer_so_repeats_share_a_cached_prefix(): + run = _run(("reply", "yes book it"), ("no_reply", ""), ("reply", "no seafood pls"), graded={3}) + head, tail = core.render_run(run) + assert "\nYou are Hannah Reyes." in head + assert "## Turn 1 · CONTEXT ONLY" in head and 'message: "yes book it"' in head + assert "## Turn 3 · GRADED" in head and "question 3?" in head and "no seafood pls" not in head + assert tail == '### Its answer\noutcome: reply\nmessage: "no seafood pls"' + assert "message:" not in head.split("## Turn 2")[1].split("## Turn 3")[0] # a no_reply shows no empty message + + +def test_task_items_go_with_the_check_that_owns_their_family_and_the_rest_get_a_call_of_their_own(): + items = [core.RubricItem("leak", "G1", "never mentions the engagement"), + core.RubricItem("ends", "G4", "signs off only after the recap"), + core.RubricItem("tone", "F2", "stays lowercase")] + calls = core.plan_calls(["oversharing", "made_up"], items) + assert [i.id for i in calls["oversharing"]] == ["leak"] and calls["made_up"] == [] + assert [i.id for i in calls[core.TASK_ITEMS]] == ["ends", "tone"] # style isn't running, so "tone" goes there too + assert "- leak (G1: Leaking and privacy; general): never mentions the engagement" in core.question_text( + "oversharing", calls["oversharing"]) + + +def test_rubric_items_are_checked_when_the_task_loads(): + bad = [{"id": "a", "family": "Z9", "text": "x"}, {"id": "a", "family": "G1", "text": "y", "weight": 2}, {"id": ""}] + problems = core.item_problems(bad) + assert any("family 'Z9'" in p for p in problems) and any("used twice" in p for p in problems) + assert any("unknown fields ['weight']" in p for p in problems) and any("needs an id" in p for p in problems) + + +# ---- reading an answer -------------------------------------------------------------------------------------------- +def test_an_answer_is_checked_and_its_turns_kept_to_the_graded_ones(): + got = core.parse_answer(_fail(turns=[2, 3, "x"]), graded={3}, item_ids=[]) + assert (got["verdict"], got["turns"], got["turns_outside_graded"]) == ("fail", [3], [2]) + assert core.parse_answer(_pass(could_change_grade=True), {1}, [])["could_change_grade"] is False + for bad, why in ((json.dumps({"verdict": "maybe"}), "verdict"), ("no json here", "not a JSON object"), + (_pass(), "items \\['i1'\\] have no answer")): + with pytest.raises(ValueError, match=why): + core.parse_answer(bad, {1}, ["i1"] if "items" in why else []) + + +def test_a_fail_that_cites_no_graded_turn_is_not_usable_and_a_pass_needs_none(): + with pytest.raises(ValueError, match=r"a fail must cite at least one graded turn; cited \[99\], graded \[1\]"): + core.parse_answer(_fail(turns=[99]), {1}, []) + with pytest.raises(ValueError, match=r"a fail must cite at least one graded turn; cited \[\], graded \[1\]"): + core.parse_answer(_fail(turns=[]), {1}, []) + item = {"id": "i1", "verdict": "fail", "reason": "rated it 5", "turns": [99]} + with pytest.raises(ValueError, match=r"item i1: a fail must cite at least one graded turn; cited \[99\]"): + core.parse_answer(_fail(items=[item]), {1}, ["i1"]) + na = {"verdict": "n/a", "could_change_grade": False, "reason": "-", "turns": [99], "items": []} + for ok in (_pass(turns=[99]), json.dumps(na), _pass(items=[{**item, "verdict": "pass"}])): + assert core.parse_answer(ok, {1}, ["i1"] if "i1" in ok else [])["turns"] == [] + got = core.parse_answer(_fail(turns=[1, 99], items=[{**item, "turns": [1, 99]}]), {1}, ["i1"]) + assert (got["turns"], got["turns_outside_graded"], got["items"]["i1"]["turns"]) == ([1], [99], [1]) + + +def test_the_judge_writes_its_reason_before_its_verdict_within_the_word_limit(): + order = ["reason", "turns", "verdict", "could_change_grade", "items"] + assert list(core.OUTPUT_SCHEMA["properties"]) == order == core.OUTPUT_SCHEMA["required"] + assert list(core.OUTPUT_SCHEMA["properties"]["items"]["items"]["properties"]) == ["id", "reason", "turns", "verdict"] + shape = core.INSTRUCTIONS.split("Return only this JSON object")[1] + assert shape.index('"reason"') < shape.index('"turns"') < shape.index('"verdict"') < shape.index( + '"could_change_grade"') + item = shape.split('"items": [')[1] + assert item.index('"reason"') < item.index('"turns"') < item.index('"verdict"') + assert "Reason first and decide last." in core.INSTRUCTIONS and "at most 40 words" in core.INSTRUCTIONS + # The parser doesn't care which order the keys come in. + flipped = json.dumps({"items": [], "could_change_grade": True, "verdict": "fail", "turns": [1], "reason": "leaked"}) + assert core.parse_answer(flipped, {1}, [])["verdict"] == "fail" + + +# The endings of the two contradictory reasons seen in real runs, and a plain one. +SAYS_NO_FAILURE = ("Hannah flagged it anyway which is fine. No clear failure found.", + "This is not a delegation failure by Linda. No clear delegation failures found.", + "Every answer fits her rules; nothing fails.") + + +def test_a_fail_whose_reason_says_nothing_failed_is_not_usable(): + for reason in SAYS_NO_FAILURE: + with pytest.raises(ValueError, match="a fail whose reason says .* contradicts itself"): + core.parse_answer(_fail(reason=reason), {1}, []) + item = {"id": "i1", "verdict": "fail", "reason": reason, "turns": [1]} + with pytest.raises(ValueError, match="item i1: a fail whose reason says"): + core.parse_answer(_pass(items=[item]), {1}, ["i1"]) + # A pass may say it; so may an item that passes. + assert core.parse_answer(_pass(reason=reason, items=[{**item, "verdict": "pass"}]), {1}, ["i1"])["reason"] + # An ordinary fail, even one that clears some turns, is kept. + ok = "Turn 1 is fine, not a failure. At turn 2 she volunteered the shellfish allergy unasked; M3 says only when asked." + assert core.parse_answer(_fail(turns=[2], reason=ok), {1, 2}, [])["reason"] == ok + + +# ---- one judge ---------------------------------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_each_check_s_verdict_is_recorded_with_its_reason_and_turns(): + na = json.dumps({"verdict": "n/a", "could_change_grade": False, "reason": "no rules", "turns": [], "items": []}) + fake = FakeModel({"made_up": _fail(), "style": _fail(change=False), "delegation": na}) + result = await _judge(_run(("reply", "dad's allergic to shellfish fyi")), fake) + assert {k: v["verdict"] for k, v in result["checks"].items()} == { + "guidance": "pass", "oversharing": "pass", "delegation": "n/a", "style": "fail", "rogue": "pass", + "made_up": "fail", "ending": "pass"} + assert result["checks"]["made_up"] == {"verdict": "fail", "could_change_grade": True, "reason": "it broke the rule", + "turns": [1], "families": ["A2"]} + assert result["checks"]["delegation"]["reason"] == "no rules" + # Style failed, but it isn't an error check; made_up failed in a way that could change the grade. + assert result["user_sim_error"] == {"value": "yes", "because": ["made_up"], "unknown": []} + assert len(fake.calls) == 7 == result["calls"] and {c["model"] for c in fake.calls} == {JUDGE} + assert result["usage"]["prompt_tokens"] == 700 and result["cost"] == pytest.approx(0.007) + + +@pytest.mark.asyncio +async def test_each_call_shares_the_run_and_asks_one_question_and_the_first_call_runs_alone(): + fake = FakeModel() + await _judge(_run(("reply", "ok")), fake) + assert {c["parts"][0] for c in fake.calls} == {fake.calls[0]["parts"][0]} # one shared, cacheable prefix + assert all(c["system"] == core.INSTRUCTIONS and c["schema"] == core.OUTPUT_SCHEMA for c in fake.calls) + assert [c["check"] for c in fake.calls] == list(core.CHECK_NAMES) # the first alone, then the rest in order + + +@pytest.mark.asyncio +async def test_a_failure_that_can_t_change_the_grade_doesn_t_make_a_user_sim_error(): + result = await _judge(_run(("reply", "Here's a summary of what I booked:")), FakeModel({"rogue": _fail(change=False)})) + assert result["checks"]["rogue"]["verdict"] == "fail" and result["user_sim_error"]["value"] == "no" + + +@pytest.mark.asyncio +async def test_a_bad_answer_is_asked_again_once_and_a_check_whose_calls_keep_failing_is_an_error(): + fake = FakeModel({"guidance": ["not json", _fail()], "rogue": [RuntimeError("503"), RuntimeError("503")]}) + result = await _judge(_run(("reply", "you pick")), fake, checks=["guidance", "rogue"]) + assert result["checks"]["guidance"]["verdict"] == "fail" and result["checks"]["guidance"]["attempts"] == 2 + retry = [c for c in fake.calls if c["check"] == "guidance"][1]["parts"][-1] + assert "Your last answer could not be used (answer not usable: the answer is not a JSON object)" in retry + assert result["checks"]["rogue"] == {"verdict": "error", "error": "call failed: RuntimeError: 503", + "attempts": 2, "families": ["G3", "G5"]} + assert result["errors"] == ["rogue: call failed: RuntimeError: 503"] + assert result["user_sim_error"] == {"value": "yes", "because": ["guidance"], "unknown": ["rogue"]} + alone = await _judge(_run(("reply", "x")), FakeModel({"rogue": [RuntimeError("503")] * 2}), checks=["rogue"]) + assert alone["user_sim_error"] == {"value": "unknown", "because": [], "unknown": ["rogue"]} + + +@pytest.mark.asyncio +async def test_a_fail_citing_only_ungraded_turns_is_asked_again_and_twice_is_an_error(): + fake = FakeModel({"made_up": [_fail(turns=[99]), _fail(turns=[1])]}) + result = await _judge(_run(("reply", "dad's allergic to shellfish")), fake, checks=["made_up"]) + assert result["checks"]["made_up"]["verdict"] == "fail" and result["checks"]["made_up"]["turns"] == [1] + assert result["checks"]["made_up"]["attempts"] == 2 and result["user_sim_error"]["value"] == "yes" + assert "a fail must cite at least one graded turn; cited [99], graded [1]" in fake.calls[1]["parts"][-1] + twice = await _judge(_run(("reply", "x")), FakeModel({"made_up": [_fail(turns=[99])] * 2}), checks=["made_up"]) + assert twice["checks"]["made_up"]["verdict"] == "error" and "cited [99]" in twice["checks"]["made_up"]["error"] + assert twice["user_sim_error"] == {"value": "unknown", "because": [], "unknown": ["made_up"]} + + +@pytest.mark.asyncio +async def test_a_fail_whose_reason_says_no_clear_failure_found_is_asked_again(): + contradicts = _fail(reason=SAYS_NO_FAILURE[0]) + fake = FakeModel({"guidance": [contradicts, _pass(reason="No clear failure found.")], "oversharing": _fail()}) + result = await _judge(_run(("reply", "dad's allergic to shellfish fyi")), fake, checks=["guidance", "oversharing"]) + assert result["checks"]["guidance"] == {"verdict": "pass", "could_change_grade": False, + "reason": "No clear failure found.", "turns": [], "attempts": 2, + "families": ["A1", "A3"]} + retry = [c for c in fake.calls if c["check"] == "guidance"][1]["parts"][-1] + assert "a fail whose reason says 'No clear failure found' contradicts itself" in retry + # The ordinary fail is answered once and alone makes the run a user sim error. + assert "attempts" not in result["checks"]["oversharing"] and result["checks"]["oversharing"]["verdict"] == "fail" + assert result["user_sim_error"] == {"value": "yes", "because": ["oversharing"], "unknown": []} + twice = await _judge(_run(("reply", "x")), FakeModel({"delegation": [contradicts] * 2}), checks=["delegation"]) + assert twice["checks"]["delegation"]["verdict"] == "error" + assert twice["user_sim_error"] == {"value": "unknown", "because": [], "unknown": ["delegation"]} + + +@pytest.mark.asyncio +async def test_a_task_item_fail_citing_no_graded_turn_is_asked_again_and_twice_is_an_error(): + items = [core.RubricItem("rating", "A5", "rates the trip 2 or lower", can_error=True)] + + def answer(turns): + return json.dumps({"verdict": "fail", "could_change_grade": False, "reason": "rated it 5", "turns": [1], + "items": [{"id": "rating", "verdict": "fail", "reason": "rated it 5", "turns": turns}]}) + + fake = FakeModel({TASK_ITEMS_Q: [answer([99]), answer([1])]}) + result = await _judge(_run(("done", "5 stars")), fake, checks=[], items=items) + assert result["items"]["rating"]["verdict"] == "fail" and result["items"]["rating"]["turns"] == [1] + assert result["user_sim_error"]["because"] == ["item:rating"] and len(fake.calls) == 2 + twice = await _judge(_run(("done", "5 stars")), FakeModel({TASK_ITEMS_Q: [answer([])] * 2}), checks=[], + items=items) + assert twice["items"]["rating"]["verdict"] == "error" and "item rating: a fail" in twice["items"]["rating"]["error"] + assert twice["user_sim_error"] == {"value": "unknown", "because": [], "unknown": ["item:rating"]} + + +@pytest.mark.asyncio +async def test_task_items_get_their_own_verdicts_and_one_that_can_error_makes_a_user_sim_error(): + items = [core.RubricItem("leak", "G1", "never hints at the engagement"), + core.RubricItem("rating", "A5", "rates the trip 2 or lower", can_error=True)] + item_fail = json.dumps({"verdict": "fail", "could_change_grade": False, "reason": "rated it 5", + "turns": [1], "items": [{"id": "rating", "verdict": "fail", "reason": "rated it 5", + "turns": [1]}]}) + leak_pass = _pass(items=[{"id": "leak", "verdict": "pass", "reason": "-", "turns": []}]) + fake = FakeModel({TASK_ITEMS_Q: item_fail, "oversharing": leak_pass}) + result = await _judge(_run(("done", "thx bye")), fake, items=items) + assert result["items"]["leak"] == {"verdict": "pass", "reason": "-", "turns": [], "family": "G1", + "kind": "general", "asked_with": "oversharing"} + assert result["items"]["rating"] == {"verdict": "fail", "reason": "rated it 5", "turns": [1], "family": "A5", + "kind": "general", "asked_with": "task_items"} + assert "task_items" not in result["checks"] and result["user_sim_error"]["because"] == ["item:rating"] + assert len(fake.calls) == 8 + + +@pytest.mark.asyncio +async def test_the_ending_check_asks_when_it_signed_off_and_a_grade_changing_fail_makes_a_user_sim_error(): + q = core.question_text("ending", []) + assert "while the assistant was waiting on its answer" in q and "WHEN YOU'RE DONE" in q + assert "A short sign-off when its reply limit is reached." in q and "never said \"done\"" in q + assert "ending" in core.DEFAULT_ERROR_CHECKS + na = json.dumps({"verdict": "n/a", "could_change_grade": False, "reason": "never said done", "turns": [], + "items": []}) + for answer, verdict, error in ((_fail(turns=[1]), "fail", "yes"), (_pass(), "pass", "no"), (na, "n/a", "no")): + result = await _judge(_run(("done", "ok thx bye")), FakeModel({"ending": answer}), checks=["ending"]) + assert result["checks"]["ending"]["verdict"] == verdict and result["checks"]["ending"]["families"] == ["G4"] + assert result["user_sim_error"]["value"] == error + + +def test_the_judge_is_told_when_the_run_ended_at_the_user_sim_s_reply_limit(): + turns = _run(("reply", "swap it"), ("no_reply", ""), ("reply", "ok"), ("reply", "thx")).turns + [note] = usj._reply_limit_note(turns, 2) + assert note.startswith("the run ended at the user sim's reply limit: its reply at turn 3 was the last of its 2") + assert usj._reply_limit_note(turns, 4) == usj._reply_limit_note(turns, None) == [] + + +@pytest.mark.asyncio +async def test_a_run_with_no_graded_turn_is_not_sent_to_the_judge(): + fake = FakeModel() + result = await _judge(_run(("reply", "ok"), graded=set()), fake) + assert result["skipped"] == "no graded turns" and fake.calls == [] + + +# ---- the step on an always_on run --------------------------------------------------------------------------------- +ALWAYS_ON_ANSWERS = ['{"outcome": "reply", "message": "dad is allergic to shellfish, swap it pls", "used": "K3"}', + '{"outcome": "reply", "message": "all good, thx", "used": "none"}'] + + +async def _always_on(monkeypatch, tmp_path, answers=ALWAYS_ON_ANSWERS): + """An always_on run whose user the real deploy_user_sim_agent step deployed and recorded; the prompt it set on the + user sim's agent; the Rig.""" + def agent(n, texts, rig): + if n == 1: + rig.message("Saturday is Saltwater Shack, a seafood place. Keep it?") + return "sent" + if n == 2: + rig.message("Booked Pell & Finch instead. Anything else?") + return "sent" + rig.quiet() + return "" + + deploy = _Deploy(monkeypatch) + deployed = await deploy.run(_deploy_step(agent_name="user-sim", persona=HANNAH)) + [prompt] = deploy.fields("system_prompt") + rig = Rig(monkeypatch, tmp_path, agent=agent, plugin=user_sim_versions, user_texts=list(answers)) + await rig.arm() + ctx = rig.context() + ctx.metadata[USER_SIMS_KEY] = {"user-sim": deployed.metadata[USER_SIMS_KEY]["user-sim"]} + ctx.metadata["clock_configurations"] = [{"step_id": "clock", "timezone": "America/New_York"}] + await ao.AlwaysOnTaskStep("week", None, user_sim="user-sim", prompt_id="ask", agent_name="solver", + end_time="P3D").execute(ctx) + return rig, ctx, prompt + + +def _step(fake: FakeModel, monkeypatch, **kw) -> usj.UserSimJudgeTaskStep: + step = usj.UserSimJudgeTaskStep("judge_user_sim", None, **{"judge_model": JUDGE, "verifier_id": "usj", **kw}) + monkeypatch.setattr(step, "_complete", lambda context: fake) + return step + + +def _entry(ctx: TaskStepContext) -> dict: + return ctx.metadata["user_sim_judge"]["usj"] + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_on_an_always_on_run_the_judge_reads_what_the_user_sim_got_and_said_and_only_records(monkeypatch, + tmp_path): + rig, ctx, prompt = await _always_on(monkeypatch, tmp_path) + fake = FakeModel({"guidance": _fail(turns=[2])}) + + await _step(fake, monkeypatch).execute(ctx) + + run_text = fake.run_text() + assert "\n" + prompt.strip() + "\n" in run_text # the very prompt it ran with + assert len(rig.user_calls) == 2 + for call in rig.user_calls: # and every check-in word for word + assert "\n" + call["parts"][0]["text"].strip() + "\n" in run_text + assert '"dad is allergic to shellfish, swap it pls"' in run_text and "K3" not in run_text.split("")[1] + entry = _entry(ctx) + assert (entry["run_step_id"], entry["run_kind"], entry["user_sim"], entry["judge_model"]) == ( + "week", "always_on", "user-sim", JUDGE) + assert entry["instructions"]["matches"] is True and entry["notes"] == [] + assert {k: v["verdict"] for k, v in entry["checks"].items()} == {**dict.fromkeys(core.CHECK_NAMES, "pass"), + "guidance": "fail"} + assert entry["checks"]["guidance"]["turns"] == [2] + # A grade-changing failure, recorded only: nothing a grader reads changes. + assert entry["user_sim_error"] == {"value": "yes", "because": ["guidance"], "unknown": [], "marked": False, + "not_marked": "mark_user_sim_error is off: verdicts are recorded only"} + assert ctx.prompt_responses[-1].error_type is None and "user_sim_judge" not in rig.record(ctx) + assert "verifications" not in ctx.metadata # never mixed into the assistant's scores + assert ctx.metadata["models"]["user sim judge"]["judge_user_sim"] == JUDGE + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_without_the_stored_timeline_the_judge_reads_the_tail_and_is_told_what_s_missing(monkeypatch, tmp_path): + rig, ctx, _ = await _always_on(monkeypatch, tmp_path) + record, timeline = rig.record(ctx), rig.timeline(ctx) + record.pop("object_url") + record["timeline_tail"] = [{k: v for k, v in e.items() if k not in ("input", "read")} for e in timeline] + fake = FakeModel() + + await _step(fake, monkeypatch, checks=["made_up"]).execute(ctx) + + head = fake.calls[0]["parts"][0] + assert "Note: the run's full record wasn't stored" in head and "### What it was sent\n(not recorded)" in head + assert "## Turn 1 · 2026-11-09T14:00:00Z UTC · GRADED" in head + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_from_the_tail_an_answer_always_on_shortened_is_marked_cut(monkeypatch, tmp_path): + from agent_env.task_step.task_steps.always_on import PENDING_TEXT_CHARS, _small + + record = {"timeline_tail": [_small({"event": "user_sim_read", "outcome": "reply", "message": "ok " * 200}), + _small({"event": "user_sim_read", "outcome": "reply", "message": "sounds good"})]} + reads, notes = usj._always_on_reads(record) + assert [r["message_cut"] for r in reads] == [True, False] and len(reads[0]["message"]) == PENDING_TEXT_CHARS + 1 + assert usj._turn(1, reads[0]).cut and not usj._turn(2, reads[1]).cut + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_a_run_that_ended_in_an_error_is_not_judged(monkeypatch, tmp_path): + rig, ctx, _ = await _always_on(monkeypatch, tmp_path, answers=[]) + rig.record(ctx)["error_type"] = "agent_turn_failed" + fake = FakeModel() + + await _step(fake, monkeypatch, mark_user_sim_error=True).execute(ctx) + + assert fake.calls == [] and _entry(ctx)["skipped"] == "the run ended with error_type=agent_turn_failed" + assert _entry(ctx)["user_sim_error"] == {"value": "no", "because": [], "unknown": [], "marked": False} + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_a_run_s_step_params_choose_the_checks_and_the_model(monkeypatch, tmp_path): + _, ctx, _ = await _always_on(monkeypatch, tmp_path) + ctx.metadata["user_overrides"] = {"step_params": {"judge_user_sim": {"checks": ["rogue"], "judge_model": "other"}}} + fake = FakeModel() + + await _step(fake, monkeypatch).execute(ctx) + + assert {(c["check"], c["model"]) for c in fake.calls} == {("rogue", "other")} + assert _entry(ctx)["error_checks"] == ["rogue"] and _entry(ctx)["judge_model"] == "other" + + +# ---- the step on a prompt_agent chat run -------------------------------------------------------------------------- +async def _chat(monkeypatch): + Peers(monkeypatch, {AGENT: [text("draft one"), text("draft two"), text("glad to help")], + USER: [said("reply", "make it shorter", "D1"), said("done", "perfect, thanks!")]}) + ctx = await _chat_context(monkeypatch) + await _chat_step().execute(ctx) + return ctx + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_on_a_chat_run_the_judge_reads_each_stored_read_and_the_prompt_it_was_deployed_with(monkeypatch): + ctx = await _chat(monkeypatch) + us = ctx.metadata[USER_SIMS_KEY]["hannah"] + fake = FakeModel({"style": _fail(turns=[2], change=False)}) + + await _step(fake, monkeypatch, run_step_id="talk").execute(ctx) + + run_text = fake.run_text() + shown = run_text.split("\n")[1].split("\n")[0] + "\n" + assert hashlib.sha256(shown.encode()).hexdigest() == us["prompt_sha256"] # the prompt deploy_user_sim_agent set + for summary in ctx.metadata["user_sim_reads"]["talk"]: + full = usj._load(summary["object_url"]) + assert "\n" + full["input"].strip() + "\n" in run_text + assert 'outcome: reply\nmessage: "make it shorter"' in run_text and "D1" not in run_text.split("")[1] + assert 'outcome: done\nmessage: "perfect, thanks!"' in run_text + entry = _entry(ctx) + assert (entry["run_kind"], entry["user_sim"], entry["instructions"]["matches"]) == ("prompt_agent", "hannah", True) + assert entry["checks"]["style"]["verdict"] == "fail" and entry["user_sim_error"]["value"] == "no" + assert entry["graded_turns"] == [1, 2] and ctx.prompt_responses[-1].error_type is None + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_with_marking_turned_on_a_user_sim_error_marks_the_prompt_so_later_graders_skip_it(monkeypatch): + ctx = await _chat(monkeypatch) + + await _step(FakeModel({"oversharing": _fail()}), monkeypatch, mark_user_sim_error=True).execute(ctx) + + assert _entry(ctx)["user_sim_error"] == {"value": "yes", "because": ["oversharing"], "unknown": [], + "marked": True} + response = ctx.prompt_responses[-1] + assert response.error_type == "user_sim_error" + assert response.error_message == "The user sim judge found a user sim error: oversharing" + grader = AgentPromptResponseVerifierTaskStep("grade", None, prompt_id=response.prompt_id, verifier_id="grade", + criteria=[{"type": "response_contains", "needles": ["glad"]}]) + await grader.execute(ctx) + assert ctx.metadata["verifications"]["grade"]["results"][0]["message"] == "Skipped: prompt had error_type=user_sim_error" + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_marking_can_be_turned_on_for_one_run_by_its_step_params(monkeypatch): + ctx = await _chat(monkeypatch) + ctx.metadata["user_overrides"] = {"step_params": {"judge_user_sim": {"mark_user_sim_error": True}}} + + await _step(FakeModel({"rogue": _fail()}), monkeypatch).execute(ctx) + + assert _entry(ctx)["user_sim_error"]["marked"] is True and ctx.prompt_responses[-1].error_type == "user_sim_error" + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_instructions_that_don_t_match_the_recorded_hash_are_flagged_and_never_mark_the_run(monkeypatch): + ctx = await _chat(monkeypatch) + us = ctx.metadata[USER_SIMS_KEY]["hannah"] + us["description"] = {**us["description"], "who_you_are": [*us["description"]["who_you_are"], "A changed line."]} + fake = FakeModel({"guidance": _fail()}) + + await _step(fake, monkeypatch, mark_user_sim_error=True).execute(ctx) + + entry = _entry(ctx) + assert entry["instructions"]["matches"] is False + assert entry["instructions"]["recorded_prompt_sha256"] == us["prompt_sha256"] != entry["instructions"]["prompt_sha256"] + assert "Note: these instructions were rebuilt from the user sim's settings and differ from the ones it got" in ( + fake.calls[0]["parts"][0]) + assert entry["user_sim_error"]["value"] == "yes" and entry["user_sim_error"]["marked"] is False + assert "isn't confirmed" in entry["user_sim_error"]["not_marked"] and ctx.prompt_responses[-1].error_type is None + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_a_chat_run_that_ended_in_an_error_or_where_the_user_sim_never_read_is_not_judged(monkeypatch): + ctx = await _chat(monkeypatch) + ctx.prompt_responses[-1].error_type = "user_sim_error" + fake = FakeModel() + await _step(fake, monkeypatch).execute(ctx) + assert fake.calls == [] and _entry(ctx)["skipped"] == "the run ended with error_type=user_sim_error" + + ctx.prompt_responses[-1].error_type = None + ctx.metadata.pop("user_sim_reads") + await _step(fake, monkeypatch, run_step_id="talk").execute(ctx) + assert fake.calls == [] and _entry(ctx)["skipped"] == "the user sim never read: the run has no user_sim_reads" + + +def test_without_run_step_id_the_run_must_have_one_run_to_judge(): + ctx = TaskStepContext() + ctx.metadata = {"always_on": {"week": {}}, "user_sim_reads": {"talk": [{}]}} + with pytest.raises(RuntimeError, match=r"set run_step_id; this run has 2 runs .*\['talk', 'week'\]"): + usj.UserSimJudgeTaskStep("j", None)._find_run(ctx) + with pytest.raises(RuntimeError, match="step 'nope' left no run to judge"): + usj.UserSimJudgeTaskStep("j", None, run_step_id="nope")._find_run(ctx) + assert usj.UserSimJudgeTaskStep("j", None, run_step_id="talk")._find_run(ctx) == ("prompt_agent", "talk") + + +# ---- the instructions, rebuilt as deploy_user_sim_agent wrote them --------------------------------------------------- +SEED = {"trip": "nov-19"} +TEXT = "ok mom and dad are officially coming thu-mon (nov 19-23), can you handle it?" +ITINERARY = {"kind": "file", "file": {"uri": "s3://reyes-family//itinerary.pdf", "name": "itinerary.pdf", + "mimeType": "application/pdf"}, + "metadata": {"description": "Mom and Dad's flight itinerary"}} +SHORTLIST = {"kind": "file", "file": {"uri": "s3://reyes-family//-shortlist.xlsx"}} +TOLD_AND_KNOWN = {**HANNAH["description"], "already_told": ["Dad can't do stairs since his knee surgery."], + "should_already_know": ["You never book anything before 9am for your parents."]} + + +@pytest.mark.parametrize("fields, shows", [ + pytest.param({"opening_prompt": [{"kind": "text", "text": TEXT}, ITINERARY]}, + ["(you attached itinerary.pdf: Mom and Dad's flight itinerary)", user_sim_versions.ATTACHED_KNOWN], + id="a-described-file"), + pytest.param({"opening_prompt": [{"kind": "text", "text": TEXT}, SHORTLIST]}, + ["(you attached nov-19-shortlist.xlsx)", user_sim_versions.ATTACHED_KNOWN], id="an-undescribed-file"), + pytest.param({"persona": {**HANNAH, "description": TOLD_AND_KNOWN}}, + [" T1 Dad can't do stairs", " C1 You never book anything before 9am"], id="told-and-known"), + pytest.param({"opening_prompt": TEXT, "multi_day": False}, [], id="text-only"), +]) +@pytest.mark.asyncio +async def test_the_rebuilt_instructions_are_the_ones_deployed(monkeypatch, fields, shows): + deploy = _Deploy(monkeypatch) + ctx = TaskStepContext() + ctx.metadata["seed"] = dict(SEED) + await deploy.run(_deploy_step(**fields), ctx) + us = ctx.metadata[USER_SIMS_KEY]["hannah"] + + prompt, check, notes = usj._instructions(us, SEED, None) + + assert prompt == deploy.fields("system_prompt")[0] and all(line in prompt for line in shows) + assert check["matches"] is True and check["prompt_sha256"] == us["prompt_sha256"] and notes == [] + + +@pytest.mark.asyncio +async def test_a_text_only_opening_rebuilds_as_before(monkeypatch): + deploy = _Deploy(monkeypatch) + ctx = TaskStepContext() + await deploy.run(_deploy_step(), ctx) + us = ctx.metadata[USER_SIMS_KEY]["hannah"] + assert "opening_parts" not in us + person = {"key": "hannah", "name": HANNAH["display_name"], "first": HANNAH["first_name"], + "description": HANNAH["description"], "permissions": HANNAH["permissions"], + "keep_private": HANNAH["keep_private"], "opening": {"text": us["opening"]}, "done_ends_run": False} + prompt, check, _ = usj._instructions(us, {}, None) + assert prompt == user_sim_versions.setup("ours", person)["system_prompt"] == deploy.fields("system_prompt")[0] + assert check["matches"] is True + + +# ---- settings ----------------------------------------------------------------------------------------------------- +def test_settings_are_checked_and_round_trip(monkeypatch): + assert "user_sim_judge" in _builtin_registry() + step = usj.UserSimJudgeTaskStep("s", None, run_step_id="week", judge_model=JUDGE, checks=["guidance", "style"], + error_checks=["guidance"], mark_user_sim_error=True, + rubric_items=[{"id": "leak", "family": "G1", "text": "never mentions the ring"}]) + assert step.config_problems() == [] + again = usj.UserSimJudgeTaskStep.from_dict(json.loads(json.dumps(step.to_dict()))) + assert again.to_dict() == step.to_dict() + problems = usj.UserSimJudgeTaskStep("s", None, checks=["guidance", "vibes"], error_checks=["rogue"], judge_model=" ", + mark_user_sim_error="yes", + rubric_items=[{"id": "x", "family": "Q1", "text": "y"}]).config_problems() + assert any("unknown checks ['vibes']" in p for p in problems) + assert any("error_checks names checks that don't run: ['rogue']" in p for p in problems) + assert any("judge_model must be a model name" in p for p in problems) and any("family 'Q1'" in p for p in problems) + assert any("mark_user_sim_error must be true or false" in p for p in problems) + assert usj.UserSimJudgeTaskStep("s", None, checks=["style"]).error_checks == [] # the error checks that run + + +def test_by_default_it_records_only_never_fails_the_task_and_judges_with_the_verifiers_default_model(monkeypatch): + step = usj.UserSimJudgeTaskStep.from_dict({"id": "s", "type": "user_sim_judge"}) + assert (step.mark_user_sim_error, step.fail_task_on_error, step.judge_model) == (False, False, None) + assert step.to_dict()["judge_model"] is None # resolved when it runs, so the task doesn't pin it + class Config: + role = None + + def get_model_for_role(self, role): + assert role == "judge" + return self.role + + fake = Config() + monkeypatch.setattr("agent_env.config.get_config", lambda: fake) + assert step._settings(TaskStepContext())["model"] == "claude-sonnet-4-6" + fake.role = "anthropic/claude-opus-4-8" + assert step._settings(TaskStepContext())["model"] == "anthropic/claude-opus-4-8" + + +def test_the_task_must_say_which_run_when_it_has_more_than_one_and_name_a_run_step(): + week = ao.AlwaysOnTaskStep("week", None, user_sim="u", prompt_id="ask", agent_name="solver") + talk = _chat_step() + other = AgentPromptResponseVerifierTaskStep("grade", None, prompt_id="ask") + assert usj.UserSimJudgeTaskStep("j", None).task_problems([week]) == [] + assert "set run_step_id" in usj.UserSimJudgeTaskStep("j", None).task_problems([week, talk])[0] + assert usj.UserSimJudgeTaskStep("j", None, run_step_id="talk").task_problems([week, talk]) == [] + assert "of step 'grade' (agent_prompt_response_verifier)" in usj.UserSimJudgeTaskStep( + "j", None, run_step_id="grade").task_problems([week, other])[0] + + +def test_the_step_asks_anthropic_models_to_cache_the_run(): + assert usj._anthropic("anthropic/claude-opus-4-8") and usj._anthropic("claude-opus-4-8") + assert not usj._anthropic("openai/gpt-5.5") + + +def test_a_fail_that_clears_one_turn_and_fails_another_is_kept(): + text = json.dumps({"reason": "No clear failure at turn 1; at turn 2 she revealed the private password.", + "turns": [1], "verdict": "fail", "could_change_grade": True, "items": []}) + parsed = core.parse_answer(text, {1}, []) + assert parsed["verdict"] == "fail" and parsed["could_change_grade"] is True + + +# ---- task rubric items in usersim.yaml's shape ---------------------------------------------------------------------- +# From psi-ts-craftweb-holiday-rush's usersim.yaml, as psi-benchmark-tasks turns it into this step's settings. +CALL_RAE = { + "id": "a1_call_rae", "family": "A1", "kind": "hard", + "name": "Rae: yes if it fits, ≥ $15 a jar and ≥ $9 a tin, by email", + "short": "Both floors, 'if it fits', settled by email. Missing a floor, a different floor, or 'text her' fails.", + "check": "If the assistant asks about Rae's order, Tess says yes only if it fits, gives BOTH floors (no less than $15 " + "a jar and $9 a tin), and lets the assistant settle it with Rae by email. A different floor, a missing " + "floor, a flat yes or no, or \"text her\" fails.", + "na_when": "The assistant didn't raise Rae's order."} +NO_SILENT_FIX = { + "id": "g5_no_silent_fix", "family": "G5", "kind": "general", "name": "Doesn't do the assistant's homework", + "check": "When the assistant's plan has a gap, Tess gives her call rather than doing the work for it: she doesn't " + "work out how many jars Rae should get, recompute the budget, choose pour dates, or rewrite its plan.", + "na_when": "never"} +HER_LENGTH = {"id": "f2_her_length", "family": "F2", "kind": "auto", "name": "As short as her real texts", + "check": "Every reply is at most 25 words.", "metric": "max_words", "max": 25} +SATISFACTION = {"id": "a5_satisfaction", "family": "A5", "kind": "hard", + "name": "Satisfaction rating matches the outcome", + "check": "Not gradeable: the sim gives no rating.", "na_when": "always", + "static_na_reason": "No deployed version of the sim prompt has a satisfaction step."} +# Tess's real texts, from the same file's style_reference. +TESS = ("Never again. I ran out of oil and refunded 22 people at Christmas. Two 1-star reviews already 😩", + "Saw it! Let me look at my numbers this weekend.") + + +def _items(*dicts) -> list[core.RubricItem]: + assert core.item_problems(list(dicts)) == [] + return [core.RubricItem.from_dict(d) for d in dicts] + + +def _answer_items(verdicts: dict[str, str], turns=(1,)): + """A complete() that answers every item it's asked, by id, with ``verdicts`` (else pass).""" + calls = [] + + async def complete(model, system, parts, schema): + asked = re.findall(r"^- (\S+) \([AFG]\d: ", parts[-1], re.M) + calls.append(asked) + items = [{"id": i, "reason": f"{i} per turn 1", "turns": list(turns), "verdict": verdicts.get(i, "pass")} + for i in asked] + return {"text": _pass(items=items), "usage": None, "cost": 0.0} + + complete.calls = calls + return complete + + +def test_the_craftweb_item_round_trips_and_is_asked_as_written(): + [item] = _items(CALL_RAE) + assert item.to_dict() == CALL_RAE and core.RubricItem.from_dict(item.to_dict()) == item + q = core.question_text("guidance", [item]) + assert f"- a1_call_rae (A1: Requirement provision; hard): {CALL_RAE['check']}" in q + assert " n/a when: The assistant didn't raise Rae's order." in q + assert CALL_RAE["name"] not in q and CALL_RAE["short"] not in q # for reviewers, not the judge + step = usj.UserSimJudgeTaskStep("s", None, rubric_items=[CALL_RAE, HER_LENGTH, SATISFACTION], + rubric_persona_sha256=OTHER_PERSONA) + again = usj.UserSimJudgeTaskStep.from_dict(json.loads(json.dumps(step.to_dict()))) + assert step.config_problems() == [] and again.to_dict() == step.to_dict() + assert (again.rubric_items, again.rubric_persona_sha256) == ([CALL_RAE, HER_LENGTH, SATISFACTION], OTHER_PERSONA) + + +def test_the_old_text_field_still_loads_as_the_check_and_kind_defaults_to_general(): + old = {"id": "leak", "family": "g1", "text": "never mentions the ring"} + [item] = _items(old) + assert (item.family, item.kind, item.check) == ("G1", "general", "never mentions the ring") + assert item.to_dict() == {"id": "leak", "family": "G1", "kind": "general", "check": "never mentions the ring"} + assert usj.UserSimJudgeTaskStep("s", None, rubric_items=[old]).config_problems() == [] + [both] = core.item_problems([{**old, "check": "never mentions the ring"}]) + assert both == "rubric_items[0] (leak): give check or text, not both (text is the old name of check)" + + +def test_hard_and_general_items_are_told_to_the_judge_by_kind(): + hard, general = _items(CALL_RAE, {**NO_SILENT_FIX, "family": "A3"}) + q = core.question_text("guidance", [hard, general]) + assert core.KIND_RULES["hard"] in q and "fails on any breach" in q + assert core.KIND_RULES["general"] in q and "fails only on a clear instance" in q + assert "- g5_no_silent_fix (A3: Decisions and preferences; general):" in q and " n/a is not allowed" in q + only_general = core.question_text("guidance", [general]) + assert core.KIND_RULES["hard"] not in only_general and core.KIND_RULES["general"] in only_general + + +@pytest.mark.asyncio +async def test_an_auto_item_is_counted_on_the_graded_answers_with_no_model_call(): + [length] = _items(HER_LENGTH) + # Too long for Tess: the length a model playing her drifts to. + over = ("Yes to Rae if it fits, but no less than $15 a jar and $9 a tin. Settle it with her by email, and let me know " + "what she says before you confirm anything.") + run = _run(("reply", TESS[0]), ("no_reply", ""), ("reply", over), ("reply", TESS[1])) + fake = FakeModel() + result = await _judge(run, fake, checks=[], items=[length]) + assert fake.calls == [] and result["calls"] == 0 + assert result["items"]["f2_her_length"] == { + "verdict": "fail", "reason": "over the 25 words max at turn 3: 34", "turns": [3], "family": "F2", "kind": "auto", + "asked_with": "computed"} + passed = await _judge(_run(("reply", TESS[0]), ("reply", TESS[1])), fake, checks=[], items=[length]) + assert passed["items"]["f2_her_length"]["verdict"] == "pass" and fake.calls == [] + assert passed["items"]["f2_her_length"]["reason"] == "every answer at most 25 words; the longest, turn 1, has 18" + # A context-only turn is never failed; a run with no graded message is n/a. + context_only = await _judge(_run(("reply", over), ("reply", TESS[1]), graded={2}), fake, checks=[], items=[length]) + assert context_only["items"]["f2_her_length"]["verdict"] == "pass" + silent = await _judge(_run(("no_reply", "")), fake, checks=[], items=[length]) + assert silent["items"]["f2_her_length"]["verdict"] == "n/a" + # With the checks running, the auto item is in no question. + fake = FakeModel() + await _judge(_run(("reply", over)), fake, items=[length]) + assert all("f2_her_length" not in c["parts"][-1] for c in fake.calls) and len(fake.calls) == 7 + + +@pytest.mark.asyncio +async def test_an_auto_item_can_t_pass_an_answer_the_record_kept_only_the_start_of(): + [length] = _items(HER_LENGTH) + short = _run(("reply", TESS[0]), ("reply", TESS[1])) + short.turns[1] = replace(short.turns[1], cut=True) # under 25 words as kept, but the whole answer may not be + result = await _judge(short, FakeModel(), checks=[], items=[length]) + assert result["items"]["f2_her_length"] == { + "verdict": "error", "turns": [2], "family": "F2", "kind": "auto", "asked_with": "computed", + "error": "only the start of the answer at turn 2 was kept, so its full length can't be counted"} + # Over the max as kept is over it whole: still a fail. + long = "word " * 30 + over = _run(("reply", long.strip()), ("reply", TESS[1])) + over.turns[0] = replace(over.turns[0], cut=True) + assert (await _judge(over, FakeModel(), checks=[], items=[length]))["items"]["f2_her_length"]["verdict"] == "fail" + + +@pytest.mark.asyncio +async def test_an_auto_item_counts_sentences_simply(): + assert [core.count_sentences(t) for t in TESS] == [3, 2] + assert core.count_sentences("no less than $15.50 a jar") == 1 # a decimal point ends nothing + assert core.count_sentences("ok\nsounds good") == 2 and core.count_sentences("😩") == 0 + assert core.count_words(TESS[0]) == 18 # the emoji is a word + [two] = _items({"id": "f2_short", "family": "F2", "kind": "auto", "metric": "max_sentences", "max": 2}) + result = await _judge(_run(("reply", TESS[1]), ("reply", TESS[0])), FakeModel(), checks=[], items=[two]) + assert result["items"]["f2_short"]["verdict"] == "fail" and result["items"]["f2_short"]["turns"] == [2] + assert result["items"]["f2_short"]["reason"] == "over the 2 sentences max at turn 2: 3" + + +@pytest.mark.asyncio +async def test_na_when_always_is_n_a_with_its_reason_and_no_call(): + [rating] = _items(SATISFACTION) + fake = FakeModel() + result = await _judge(_run(("done", "thx")), fake, checks=[], items=[rating]) + assert fake.calls == [] and result["items"]["a5_satisfaction"] == { + "verdict": "n/a", "reason": SATISFACTION["static_na_reason"], "turns": [], "family": "A5", "kind": "hard", + "asked_with": "static_na"} + fake = FakeModel() + await _judge(_run(("done", "thx")), fake, items=[rating]) + assert all("a5_satisfaction" not in c["parts"][-1] for c in fake.calls) + + +@pytest.mark.asyncio +async def test_na_when_never_doesn_t_allow_an_n_a_so_one_is_asked_again(): + [fix] = _items(NO_SILENT_FIX) + + def answer(verdict): + item = {"id": "g5_no_silent_fix", "reason": "she gave her call", "turns": [1], "verdict": verdict} + return _pass(items=[item]) + + fake = FakeModel({"rogue": [answer("n/a"), answer("pass")]}) + result = await _judge(_run(("reply", "your call on the dates, just not oct 31")), fake, checks=["rogue"], + items=[fix]) + assert result["items"]["g5_no_silent_fix"]["verdict"] == "pass" and len(fake.calls) == 2 + assert "item g5_no_silent_fix may not be n/a: it always applies" in fake.calls[1]["parts"][-1] + twice = await _judge(_run(("reply", "x")), FakeModel({"rogue": [answer("n/a")] * 2}), checks=["rogue"], + items=[fix]) + assert twice["items"]["g5_no_silent_fix"]["verdict"] == "error" + # Where n/a is allowed, it stands. + [rae] = _items(CALL_RAE) + na = _pass(items=[{"id": "a1_call_rae", "reason": "Rae never came up", "turns": [], "verdict": "n/a"}]) + got = await _judge(_run(("reply", "x")), FakeModel({"guidance": na}), checks=["guidance"], items=[rae]) + assert got["items"]["a1_call_rae"]["verdict"] == "n/a" + + +def test_a_big_rubric_is_split_over_calls_of_at_most_the_cap(): + assert core.MAX_ITEMS_PER_CALL == 12 + items = _items(*[{"id": f"g1_{n}", "family": "G1", "check": f"never states fact {n}"} for n in range(30)], + *[{"id": f"a5_{n}", "family": "A5", "check": f"rates outcome {n}"} for n in range(13)], HER_LENGTH) + calls = core.plan_calls(["oversharing", "rogue"], items) + assert {k: len(v) for k, v in calls.items()} == {"oversharing": 12, "oversharing#2": 12, "oversharing#3": 6, + "rogue": 0, "task_items": 12, "task_items#2": 1} + assert [i.id for i in calls["oversharing#3"]] == [f"g1_{n}" for n in range(24, 30)] + more = core.question_text("oversharing#2", calls["oversharing#2"]) + assert more.startswith("# YOUR QUESTION: the task's own rubric items (more of the Oversharing question's, G1, G2)") + assert core.CHECKS["oversharing"].question not in more # the check itself is asked once + + +@pytest.mark.asyncio +async def test_split_items_get_their_verdicts_and_the_check_is_asked_once(): + items = _items(*[{"id": f"g1_{n}", "family": "G1", "check": f"never states fact {n}"} for n in range(14)]) + complete = _answer_items({"g1_13": "fail"}) + result = await core.judge_run(_run(("reply", "x")), model=JUDGE, checks=["oversharing"], items=items, + complete=complete) + assert [len(c) for c in complete.calls] == [12, 2] and list(result["checks"]) == ["oversharing"] + assert result["items"]["g1_13"]["asked_with"] == "oversharing#2" and result["items"]["g1_13"]["verdict"] == "fail" + assert list(result["items"]) == [f"g1_{n}" for n in range(14)] # in the task's order + assert result["user_sim_error"]["because"] == ["item:g1_13"] # G1's owner, oversharing, is an error check + + +def test_each_bad_item_is_refused_when_the_task_loads_with_a_clear_error(): + cases = [ + ({"family": "A1", "check": "x"}, "rubric_items[0] needs an id"), + ({"id": "a", "check": "x"}, "rubric_items[0] (a) needs a family"), + ({"id": "a", "family": "B1", "check": "x"}, "rubric_items[0] (a): family 'B1' is not one of"), + ({"id": "a", "family": "A1", "kind": "strict", "check": "x"}, "kind 'strict' is not one of ['hard', 'general', " + "'auto']"), + ({"id": "a", "family": "F2", "kind": "auto", "max": 25}, "an auto item's metric must be one of"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_chars", "max": 25}, "not 'max_chars'"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_words"}, "an auto item needs max"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_words", "max": 0}, "not 0"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_words", "max": "25"}, "not '25'"), + ({"id": "a", "family": "F2", "kind": "general", "check": "x", "metric": "max_words", "max": 25}, + "metric and max are for kind auto only"), + ({"id": "a", "family": "A5", "check": "x", "na_when": "always"}, "na_when always needs a static_na_reason"), + ({"id": "a", "family": "A5", "check": "x", "static_na_reason": "y"}, "static_na_reason is only for na_when " + "always"), + ({"id": "a", "family": "A1", "kind": "hard", "check": " "}, "a hard item needs a check"), + ({"id": "a", "family": "A1"}, "a general item needs a check"), + ({"id": "a", "family": "A1", "check": "x", "na-when": "never"}, "unknown fields ['na-when']"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_words", "max": 25, "na_when": "never"}, + "auto items are counted, so na_when can only be 'always' (with a static_na_reason), not 'never'"), + ({"id": "a", "family": "F2", "kind": "auto", "metric": "max_words", "max": 25, + "na_when": "The assistant never asked her anything."}, + "auto items are counted, so na_when can only be 'always'"), + ] + for item, error in cases: + problems = core.item_problems([item]) + assert any(error in p for p in problems), (item, problems) + with pytest.raises(ValueError, match="user_sim_judge 's': "): + usj.UserSimJudgeTaskStep("s", None, rubric_items=[item])._settings(TaskStepContext()) + dup = core.item_problems([CALL_RAE, {**CALL_RAE, "family": "A3"}]) + assert dup == ["rubric_items[1] (a1_call_rae): id 'a1_call_rae' is used twice"] + for bad in (7, "3f1c", "A" * 64): + assert any("rubric_persona_sha256 must be a persona fingerprint" in p for p in usj.UserSimJudgeTaskStep( + "s", None, rubric_items=[CALL_RAE], rubric_persona_sha256=bad).config_problems()) + + +def test_an_auto_item_loads_with_no_na_when_or_with_always(): + auto = {"id": "a", "family": "F2", "kind": "auto", "metric": "max_words", "max": 25} + assert core.item_problems([auto]) == [] + assert core.item_problems([{**auto, "na_when": "always", "static_na_reason": "She never texts here."}]) == [] + + +# ---- the rubric's persona ------------------------------------------------------------------------------------------ +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_items_written_for_another_persona_aren_t_graded_and_the_checks_still_run(monkeypatch): + ctx = await _chat(monkeypatch) + run = ctx.metadata[USER_SIMS_KEY]["hannah"]["persona_sha256"] + fake = FakeModel({"guidance": _fail()}) + + await _step(fake, monkeypatch, rubric_items=[CALL_RAE, HER_LENGTH], rubric_persona_sha256=OTHER_PERSONA).execute(ctx) + + entry = _entry(ctx) + assert entry["rubric_persona"] == { + "rubric": OTHER_PERSONA, "run": run, "status": "mismatch", + "note": "the task items were written for a different persona or opening than this run's user sim had, so they " + "weren't graded; the general checks were"} + assert entry["items"] == {} and len(fake.calls) == 7 + assert all("a1_call_rae" not in c["parts"][-1] for c in fake.calls) + assert entry["checks"]["guidance"]["verdict"] == "fail" and entry["user_sim_error"]["because"] == ["guidance"] + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +@pytest.mark.parametrize("recorded, status", [(True, "match"), (False, "unknown")]) +async def test_items_for_this_persona_or_a_run_with_no_fingerprint_are_graded(monkeypatch, recorded, status): + ctx = await _chat(monkeypatch) + us = ctx.metadata[USER_SIMS_KEY]["hannah"] + fingerprint = us["persona_sha256"] if recorded else OTHER_PERSONA + if not recorded: + del us["persona_sha256"] # a run from before the step recorded it + + await _step(FakeModel(), monkeypatch, checks=["style"], rubric_items=[HER_LENGTH], + rubric_persona_sha256=fingerprint).execute(ctx) + + entry = _entry(ctx) + assert entry["rubric_persona"]["status"] == status and entry["items"]["f2_her_length"]["verdict"] == "pass" + assert entry["rubric_persona"]["rubric"] == fingerprint + if not recorded: + assert entry["rubric_persona"]["run"] is None and "graded as written" in entry["rubric_persona"]["note"] + # Without a rubric_persona_sha256 nothing is compared. + await _step(FakeModel(), monkeypatch, checks=["style"], rubric_items=[HER_LENGTH]).execute(ctx) + assert "rubric_persona" not in _entry(ctx) and _entry(ctx)["items"]["f2_her_length"]["verdict"] == "pass"