feat(verifiers): user_sim_judge grades the user sim of a run, record-only

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",

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,

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 ``<seed>`` 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 ``<seed>`` 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,

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."""