diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py index aff79e159..661190ef4 100644 --- a/src/agent_env/task/episode/rules.py +++ b/src/agent_env/task/episode/rules.py @@ -167,6 +167,9 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s "agent_prompt_response_verifier", "verify_mcp_env_assessment", # It skips an agent that isn't deployed. "teardown_sandboxes", + # It talks only to a user sim deploy_user_sim_agent deployed (its load check), which that step's check keeps from + # being played from outside. + "rate_user_satisfaction", # They act on an outside agent's proxy, in its sandbox, as on the agent; install_agent installs the proxy in its place. "add_skills", "install_agent", "load_artifact", "run_code", "verify_sandbox", }) diff --git a/src/agent_env/task_step/registry.py b/src/agent_env/task_step/registry.py index dca80e64c..94df3ede2 100644 --- a/src/agent_env/task_step/registry.py +++ b/src/agent_env/task_step/registry.py @@ -65,6 +65,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.always_on import AlwaysOnTaskStep from agent_env.task_step.task_steps.agent.deploy_user_sim_agent import DeployUserSimAgentTaskStep + from agent_env.task_step.task_steps.agent.rate_user_satisfaction import RateUserSatisfactionTaskStep from agent_env.task_step.task_steps.register_env_triggers import RegisterEnvTriggersStep from agent_env.task_step.task_steps.register_agent_triggers import RegisterAgentTriggersStep from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep @@ -139,6 +140,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: _get_type(PromptAgentTaskStep): PromptAgentTaskStep, _get_type(AlwaysOnTaskStep): AlwaysOnTaskStep, _get_type(DeployUserSimAgentTaskStep): DeployUserSimAgentTaskStep, + _get_type(RateUserSatisfactionTaskStep): RateUserSatisfactionTaskStep, _get_type(RegisterEnvTriggersStep): RegisterEnvTriggersStep, _get_type(RegisterAgentTriggersStep): RegisterAgentTriggersStep, _get_type(SyncEnvClockTaskStep): SyncEnvClockTaskStep, 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 5b222fbd8..3224d2fe6 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 @@ -10,6 +10,12 @@ --user-model``, a bundle's ``user_model``), else the step's ``model``. A run-wide ``a2a_agent_id`` is ignored, so the agent under test never plays the user; this step's own ``step_params`` can still change it. +``opening_prompt`` is text, or a list of A2A text and file parts in ``prompt_agent``'s ``parts`` format, for a first +message that attaches files (any type). A file part may carry a one-line ``metadata.description`` of what it is, for +the user sim, which can't open files; the agent is sent the file without it. The parts are checked when the task loads: +at least one text part with text, each part a valid text or file part, each description one line. The seed fills +````s in text and in a file's ``uri`` and ``name``, as in any prompt's parts. + What always_on reads from the record (``metadata["user_sims"][agent_name]``): - ``name`` (the agent name), ``display_name`` and ``first_name`` (else the display name's first word): how the agent's @@ -20,32 +26,40 @@ ``response_times`` need the task's clock (``sync_env_clock``): always_on refuses to start a run that sets either with no clock armed. The user's time zone is the task clock's. - ``description``, ``permissions``, ``keep_private``: in ``person``, for the reads. -- ``opening_prompt``: the user's first message, which always_on sends; always_on has no prompt of its own. +- ``opening`` (the opening's text parts, joined) and, for an ``opening_prompt`` given as parts, ``opening_parts`` + (those parts, descriptions included): both as written, the seed not yet filled in. The user's first message, which + always_on sends, its text as one labelled line where its first text part was and each file in its place; always_on + has no prompt of its own. - ``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. +- ``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. -The built-in user sim (``user_sim_versions``) is used directly; its interface, internal to the two steps: +The built-in user sim (``user_sim_versions``) is used directly; its interface, internal to the two steps and +rate_user_satisfaction: setup(version: str, person: dict) -> {"system_prompt": str | None, "output_format": dict | None} async answer(version: str, person: dict, history: list[dict], read: dict, ask) -> dict + async rate(version: str, person: dict, satisfaction: list[str], read: dict, ask) -> dict ``person`` is ``{"key": name, "name", "first", "time_zone", "description", "permissions", "keep_private", "opening", "done_ends_run"}``, with ``time_zone`` the run's zone (the persona's, else the task clock's, else UTC) and ``opening`` the user's opening -request, ``{"at": RFC 3339 UTC, "text"}``, so ``setup`` can write it into the system prompt (this step's has no -``at``). The opening is also the first line of the first read's ``you_sent``. ``done_ends_run`` is set before -``setup``, as each read has it, since the done line is in the system prompt; the built-in user sim uses the multi-day -wording when it's missing. +request, ``{"at": RFC 3339 UTC, "text"}``, with ``"files": [{"name", "description"}]`` when it attaches files (the +description only when the part has one, the name its file's ``name``, else the last segment of its ``uri``), so +``setup`` can write it into the system prompt (this step's has no ``at``). The opening is also the first line of the +first read's ``you_sent``. ``done_ends_run`` is set before ``setup``, as each read has it, since the done line is in +the system prompt; the built-in user sim uses the multi-day wording when it's missing. ``answer`` runs once per read, at the time the user would read. ``read`` is ``{"time": RFC 3339 UTC, "assistant_messages": [{"at", "text"}], "phone": [{"at", "channel", "text"}], "you_sent": [{"at", "text"}], "done_ends_run": bool}``: the agent's message_user texts since the last read, oldest first; what triggers sent ``user_phone`` since the last read, its sender in ``channel`` (``"Gmail · Carol Reyes (Mom)"``) as in the agent's -alerts; the user's own lines (the opening request, ``user_says`` lines, ``final``) each at the time it was sent; never -anything meant for the agent; and whether done ends the run (one-sitting) or is only recorded (multi-day). The user -sim reads only after the agent has used message_user. ``history`` is every earlier ``{"read", "answer"}``, oldest -first. ``ask(text)`` sends one turn to the user sim agent (one session per run) and returns ``{"text", "structured", -"trajectory_url"}``, or ``{"text": None, "structured": None, "error"}``. +alerts; the user's own lines (the opening request, with its ``files``, ``user_says`` lines, ``final``) each at the time +it was sent; never anything meant for the agent; and whether done ends the run (one-sitting) or is only recorded +(multi-day). The user sim reads only after the agent has used message_user. ``history`` is every earlier +``{"read", "answer"}``, oldest first. ``ask(text)`` sends one turn to the user sim agent (one session per run) and +returns ``{"text", "structured", "trajectory_url"}``, or ``{"text": None, "structured": None, "error"}``. ``answer`` returns ``{"outcome": "reply" | "no_reply" | "done", "message": str, "used": str | None}``, and may add ``input``, ``raw``, ``score``, ``schema_errors``, ``turns``, ``retried``, ``first_answer``, ``shown`` and @@ -61,8 +75,18 @@ line, when it sets one, in ``you_sent``. Its answer is recorded and never delivered; a ``done`` or an error on it changes nothing. +``rate`` is the rate_user_satisfaction step's (rate_user_satisfaction.py), once an always_on or prompt_agent run is +over, never the run step's: ``satisfaction`` is that step's lines, which are never in the persona, so the system +prompt ``setup`` writes never has them. ``read`` is ``{"time": RFC 3339 UTC, "assistant_messages": [{"at", "text"}]}``, +the agent's messages the user never read (often none). It asks the user sim, in the run's session (``ask`` as above), +to rate the outcome 1-5 by those lines with a one-line reason, the answer in the same format as a read's, and returns +``{"score": 1-5 | None, "reason", "used", "schema_errors", "input", "raw", "retried", "turns"}`` (``first_answer`` +after a retry). The step records it in ``metadata["user_sim_satisfaction"][run step id]``; it never reaches the agent +or the graded conversation. + prompt_agent with ``user_agent_name`` set to the agent this step deployed reads it with the same ``answer``, in one -sitting: one read per agent reply, ``phone`` empty, ``done_ends_run`` true, its prompt this step's opening (user_sim.md). +sitting: one read per agent reply, ``phone`` empty, ``done_ends_run`` true, its prompt this step's opening: the same +text, and the same files in the same order once the seed is filled in (user_sim.md). """ from __future__ import annotations @@ -78,6 +102,7 @@ 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.deploy_agent import DeployAgentTaskStep logger = logging.getLogger(__name__) @@ -175,7 +200,7 @@ class DeployUserSimAgentTaskStep(DeployAgentTaskStep): "a2a_agent_version": FieldScope.TASK} def __init__(self, id: str, version: Optional[int], *, persona: Optional[dict] = None, model: Optional[str] = None, - multi_day: Optional[bool] = None, opening_prompt: Optional[str] = None, **deploy: Any): + multi_day: Optional[bool] = None, opening_prompt: Optional[str | list[dict]] = None, **deploy: Any): if deploy.get("system_prompt") is not None: raise ValueError(f"{self.type} '{id}': system_prompt is written from the persona and opening_prompt; don't set it") deploy.pop("system_prompt", None) @@ -191,8 +216,7 @@ def __init__(self, id: str, version: Optional[int], *, persona: Optional[dict] = if not isinstance(multi_day, bool): found.append("multi_day must be true or false: whether the run goes past the day it starts on") # Required: always_on starts the run from it, and refuses a user sim without one only after the deploy. - if not isinstance(opening_prompt, str) or not opening_prompt.strip(): - found.append("opening_prompt is required: the user's first message, which always_on starts the run with") + found += opening_problems(opening_prompt) found += persona_problems(persona) if found: raise ValueError(f"{self.type} '{self.agent_name}': " + "; ".join(found)) @@ -227,19 +251,19 @@ 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"] - opening = self._opening(context) + 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": opening} if opening else None, "done_ends_run": not self.multi_day} + "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 - def _opening(self, context: TaskStepContext) -> Optional[str]: - """``opening_prompt`` with the run's ```` placeholders filled in, as always_on fills its prompt's.""" - text = self.opening_prompt - if text: - for key, value in (context.metadata.get("seed") or {}).items(): - text = text.replace(f"<{key}>", str(value)) - return text + return apply_seed(opening_parts(self.opening_prompt), context.metadata.get("seed") or {}) def _setup(self, context: TaskStepContext) -> dict: return user_sim_versions.setup(user_sim_versions.VERSION, self.person(context)) @@ -290,9 +314,13 @@ async def _set_user(self, context: TaskStepContext, agent, setup: dict) -> TaskS "description": persona.get("description"), "permissions": persona.get("permissions"), "keep_private": persona.get("keep_private"), "schedule": persona.get("schedule"), "response_times": persona.get("response_times"), "max_replies": persona.get("max_replies"), - "opening": self.opening_prompt, + # The opening's text, as written; an opening with files also has its parts, as written. + "opening": opening_text(opening_parts(self.opening_prompt)), + **({"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, + # 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, # The run record's user sim settings, as always_on wrote them when it set the agent up itself. "setup": {"model": model, "model_from": source, "user_sim_step_model": self.model, "model_set": True, diff --git a/src/agent_env/task_step/task_steps/agent/rate_user_satisfaction.py b/src/agent_env/task_step/task_steps/agent/rate_user_satisfaction.py new file mode 100644 index 000000000..16db2efc9 --- /dev/null +++ b/src/agent_env/task_step/task_steps/agent/rate_user_satisfaction.py @@ -0,0 +1,389 @@ +"""The rate_user_satisfaction step: once an always_on or prompt_agent run is over, the user sim that played its user +rates the job 1-5 with a one-line reason, by this step's ``satisfaction`` lines (user_sim.md, "The rating at the end"). + +The lines are this step's, never the persona's, so the user sim's system prompt never has them and the run behaves as +it does without a rating. The step runs after the run step, as a step of its own, so the rating takes no time from +the run's budget, and it changes nothing in the run step. + +Which run: ``run_step_id``, else the task's only always_on or prompt_agent step; the task is refused when it loads if +that step doesn't exist, isn't one of the two, or isn't run with a user sim a deploy_user_sim_agent step deploys, when +the task has several and no ``run_step_id`` picks one, when this step doesn't come after it (``depends_on``, when set, +must name it), or when two rating steps rate the same run. ``satisfaction`` is checked as the persona's sections are: +a list of non-empty lines. + +The ask is one message to the user sim, in the session the run used, so it remembers the run: + +- always_on: the session its run record names (``metadata["always_on"][run step id]["user_sim"]["context_id"]``), + the one every read of the run used. The messages the user never read are the record's ``unread_by_user`` at the end + (from the stored timeline's end event, else the record's own bounded copy), and the ask is at the time of her last + read, or the run's end when that's later. +- prompt_agent: the run's conversation id (``metadata["a2a_conversations"][run step id]``), the session its built-in + user sim's reads use (``_UserSide._ask_built_in``). The messages the user never read are the agent's replies after + her last read (``metadata["user_sim_reads"]``), from the run's conversation, and the ask is at the task clock's time. + +The message (``user_sim_versions.render_rating``) has those unread messages, the ``satisfaction`` lines under "HOW +YOU'D RATE THE JOB" tagged S1..., and the 1-5 ask; a malformed answer gets one retry (``user_sim_versions.rate``). The +rating is recorded at ``metadata["user_sim_satisfaction"][run step id]``: ``user_sim``, ``score`` (None when it +failed), ``reason``, ``used``, ``at``, ``unread_shown``, ``schema_errors``, ``retried``, ``user_sim_error``, +``user_sim_turns``, ``user_sim_trajectory_urls``, ``rated_by`` (this step's id), ``context_id``, ``session_from`` +and ``tagged_lines`` (``{"S1": line, ...}``, the lines the ask tagged, so ``used`` resolves; a persona tag resolves by +the deploy record's ``tagged_lines``), the whole ask and raw answers in the object store at its ``object_url``. It never reaches the +agent, the graded conversation, or the verifications. + +Not rated, recorded with ``skipped`` and why: a run that ended on an error (always_on's error end reasons; a +prompt_agent run whose agent turn or user sim failed; a run step that failed), and a run whose user sim agent is no +longer deployed. A rating that fails or runs past ``timeout_seconds`` (by default one always_on read's worst case) is +recorded with its error; the step fails only when it can't find the run or its user, and by default +(``fail_task_on_error`` false) that doesn't fail the task. +""" +from __future__ import annotations + +import asyncio +import json +import logging +import time +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import ClassVar, Optional + +from a2a.types import TaskState + +from agent_env.a2a_agent import A2AAgent, conversation_store, protocol +from agent_env.a2a_agent.object_transfer import send_and_wait +from agent_env.env.gateway.constants import EXT_CLOCK_URI +from agent_env.store import get_config +from agent_env.task_step.context import DeployedAgent, PromptResponse, StepContextError, TaskStepContext +from agent_env.task_step.task_step import TaskStep, TaskStepDependency +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, DeployUserSimAgentTaskStep +from agent_env.task_step.task_steps.agent.user_sim_common import SATISFACTION_KEY, iso, person_of, task_zone, utc +from agent_env.task_step.task_steps.always_on import ( + ALWAYS_ON_KEY, + ERROR_ENDS, + READ_ASKS, + TRAJECTORY_FETCH_SECONDS, + TURN_GRACE_SECONDS, + AlwaysOnTaskStep, +) +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep, is_prompt_step + +logger = logging.getLogger(__name__) + +# One always_on read's worst case at its default user sim timeout (AlwaysOnTaskStep.read_cap_seconds). +USER_TIMEOUT_DEFAULT_SECONDS = 300 +DEFAULT_TIMEOUT_SECONDS = READ_ASKS * (USER_TIMEOUT_DEFAULT_SECONDS + TURN_GRACE_SECONDS + TRAJECTORY_FETCH_SECONDS) +SIM_FAILED_PREFIX = "The user simulator failed: " # prompt_agent's error_message for a user sim error + + +@dataclass +class _Session: + """What the step needs of the run: whose user sim, the session, what she never read, and when she's asked.""" + + session_from: str + user_sim: Optional[str] = None + context_id: Optional[str] = None + time_zone: Optional[str] = None + unread: list[dict] = field(default_factory=list) + at: Optional[str] = None + trajectory_prefix: Optional[str] = None + skipped: Optional[str] = None + + +class RateUserSatisfactionTaskStep(TaskStep): + """Once the run step is over, its user sim rates the job 1-5 by ``satisfaction``, recorded in + ``metadata["user_sim_satisfaction"][run step id]`` and sent nowhere.""" + + type: ClassVar[str] = "rate_user_satisfaction" + entity_refs = () + + def __init__(self, id: str, version: Optional[int], *, satisfaction: list[str], run_step_id: Optional[str] = None, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + poll_interval_seconds: float = PromptAgentTaskStep.DEFAULT_POLL_INTERVAL_SECONDS, + depends_on: Optional[list[TaskStepDependency]] = None, fail_task_on_error: bool = False): + """satisfaction: how the user would rate the job, one line each (what 5, 3 and 1 mean for it, what moves the + score). run_step_id: the always_on or prompt_agent step to rate; unset, the task's only one. timeout_seconds: + the longest the rating may take, both asks included.""" + super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + found = user_sim_versions.satisfaction_problems(satisfaction) + if run_step_id is not None and (not isinstance(run_step_id, str) or not run_step_id.strip()): + found.append("run_step_id must name the always_on or prompt_agent step to rate") + for name, value in (("timeout_seconds", timeout_seconds), ("poll_interval_seconds", poll_interval_seconds)): + if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0 or ( + name == "timeout_seconds" and value == 0): + found.append(f"{name} must be a number of seconds" + (", more than 0" if name == "timeout_seconds" + else ", 0 or more")) + if found: + raise ValueError(f"{self.type} '{id}': " + "; ".join(found)) + self.satisfaction = list(satisfaction) + self.run_step_id = run_step_id + self.timeout_seconds = timeout_seconds + self.poll_interval_seconds = poll_interval_seconds + + _OWN = ("satisfaction", "run_step_id", "timeout_seconds", "poll_interval_seconds") + + def to_dict(self) -> dict: + return {**super().to_dict(), **{k: getattr(self, k) for k in self._OWN}} + + @classmethod + def from_dict(cls, data: dict) -> RateUserSatisfactionTaskStep: + base = {**cls._base_from_dict(data), "fail_task_on_error": data.get("fail_task_on_error", False)} + return cls(**base, **{k: data[k] for k in cls._OWN if k in data}) + + # ---- the load check --------------------------------------------------------------------------------------------- + def _run_step(self, steps: list[TaskStep]) -> tuple[Optional[PromptAgentTaskStep], list[str]]: + """The step this one rates, or why there's none.""" + where = f"{self.type} '{self.id}'" + if self.run_step_id is not None: + run = next((s for s in steps if s.id == self.run_step_id), None) + if run is None: + return None, [f"{where}: run_step_id '{self.run_step_id}' names no step of the task"] + if not isinstance(run, PromptAgentTaskStep): + return None, [f"{where}: run_step_id '{self.run_step_id}' is a {run.type} step; name the always_on or " + "prompt_agent step to rate"] + return run, [] + runs = [s for s in steps if isinstance(s, PromptAgentTaskStep)] + if not runs: + return None, [f"{where}: the task has no always_on or prompt_agent step to rate"] + if len(runs) > 1: + return None, [f"{where}: the task has several always_on and prompt_agent steps " + f"({', '.join(s.id for s in runs)}); set run_step_id to the one to rate"] + return runs[0], [] + + def task_problems(self, steps: list[TaskStep]) -> list[str]: + run, found = self._run_step(steps) + if run is None: + return found + where = f"{self.type} '{self.id}'" + if steps.index(run) > steps.index(self): + found.append(f"{where} comes before step '{run.id}', the run it rates; put it after") + if self.depends_on is not None and run.id not in {d.task_step_id for d in self.depends_on}: + found.append(f"{where}: its depends_on must name '{run.id}', the run it rates, so it runs once that's over") + user = _user_of(run) + if user is None or not any(isinstance(s, DeployUserSimAgentTaskStep) and s.agent_name == user for s in steps): + found.append(f"{where}: the user of {run.type} step '{run.id}' isn't a user sim a deploy_user_sim_agent " + "step deploys; only that user sim can rate the run") + earlier = next((s for s in steps[:steps.index(self)] if isinstance(s, RateUserSatisfactionTaskStep) + and s._run_step(steps)[0] is run), None) + if earlier is not None: + found.append(f"{where}: step '{earlier.id}' already rates '{run.id}'; a run is rated once") + return found + + # ---- the run ---------------------------------------------------------------------------------------------------- + async def execute(self, context: TaskStepContext) -> TaskStepContext: + run_id = self.run_step_id or _only_run(context, self.id) + session = await self._session(context, run_id) + us = (context.metadata.get(USER_SIMS_KEY) or {}).get(session.user_sim) if session.user_sim else None + summary: dict = {"user_sim": session.user_sim, "score": None, "reason": None, "rated_by": self.id} + context.metadata.setdefault(SATISFACTION_KEY, {})[run_id] = summary + if session.skipped is None and us is None: + raise StepContextError(f"{self.type} '{self.id}': step '{run_id}' ran with no user sim a " + "deploy_user_sim_agent step recorded, so there's no one to ask") + agent = next((a for a in context.deployed_agents if a.agent_name == us.get("agent_name")), None) if us else None + if session.skipped is None and agent is None: + session.skipped = f"the user sim's agent '{us.get('agent_name')}' is no longer deployed" + if session.skipped is not None: + summary["skipped"] = session.skipped + logger.info(f"{self.type} '{self.id}': '{run_id}' not rated: {session.skipped}") + return context + if not session.context_id: + raise StepContextError(f"{self.type} '{self.id}': step '{run_id}' recorded no user sim session to ask in") + + person = {**person_of(us), "time_zone": session.time_zone} + read: dict = {"time": session.at or await _clock_now(context), + "assistant_messages": [{"at": m.get("at"), "text": m["text"]} for m in session.unread]} + calls: list[dict] = [] + deadline = time.monotonic() + self.timeout_seconds + + async def ask(text: str) -> dict: + got = await self._ask(agent, session, text, deadline) + calls.append(got) + return got + + got: dict = {} + error = None + try: + got = await asyncio.wait_for(user_sim_versions.rate( + us.get("user_version") or user_sim_versions.VERSION, person, self.satisfaction, read, ask), + self.timeout_seconds) or {} + except Exception as e: # TimeoutError included: the rating is recorded with its error + error = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ + error = error or (calls[-1].get("error") if calls else None) + summary.update({k: got.get(k) for k in ("score", "reason", "used", "schema_errors", "retried")}, + at=read["time"], unread_shown=len(read["assistant_messages"]), user_sim_error=error, + user_sim_turns=got.get("turns", len(calls)), + user_sim_trajectory_urls=[c.get("trajectory_url") for c in calls], + context_id=session.context_id, session_from=session.session_from, + tagged_lines=user_sim_versions.rating_tagged_lines(self.satisfaction)) + logger.info(f"{self.type} '{self.id}': '{run_id}' rated {summary['score']} (error={error})") + full = {**summary, "run_step_id": run_id, "satisfaction": self.satisfaction, "read": read, + **{k: got.get(k) for k in ("input", "raw", "first_answer")}} + try: + config = get_config() + key = (f"{config.get_artifact_key_prefix()}user_sim_ratings/instance_id={context.instance_id or 'adhoc'}/" + f"{run_id}-{self.id}-{uuid.uuid4().hex[:8]}.json") + summary["object_url"] = await asyncio.to_thread( + config.get_object_store().put, key, json.dumps(full, indent=1, default=str).encode(), + content_type="application/json") + except Exception as e: + logger.warning(f"{self.type} '{self.id}': could not store the rating: {e}") + summary["store_error"] = f"{type(e).__name__}: {str(e)[:300]}" + return context + + async def _session(self, context: TaskStepContext, run_id: str) -> _Session: + """The run's user sim session, what she never read, and whether the run is rated at all.""" + response = next((r for r in reversed(context.prompt_responses) if r.step_id == run_id), None) + prefix = response.agent_trajectory_object_prefix if response is not None else None + record = (context.metadata.get(ALWAYS_ON_KEY) or {}).get(run_id) + if record is not None: + session = _always_on_session(record) + elif response is not None: + session = await _prompt_agent_session(context, run_id, response) + else: + failed = _failed(context, run_id) + if failed is None: + raise StepContextError(f"{self.type} '{self.id}': step '{run_id}' left no run to rate; run this step " + "after it") + return _Session("none", skipped=f"the run step failed: {failed}") + session.trajectory_prefix = prefix + "user_sim/" if prefix else None + if session.skipped is None and (failed := _failed(context, run_id)) is not None: + session.skipped = f"the run step failed: {failed}" + return session + + async def _ask(self, agent: DeployedAgent, session: _Session, text: str, deadline: float) -> dict: + """One user sim turn on ``text`` in the run's session, cut off at the step's ``deadline`` (time.monotonic()): + ``{"text", "structured", "trajectory_url"}``, or ``{"text": None, "structured": None, "error"}``.""" + limit = deadline - time.monotonic() + if limit <= 0: + return {"text": None, "structured": None, "error": "the step's time limit passed"} + url = agent.a2a_url or agent.api_url + message_id = uuid.uuid4().hex + try: + task_id, result = await asyncio.wait_for(send_and_wait( + url, [{"kind": "text", "text": text}], agent=agent if agent.sandbox_id else None, + message_id=message_id, context_id=session.context_id, timeout_seconds=limit, + poll_interval_seconds=self.poll_interval_seconds), limit) + except TimeoutError as e: + return {"text": None, "structured": None, "error": f"timeout: {e}"} + status = result.get("status") or {} + if status.get("state") != TaskState.completed and str(status.get("state")) != "completed": + return {"text": None, "structured": None, "error": f"state {status.get('state')}"} + msg = status.get("message") or {} + term = protocol.TerminalResponse.from_message(msg) + raw = next((p["text"] for p in msg.get("parts") or [] if p.get("kind") == "text"), term.response_text) + structured = term.structured_output if isinstance(term.structured_output, dict) else None + return {"text": raw, "structured": structured, + "trajectory_url": await self._trajectory(agent, session, task_id, message_id, deadline)} + + async def _trajectory(self, agent: DeployedAgent, session: _Session, task_id: str, message_id: str, + deadline: float) -> Optional[str]: + """The user sim's trajectory for one turn, stored where the run stored its reads'; None when its card offers + none or the run left no trajectory prefix.""" + card = agent.a2a_card or {} + ext = A2AAgent.find_extension(card, A2AAgent.EXT_TRAJECTORY) + if not ext or not session.trajectory_prefix: + return None + try: + # prompt_agent's fetch uses nothing of its step, so the run's own code stores it. + return await asyncio.wait_for(PromptAgentTaskStep._fetch_trajectory( + None, agent.a2a_url or agent.api_url, ext, task_id, session.trajectory_prefix, message_id, + card=card, sandbox_type=agent.sandbox_type), + max(0.0, min(TRAJECTORY_FETCH_SECONDS, deadline - time.monotonic()))) + except Exception as e: + logger.warning(f"user sim trajectory fetch failed (continuing): {e}") + return None + + +# ---- the run, as the run step recorded it ---------------------------------------------------------------------------- +def _user_of(run: PromptAgentTaskStep) -> Optional[str]: + """The agent a run step talks to as its user: always_on's ``user_sim``; prompt_agent's ``user_agent_name``, unless + a ``user_a2a_url`` puts someone else there.""" + if isinstance(run, AlwaysOnTaskStep): + return run.user_sim + return run.user_agent_name if run.user_a2a_url is None else None + + +def _only_run(context: TaskStepContext, step_id: str) -> str: + """With no ``run_step_id``: the task's one run step (the load check made sure there's one), by what it left.""" + ids = list(dict.fromkeys([r.step_id for r in context.prompt_responses if r.step_id] + + [f["step_id"] for f in context.metadata.get("failed_steps") or [] + if isinstance(f, dict) and is_prompt_step({"type": f.get("step_type")})])) + if len(ids) != 1: + raise StepContextError(f"{RateUserSatisfactionTaskStep.type} '{step_id}': " + + (f"several runs ran ({', '.join(ids)}); set run_step_id" if ids + else "no always_on or prompt_agent step ran before it")) + return ids[0] + + +def _failed(context: TaskStepContext, run_id: str) -> Optional[str]: + """The run step's error, when it failed and the task went on; a failure a retry recovered from is no failure.""" + return next((f.get("error") or f.get("error_type") or "an error" for f in context.metadata.get("failed_steps") or [] + if isinstance(f, dict) and f.get("step_id") == run_id and not f.get("retried")), None) + + +def _always_on_session(record: dict) -> _Session: + """From always_on's record: its user sim session, what she never read, and her last read's time or the end's.""" + sim = record.get("user_sim") or {} + session = _Session("always_on record", user_sim=sim.get("name"), context_id=sim.get("context_id"), + time_zone=record.get("time_zone")) + if record.get("end_reason") in ERROR_ENDS: + session.skipped = f"the run ended on {record['end_reason']}" + return session + timeline = record.get("timeline_tail") or [] + if record.get("object_url"): + try: + url = record["object_url"] + timeline = json.loads(get_config().get_object_store_at(url).get(url))["timeline"] + except Exception as e: # the record's own copy is bounded, but has them + logger.warning(f"rate_user_satisfaction: could not read always_on's timeline: {e}") + end = next((e for e in reversed(timeline) if e.get("event") == "end"), {}) + unread = ((end.get("pending") or {}).get("unread_by_user") if isinstance(end.get("pending"), dict) else None) + if not isinstance(unread, list): + unread = ((record.get("pending_at_end") or {}).get("unread_by_user") or {}).get("items") or [] + session.unread = [m for m in unread if isinstance(m, dict) and isinstance(m.get("text"), str)] + times = [utc(e["virtual_time"]) for e in timeline + if e.get("event") in ("user_sim_read", "end") and e.get("virtual_time")] + session.at = iso(max(times)) if times else None + return session + + +async def _prompt_agent_session(context: TaskStepContext, run_id: str, response: PromptResponse) -> _Session: + """From prompt_agent's run: its conversation, the session its built-in user sim's reads used; the agent's replies + after her last read; the task clock's time.""" + conversation_id = (context.metadata.get("a2a_conversations") or {}).get(run_id) + doc = conversation_store.get_conversation(conversation_id) if conversation_id else None + session = _Session("prompt_agent conversation", user_sim=(doc or {}).get("source_agent_name"), + context_id=conversation_id) + if response.error_type == "user_sim_error": + session.skipped = f"the user sim failed: {(response.error_message or '').removeprefix(SIM_FAILED_PREFIX)}" + return session + if response.error_type or response.error_message: + session.skipped = "the agent's turn failed" + return session + us = (context.metadata.get(USER_SIMS_KEY) or {}).get(session.user_sim) if session.user_sim else None + if us is not None: + session.time_zone = task_zone(us, context.metadata.get("clock_configurations") or [], + f"{RateUserSatisfactionTaskStep.type} on '{run_id}'") + session.at = await _clock_now(context) + last_read = max((r.get("turn") or 0 for r in (context.metadata.get("user_sim_reads") or {}).get(run_id) or []), + default=0) + replies = [m for m in (doc or {}).get("messages") or [] if m.get("role") == "agent"] + for m in replies[last_read:]: + text = "\n".join(p.get("text", "") for p in m.get("parts") or [] if p.get("kind") == "text") + clocks = [utc(c["virtual_time"]) for c in m.get("env_clock") or [] if c.get("virtual_time")] + session.unread.append({"at": iso(min(clocks)) if clocks else session.at, "text": text}) + return session + + +async def _clock_now(context: TaskStepContext) -> str: + """The task clock's time when one is armed (the earliest), else now; RFC 3339 UTC.""" + times = [] + for env in (d for d in context.deployed_envs if d.supports(EXT_CLOCK_URI, "state")): + try: + state = await env.invoke(EXT_CLOCK_URI, "state", timeout=5) + except Exception as e: + logger.warning(f"clock state read failed for env '{env.env_id}' (continuing): {e}") + continue + if state and state.get("armed") and state.get("virtual_time"): + times.append(utc(state["virtual_time"])) + return iso(min(times, default=datetime.now(timezone.utc))) diff --git a/src/agent_env/task_step/task_steps/agent/user_sim.md b/src/agent_env/task_step/task_steps/agent/user_sim.md index 2b63d85f2..baf62e21e 100644 --- a/src/agent_env/task_step/task_steps/agent/user_sim.md +++ b/src/agent_env/task_step/task_steps/agent/user_sim.md @@ -31,12 +31,39 @@ Its own fields: | `agent_name` | required | The agent's name, and the key always-on's `user_sim` names | | `persona` | required | Who the user is: `display_name` (their full name), `first_name` (else the display name's first word), `time_zone` (their IANA zone; else the task clock's, else UTC; a clock that sets a different zone is refused), `description`, `permissions`, `keep_private`, `schedule`, `response_times`, `max_replies` (after that many replies the user stops reading) | | `model` | required | The user sim's model. This step's `step_params` `model` wins, then the run's `user_model` (`--user-model`), then this | -| `opening_prompt` | required | The user's first message. The always-on step sends it and has no prompt of its own | +| `opening_prompt` | required | The user's first message: text, or text and file parts (below). The always-on step sends it and has no prompt of its own | | `multi_day` | required, for now | Whether the run goes past the day it starts on (always-on's `end_time` falls on a later day). It picks the prompt's done line; always-on refuses a run of the other kind | `multi_day` is a stopgap: the prompt is written before the run, so the step has to be told which done line to use. It goes away once the step can learn it from the task. +**A first message with files.** `opening_prompt` can be a list of parts in `prompt_agent`'s `parts` format, text and +file parts, for a first message that attaches a photo, a PDF, a spreadsheet or any other file: + +```json +"opening_prompt": [ + {"kind": "text", "text": "ok mom and dad are officially coming thu–mon (nov 19–23), can you handle it?"}, + {"kind": "file", "file": {"uri": "s3:////itinerary.pdf", "name": "itinerary.pdf", "mimeType": "application/pdf"}, + "metadata": {"description": "Mom and Dad's flight itinerary"}} +] +``` + +The agent gets the parts on its first wake, in order: the text, joined, as the user's one labelled line where the +first text part was, and each file in its place. The files are in the user's first message of the graded +conversation too. A file goes through the same sharing as any prompt's file part (an object a configured store owns is +sent as a URL the agent can read), and the seed fills ``s in its `uri` and `name`. Whether the agent can +open a file is up to its harness. + +The user sim can't open files, so its prompt and its first read say what it attached, after its message: `"ok mom +and dad…" (you attached itinerary.pdf: Mom and Dad's flight itinerary)`. `metadata.description` is that one line; the +agent is sent the file without it. The prompt adds, under that line, that the user sim remembers only what's said of +each file, not what's in it, so asked what a file holds it says it doesn't remember or points the assistant to the +file rather than making the contents up. With no description, the user sim gets the file's name only (its `name`, else the +last part of its `uri`). The step refuses, when the task loads, an opening with no text, a part that isn't a valid +text or file part, or a description that isn't one line. The run records the opening's text in +`metadata["user_sims"][]["opening"]`, as for a text-only opening, and the parts, as written, in +`opening_parts`. + `system_prompt` is written from the persona and `opening_prompt`, so the step refuses one set directly. Each read shows times in the persona's `time_zone`. A run-wide `a2a_agent_id` never changes the user sim's agent, so the agent under test never plays the user; this step's own `step_params` `a2a_agent_id` can. An agent @@ -44,7 +71,9 @@ whose card takes no system prompt or no model fails the step, since without them **Used with prompt_agent.** A `prompt_agent` step whose `user_agent_name` is this step's `agent_name` (and sets no `user_a2a_url`) talks to the same user in one sitting: each reply of the agent is one read, with no phone. Its prompt -must be this step's `opening_prompt`, which the user's prompt quotes as what they sent (FD-5495 is to write it once). +must be this step's `opening_prompt`, which the user's prompt quotes as what they sent (FD-5495 is to write it once): +the same text, and the same files in the same order, once the seed is filled in. The check compares each file part's +`file` (`uri`, `name`, `mimeType`), not its `metadata`, so the description can stay on this step alone. `reply` is the agent's next prompt; `done` with a message reaches the agent and ends the conversation after its answer; `done` without one, or `no_reply`, ends it at once. After `max_replies` replies, the last one reaches the agent and the conversation ends after its answer; the user isn't asked again, not even by a @@ -65,13 +94,18 @@ it starts, before its first agent turn; a refused one leaves no conversation. ## The description `description` is the person in plain English, as sections of lines. Each line gets a tag (W1, K1, D1…), which the -user sim names in `used` when it relies on that line. +user sim names in `used` when it relies on that line. A tag is its section's letter and the line's place in it, so +reordering or adding lines moves it; the run records each tag with its line, as the prompt showed it, in +`metadata["user_sims"][]["tagged_lines"]` (`{"W1": "...", "D9": "..."}`), and an old run's `used` +resolves there. | Section | Tag | What goes in it | |---|---|---| | `who_you_are` | W | Who they are, the week it is, their day ("at work 9-6, asleep 11pm-7am") | | `how_you_write` | | How they text, best shown with 2–3 of their real messages | | `what_you_know` | K | Facts they give only when asked | +| `already_told` | T | Optional. Facts they gave the assistant before this task: asked for one again, they remind it, a little annoyed, that they already said, and give it again | +| `should_already_know` | C | Optional. What the assistant should know without being told: they never bring it up, and if the assistant gets it wrong they correct it briefly, a little irritated, and add nothing else | | `what_to_look_up` | L | Facts the assistant must find: say where they are ("it's in mom's email"), never the value | | `decisions` | D | What they say when the assistant asks or proposes something | | `never_supply` | N | What they never hand over (card numbers, codes, anything the assistant should find) | @@ -80,6 +114,11 @@ user sim names in `used` when it relies on that line. | `phone` | F | What they do with what reaches their phone | | `when_done` | | When they'd say they're done | +`already_told` and `should_already_know` are for what the assistant has from before this task. Each "should already +know" line must also be in the assistant's own instructions or notes, and each "already told" fact somewhere it can +read, or the user sim expects knowledge the assistant was never given. A fact the assistant should find in the apps +goes in `what_to_look_up` or `never_supply` instead: an "already told" fact is given again when asked. + Don't write the opening request into the description. Put it in `opening_prompt`: the prompt shows it right after the first `who_you_are` line ("Earlier you texted the assistant: …"). @@ -87,6 +126,51 @@ after the first `who_you_are` line ("Earlier you texted the assistant: …"). `tell_after`. Each is a list of rules, or `{items, then}` to add what follows. `keep_private` becomes the "PRIVATE" lines. Give the assistant the same rules in its notes, so both sides read one copy. +## The rating at the end + +A `rate_user_satisfaction` step has the user sim rate the job once the run is over: 1 to 5, with a one-line reason. +It goes after the `always_on` or `prompt_agent` step it rates, with its own lines, `satisfaction`. Write what each end +of the scale means for this task, and what moves the score for this person (the intake's User Sim Questions 6 and 7): + +```json +{"id": "rate", "type": "rate_user_satisfaction", "run_step_id": "week", "depends_on": ["week"], + "satisfaction": ["5, perfect: flights, a step-free place a walk from you, and both dinners booked, with nothing you had to chase.", + "3, fine: all booked, but you had to answer a question your rules already settled.", + "1, you'd have done it yourself: a dinner or the place is still not booked at the end.", + "Being asked for things that are in Mom's email bothers you."]} +``` + +| Field | | What it is | +|---|---|---| +| `satisfaction` | required | How the user would rate the job, one line each, tagged S1… in the ask | +| `run_step_id` | optional | The `always_on` or `prompt_agent` step to rate; unset, the task's only one | +| `timeout_seconds` | 960 | The longest the rating may take, both asks included (one always-on read's worst case) | +| `fail_task_on_error` | false | A rating step that fails doesn't fail the task | + +The lines are on this step, never in the persona, so the user sim's prompt never has them and the run is the one +tested; a `satisfaction` section in the description is refused. The run step is as it is without a rating, and the +rating takes none of always-on's time budget. + +The step sends the user sim one message, in the session the run used, so it remembers the run: always-on's, which its +record names (`metadata["always_on"][]["user_sim"]["context_id"]`), or `prompt_agent`'s conversation. The +message has any of the assistant's messages the user never read (in always-on, what was unread at the end; in +`prompt_agent`, the replies after the last read), the lines under "HOW YOU'D RATE THE JOB", and the ask: 5 is perfect, +3 fine, 1 "you'd have done it yourself", 2 and 4 in between, and something it was never told was done doesn't count as +done. The answer uses the usual format, with `score` and the reason in `message`; a malformed one gets one retry. + +The rating goes to `metadata["user_sim_satisfaction"][]`: `score` (None when it failed), `reason`, +`used`, `at`, `unread_shown`, `schema_errors`, `retried`, `user_sim_error`, `user_sim_turns`, `context_id` (the +session it asked in), `rated_by` (the rating step), `tagged_lines` (`{"S1": "...", ...}`, the lines the ask tagged, so +its `used` resolves), and `skipped` with the reason when it wasn't asked: the run ended +on an error (always-on's error ends, a `prompt_agent` agent turn or user sim that failed, a run step that failed), or +the user sim's agent is no longer deployed. The whole ask and raw answer are in the object store at its `object_url`. +It never reaches the assistant, the graded conversation, or the verifications. A user sim rates more kindly than +people do, so treat it as a signal next to the graders, not the score. + +The task is refused when it loads if the step has no run to rate (none, or several and no `run_step_id`), comes +before it or has a `depends_on` that doesn't name it, if the run's user isn't a user sim a `deploy_user_sim_agent` +step deploys, if two rating steps rate the same run, or if `satisfaction` isn't a list of non-empty lines. + ## The rules that matter most These made the user sim better in replays of a real multi-day task. The instructions drove more of the gain than how 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 b44276c87..7cd09efa2 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 @@ -1,8 +1,12 @@ """What always_on and prompt_agent share when they read a user deployed by deploy_user_sim_agent.""" from __future__ import annotations +import copy from datetime import datetime, timezone -from typing import Optional +from typing import Any, Optional + +# Where rate_user_satisfaction records a run's rating by its user sim: metadata[SATISFACTION_KEY][run step id]. +SATISFACTION_KEY = "user_sim_satisfaction" def person_of(us: dict) -> dict: @@ -37,3 +41,90 @@ def utc(text: str) -> datetime: def iso(value: Optional[datetime]) -> Optional[str]: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") if value else None + + +# ---- the opening: the user's first message --------------------------------------------------------------------------- +# ``opening_prompt`` is text, or a list of A2A text and file parts (prompt_agent's ``parts``). A file part may carry a +# one-line ``metadata.description`` for the user sim, which can't open files; the agent is sent the file without it. +OPENING_KINDS = ("text", "file") +DESCRIPTION = "description" + + +def opening_parts(opening: Any) -> list[dict]: + """``opening_prompt`` as parts, a copy: text is one text part.""" + if isinstance(opening, str): + return [{"kind": "text", "text": opening}] + return copy.deepcopy(list(opening or [])) + + +def recorded_opening(us: dict) -> list[dict]: + """The opening a deploy_user_sim_agent record holds, as parts, its seed not yet filled in.""" + return opening_parts(us["opening_parts"] if us.get("opening_parts") is not None else us.get("opening") or "") + + +def opening_text(parts: list[dict]) -> str: + """The opening's text parts joined, as the user's one line of text.""" + return "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text") + + +def file_name(part: dict) -> str: + """What the user calls a file part: its name, else the last segment of its address.""" + file = part.get("file") or {} + return file.get("name") or (file.get("uri") or "").rstrip("/").rsplit("/", 1)[-1] or "a file" + + +def attached(parts: list[dict]) -> list[dict]: + """The files the opening attaches, as the user sim is told of them: ``{"name", "description"}``, the description + only when the part has one.""" + out = [] + for part in parts: + if part.get("kind") == "file": + description = (part.get("metadata") or {}).get(DESCRIPTION) + out.append({"name": file_name(part), **({"description": description} if description else {})}) + return out + + +def for_agent(part: dict) -> dict: + """A part as the agent is sent it: without the description, which is for the user sim.""" + metadata = part.get("metadata") + if not isinstance(metadata, dict) or DESCRIPTION not in metadata: + return part + rest = {k: v for k, v in metadata.items() if k != DESCRIPTION} + return {**{k: v for k, v in part.items() if k != "metadata"}, **({"metadata": rest} if rest else {})} + + +def opening_problems(opening: Any) -> list[str]: + """What's wrong with ``opening_prompt``, one message each; empty when it's fine.""" + required = "opening_prompt is required: the user's first message, which always_on starts the run with" + if opening is None or isinstance(opening, str): + return [] if isinstance(opening, str) and opening.strip() else [required] + if not isinstance(opening, list): + return ["opening_prompt must be text, or a list of text and file parts"] + from a2a.types import FilePart, TextPart + from pydantic import ValidationError + + out = [] + for i, part in enumerate(opening): + where = f"opening_prompt[{i}]" + if not isinstance(part, dict): + out.append(f"{where} must be a part, an object") + continue + if part.get("kind") not in OPENING_KINDS: + out.append(f"{where} is a {part.get('kind')!r} part; the user's first message takes text and file parts") + continue + try: + (TextPart if part["kind"] == "text" else FilePart).model_validate(part) + except ValidationError as e: + why = "; ".join(f"{'.'.join(map(str, err['loc']))}: {err['msg']}" for err in e.errors()) + out.append(f"{where} is not a valid A2A {part['kind']} part ({why})") + continue + metadata = part.get("metadata") + if part["kind"] == "file" and isinstance(metadata, dict) and DESCRIPTION in metadata: + description = metadata[DESCRIPTION] + if not isinstance(description, str) or not description.strip() or "\n" in description.strip(): + out.append(f"{where}'s metadata.description must be one line of text: what the user sim is told " + "the file is") + if not opening_text([p for p in opening if isinstance(p, dict)]).strip(): + out.append("opening_prompt needs a text part with text: the user's first message, which always_on starts " + "the run with") + return out diff --git a/src/agent_env/task_step/task_steps/agent/user_sim_versions.py b/src/agent_env/task_step/task_steps/agent/user_sim_versions.py index fb3a8ed19..55eee2d03 100644 --- a/src/agent_env/task_step/task_steps/agent/user_sim_versions.py +++ b/src/agent_env/task_step/task_steps/agent/user_sim_versions.py @@ -7,7 +7,15 @@ answer format. Its done line follows ``person["done_ends_run"]``, the multi-day wording when that's absent. ``answer`` renders one read as one labelled message, asks the user sim, and parses its answer; a malformed answer gets one retry in the same session. What earlier reads showed, and when the user last wrote, come from ``history``; -keys of ``read`` it doesn't know are ignored. +keys of ``read`` it doesn't know are ignored. A line the user sent with files (the opening's ``files``, +``{"name", "description"}``) shows each after its text, "(you attached itinerary.pdf: Mom and Dad's flight +itinerary)", or by name alone with no description: the user sim can't open them, and the opening's line says it +knows only that much of them. +``tagged_lines`` is each tag the system prompt shows with its line, which deploy_user_sim_agent records so an answer's +``used`` resolves against the lines the user sim saw. +``rate`` asks the user sim, once the run is over, to rate the outcome 1-5 by the ``satisfaction`` lines it is given +(the rate_user_satisfaction step's, never the persona's, so the system prompt never has them), with a one-line +reason. ``problems`` is the load check for the settings.""" from __future__ import annotations @@ -24,6 +32,10 @@ ("who_you_are", "WHO YOU ARE", "W"), ("how_you_write", "HOW YOU WRITE", ""), ("what_you_know", "WHAT YOU KNOW (to use only when the assistant asks about it)", "K"), + ("already_told", "THINGS YOU ALREADY TOLD THE ASSISTANT (if it asks for one again, remind it, a little annoyed, " + "that you already told it, and tell it again)", "T"), + ("should_already_know", "THINGS THE ASSISTANT SHOULD ALREADY KNOW (never bring these up; if it gets one wrong, " + "correct it briefly, a little irritated, and add nothing else)", "C"), ("what_to_look_up", "WHAT TO LOOK UP (say where; don't give the value)", "L"), ("decisions", "YOUR DECISIONS (when the assistant asks)", "D"), ("never_supply", "NEVER SUPPLY (the assistant must find these itself)", "N"), @@ -46,6 +58,12 @@ BROKEN_RULE_LINE = ('If it does an "ask first" or a "never" without asking and tells you, say once, briefly, that ' "you're not happy about it.") OPENING_LINE = 'Earlier you texted the assistant:\n"{text}"' +# A file the user sent with a line, after its text: the user sim can't open it, so it's told what it attached. +ATTACHED = "(you attached {name}: {description})" +ATTACHED_NAME_ONLY = "(you attached {name})" +# Under an opening with files, so the user sim never makes up what one holds. +ATTACHED_KNOWN = ("You remember only what's said here about each file you attached, not what's in it; if the " + "assistant asks what's in one, say you don't remember or point it to the file.") OUTCOMES = ("reply", "no_reply", "done") ANSWER_LINE = '{"outcome": "reply" | "no_reply" | "done", "message": "...", "used": "..."}' @@ -75,6 +93,22 @@ EARLIER_MAX = 20 EARLIER_CUT = " (older messages not shown)" +# The rating after the run (``rate``), asked by the rate_user_satisfaction step with that step's ``satisfaction`` lines, +# shown in the ask under RATING_TITLE, tagged S1... Its answer fits OUTPUT_FORMAT, which the agent was given at deploy. +SCORES = (1, 2, 3, 4, 5) +RATING_TITLE = "HOW YOU'D RATE THE JOB" +RATING_TAG = "S" +RATING_LINE = '{"outcome": "done", "score": 1 | 2 | 3 | 4 | 5, "message": "...", "used": "..."}' +RATING_OVER = "The job is over: your AI assistant won't see this, or anything you write from now on." +RATING_UNREAD = "\nNEW FROM YOUR AI ASSISTANT (you haven't read this yet):" +RATING_ASK = ("As {first}, rate how the job went, from everything you've seen: 5 if it was done perfectly, 3 if it " + "wasn't perfect but was fine, 1 if you'd have done it yourself; 2 and 4 are in between. HOW YOU'D RATE " + "THE JOB says what those mean for this job. Something you were never told was done doesn't count as " + "done.") +RATING_ANSWER = ('In "message", give your reason in one line. In "used", write the tag of the line it rests on, or ' + '"none". Answer in JSON: ' + RATING_LINE) +RATING_RETRY = "Your last answer wasn't in the required format. Answer again, only as JSON: " + RATING_LINE + # The kinds of line the user sent; a you_sent item of any other kind (an alert, a nudge) is never shown as theirs. OWN_LINE_KINDS = frozenset({"opening", "user_says", "final"}) @@ -135,12 +169,19 @@ def rule_lines(permissions: dict) -> list[str]: return lines +def attached_note(files: Optional[list[dict]]) -> str: + """What a line's ``files`` (``{"name", "description"}``) add after its text: nothing, when it has none.""" + return "".join(" " + (ATTACHED.format(**f) if f.get("description") else ATTACHED_NAME_ONLY.format(name=f["name"])) + for f in files or []) + + def _who(person: dict) -> list[str]: """Who they are, with the opening request right after the first line.""" lines = list(person["description"].get("who_you_are") or []) - opening = (person.get("opening") or {}).get("text") - if opening: - lines.insert(min(1, len(lines)), OPENING_LINE.format(text=opening)) + opening = person.get("opening") or {} + if opening.get("text"): + line = OPENING_LINE.format(text=opening["text"]) + attached_note(opening.get("files")) + lines.insert(min(1, len(lines)), line + ("\n" + ATTACHED_KNOWN if opening.get("files") else "")) return lines @@ -149,16 +190,35 @@ def _section(out: list[str], title: str, tag: str, lines: list[str]) -> None: out += ["", title] + [tag_line(f"{tag}{i}" if tag else "", text) for i, text in enumerate(lines, 1)] -def system_prompt(person: dict) -> str: - out = _preamble(person) +def _sections(person: dict) -> list[tuple[str, str, list[str]]]: + """Each section the system prompt shows, in order: its title, its tag letter ("" for none) and its lines.""" + out = [] for key, title, tag in SECTIONS: - _section(out, title, tag, _who(person) if key == "who_you_are" else list(person["description"].get(key) or [])) + out.append((title, tag, _who(person) if key == "who_you_are" else list(person["description"].get(key) or []))) if key == RULES_AFTER: - _section(out, RULES_TITLE, "A", rule_lines(person.get("permissions") or {})) - _section(out, PRIVATE_TITLE, "P", list(person.get("keep_private") or [])) + out.append((RULES_TITLE, "A", rule_lines(person.get("permissions") or {}))) + out.append((PRIVATE_TITLE, "P", list(person.get("keep_private") or []))) + return out + + +def system_prompt(person: dict) -> str: + out = _preamble(person) + for title, tag, lines in _sections(person): + _section(out, title, tag, lines) return "\n".join(out) + "\n" +def tagged_lines(person: dict) -> dict[str, str]: + """Each tag the system prompt shows, with its line as written there ("W2": the opening request, ...): what an + answer's ``used`` names, recorded with the run so it resolves however the persona's lines change later.""" + return {f"{tag}{i}": text for _, tag, lines in _sections(person) if tag for i, text in enumerate(lines, 1)} + + +def rating_tagged_lines(satisfaction: list[str]) -> dict[str, str]: + """The rating ask's tags, ``S1``..., with their ``satisfaction`` lines, as ``tagged_lines`` for the rating.""" + return {f"{RATING_TAG}{i}": text for i, text in enumerate(satisfaction, 1)} + + # ---- the load check ----------------------------------------------------------------------------------------------- def _lines(value: Any) -> bool: return isinstance(value, list) and all(isinstance(v, str) and v.strip() for v in value) @@ -180,7 +240,9 @@ def problems(person: Any) -> list[str]: else: unknown = sorted(set(description) - set(SECTION_KEYS)) if unknown: - out.append(f"description has unknown sections {unknown}; the sections are {list(SECTION_KEYS)}") + out.append(f"description has unknown sections {unknown}; the sections are {list(SECTION_KEYS)}" + + ("; satisfaction goes on the rate_user_satisfaction step, so the user sim's prompt never has " + "it" if "satisfaction" in unknown else "")) out += [f"description.{key} must be a list of non-empty lines" for key in SECTION_KEYS if key in description and not _lines(description[key])] if not any(_lines(description.get(key)) and description[key] for key in SECTION_KEYS): @@ -250,6 +312,11 @@ def _stamp(t: datetime, now: datetime) -> str: return f"{t.strftime('%a %b')} {t.day}, {_hm(t)}" +def _message(m: dict, stamp: str) -> str: + """One of the assistant's messages: its time, then its text indented under it.""" + return f" [{stamp}]\n" + "\n".join(" " + line if line.strip() else "" for line in m["text"].strip().splitlines()) + + def render(person: dict, history: list[dict], read: dict) -> Optional[str]: """The message the user sim gets for ``read``, or None when the assistant sent nothing. An item without ``at`` shows at the time of the read it came in.""" @@ -295,10 +362,9 @@ def dated(r: dict, items: list) -> list[dict]: # an earlier read's items, each out.append(EARLIER_CUT) if own: out.append("\nYOU SENT YOUR AI ASSISTANT:") - out += [f" [{_stamp(at(s), now)}] {s['text']}" for s in own] + out += [f" [{_stamp(at(s), now)}] {s['text']}{attached_note(s.get('files'))}" for s in own] out.append("\nNEW FROM YOUR AI ASSISTANT (this is what you're answering):") - out += [f" [{_stamp(at(m), now)}]\n" + "\n".join(" " + line if line.strip() else "" - for line in m["text"].strip().splitlines()) for m in messages] + out += [_message(m, _stamp(at(m), now)) for m in messages] out.append("\n" + CLOSING.format(first=_first(person))) out.append("Answer in JSON: " + ANSWER_LINE) return "\n".join(out) @@ -392,3 +458,80 @@ async def answer(version: str, person: dict, history: list[dict], read: dict, def _parsed(got: dict) -> dict: structured = got.get("structured") return parse(structured if isinstance(structured, dict) else got.get("text")) + + +# ---- the rating after the run ------------------------------------------------------------------------------------- +def satisfaction_problems(lines: Any) -> list[str]: + """What's wrong with the lines a rating goes by: one message, or none.""" + if not (_lines(lines) and lines): + return ["satisfaction must be a list of non-empty lines: what each end of the scale means for this job"] + return [] + + +def render_rating(person: dict, satisfaction: list[str], read: dict) -> str: + """The message that asks for the rating at ``read["time"]``: the assistant's messages the user hasn't read + (``read["assistant_messages"]``, maybe none), the ``satisfaction`` lines tagged S1..., then the ask.""" + zone = ZoneInfo(person.get("time_zone") or "UTC") + now = _local(read["time"], zone) + out = [f"It's {now.strftime('%a, %b')} {now.day}, {_hm(now)}. {RATING_OVER}"] + messages = [m for m in read.get("assistant_messages") or [] if (m.get("text") or "").strip()] + if messages: + out.append(RATING_UNREAD) + out += [_message(m, _stamp(_local(m.get("at") or read["time"], zone), now)) for m in messages] + _section(out, RATING_TITLE, RATING_TAG, list(satisfaction)) + out.append("\n" + RATING_ASK.format(first=_first(person))) + out.append(RATING_ANSWER) + return "\n".join(out) + + +def rating_problems(obj: Any) -> list[str]: + out = answer_problems(obj) + if out: + return out + if "score" not in obj: + return ["missing score"] + if obj["score"] not in SCORES: + out.append(f"score {obj['score']!r} is not one of {list(SCORES)}") + if not obj["message"].strip(): + out.append("message must give the reason") + return out + + +def parse_rating(raw: Any) -> dict: + """The rating as the step records it: ``score`` (1-5, None when malformed), ``reason``, ``used`` and + ``schema_errors``.""" + obj = _object(raw) + errors = rating_problems(obj) + if errors: + return {"score": None, "reason": None, "used": None, "schema_errors": errors} + return {"score": obj["score"], "reason": obj["message"].strip(), "used": obj.get("used"), "schema_errors": []} + + +async def rate(version: str, person: dict, satisfaction: list[str], read: dict, + ask: Callable[[str], Awaitable[dict]]) -> dict: + """The user's rating of the run once it's over, by the ``satisfaction`` lines: one ask, in the run's session, with + one retry when the answer is malformed. ``read`` is ``{"time": RFC 3339 UTC, "assistant_messages": [{"at", + "text"}]}``, the messages the user hasn't read. Returns ``score``, ``reason``, ``used``, ``schema_errors``, + ``input``, ``raw``, ``retried``, ``turns`` and, after a retry, ``first_answer``. The rating is never a reply: the + caller records it and sends it nowhere.""" + _check(version) + found = satisfaction_problems(satisfaction) + if found: + raise ValueError("; ".join(found)) + text = render_rating(person, satisfaction, read) + got = await ask(text) + first = _parsed_rating(got) + if not first["schema_errors"] or got.get("error"): + return {**first, "input": text, "raw": got.get("text"), "retried": False, "turns": 1} + again = await ask(RATING_RETRY) + second = _parsed_rating(again) + if second["schema_errors"]: + second["schema_errors"] = ([f"first answer: {e}" for e in first["schema_errors"]] + + [f"answer after the retry: {e}" for e in second["schema_errors"]]) + return {**second, "input": text, "raw": again.get("text"), "retried": True, "turns": 2, + "first_answer": {"raw": got.get("text"), "schema_errors": first["schema_errors"]}} + + +def _parsed_rating(got: dict) -> dict: + structured = got.get("structured") + return parse_rating(structured if isinstance(structured, dict) else got.get("text")) diff --git a/src/agent_env/task_step/task_steps/always_on.py b/src/agent_env/task_step/task_steps/always_on.py index bb60c912a..a31e0586f 100644 --- a/src/agent_env/task_step/task_steps/always_on.py +++ b/src/agent_env/task_step/task_steps/always_on.py @@ -5,9 +5,11 @@ deploys the agent that plays the user, with its system prompt and model set, and records the user in the run's metadata under its ``agent_name``. The ``always_on`` step names it with ``user_sim`` and runs the loop: -- The first wake is the user's request. Each wake opens with what came due for the agent on notifications/v1 (a - trigger's ``notify``, a scheduled check), labelled ``[time · channel]``, then the user's own messages, each labelled - with its time; while the task's clock is set, the agent can read it with the gateway's ``get_time``. +- The first wake is the user's request, with the files its opening attaches in their place (its timeline ``wake`` + event names them in ``files``, by ``name`` and ``uri``). Each wake opens with what came due for the agent on + notifications/v1 (a trigger's ``notify``, a scheduled check), labelled ``[time · channel]``, then the user's own + messages, each labelled with its time; while the task's clock is set, the agent can read it with the gateway's + ``get_time``. - Only what the agent sends with the gateway's ``message_user`` tool (user-chat/v1) reaches the user; its plain reply goes to the log. Each wake should end with ``message_user`` or ``stay_quiet``: one that ends with closing text and neither gets one ``nudge`` turn. ``stay_quiet`` and the nudge are on unless the task turns them off; the rule itself is @@ -63,7 +65,8 @@ reply that never reached it (her last read's, or one still held at the end) has ``delivered`` false and no ``user_reads_at``. - The run's metadata keeps a summary in ``metadata["always_on"][step_id]`` (small enough for a running job to report - back); the full timeline goes to the object store, at the summary's ``object_url``. + back); the full timeline goes to the object store, at the summary's ``object_url``. Its ``user_sim`` names the user + sim's session (``context_id``), which rate_user_satisfaction asks in once the run is over. - An always_on_agent_verifier (``verifiers/always_on_agent_verifier.py``) grades the step's prompt from the user's chat and the run's context, the run wake by wake among it (``render_for_judge``); a plain rubrics_verifier on it is refused when the task loads. @@ -115,6 +118,7 @@ from agent_env.task_step.task_steps.notification_delivery import delivered_parts, label_time, take_due from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.agent.user_sim_common import iso as _iso, person_of, task_zone, utc as _utc +from agent_env.task_step.task_steps.agent.user_sim_common import attached, file_name, for_agent, recorded_opening logger = logging.getLogger(__name__) @@ -488,7 +492,8 @@ async def go(self) -> None: raise ValueError(f"always_on: the user sim '{step.user_sim}' has no opening_prompt, the user's first " "message; give the deploy_user_sim_agent step one") now = await self.now() - self.opening = {"at": _iso(now), "text": self._effective_prompt()} + files = attached(self._prompt_parts()) + self.opening = {"at": _iso(now), "text": self._effective_prompt(), **({"files": files} if files else {})} self.person["opening"] = self.opening t0 = await self._t0() or now self.end = _end_instant(step.end_time, t0) if step.end_time else None @@ -1122,19 +1127,19 @@ async def _last_read(self) -> None: def _prompt_parts(self) -> list[dict]: """The user's first message, the user sim's opening, as A2A parts, its seed applied.""" - return self.step._apply_seed([{"kind": "text", "text": self.us["opening"]}], self.ctx.metadata.get("seed", {})) + return self.step._apply_seed(recorded_opening(self.us), self.ctx.metadata.get("seed", {})) def _effective_prompt(self) -> str: return "\n".join(p.get("text", "") for p in self._prompt_parts() if p.get("kind") == "text") def _opening_parts(self, labelled) -> list[dict]: """The prompt as the opening wake and the user's chat carry it: its text as one labelled line from the user - (``labelled(text)``), where its first text part was, and every other part (a file, data) as it is, in the - prompt's order, as prompt_agent sends them.""" + (``labelled(text)``), where its first text part was, and every file part as it is, in the prompt's order, as + prompt_agent sends them, without the description that is for the user sim.""" out, text = [], self._effective_prompt() for part in self._prompt_parts(): if part.get("kind") != "text": - out.append(part) + out.append(for_agent(part)) elif text is not None: out.append(labelled(text)) text = None @@ -1252,8 +1257,12 @@ async def wake(self, parts: list[dict], woke_by: list[str], notifications: Optio def record() -> None: self.per_turn_parts.append(self._kept_parts(parts)) + # The files a wake carries (the opening's), by name and address; their descriptions are the user sim's. + files = [{"name": file_name(p), "uri": (p.get("file") or {}).get("uri")} + for p in parts if p.get("kind") == "file"] self.log("wake", n=self.wakes, virtual_time=_iso(sent_at), woke_by=woke_by, notifications=notifications or [], - last=last, parts_text="\n".join(p.get("text", "") for p in parts if p.get("kind") == "text")) + last=last, parts_text="\n".join(p.get("text", "") for p in parts if p.get("kind") == "text"), + **({"files": files} if files else {})) t = time.time() try: task_id, result = await self._agent_turn(parts, msg_id, record) @@ -1685,7 +1694,8 @@ async def finish(self) -> None: record = {"step_id": step.id, "end_reason": self.end_reason, "end_note": self.end_note, "error": self.error, "error_type": self.error_type, "counts": counts, "user_sim": {"name": step.user_sim, "version": self.version, - "agent_name": self.us.get("agent_name"), **self.user_setup}, + "agent_name": self.us.get("agent_name"), "context_id": self.user_context_id, + **self.user_setup}, "end_time": _iso(self.end) if self.end else None, "time_zone": self.zone, "multi_day": self.multi_day, "real_seconds": round(time.time() - self.t0_real, 1), "timeline_events": len(self.timeline), "pending_at_end": _bounded(self.pending)} diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 8098f1a4b..f260e0e8f 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -43,6 +43,7 @@ from agent_env.task_step.task_steps.post_turn_hooks import PostTurnHooks, run_post_turn_hooks, validate_hook_specs from agent_env.task_step.task_steps.agent import user_sim_versions from agent_env.task_step.task_steps.agent.user_sim_common import iso, person_of, task_zone, utc +from agent_env.task_step.task_steps.agent.user_sim_common import attached, for_agent, opening_text, recorded_opening from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import ( agent_error_text, fetch_container_logs, @@ -109,6 +110,28 @@ def _duplicates_prompt_text(parts: list[dict], prompt_text: Optional[str]) -> bo return parts == [{"kind": "text", "text": prompt_text}] +def apply_seed(parts: list[dict], seed: dict) -> list[dict]: + """Return a deep-copied parts list with seed substitutions applied. + `` placeholders are replaced inside text parts' `text` and inside + file parts' `uri`/`name` (so tasks-as-templates with an object URL + ending `seeds//x.png` resolve per-run).""" + if not seed: + return parts + out = copy.deepcopy(parts) + for p in out: + kind = p.get("kind") + if kind == "text": + for k, v in seed.items(): + p["text"] = p["text"].replace(f"<{k}>", str(v)) + elif kind == "file": + f = p.get("file") or {} + for field in ("uri", "name"): + if field in f and isinstance(f[field], str): + for k, v in seed.items(): + f[field] = f[field].replace(f"<{k}>", str(v)) + return out + + def _file_uris(parts: list[dict]) -> list[str]: files = (part.get("file") for part in parts if part.get("kind") == "file") return [file["uri"] for file in files if isinstance(file, dict) and isinstance(file.get("uri"), str)] @@ -251,25 +274,7 @@ def _effective_parts(self) -> list[dict]: return [{"kind": "text", "text": self.prompt}] def _apply_seed(self, parts: list[dict], seed: dict) -> list[dict]: - """Return a deep-copied parts list with seed substitutions applied. - `` placeholders are replaced inside text parts' `text` and inside - file parts' `uri`/`name` (so tasks-as-templates with an object URL - ending `seeds//x.png` resolve per-run).""" - if not seed: - return parts - out = copy.deepcopy(parts) - for p in out: - kind = p.get("kind") - if kind == "text": - for k, v in seed.items(): - p["text"] = p["text"].replace(f"<{k}>", str(v)) - elif kind == "file": - f = p.get("file") or {} - for field in ("uri", "name"): - if field in f and isinstance(f[field], str): - for k, v in seed.items(): - f[field] = f[field].replace(f"<{k}>", str(v)) - return out + return apply_seed(parts, seed) def to_dict(self) -> dict: base = super().to_dict() @@ -428,12 +433,18 @@ def _built_in_user_sim(self, context: TaskStepContext) -> Optional[tuple[dict, s "each reply at once); drop them from that step's persona") seed = context.metadata.get("seed", {}) parts = self._apply_seed(self._effective_parts(), seed) - deployed_with = self._apply_seed([{"kind": "text", "text": record.get("opening") or ""}], seed)[0]["text"] + deployed_with = self._apply_seed(recorded_opening(record), seed) # TODO(FD-5495): the first message is written twice, here and as the user sim's opening_prompt; write it once. - if (_joined_text(parts) or "").strip() != deployed_with.strip(): + if (_joined_text(parts) or "").strip() != opening_text(deployed_with).strip(): # Its prompt quotes the opening it was deployed with as what the user sent; another one would contradict it. raise ValueError(f"prompt_agent '{self.id}': its prompt isn't the opening_prompt its user " f"'{self.user_agent_name}' was deployed with; give both the same text") + mine, theirs = _files(parts), _files(deployed_with) + if mine != theirs: + # The user's prompt says what they attached; the agent must be sent those files, in that order. + raise ValueError(f"prompt_agent '{self.id}': its prompt's files ({_described(mine)}) aren't the files of " + f"the opening_prompt its user '{self.user_agent_name}' was deployed with " + f"({_described(theirs)}); give both the same files in the same order") clocks = context.metadata.get("clock_configurations") or [] return record, task_zone(record, clocks, f"prompt_agent '{self.id}'") @@ -516,6 +527,8 @@ async def _converse( for key, value in seed.items(): system_prompt = system_prompt.replace(f"<{key}>", str(value)) initial_parts = self._apply_seed(self._effective_parts(), seed) + if built_in is not None: # a file part's description, copied from the opening_prompt, is for the user sim + initial_parts = [for_agent(p) for p in initial_parts] prompt_text = initial_parts[0]["text"] if self.prompt is not None else None model, effort = self._resolve_model_and_effort(context) @@ -1259,7 +1272,11 @@ async def start(cls, step: PromptAgentTaskStep, context: TaskStepContext, record initial_parts: list[dict], clock_envs: list) -> _BuiltInUserSim: """The user ``record`` describes, already checked by ``_built_in_user_sim``, in time zone ``zone``.""" text = _joined_text(initial_parts) - opening = {"at": await _clock_now(step, clock_envs), "text": text} if text else None + # What the user attached, as their prompt says: the opening's files, with their descriptions, which + # _built_in_user_sim checked are the ones this step sends. + files = attached(step._apply_seed(recorded_opening(record), context.metadata.get("seed", {}))) + opening = ({"at": await _clock_now(step, clock_envs), "text": text, **({"files": files} if files else {})} + if text else None) person = {**person_of(record), "max_replies": record.get("max_replies"), "time_zone": zone, "opening": opening, "done_ends_run": True} return cls(step, context, person, opening, clock_envs) @@ -1332,6 +1349,19 @@ def _agent_text(solver_message: str, agent_parts: Optional[list[dict]]) -> str: return _joined_text(agent_parts or []) or solver_message +def _files(parts: list[dict]) -> list[dict]: + """The file parts' files, in order, as the match check on the user's opening compares them.""" + return [p.get("file") or {} for p in parts if p.get("kind") == "file"] + + +def _described(files: list[dict]) -> str: + """Files named for an error: each one's name, address and type, or its size when it is sent inline.""" + def one(f: dict) -> str: + where = f.get("uri") or (f"{len(f['bytes'])} bytes inline" if isinstance(f.get("bytes"), str) else "?") + return " ".join(x for x in (f.get("name"), f"<{where}>", f.get("mimeType") or f.get("mime_type")) if x) + return "; ".join(map(one, files)) or "none" + + def _joined_text(parts: list[dict]) -> Optional[str]: """The prompt's text parts joined, as always_on reads them; None when there are none.""" return "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text") or None diff --git a/src/agent_env/training/harness.py b/src/agent_env/training/harness.py index 00ca38851..7f09217ac 100644 --- a/src/agent_env/training/harness.py +++ b/src/agent_env/training/harness.py @@ -35,6 +35,7 @@ from agent_env.task.task import Task, recorded_scores from agent_env.task.teardown import teardown_run from agent_env.task_step.context import TaskStepContext, sandbox_overrides +from agent_env.task_step.task_steps.agent.user_sim_common import opening_parts, opening_text from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.prompt_agent import is_prompt_step, user_sim_agents from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep @@ -233,8 +234,9 @@ def _index_named(prompt: Any, ids: list[str]) -> int: def _first_prompt(task: Task) -> str: """The user's first message to the agent under test: its step's prompt, else the opening of the user sim the step - names (an always_on step's ``user_sim``, a prompt_agent step's ``user_agent_name``), as it was written. A step that - prompts an agent another step talks to as its user (the user sim, say a briefing) isn't the agent under test's.""" + names (an always_on step's ``user_sim``, a prompt_agent step's ``user_agent_name``), as it was written (its + text, for an opening with files). A step that prompts an agent another step talks to as its user (the user sim, + say a briefing) isn't the agent under test's.""" users = user_sim_agents(task.steps) step = next((each for each in task.steps if is_prompt_step(each) and getattr(each, "agent_name", None) not in users), None) @@ -245,7 +247,8 @@ def _first_prompt(task: Task) -> str: user = getattr(step, "user_sim", None) or getattr(step, "user_agent_name", None) sim = next((each for each in task.steps if user and getattr(each, "agent_name", None) == user and getattr(each, "opening_prompt", None)), None) - return getattr(sim, "opening_prompt", None) or "" + opening = getattr(sim, "opening_prompt", None) or "" + return opening if isinstance(opening, str) else opening_text(opening_parts(opening)) def _free_port() -> int: 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 2992a85ba..c92557584 100644 --- a/tst/unit/task_step/test_deploy_user_sim_agent.py +++ b/tst/unit/task_step/test_deploy_user_sim_agent.py @@ -8,6 +8,7 @@ import hashlib import json import logging +import re from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -96,6 +97,30 @@ async def test_the_prompt_set_on_the_user_s_agent_is_the_tested_one_byte_for_byt assert context.metadata["models"]["user simulator"]["user"] == "sim-model" +@pytest.mark.asyncio +async def test_the_record_keeps_each_tag_s_line_so_an_old_used_resolves_after_the_persona_is_reordered(monkeypatch): + record = await _published(monkeypatch, _step()) + seen = record["tagged_lines"] + # Every tag the prompt shows, each with its line as the prompt shows it. + assert list(seen) == re.findall(r"^ ([A-Z]\d+) ", PROMPT, re.M) + assert all(f" {tag:<4}{line.split(chr(10))[0]}" in PROMPT for tag, line in seen.items()) + decisions = HANNAH["description"]["decisions"] + assert seen["D9"] == decisions[8] + assert seen["W2"] == f'Earlier you texted the assistant:\n"{PERSON["opening"]["text"]}"' + used = "D9" # what the user sim answered in that run + + # The task writer later reorders the decisions and adds one at the top: D9 now names another line. + edited = {**HANNAH, "description": {**HANNAH["description"], + "decisions": ["Anything over $600 for the weekend: ask you first.", + *reversed(decisions)]}} + now = (await _published(monkeypatch, _step(persona=edited)))["tagged_lines"] + assert now["D9"] != decisions[8] + # The old run's record still resolves its used to the line its user sim saw. + assert seen[used] == decisions[8] + # Recording the map leaves the prompt as tested. + assert record["prompt_sha256"] == hashlib.sha256(PROMPT.encode()).hexdigest() + + @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/test_user_sim_opening_files.py b/tst/unit/task_step/test_user_sim_opening_files.py new file mode 100644 index 000000000..e16faa3d0 --- /dev/null +++ b/tst/unit/task_step/test_user_sim_opening_files.py @@ -0,0 +1,326 @@ +"""The user's first message with files (FD-5494), on the parents' visit task with an itinerary PDF: deploy_user_sim_agent +takes ``opening_prompt`` as text and file parts and tells the user sim what it attached; always_on sends the files on +the first wake in their place and the graded conversation has them; prompt_agent's match check compares the files; +the seed fills a file's address; a text-only opening is unchanged.""" + +from __future__ import annotations + +import copy +import json + +import pytest + +from agent_env.a2a_agent import conversation_store +from agent_env.a2a_agent.object_transfer import readable_parts +from agent_env.config import configure +from agent_env.task_step.context import TaskStepContext +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, DeployUserSimAgentTaskStep +from agent_env.task_step.task_steps.agent.user_sim_common import attached, file_name, for_agent +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from tst.unit.task_step.always_on_harness import Rig +from tst.unit.task_step.test_deploy_user_sim_agent import HANNAH, PERSON, PROMPT, _Deploy, _published, _step +from tst.unit.task_step.test_prompt_agent_built_in_user_sim import USER, _context, _full, _reads, said +from tst.unit.task_step.test_prompt_agent_user_side import AGENT, Peers, exchange, text +from tst.util.granting_object_store import GRANT_ORIGIN, GrantingObjectStore + +TEXT = PERSON["opening"]["text"] +SEED = {"trip": "nov-19"} +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//hotel-shortlist.xlsx", + "name": "hotel-shortlist.xlsx"}} +OPENING = [{"kind": "text", "text": TEXT}, ITINERARY, SHORTLIST] +NOTE = "(you attached itinerary.pdf: Mom and Dad's flight itinerary) (you attached hotel-shortlist.xlsx)" +KNOWN = ("You remember only what's said here about each file you attached, not what's in it; if the assistant asks " + "what's in one, say you don't remember or point it to the file.") + + +def _sent(part: dict) -> dict: + """A file part as the agent is sent it this run: the seed filled in, the description dropped.""" + file = {**part["file"], "uri": part["file"]["uri"].replace("", SEED["trip"])} + return {"kind": "file", "file": file} + + +def _seeded() -> TaskStepContext: + context = TaskStepContext() + context.metadata["seed"] = dict(SEED) + return context + + +# ---- deploy_user_sim_agent ------------------------------------------------------------------------------------------ +@pytest.mark.asyncio +async def test_a_text_only_opening_is_recorded_and_prompted_as_before(monkeypatch): + record = await _published(monkeypatch, _step()) + assert record["opening"] == TEXT and "opening_parts" not in record + deploy = _Deploy(monkeypatch) + await deploy.run(_step(opening_prompt=[{"kind": "text", "text": TEXT}])) + assert deploy.fields("system_prompt") == [PROMPT] + assert user_sim_versions.ATTACHED_KNOWN == KNOWN and KNOWN not in PROMPT + + +@pytest.mark.asyncio +async def test_the_prompt_says_what_the_user_attached_by_description_else_by_name(monkeypatch): + deploy = _Deploy(monkeypatch) + context = await deploy.run(_step(opening_prompt=OPENING), _seeded()) + + assert PROMPT.count(f'"{TEXT}"\n') == 1 + # Under the opening's line, hanging in W2: it knows only what's said of the files, never what's in them. + assert deploy.fields("system_prompt") == [PROMPT.replace(f'"{TEXT}"\n', f'"{TEXT}" {NOTE}\n {KNOWN}\n')] + record = context.metadata[USER_SIMS_KEY]["hannah"] + # The text as before, so text readers keep working; the parts as written, the seed still to fill in. + assert record["opening"] == TEXT + assert record["opening_parts"] == OPENING + assert record["tagged_lines"]["W2"] == f'Earlier you texted the assistant:\n"{TEXT}" {NOTE}\n{KNOWN}' + + +@pytest.mark.asyncio +async def test_the_seed_fills_a_file_s_name_in_what_the_user_sim_is_told(monkeypatch): + deploy = _Deploy(monkeypatch) + named = {**ITINERARY, "file": {**ITINERARY["file"], "name": "-itinerary.pdf"}} + await deploy.run(_step(opening_prompt=[OPENING[0], named]), _seeded()) + [prompt] = deploy.fields("system_prompt") + assert f'"{TEXT}" (you attached nov-19-itinerary.pdf: Mom and Dad\'s flight itinerary)\n' in prompt + + +def test_a_step_with_an_opening_of_parts_round_trips(): + step = _step(opening_prompt=OPENING) + assert DeployUserSimAgentTaskStep.from_dict(step.to_dict()).to_dict() == step.to_dict() + assert step.to_dict()["opening_prompt"] == OPENING + + +@pytest.mark.parametrize("opening, problem", [ + ([ITINERARY], "opening_prompt needs a text part with text"), + ([{"kind": "text", "text": " "}, ITINERARY], "opening_prompt needs a text part with text"), + ([], "opening_prompt needs a text part with text"), + ({"text": TEXT}, "opening_prompt must be text, or a list of text and file parts"), + ([{"kind": "text", "text": TEXT}, "itinerary.pdf"], "opening_prompt[1] must be a part, an object"), + ([{"kind": "text", "text": TEXT}, {"kind": "data", "data": {"flight": "UA 1534"}}], + "opening_prompt[1] is a 'data' part; the user's first message takes text and file parts"), + ([{"kind": "text", "text": TEXT}, {"kind": "file", "file": {"name": "itinerary.pdf"}}], + "opening_prompt[1] is not a valid A2A file part (file.FileWithBytes.bytes: Field required; " + "file.FileWithUri.uri: Field required)"), + ([{"kind": "text"}], "opening_prompt[0] is not a valid A2A text part (text: Field required)"), + ([{"kind": "text", "text": TEXT}, {**ITINERARY, "metadata": {"description": "Mom and Dad's\nflights"}}], + "opening_prompt[1]'s metadata.description must be one line of text"), + ([{"kind": "text", "text": TEXT}, {**ITINERARY, "metadata": {"description": " "}}], + "opening_prompt[1]'s metadata.description must be one line of text"), + ([{"kind": "text", "text": TEXT}, {**ITINERARY, "metadata": {"description": ["flights"]}}], + "opening_prompt[1]'s metadata.description must be one line of text"), +], ids=["no-text", "blank-text", "empty", "not-a-list", "not-a-part", "data-part", "file-without-address", + "text-without-text", "two-line-description", "blank-description", "description-not-text"]) +def test_a_bad_opening_is_refused_when_the_task_loads(opening, problem): + with pytest.raises(ValueError, match="deploy_user_sim_agent 'hannah'") as e: + _step(opening_prompt=opening) + assert problem in str(e.value) + + +def test_a_file_is_named_by_its_name_else_its_address_and_only_its_description_is_kept_from_the_agent(): + assert file_name({"kind": "file", "file": {"uri": "s3://reyes-family/scans/boarding-pass.png"}}) \ + == "boarding-pass.png" + assert file_name({"kind": "file", "file": {"bytes": "aGk="}}) == "a file" + assert attached(OPENING) == [{"name": "itinerary.pdf", "description": "Mom and Dad's flight itinerary"}, + {"name": "hotel-shortlist.xlsx"}] + tagged = {**ITINERARY, "metadata": {**ITINERARY["metadata"], "source": "intake"}} + assert for_agent(tagged) == {**ITINERARY, "metadata": {"source": "intake"}} + assert for_agent(ITINERARY) == {"kind": "file", "file": ITINERARY["file"]} + assert for_agent(SHORTLIST) is SHORTLIST + + +def test_a_read_shows_what_the_user_attached_after_their_line(): + person = {**PERSON, "opening": None} + sent = {"at": "2026-11-16T15:50:00Z", "text": TEXT, + "files": [{"name": "itinerary.pdf", "description": "Mom and Dad's flight itinerary"}, + {"name": "hotel-shortlist.xlsx"}]} + read = {"time": "2026-11-16T16:05:00Z", "assistant_messages": [{"at": "2026-11-16T16:05:00Z", "text": "On it."}], + "phone": [], "you_sent": [sent], "done_ends_run": False} + without = {**read, "you_sent": [{k: v for k, v in sent.items() if k != "files"}]} + assert user_sim_versions.render(person, [], read) == user_sim_versions.render(person, [], without).replace( + f" [7:50 AM] {TEXT}\n", f" [7:50 AM] {TEXT} {NOTE}\n") + + +# ---- always_on ---------------------------------------------------------------------------------------------------- +async def _always_on_context(monkeypatch, rig: Rig, opening) -> TaskStepContext: + """The Rig's world with Hannah as deploy_user_sim_agent records her, her first message ``opening``.""" + record = await _published(monkeypatch, _step(agent_name="user-sim", opening_prompt=opening, multi_day=False, + persona={**HANNAH, "time_zone": "America/Los_Angeles"}), _seeded()) + ctx = rig.context() + ctx.metadata[USER_SIMS_KEY] = {"user-sim": record} + ctx.metadata["seed"] = dict(SEED) + ctx.metadata["clock_configurations"] = [{"step_id": "clock", "timezone": "America/Los_Angeles"}] + return ctx + + +def _always_on() -> ao.AlwaysOnTaskStep: + return ao.AlwaysOnTaskStep("week", None, user_sim="user-sim", prompt_id="ask", agent_name="solver") + + +def _asks_once(n, texts, rig): + if n == 1: + rig.message("Booking their flights now. Any seat preference?") + return "sent" + rig.quiet() + return "" + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_always_on_sends_the_files_on_the_first_wake_in_their_place_and_the_chat_and_the_user_sim_have_them( + monkeypatch, tmp_path): + rig = Rig(monkeypatch, tmp_path, agent=_asks_once, plugin=user_sim_versions, t0="2026-11-16T15:50:00Z", + user_texts=['{"outcome": "done", "message": "aisle for dad, thanks!", "used": "none"}']) + await rig.arm() + later = {"kind": "text", "text": "the hotels i looked at are in the sheet"} + ctx = await rig.run(_always_on(), await _always_on_context(monkeypatch, rig, [*OPENING[:2], later, SHORTLIST])) + + line = f"[Mon Nov 16, 3:50 PM UTC] Hannah: {TEXT}\n{later['text']}" + opening = rig.parts[0] + assert [p.get("text") for p in opening if p.get("kind") == "text"].count(line) == 1 + at = next(i for i, p in enumerate(opening) if p.get("text") == line) + assert opening[at + 1:] == [_sent(ITINERARY), _sent(SHORTLIST)] # in place, the seed filled in, no description + [first_turn, *_] = [m for m in rig.chat(ctx) if m["role"] == "user"] + assert [p.get("kind") for p in first_turn["parts"]] == ["text", "file", "file"] + assert first_turn["parts"][1:] == [_sent(ITINERARY), _sent(SHORTLIST)] + [call] = rig.user_calls + assert (f"\nYOU SENT YOUR AI ASSISTANT:\n [7:50 AM] {TEXT}\n{later['text']} {NOTE}\n" + in call["parts"][0]["text"]) + # The opening wake's timeline event names its files, by name and address as the agent was sent them; no + # description, which is the user sim's. Later wakes carry no files and name none. + [first_wake, *later_wakes] = rig.events(ctx, "wake") + assert first_wake["files"] == [{"name": "itinerary.pdf", "uri": "s3://reyes-family/nov-19/itinerary.pdf"}, + {"name": "hotel-shortlist.xlsx", "uri": "s3://reyes-family/nov-19/hotel-shortlist.xlsx"}] + assert "Mom and Dad's flight itinerary" not in json.dumps(first_wake) + assert later_wakes and all("files" not in w for w in later_wakes) + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_always_on_with_a_text_only_opening_sends_and_shows_only_the_text(monkeypatch, tmp_path): + rig = Rig(monkeypatch, tmp_path, agent=_asks_once, plugin=user_sim_versions, t0="2026-11-16T15:50:00Z", + user_texts=['{"outcome": "done", "message": "aisle for dad, thanks!", "used": "none"}']) + await rig.arm() + ctx = await rig.run(_always_on(), await _always_on_context(monkeypatch, rig, TEXT)) + + assert {p.get("kind") for p in rig.parts[0]} == {"text"} + assert all("files" not in w for w in rig.events(ctx, "wake")) + [first_turn, *_] = [m for m in rig.chat(ctx) if m["role"] == "user"] + assert first_turn["texts"] == [f"[Mon Nov 16, 3:50 PM UTC] Hannah: {TEXT}"] + [call] = rig.user_calls + assert f"\nYOU SENT YOUR AI ASSISTANT:\n [7:50 AM] {TEXT}\n\nNEW FROM" in call["parts"][0]["text"] + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_an_opening_file_the_store_owns_reaches_the_agent_as_a_grant_like_any_prompt_file(monkeypatch, tmp_path): + store = GrantingObjectStore(str(tmp_path / "objects"), reaches=True) + configure(object_store=store) + url = store.put("trips/nov-19/itinerary.pdf", b"%PDF") + rig = Rig(monkeypatch, tmp_path, agent=_asks_once, plugin=user_sim_versions, t0="2026-11-16T15:50:00Z", + user_texts=['{"outcome": "done", "message": "aisle for dad, thanks!", "used": "none"}']) + scripted = ao.send_and_wait + + async def send_and_wait(a2a_url, parts, *, agent=None, timeout_seconds=600, **kw): + # The real send_and_wait's sharing, in front of the Rig's scripted agent. + if agent is None: + return await scripted(a2a_url, parts, timeout_seconds=timeout_seconds, **kw) + async with readable_parts(parts, a2a_url=a2a_url, card=agent.a2a_card, sandbox_type=agent.sandbox_type, + expires_in=timeout_seconds) as sent: + return await scripted(a2a_url, sent, agent=agent, timeout_seconds=timeout_seconds, **kw) + + monkeypatch.setattr(ao, "send_and_wait", send_and_wait) + await rig.arm() + owned = {**ITINERARY, "file": {**ITINERARY["file"], "uri": store.object_url("trips//itinerary.pdf")}} + ctx = await rig.run(_always_on(), await _always_on_context(monkeypatch, rig, [OPENING[0], owned])) + + assert rig.parts[0][-1] == {"kind": "file", "file": {**ITINERARY["file"], + "uri": f"{GRANT_ORIGIN}/trips/nov-19/itinerary.pdf?sig=read"}} + [first_turn, *_] = [m for m in rig.chat(ctx) if m["role"] == "user"] + assert first_turn["parts"][-1]["file"]["uri"] == url # the conversation keeps the object's own address + + +# ---- prompt_agent with the built-in user sim ---------------------------------------------------------------------- +def _prompt_agent(parts: list[dict]) -> PromptAgentTaskStep: + return PromptAgentTaskStep(id="talk", version=None, agent_name="solver", user_agent_name="hannah", parts=parts, + poll_interval_seconds=0, max_conversation_turns=3) + + +async def _hannah(monkeypatch, opening) -> TaskStepContext: + ctx = await _context(monkeypatch, opening_prompt=opening) + ctx.metadata["seed"] = dict(SEED) + return ctx + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_prompt_agent_sends_the_opening_s_files_and_the_user_sim_knows_what_it_attached(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("Booking their flights now.")], USER: [said("done")]}) + ctx = await _hannah(monkeypatch, OPENING) + # The same files, without the description, which is the user sim's. + sent = [{"kind": "text", "text": TEXT}, for_agent(ITINERARY), SHORTLIST] + + async def armed_clock(self, envs): # one armed clock, standing still at 7:50 AM in San Francisco + return {"env_clock": [{"env_id": "env", "env_step_id": "clock", "virtual_time": "2026-11-16T15:50:00+00:00"}]} + + monkeypatch.setattr(PromptAgentTaskStep, "_env_clock", armed_clock) + await _prompt_agent(sent).execute(ctx) + + first = _full(_reads(ctx)[0]) + assert peers.requests(ctx)[:2] == exchange( + AGENT, [{"kind": "text", "text": TEXT}, _sent(ITINERARY), _sent(SHORTLIST)], "", "", "task-1") + assert first["read"]["you_sent"][0]["files"] == [ + {"name": "itinerary.pdf", "description": "Mom and Dad's flight itinerary"}, {"name": "hotel-shortlist.xlsx"}] + assert f"\nYOU SENT YOUR AI ASSISTANT:\n [3:50 PM] {TEXT} {NOTE}\n" in first["input"] # Hannah's zone: UTC + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_prompt_agent_with_the_opening_s_parts_copied_sends_and_stores_them_without_the_description(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("Booking their flights now.")], USER: [said("done")]}) + ctx = await _hannah(monkeypatch, OPENING) + await _prompt_agent(copy.deepcopy(OPENING)).execute(ctx) # the opening_prompt's parts, descriptions and all + + assert peers.requests(ctx)[:2] == exchange( + AGENT, [{"kind": "text", "text": TEXT}, _sent(ITINERARY), _sent(SHORTLIST)], "", "", "task-1") + doc = conversation_store.get_conversation(ctx.metadata["a2a_conversations"]["talk"]) + assert doc["messages"][0]["parts"] == [{"kind": "text", "text": TEXT}, _sent(ITINERARY), _sent(SHORTLIST)] + assert ctx.prompt_responses[0].source_agent_per_turn_prompt_parts[0] == doc["messages"][0]["parts"] + assert "Mom and Dad's flight itinerary" not in json.dumps([r[2] for r in peers.raw if r[1].startswith(AGENT)]) + assert _full(_reads(ctx)[0])["read"]["you_sent"][0]["files"] == [ # the user sim still knows what it attached + {"name": "itinerary.pdf", "description": "Mom and Dad's flight itinerary"}, {"name": "hotel-shortlist.xlsx"}] + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +@pytest.mark.parametrize("files, mine", [ + ([SHORTLIST, ITINERARY], "hotel-shortlist.xlsx ; itinerary.pdf " + " application/pdf"), + ([ITINERARY], "itinerary.pdf application/pdf"), + ([{**ITINERARY, "file": {**ITINERARY["file"], "uri": "s3://reyes-family//itinerary-v2.pdf"}}, SHORTLIST], + "itinerary.pdf application/pdf; hotel-shortlist.xlsx " + ""), + ([], "none"), +], ids=["order", "missing", "other-address", "text-only"]) +async def test_prompt_agent_refuses_a_prompt_whose_files_aren_t_the_opening_s(monkeypatch, files, mine): + peers = Peers(monkeypatch, {}) + ctx = await _hannah(monkeypatch, OPENING) + with pytest.raises(ValueError) as e: + await _prompt_agent([{"kind": "text", "text": TEXT}, *copy.deepcopy(files)]).execute(ctx) + assert str(e.value) == ( + f"prompt_agent 'talk': its prompt's files ({mine}) aren't the files of the opening_prompt its user 'hannah' " + "was deployed with (itinerary.pdf application/pdf; " + "hotel-shortlist.xlsx ); give both the same files in the same " + "order") + assert peers.raw == [] + + +@pytest.mark.usefixtures("local_stores") +@pytest.mark.asyncio +async def test_prompt_agent_refuses_files_the_text_only_opening_never_had(monkeypatch): + Peers(monkeypatch, {}) + ctx = await _hannah(monkeypatch, TEXT) + with pytest.raises(ValueError, match=r"its prompt's files \(itinerary.pdf " + r"application/pdf\) aren't the files of the opening_prompt its user 'hannah' " + r"was deployed with \(none\)"): + await _prompt_agent([{"kind": "text", "text": TEXT}, ITINERARY]).execute(ctx) diff --git a/tst/unit/task_step/test_user_sim_satisfaction.py b/tst/unit/task_step/test_user_sim_satisfaction.py new file mode 100644 index 000000000..8c9a39e40 --- /dev/null +++ b/tst/unit/task_step/test_user_sim_satisfaction.py @@ -0,0 +1,529 @@ +"""The user sim's rating of the run (FD-5491), a step of its own: rate_user_satisfaction runs after an always_on or +prompt_agent run and asks the user sim once, in the session the run used, to rate the job 1-5 with a reason, by the +step's satisfaction lines. The lines are never in the persona or the user sim's prompt, the run steps are as they were +without a rating, and the rating is recorded, never delivered, and never fails the task.""" + +from __future__ import annotations + +import asyncio +import copy +import json +from pathlib import Path + +import pytest + +from agent_env.a2a_agent import conversation_store +from agent_env.store import get_config +from agent_env.task.task import Task +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 rate_user_satisfaction as rus +from agent_env.task_step.task_steps.agent import user_sim_versions as us +from agent_env.task_step.task_steps.agent.deploy_user_sim_agent import persona_problems +from agent_env.task_step.task_steps.agent.rate_user_satisfaction import RateUserSatisfactionTaskStep +from agent_env.task_step.task_steps.agent.user_sim_common import SATISFACTION_KEY +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from tst.unit.task_step.always_on_harness import Fail, Rig +from tst.unit.task_step.test_deploy_user_sim_agent import DESCRIPTION, HANNAH, _always_on, _Deploy, _rig_context +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 AGENT, USER, Peers, _context, _step, said, text + +pytestmark = pytest.mark.usefixtures("local_stores") + +DATA = Path(__file__).parents[2] / "data" / "user_sim" +PERSON = json.loads((DATA / "hannah_person.json").read_text()) +PROMPT = (DATA / "hannah_system_prompt.txt").read_text() +SATISFIED = ["5, perfect: flights, a step-free place a walk from you, and both dinners booked, with nothing you had " + "to chase.", + "3, fine: all booked, but you had to answer a question your rules already settled.", + "1, you'd have done it yourself: a dinner or the place is still not booked at the end.", + "Being asked for things that are in Mom's email bothers you."] +LINDA_LINES = ["5: the house and the dinner booked.", "1: nothing booked."] +NOTICE = ao.FINAL_REMINDER.format(first="Linda") + + +def rated(score, message="all booked, no chasing", used="S1", outcome="done") -> dict: + return {"outcome": outcome, "score": score, "message": message, "used": used} + + +def _rater(**fields) -> RateUserSatisfactionTaskStep: + return RateUserSatisfactionTaskStep.from_dict({"id": "rate", "type": "rate_user_satisfaction", + "satisfaction": LINDA_LINES, "poll_interval_seconds": 0, **fields}) + + +def _asker(*answers): + sent = [] + + async def ask(text): + sent.append(text) + got = answers[len(sent) - 1] + if isinstance(got, dict) and "error" in got: + return got + return {"text": got if isinstance(got, str) else json.dumps(got), + "structured": got if isinstance(got, dict) else None, "trajectory_url": None} + + return ask, sent + + +NO_UNREAD = {"time": "2026-11-24T05:00:00Z", "assistant_messages": []} + + +# ---- the user sim's prompt and persona: no satisfaction lines, ever ---------------------------------------------------- +def test_the_user_sims_prompt_is_the_tested_one_and_never_has_s_lines(): + prompt = us.setup("ours", PERSON)["system_prompt"] + assert prompt == PROMPT and "HOW YOU'D RATE" not in prompt and " S1" not in prompt + assert all(key != "satisfaction" for key, _, _ in us.SECTIONS) + + +@pytest.mark.asyncio +async def test_with_a_rating_step_the_deployed_prompt_is_still_the_tested_one(monkeypatch): + deploy = _Deploy(monkeypatch) + step = _deploy_step() + Task("t", None, [step, PromptAgentTaskStep(id="talk", version=None, prompt=PERSON["opening"]["text"], + agent_name="solver", user_agent_name="hannah"), + _rater(satisfaction=SATISFIED)]) + await deploy.run(step) + assert deploy.fields("system_prompt") == [PROMPT] + + +def test_a_persona_with_satisfaction_lines_is_refused_and_pointed_at_the_step(): + person = copy.deepcopy(PERSON) + person["description"]["satisfaction"] = SATISFIED + [problem] = [p for p in us.problems(person) if "unknown sections" in p] + assert "satisfaction goes on the rate_user_satisfaction step" in problem + with pytest.raises(ValueError, match="rate_user_satisfaction"): + _deploy_step(persona={**HANNAH, "description": person["description"]}) + assert any("satisfaction" in p for p in persona_problems({**HANNAH, "description": person["description"]})) + + +# ---- the ask ------------------------------------------------------------------------------------------------------------ +def test_the_rating_s_tag_map_is_the_tags_the_ask_shows(): + ask = us.render_rating(PERSON, SATISFIED, NO_UNREAD) + lines = us.rating_tagged_lines(SATISFIED) + assert list(lines) == ["S1", "S2", "S3", "S4"] + assert all(f"\n {tag:<4}{line}\n" in ask + "\n" for tag, line in lines.items()) + + +def test_the_rating_ask_word_for_word(): + read = {"time": "2026-11-24T05:00:00Z", "assistant_messages": [ + {"at": "2026-11-24T04:55:00Z", "text": "All set: flights, the Castro Street flat, both dinners.\n\nEnjoy!"}]} + assert us.render_rating(PERSON, SATISFIED, read) == ( + "It's Mon, Nov 23, 9:00 PM. The job is over: your AI assistant won't see this, or anything you write from now " + "on.\n" + "\nNEW FROM YOUR AI ASSISTANT (you haven't read this yet):\n" + " [8:55 PM]\n" + " All set: flights, the Castro Street flat, both dinners.\n" + "\n" + " Enjoy!\n" + "\nHOW YOU'D RATE THE JOB\n" + " S1 5, perfect: flights, a step-free place a walk from you, and both dinners booked, with nothing you had to " + "chase.\n" + " S2 3, fine: all booked, but you had to answer a question your rules already settled.\n" + " S3 1, you'd have done it yourself: a dinner or the place is still not booked at the end.\n" + " S4 Being asked for things that are in Mom's email bothers you.\n" + "\nAs Hannah, rate how the job went, from everything you've seen: 5 if it was done perfectly, 3 if it wasn't " + "perfect but was fine, 1 if you'd have done it yourself; 2 and 4 are in between. HOW YOU'D RATE THE JOB says " + "what those mean for this job. Something you were never told was done doesn't count as done.\n" + 'In "message", give your reason in one line. In "used", write the tag of the line it rests on, or "none". ' + 'Answer in JSON: {"outcome": "done", "score": 1 | 2 | 3 | 4 | 5, "message": "...", "used": "..."}') + no_unread = us.render_rating(PERSON, SATISFIED, NO_UNREAD) + assert "NEW FROM" not in no_unread and no_unread.startswith("It's Mon, Nov 23, 9:00 PM. The job is over") + assert "\n\nHOW YOU'D RATE THE JOB\n S1 5, perfect" in no_unread + + +@pytest.mark.asyncio +async def test_a_rating_is_asked_once_and_parsed(): + ask, sent = _asker(rated(4, " all booked, but it asked about the dates ", "S2")) + got = await us.rate("ours", PERSON, SATISFIED, NO_UNREAD, ask) + assert len(sent) == 1 and sent[0] == got["input"] + assert (got["score"], got["reason"], got["used"], got["schema_errors"], got["retried"], got["turns"]) == ( + 4, "all booked, but it asked about the dates", "S2", [], False, 1) + + +@pytest.mark.parametrize("raw, error", [ + (rated(7), "score 7 is not one of [1, 2, 3, 4, 5]"), + (rated(0), "score 0 is not one of [1, 2, 3, 4, 5]"), + (rated(True), "score must be a whole number"), + ({"outcome": "done", "message": "fine"}, "missing score"), + (rated(3, " "), "message must give the reason"), + ("a solid 4", "the answer is not a JSON object"), +]) +def test_a_malformed_rating_has_no_score_and_says_why(raw, error): + got = us.parse_rating(raw) + assert got["score"] is None and error in got["schema_errors"], got + + +@pytest.mark.asyncio +async def test_a_malformed_rating_gets_one_retry(): + ask, sent = _asker("a solid 4", rated(4)) + got = await us.rate("ours", PERSON, SATISFIED, NO_UNREAD, ask) + assert sent[1] == us.RATING_RETRY and got["score"] == 4 and got["retried"] is True and got["turns"] == 2 + assert got["first_answer"] == {"raw": "a solid 4", "schema_errors": ["the answer is not a JSON object"]} + + ask, sent = _asker("a solid 4", rated(9)) + got = await us.rate("ours", PERSON, SATISFIED, NO_UNREAD, ask) + assert len(sent) == 2 and got["score"] is None + assert got["schema_errors"] == ["first answer: the answer is not a JSON object", + "answer after the retry: score 9 is not one of [1, 2, 3, 4, 5]"] + + +@pytest.mark.asyncio +async def test_a_user_sim_error_on_the_rating_is_not_retried(): + ask, sent = _asker({"text": None, "structured": None, "error": "timeout"}) + got = await us.rate("ours", PERSON, SATISFIED, NO_UNREAD, ask) + assert len(sent) == 1 and got["score"] is None and got["retried"] is False + + +# ---- the step's settings and the load check ----------------------------------------------------------------------------- +@pytest.mark.parametrize("lines", ["5: all booked", [], ["5: all booked", " "], [5, 3, 1], None]) +def test_bad_satisfaction_lines_are_refused(lines): + with pytest.raises(ValueError, match="satisfaction must be a list of non-empty lines"): + _rater(satisfaction=lines) + + +@pytest.mark.parametrize("fields, problem", [ + ({"timeout_seconds": 0}, "timeout_seconds must be a number of seconds, more than 0"), + ({"timeout_seconds": True}, "timeout_seconds must be a number of seconds"), + ({"run_step_id": " "}, "run_step_id must name"), +]) +def test_bad_settings_are_refused(fields, problem): + with pytest.raises(ValueError, match=problem): + _rater(**fields) + + +def test_the_step_round_trips_is_registered_and_never_fails_the_task_by_default(): + step = _rater(run_step_id="week", depends_on=["week"]) + again = RateUserSatisfactionTaskStep.from_dict(step.to_dict()) + assert again.to_dict() == step.to_dict() + assert (again.satisfaction, again.run_step_id, again.fail_task_on_error) == (LINDA_LINES, "week", False) + assert again.timeout_seconds == ao.AlwaysOnTaskStep("w", None, user_sim="u", prompt_id="p").read_cap_seconds() + assert _builtin_registry()["rate_user_satisfaction"] is RateUserSatisfactionTaskStep + + +def _sim(): + return _deploy_step(agent_name="user-sim") + + +def _talk(**kw): + return PromptAgentTaskStep(id="talk", version=None, prompt="plan it", agent_name="solver", + **{"user_agent_name": "user-sim", **kw}) + + +def test_the_only_run_step_is_rated_by_default_and_a_named_one_is_checked(): + Task("t", None, [_sim(), _always_on(end_time="P3D"), _rater()]) + Task("t", None, [_sim(), _always_on(end_time="P3D"), _rater(run_step_id="week", depends_on=["week"])]) + Task("t", None, [_sim(), _talk(), _rater()]) + + +@pytest.mark.parametrize("steps, problem", [ + (lambda: [_sim(), _rater()], "the task has no always_on or prompt_agent step to rate"), + (lambda: [_sim(), _always_on(end_time="P3D"), _talk(), _rater()], + "several always_on and prompt_agent steps (week, talk); set run_step_id"), + (lambda: [_sim(), _always_on(end_time="P3D"), _rater(run_step_id="later")], + "run_step_id 'later' names no step of the task"), + (lambda: [_sim(), _always_on(end_time="P3D"), _rater(run_step_id="user")], + "run_step_id 'user' is a deploy_user_sim_agent step"), + (lambda: [_sim(), _rater(), _always_on(end_time="P3D")], "comes before step 'week'"), + (lambda: [_sim(), _always_on(end_time="P3D"), _rater(depends_on=["user"])], "its depends_on must name 'week'"), + (lambda: [_sim(), _talk(user_agent_name="someone-else"), _rater()], + "the user of prompt_agent step 'talk' isn't a user sim a deploy_user_sim_agent step deploys"), + (lambda: [_sim(), _talk(user_a2a_url="http://human.test"), _rater()], + "the user of prompt_agent step 'talk' isn't a user sim"), + (lambda: [_sim(), _always_on(end_time="P3D"), _rater(), _rater(id="again")], + "step 'rate' already rates 'week'; a run is rated once"), +]) +def test_a_task_the_step_cant_rate_is_refused_when_it_loads(steps, problem): + with pytest.raises(ValueError, match="rate_user_satisfaction") as e: + Task("t", None, steps()) + assert problem in str(e.value) + + +# ---- after an always_on run --------------------------------------------------------------------------------------------- +def _wraps_up(n, texts, rig): + if n == 1: + rig.message("Blue house or green?") + return "sent" + if NOTICE in texts: + rig.message("RESULT: blue house booked, dinner at La Ola") + return "sent" + rig.quiet() + return "" + + +def _reply(message: str) -> str: + return json.dumps({"outcome": "reply", "message": message, "used": "W1"}) + + +async def _always_on_run(monkeypatch, tmp_path, *user_texts, agent=_wraps_up, persona=None, **settings): + rig = Rig(monkeypatch, tmp_path, agent=agent, plugin=us, user_texts=user_texts) + monkeypatch.setattr(rus, "send_and_wait", rig._send_and_wait) + await rig.arm() + persona = {"display_name": "Linda Murphy", "description": DESCRIPTION, **(persona or {})} + ctx = await _rig_context(monkeypatch, rig, persona=persona) + ctx = await rig.run(_always_on(**{"end_time": "P3D", **settings}), ctx) + return rig, ctx + + +def _delivered(rig, ctx) -> str: + """Everything the agent was sent, the graded chat, the responses and the verifications.""" + return (json.dumps(rig.sent) + json.dumps(rig.chat(ctx)) + json.dumps([r.response for r in ctx.prompt_responses]) + + json.dumps(ctx.metadata.get(ao.HOOK_RECORDS_KEY)) + json.dumps(ctx.metadata.get("verifications"))) + + +@pytest.mark.asyncio +async def test_an_always_on_task_without_the_step_asks_and_records_nothing_more(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks")) + assert len(rig.user_calls) == 2 and SATISFACTION_KEY not in ctx.metadata + assert "HOW YOU'D RATE" not in json.dumps(rig.user_calls) + + +@pytest.mark.asyncio +async def test_after_always_on_one_ask_in_the_runs_session_recorded_and_never_delivered(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks"), + json.dumps(rated(5, "house and dinner booked", "S1"))) + assert len(rig.user_calls) == 2 # the run asks nothing of a rating + before = _delivered(rig, ctx), len(ctx.prompt_responses), len(rig.timeline(ctx)) + + await _rater().execute(ctx) + + assert len(rig.user_calls) == 3 + ask = rig.user_calls[2]["parts"][0]["text"] + record = rig.record(ctx) + assert record["user_sim"]["context_id"] == rig.user_calls[0]["context_id"] + assert len({c["context_id"] for c in rig.user_calls}) == 1 # the run's one user sim session + assert us.RATING_ASK.format(first="Linda") in ask + assert "\nHOW YOU'D RATE THE JOB\n S1 5: the house and the dinner booked.\n S2 1: nothing booked.\n" in ask + assert "NEW FROM YOUR AI ASSISTANT (you haven't read" not in ask # the last read already showed the wrap-up + + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert (rating["user_sim"], rating["score"], rating["reason"], rating["used"], rating["rated_by"]) == ( + "user-sim", 5, "house and dinner booked", "S1", "rate") + assert rating["at"] == rig.events(ctx, "user_sim_read")[-1]["virtual_time"] + assert (rating["schema_errors"], rating["retried"], rating["user_sim_error"], rating["user_sim_turns"]) == ( + [], False, None, 1) + assert (rating["context_id"], rating["session_from"], rating["unread_shown"]) == ( + record["user_sim"]["context_id"], "always_on record", 0) + stored = json.loads(get_config().get_object_store().get(rating["object_url"])) + assert stored["input"] == ask and stored["satisfaction"] == LINDA_LINES + # The S tags the ask showed, with their lines, so used resolves however the step's lines change later. + assert rating["tagged_lines"] == {"S1": LINDA_LINES[0], "S2": LINDA_LINES[1]} + assert rating["tagged_lines"][rating["used"]] == "5: the house and the dinner booked." + + # Never delivered: the agent's wakes, the graded chat, the responses, the hook records and the verifications are as + # the run left them. + assert (_delivered(rig, ctx), len(ctx.prompt_responses), len(rig.timeline(ctx))) == before + assert "house and dinner booked" not in _delivered(rig, ctx) + assert record["counts"]["user_sim_reads"] == 2 and record["counts"]["user_replies"] == 2 + + +@pytest.mark.asyncio +async def test_after_always_on_the_messages_she_never_read_come_with_the_ask(monkeypatch, tmp_path): + """After max_replies she stops reading, so the wrap-up is unread: it comes with the rating.""" + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), json.dumps(rated(4, "booked, mostly")), + persona={"max_replies": 1}) + assert len(rig.user_calls) == 1 + await _rater().execute(ctx) + ask = rig.user_calls[-1]["parts"][0]["text"] + assert "NEW FROM YOUR AI ASSISTANT (you haven't read this yet):" in ask + assert "RESULT: blue house booked, dinner at La Ola" in ask + assert ask.index("RESULT: blue house") < ask.index("HOW YOU'D RATE THE JOB") < ask.index("As Linda, rate") + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert rating["score"] == 4 and rating["unread_shown"] == 1 + assert rig.record(ctx)["counts"]["user_replies"] == 1 + + +@pytest.mark.asyncio +async def test_after_always_on_the_unread_messages_come_from_the_record_when_its_timeline_cant_be_read( + monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), json.dumps(rated(4, "booked")), + persona={"max_replies": 1}) + rig.record(ctx)["object_url"] = "file:///nowhere/timeline.json" + await _rater().execute(ctx) + assert "RESULT: blue house booked" in rig.user_calls[-1]["parts"][0]["text"] + assert ctx.metadata[SATISFACTION_KEY]["week"]["score"] == 4 + + +@pytest.mark.asyncio +async def test_a_malformed_answer_gets_one_retry_in_the_same_session(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks"), + "a solid 4", json.dumps(rated(4))) + await _rater().execute(ctx) + assert [c["parts"][0]["text"] for c in rig.user_calls[3:]] == [us.RATING_RETRY] + assert len({c["context_id"] for c in rig.user_calls}) == 1 + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert (rating["score"], rating["retried"], rating["user_sim_turns"], rating["schema_errors"]) == (4, True, 2, []) + stored = json.loads(get_config().get_object_store().get(rating["object_url"])) + assert stored["first_answer"]["raw"] == "a solid 4" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("answers, error", [(("not json", "still not json"), None), ((), "timeout")]) +async def test_a_rating_that_fails_is_recorded_and_the_step_goes_on(monkeypatch, tmp_path, answers, error): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks"), *answers) + if error: + async def times_out(url, parts, **kw): + raise TimeoutError("the user sim took too long") + + monkeypatch.setattr(rus, "send_and_wait", times_out) + await _rater().execute(ctx) + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert rating["score"] is None + if error: + assert rating["user_sim_error"].startswith("timeout") + else: + assert rating["retried"] is True and len(rating["schema_errors"]) == 2 + + +@pytest.mark.asyncio +async def test_a_rating_past_its_timeout_is_recorded(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks")) + + async def hangs(url, parts, **kw): + await asyncio.sleep(5) + + monkeypatch.setattr(rus, "send_and_wait", hangs) + await _rater(timeout_seconds=0.05).execute(ctx) + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert rating["score"] is None and rating["user_sim_error"].lower().startswith("timeout") + + +@pytest.mark.asyncio +async def test_an_always_on_run_that_ended_on_an_error_is_not_rated(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, agent=lambda n, texts, rig: Fail()) + await _rater().execute(ctx) + assert rig.user_calls == [] + assert ctx.metadata[SATISFACTION_KEY]["week"] == {"user_sim": "user-sim", "score": None, "reason": None, + "rated_by": "rate", + "skipped": "the run ended on agent_turn_failed"} + + +@pytest.mark.asyncio +async def test_a_run_whose_user_sim_agent_is_gone_is_not_rated(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks")) + ctx.deployed_agents = [a for a in ctx.deployed_agents if a.agent_name != "user-sim"] + await _rater().execute(ctx) + assert len(rig.user_calls) == 2 + assert ctx.metadata[SATISFACTION_KEY]["week"]["skipped"] == "the user sim's agent 'user-sim' is no longer deployed" + + +@pytest.mark.asyncio +async def test_a_run_step_that_failed_and_let_the_task_go_on_is_not_rated(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks")) + ctx.metadata["failed_steps"] = [{"step_id": "week", "step_type": "always_on", "error": "it broke"}] + await _rater().execute(ctx) + assert ctx.metadata[SATISFACTION_KEY]["week"]["skipped"] == "the run step failed: it broke" + + +@pytest.mark.asyncio +async def test_a_run_step_that_failed_once_and_succeeded_on_its_retry_is_rated(monkeypatch, tmp_path): + rig, ctx = await _always_on_run(monkeypatch, tmp_path, _reply("blue!"), _reply("amazing, thanks"), + json.dumps(rated(5, "house and dinner booked", "S1"))) + ctx.metadata["failed_steps"] = [{"step_id": "week", "step_type": "always_on", "error": "it broke", + "is_fatal": True, "retried": True}] + await _rater().execute(ctx) + rating = ctx.metadata[SATISFACTION_KEY]["week"] + assert "skipped" not in rating and rating["score"] == 5 and len(rig.user_calls) == 3 + + +@pytest.mark.asyncio +async def test_a_step_with_no_run_before_it_fails(): + from agent_env.task_step.context import StepContextError, TaskStepContext + + with pytest.raises(StepContextError, match="no always_on or prompt_agent step ran before it"): + await _rater().execute(TaskStepContext()) + + +# ---- after a prompt_agent chat ------------------------------------------------------------------------------------------ +def _user_asks(peers) -> list[tuple[str, str]]: + """Each message the user sim got: its text and its session.""" + return [(r[2]["params"]["message"]["parts"][0]["text"], r[2]["params"]["message"]["contextId"]) + for r in peers.raw if r[1] == f"{USER}/a2a" and r[2]["method"] == "message/send"] + + +def _hannah_rater(**fields): + return _rater(run_step_id="talk", satisfaction=SATISFIED, **fields) + + +@pytest.mark.asyncio +async def test_a_prompt_agent_task_without_the_step_asks_nothing_more(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one")], USER: [said("no_reply")]}) + ctx = await _context(monkeypatch) + await _step().execute(ctx) + assert [r[1] for r in peers.requests(ctx)] == [f"{AGENT}/a2a"] * 2 + [f"{USER}/a2a"] * 2 + assert SATISFACTION_KEY not in ctx.metadata + + +@pytest.mark.asyncio +async def test_after_a_prompt_agent_chat_one_ask_in_its_session_with_the_reply_she_never_read(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one"), text("glad to help")], + USER: [said("done", "perfect, thanks!"), text(json.dumps(rated(5, "booked")))]}) + ctx = await _context(monkeypatch) + result = await _step().execute(ctx) + conversation = copy.deepcopy(conversation_store.get_conversation(ctx.metadata["a2a_conversations"]["talk"])) + responses = [r.to_dict() for r in result.prompt_responses] + + await _hannah_rater().execute(ctx) + + asks = _user_asks(peers) + assert len(asks) == 2 + (read, read_session), (ask, ask_session) = asks + assert ask_session == read_session == ctx.metadata["a2a_conversations"]["talk"] + assert us.RATING_ASK.format(first="Hannah") in ask and "\nHOW YOU'D RATE THE JOB\n S1 5, perfect" in ask + assert "NEW FROM YOUR AI ASSISTANT (you haven't read this yet):\n [" in ask and "glad to help" in ask + assert "draft one" not in ask # she read it + rating = ctx.metadata[SATISFACTION_KEY]["talk"] + assert (rating["user_sim"], rating["score"], rating["reason"], rating["unread_shown"]) == ("hannah", 5, "booked", 1) + assert (rating["context_id"], rating["session_from"]) == (read_session, "prompt_agent conversation") + stored = json.loads(get_config().get_object_store().get(rating["object_url"])) + assert stored["input"] == ask and stored["score"] == 5 + # Never delivered: nothing more went to the agent, and the conversation, the reads, the responses and the + # verifications are as the chat left them. + assert [r for r in peers.raw if r[1] == f"{AGENT}/a2a"][-1][2]["method"] == "tasks/get" + assert len([r for r in peers.raw if r[1] == f"{AGENT}/a2a"]) == 4 + assert conversation_store.get_conversation(ctx.metadata["a2a_conversations"]["talk"]) == conversation + assert len(ctx.metadata["user_sim_reads"]["talk"]) == 1 + assert [r.to_dict() for r in ctx.prompt_responses] == responses and "verifications" not in ctx.metadata + + +@pytest.mark.asyncio +async def test_after_a_prompt_agent_chat_she_read_to_the_end_nothing_is_unread(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one")], USER: [said("no_reply"), text(json.dumps(rated(3)))]}) + ctx = await _context(monkeypatch) + await _step().execute(ctx) + await _hannah_rater().execute(ctx) + assert "NEW FROM" not in _user_asks(peers)[1][0] + assert ctx.metadata[SATISFACTION_KEY]["talk"]["score"] == 3 + assert ctx.metadata[SATISFACTION_KEY]["talk"]["unread_shown"] == 0 + + +@pytest.mark.asyncio +async def test_after_a_prompt_agent_chat_a_malformed_answer_gets_one_retry(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one")], + USER: [said("no_reply"), text("a solid 4"), text(json.dumps(rated(4)))]}) + ctx = await _context(monkeypatch) + await _step().execute(ctx) + await _hannah_rater().execute(ctx) + asks = _user_asks(peers) + assert asks[2][0] == us.RATING_RETRY and len({session for _, session in asks}) == 1 + assert ctx.metadata[SATISFACTION_KEY]["talk"]["score"] == 4 and ctx.metadata[SATISFACTION_KEY]["talk"]["retried"] + + +@pytest.mark.asyncio +async def test_a_prompt_agent_chat_the_user_sim_broke_is_not_rated(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one")], USER: [("failed", said("reply", "x"))]}) + ctx = await _context(monkeypatch) + await _step().execute(ctx) + await _hannah_rater().execute(ctx) + assert len(_user_asks(peers)) == 1 + rating = ctx.metadata[SATISFACTION_KEY]["talk"] + assert rating["skipped"] == "the user sim failed: state failed" and rating["score"] is None + + +@pytest.mark.asyncio +async def test_a_prompt_agent_chat_whose_agent_turn_failed_is_not_rated(monkeypatch): + Peers(monkeypatch, {AGENT: [("failed", text("it broke"))], USER: []}) + ctx = await _context(monkeypatch) + with pytest.raises(RuntimeError, match="A2A task failed"): + await _step().execute(ctx) + await _hannah_rater().execute(ctx) + assert ctx.metadata[SATISFACTION_KEY]["talk"]["skipped"] == "the agent's turn failed" diff --git a/tst/unit/task_step/test_user_sim_versions.py b/tst/unit/task_step/test_user_sim_versions.py index d183b35b1..8b810d0b6 100644 --- a/tst/unit/task_step/test_user_sim_versions.py +++ b/tst/unit/task_step/test_user_sim_versions.py @@ -71,6 +71,33 @@ def test_sections_with_nothing_in_them_are_left_out(): assert prompt.endswith("\n\nWHO YOU ARE\n W1 Hannah Reyes.\n") +TOLD = ["Dad can't do stairs since his knee surgery."] +KNOWN = ["You never book anything before 9am for your parents."] + + +def test_things_already_told_and_things_it_should_know_show_tagged_after_what_you_know(): + person = _person() + person["description"].update(already_told=TOLD, should_already_know=KNOWN) + prompt = us.system_prompt(person) + told = ("\n\nTHINGS YOU ALREADY TOLD THE ASSISTANT (if it asks for one again, remind it, a little annoyed, that you " + "already told it, and tell it again)\n T1 Dad can't do stairs since his knee surgery.\n") + known = ("\n\nTHINGS THE ASSISTANT SHOULD ALREADY KNOW (never bring these up; if it gets one wrong, correct it " + "briefly, a little irritated, and add nothing else)\n C1 You never book anything before 9am for your " + "parents.\n") + assert told in prompt and known in prompt + assert prompt.index("\nWHAT YOU KNOW") < prompt.index(told) < prompt.index(known) < prompt.index("\nWHAT TO LOOK UP") + lines = us.tagged_lines(person) + assert lines["T1"] == TOLD[0] and lines["C1"] == KNOWN[0] + assert us.problems(person) == [] + + +def test_without_them_the_prompt_and_tags_are_as_before(): + assert not {"already_told", "should_already_know"} & set(PERSON["description"]) + assert us.system_prompt(PERSON) == PROMPT + assert "ALREADY" not in PROMPT + assert not any(tag[0] in "TC" for tag in us.tagged_lines(PERSON)) + + def test_a_line_s_later_lines_hang_under_its_text(): assert us.tag_line("W2", 'Earlier you texted:\n"hi"') == ' W2 Earlier you texted:\n "hi"' assert us.tag_line("", "Answer short.\nNo emojis.") == " Answer short.\n No emojis." @@ -89,6 +116,14 @@ def test_the_tested_settings_have_no_problems(): ({"description": {"who_you_are": []}}, "description is empty"), ({"description": {"who_you_are": ["Hannah."], "events": ["Tue: an email"]}}, "unknown sections ['events']"), ({"description": {"who_you_are": ["Hannah.", ""]}}, "description.who_you_are must be a list of non-empty lines"), + ({"description": {"who_you_are": ["Hannah."], "already_told": "Dad can't do stairs."}}, + "description.already_told must be a list of non-empty lines"), + ({"description": {"who_you_are": ["Hannah."], "already_told": [" "]}}, + "description.already_told must be a list of non-empty lines"), + ({"description": {"who_you_are": ["Hannah."], "should_already_know": [3]}}, + "description.should_already_know must be a list of non-empty lines"), + ({"description": {"who_you_are": ["Hannah."], "should_already_know": ["Nothing before 9am.", ""]}}, + "description.should_already_know must be a list of non-empty lines"), ({"permissions": ["book flights"]}, "permissions must be an object"), ({"permissions": {"just_do": ["x"], "maybe": ["y"]}}, "permissions has unknown kinds ['maybe']"), ({"permissions": {"never": "text her mom"}}, "permissions.never must be a list of rules"), diff --git a/tst/unit/training/harness_test.py b/tst/unit/training/harness_test.py index d0999762e..99487146e 100644 --- a/tst/unit/training/harness_test.py +++ b/tst/unit/training/harness_test.py @@ -272,6 +272,16 @@ def user_sim(name, opening): assert harness._first_prompt(SimpleNamespace(steps=[user_sim("nora", "Find me a sofa."), unnamed])) == "" +def test_a_rows_prompt_from_an_opening_with_files_is_its_text(): + opening = [{"kind": "text", "text": "Find me a sofa."}, + {"kind": "file", "file": {"uri": "s3://nora/room.jpg", "name": "room.jpg"}}, + {"kind": "text", "text": "Something that fits this corner."}] + sim = SimpleNamespace(type="deploy_user_sim_agent", agent_name="nora", opening_prompt=opening) + always_on = SimpleNamespace(type="always_on", prompt=None, user_sim="nora") + assert harness._first_prompt(SimpleNamespace(steps=[sim, always_on])) == \ + "Find me a sofa.\nSomething that fits this corner." + + def test_a_rows_prompt_skips_a_briefing_sent_to_the_user_sim(): sim = SimpleNamespace(type="deploy_user_sim_agent", agent_name="nora", opening_prompt="Find a sofa.") briefing = SimpleNamespace(type="prompt_agent", agent_name="nora", prompt="You are Nora; stay in character.")