From 6bbac7d1659ff125f4e430611c1921d7fb63e886 Mon Sep 17 00:00:00 2001 From: t Date: Tue, 10 Nov 2026 11:19:15 -0500 Subject: [PATCH 1/3] feat(prompt_agent): an opt-in gate on user-sim messages prompt_agent's new user_sim_gate checks each LLM user-sim message before the agent receives it: phrase patterns for the assistant's side of a conversation, and optionally a small judge model that decides only whose voice the message is in. A message that fails is held back and the user-sim is asked to rewrite it, up to max_rewrites times; on_fail then fails the step, ends the conversation, or sends it anyway. Every verdict is recorded in context.metadata["user_sim_gate"]. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 73 ++++- .../task_step/task_steps/user_sim_gate.py | 256 ++++++++++++++++++ .../task_step/test_prompt_agent_user_side.py | 128 +++++++++ tst/unit/task_step/test_user_sim_gate.py | 71 +++++ 4 files changed, 527 insertions(+), 1 deletion(-) create mode 100644 src/agent_env/task_step/task_steps/user_sim_gate.py create mode 100644 tst/unit/task_step/test_user_sim_gate.py 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 c11fb716b..3bce8bae4 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -41,6 +41,7 @@ from agent_env.task_step.task_step import RetryConfig, TaskStep, TaskStepDependency from agent_env.task_step.task_steps.notification_delivery import delivered_parts, next_message, take_due 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.user_sim_gate import UserSimGate, rewrite_request from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import ( agent_error_text, fetch_container_logs, @@ -177,6 +178,7 @@ def __init__( snapshot_config: Optional[SnapshotConfig] = None, live_trajectory: bool = False, post_turn_hooks: Optional[list[dict]] = None, + user_sim_gate: Optional[UserSimGate | dict] = None, ): # `prompt` is the legacy text-only field; `parts` is the multimodal # form that's the authoritative A2A wire shape. Exactly one must be set: @@ -223,6 +225,10 @@ def __init__( self.snapshot_config = snapshot_config self.live_trajectory = live_trajectory self.post_turn_hooks = post_turn_hooks + # Checks each user-sim message before the agent gets it (user_sim_gate.py). A dict is accepted + # too, as JSON task files and the kwargs form of put() carry it. + self.user_sim_gate = (UserSimGate.from_dict(user_sim_gate) if isinstance(user_sim_gate, dict) + else user_sim_gate) @staticmethod def _validate_parts(parts: list[dict]) -> None: @@ -300,6 +306,8 @@ def to_dict(self) -> dict: base["live_trajectory"] = True if self.post_turn_hooks is not None: base["post_turn_hooks"] = self.post_turn_hooks + if self.user_sim_gate is not None: + base["user_sim_gate"] = self.user_sim_gate.to_dict() return base @classmethod @@ -337,6 +345,7 @@ def from_dict(cls, data: dict) -> PromptAgentTaskStep: ), live_trajectory=data.get("live_trajectory", False), post_turn_hooks=data.get("post_turn_hooks"), + user_sim_gate=data.get("user_sim_gate"), ) def _trajectory_prefix(self) -> str: @@ -574,7 +583,11 @@ async def _converse( self, context, url=user_url, user_sim=user_sim, is_user_sim=is_user_sim, conversation_id=conversation_id, decide_url=decide_url, shareable=sent_to_target, decides_quietly=card_request_accepts(decide_method.get("request") or {}, _QUIET_DECIDE), + opening_text="\n".join(p.get("text", "") for p in initial_parts if p.get("kind") == "text"), ) + if self.user_sim_gate is not None and not is_user_sim: + logger.warning(f"prompt_agent '{self.id}': user_sim_gate is set but the user side isn't a user-sim " + "agent-env deployed; no message is checked") try: for turn in range(self.max_conversation_turns): @@ -764,6 +777,9 @@ def record_turn() -> None: structured_output=structured_output, )) + if user_side.gate_failure is not None: + raise RuntimeError(f"user_sim_gate: {user_side.gate_failure}") + if final_state == TaskState.failed: detail_parts = [] if final_terminal.error_type: @@ -1031,7 +1047,8 @@ class _UserSide: def __init__(self, step: PromptAgentTaskStep, context: TaskStepContext, *, url: Optional[str], user_sim: Optional[DeployedAgent], is_user_sim: bool, conversation_id: str, - decide_url: Optional[str], decides_quietly: bool, shareable: set[str]) -> None: + decide_url: Optional[str], decides_quietly: bool, shareable: set[str], + opening_text: str = "") -> None: self._step = step self._context = context self._url = url @@ -1041,9 +1058,23 @@ def __init__(self, step: PromptAgentTaskStep, context: TaskStepContext, *, url: self._decide_url = decide_url self._decides_quietly = decides_quietly self._shareable = shareable + # The conversation as the user-sim gate reads it: (role, text), role "user" or "assistant". + self._history: list[tuple[str, str]] = [("user", opening_text)] if opening_text else [] + self._gate = step.user_sim_gate if is_user_sim else None + self._gate_call_config = None + # Set once a user-sim message fails the gate after its rewrites, with on_fail "fail". + self.gate_failure: Optional[str] = None async def ask(self, number: int, solver_message: str, agent_parts: list[dict]) -> dict: """The user side's answer: ``{"message": parts}``, or ``{"end": reason}`` with any closing message to record.""" + self._history.append(("assistant", solver_message or "")) + answer = await self._answer(number, agent_parts, solver_message) + if "message" in answer and "end" not in answer: + self._history.append(("user", "\n".join(p.get("text", "") for p in answer["message"] + if p.get("kind") == "text"))) + return answer + + async def _answer(self, number: int, agent_parts: list[dict], solver_message: str) -> dict: if self._decide_url is not None: env_triggers = await self._step._read_env_triggers(self._context) self._record("env_trigger_snapshots", {"turn": number, "envs": env_triggers}) @@ -1076,6 +1107,46 @@ async def triggers(self, number: int, solver_message: str) -> Optional[dict]: return answer async def _ask_peer(self, number: int, agent_parts: list[dict]) -> dict: + answer = await self._send_to_peer(number, agent_parts) + if self._gate is None or "end" in answer: + return answer + return await self._gated(number, answer) + + async def _gated(self, number: int, answer: dict) -> dict: + """Check the user-sim's message; ask for a rewrite while it fails, up to the gate's limit.""" + gate = self._gate + if self._gate_call_config is None and gate.judge_model: + config = get_config() + overrides = self._context.metadata.get("user_overrides", {}) + self._gate_call_config = config.resolve_model_call( + gate.judge_model, + default_api_key=config.resolve_reference( + overrides.get("judge_litellm_api_key") or overrides.get("litellm_api_key")), + base_override=overrides.get("judge_litellm_base_url") or overrides.get("litellm_base_url"), + ) + for attempt in range(gate.max_rewrites + 1): + text = "\n".join(p.get("text", "") for p in answer["message"] if p.get("kind") == "text") + verdict = await gate.check(text, self._history, call_config=self._gate_call_config) + self._record("user_sim_gate", {"turn": number, "attempt": attempt, "chars": len(text), **verdict.to_dict()}) + if verdict.passed: + return answer + logger.warning(f"Turn {number}: user-sim message failed the gate (attempt {attempt + 1}/" + f"{gate.max_rewrites + 1}): {'; '.join(verdict.reasons())[:300]}") + if attempt == gate.max_rewrites: + break + answer = await self._send_to_peer(number, [{"kind": "text", "text": rewrite_request(verdict)}]) + if "end" in answer: + return answer + if gate.on_fail == "send": + return answer + self._context.metadata.setdefault("user_sim_invalid", {})[self._step.id] = { + "turn": number, "reasons": verdict.reasons()} + if gate.on_fail == "fail": + self.gate_failure = (f"user-sim message at turn {number} still reads as the assistant's after " + f"{gate.max_rewrites} rewrite(s): {'; '.join(verdict.reasons())[:500]}") + return {"end": "user_sim_invalid"} + + async def _send_to_peer(self, number: int, agent_parts: list[dict]) -> dict: step = self._step if self._url is None: self._url = get_config().get_default_human_a2a_url() diff --git a/src/agent_env/task_step/task_steps/user_sim_gate.py b/src/agent_env/task_step/task_steps/user_sim_gate.py new file mode 100644 index 000000000..93880ce55 --- /dev/null +++ b/src/agent_env/task_step/task_steps/user_sim_gate.py @@ -0,0 +1,256 @@ +"""A check on every simulated-user message before prompt_agent sends it to the agent. + +An LLM playing the user can slip into the assistant's role: it answers its own question, reports work as +done, or offers to do the agent's job. Once one such message is in the conversation the rest of the run +is no longer a test of the agent. The gate checks each user-sim message, asks the user-sim to rewrite one +that fails, and, if the rewrites fail too, ends the conversation and marks the run. + +Two checks, either of which can fail a message: + +- ``patterns``: phrases and shapes that belong to the assistant's side (a report of work done, a + formatted deliverable, offers to do the work). Free, and deliberately narrow. +- ``judge_model``: one model call that reads the last turns and the candidate message and says whether + the message is written as the user. + +Opt in with ``user_sim_gate`` on a ``prompt_agent`` step that has a user-sim:: + + {"judge_model": "", "max_rewrites": 2, "on_fail": "fail"} +""" +from __future__ import annotations + +import json +import logging +import re +from dataclasses import asdict, dataclass, field +from typing import Optional + +logger = logging.getLogger(__name__) + +ON_FAIL = ("fail", "end", "send") + +# ── Patterns ────────────────────────────────────────────────────────────────────────────────────────── + +_REPORT_OPENING = re.compile( + r"^\s*(?:\*\*)?(?:done\b|all done|checked\b|short (?:answer|version)\b" + r"|here'?s (?:what (?:i|we) (?:found|did|checked|changed)|the (?:full |short |whole )?(?:picture|summary|breakdown" + r"|rundown|update|status)|where (?:things|we|it|you) stands?|how (?:it|things) (?:stands?|looks?))" + r"|summary\b|update:|findings\b|tl;?dr\b)", + re.I, +) +_WORK_DONE = re.compile( + r"\bI(?:'ve| have)? (?:checked|reviewed|pulled|updated|put together|drafted|flagged|verified|" + r"cross-checked|went through|compared|audited|analy[sz]ed|reconciled|fixed|created|added|" + r"rewrote|built|calculated|ran)\b", + re.I, +) +_NOW_LIVE = re.compile(r"\b(?:now live|is now (?:up|live|updated)|has been (?:updated|created|sent|posted))\b", re.I) +_DELIVERED = re.compile( + r"\b(?:(?:emails?|messages?|drafts?|replies|invites?|notes?) (?:sent|drafted|posted|scheduled)\b|incorporated\b|" + r"(?:the|a|my) (?:cleaner|updated|revised|corrected|final) version\b)", + re.I, +) +_AS_ASSISTANT = re.compile( + r"\b(?:as an ai|i'?m an? (?:ai|assistant|language model)|my previous (?:answer|response|reply)|" + r"i apologi[sz]e for (?:the|my)|let me know if you'?d like me to|" + r"would you like me to|want me to (?:draft|send|update|create|go ahead)|" + r"standing by|ready when you are|parked until|" + r"anything else (?:you(?:'d)? (?:want|like) (?:me )?to|to (?:cover|look at|dig into))|" + r"what do you want (?:me )?to do next|should i (?:send|draft|go ahead)|" + r"i'll (?:draft|fold|pull|search|put together|pick (?:it |this )?up|rerun|re-run)|" + r"i can (?:send|draft|search|pull) (?:it|that|them)|" + r"i (?:don't|do not) have (?:the )?(?:tools|access)|corrections (?:noted|taken)|" + r"(?:send|drop) me (?:whatever|that phrase|the results)|" + r"how (?:can|may) i (?:help|assist)|i'?m (?:here|ready) to help|what would you like (?:me )?to (?:work on|do) today)\b", + re.I, +) +_TABLE = re.compile(r"^\s*\|?\s*:?-{3,}:?\s*\|", re.M) +_HEADER = re.compile(r"^\s*#{1,4} \S", re.M) +_BOLD_BULLET = re.compile(r"^\s*(?:[-*•]|\d+\.)\s+\*\*", re.M) + + +def pattern_reasons(text: str) -> list[str]: + """Why ``text`` reads as the assistant's message, one reason per pattern that matched; empty if none.""" + reasons = [] + n = len(text) + if _AS_ASSISTANT.search(text): + reasons.append("it talks like the assistant (offers to do the work, waits on the user, or speaks as the AI)") + if _REPORT_OPENING.search(text) and n > 300: + reasons.append("it opens like a report of work done") + if _NOW_LIVE.search(text): + reasons.append("it says something is now live or updated") + if _DELIVERED.search(text) and n > 300: + reasons.append("it reports something delivered or revised") + if (_TABLE.search(text) or len(_HEADER.findall(text)) >= 2 or len(_BOLD_BULLET.findall(text)) >= 3) and n > 600: + reasons.append("it is a formatted deliverable (a table, headings or bold bullet points)") + work = _WORK_DONE.findall(text) + if work and (n > 500 or len(work) >= 2): + reasons.append("it reports checking or building things itself") + return reasons + + +# ── Judge ───────────────────────────────────────────────────────────────────────────────────────────── + +_JUDGE_PROMPT = """You check one message in a test conversation. A model is playing a USER who gave an AI ASSISTANT \ +a task. The assistant does the work; the user asks, answers questions the assistant put to them, decides, approves, \ +pushes back, or ends the conversation. + +Decide whether the CANDIDATE message is written as the user. It is NOT written as the user when it does the \ +assistant's job: it answers the question the user should be asking, reports work as done or findings it produced \ +("Done. Here's what I found", "I've updated the page"), presents analysis or a deliverable, apologizes for "my \ +previous answer", offers to do work ("I'll draft it", "Standing by", "Want me to send it?"), or speaks as the AI. + +It IS written as the user when it asks, instructs, approves, pushes back, answers a question the assistant asked, \ +or shares the user's own knowledge, even if it is long, formatted, quotes the assistant's numbers, or pastes an \ +email or document. Length and markdown alone are not evidence. + +Judge only whose voice the message is in. A user message that repeats an earlier one, ignores or contradicts what \ +the assistant said, states a decision or a fact the user holds, or seems off topic is still written as the user. + +CONVERSATION SO FAR (most recent last): +{history} + +CANDIDATE (the user's next message): +<<< +{candidate} +>>> + +Answer in JSON: {{"in_role": true or false, "reason": "one sentence citing the candidate"}}""" + +_JUDGE_SCHEMA = { + "type": "object", + "properties": {"in_role": {"type": "boolean"}, "reason": {"type": "string"}}, + "required": ["in_role", "reason"], + "additionalProperties": False, +} + + +def render_history(history: list[tuple[str, str]], turns: int, max_chars: int) -> str: + """The last ``turns`` messages as ``USER:`` / ``ASSISTANT:`` blocks, each cut to ``max_chars``.""" + out = [] + for role, text in history[-turns:]: + body = text if len(text) <= max_chars else text[:max_chars] + f"\n[… {len(text) - max_chars} more characters]" + out.append(f"{role.upper()}:\n{body}") + return "\n\n".join(out) or "(nothing yet)" + + +def parse_judge(content: str) -> Optional[dict]: + """The judge's JSON object, tolerating text around it; None if there's none with a boolean ``in_role``.""" + start = content.find("{") + if start < 0: + return None + try: + obj = json.JSONDecoder().raw_decode(content[start:])[0] + except ValueError: + return None + return obj if isinstance(obj, dict) and isinstance(obj.get("in_role"), bool) else None + + +# ── The gate ────────────────────────────────────────────────────────────────────────────────────────── + +@dataclass +class UserSimGate: + """How ``prompt_agent`` checks user-sim messages; see the module docstring.""" + + patterns: bool = True + judge_model: Optional[str] = None + max_rewrites: int = 2 + # What happens when a message still fails after its rewrites: "fail" ends the conversation and fails the + # step, "end" ends the conversation and lets the step finish, "send" sends it anyway (record only). + on_fail: str = "fail" + context_turns: int = 4 + # The judge alone fails only a message at least this long; a shorter one also needs a pattern. Swaps are long or + # match a pattern, and without the scenario the judge misreads short scripted answers that correct the agent. + judge_min_chars: int = 300 + context_chars: int = 3000 + judge_timeout_seconds: float = 60 + + def __post_init__(self) -> None: + if self.on_fail not in ON_FAIL: + raise ValueError(f"user_sim_gate.on_fail must be one of {ON_FAIL}, not {self.on_fail!r}") + if self.max_rewrites < 0: + raise ValueError("user_sim_gate.max_rewrites must be >= 0") + if not self.patterns and not self.judge_model: + raise ValueError("user_sim_gate needs patterns, a judge_model, or both") + + @classmethod + def from_dict(cls, data: Optional[dict]) -> Optional[UserSimGate]: + if not data: + return None + known = set(cls.__dataclass_fields__) + unknown = set(data) - known + if unknown: + raise ValueError(f"user_sim_gate: unknown keys {sorted(unknown)}") + return cls(**data) + + def to_dict(self) -> dict: + return asdict(self) + + async def check(self, candidate: str, history: list[tuple[str, str]], *, call_config=None) -> GateVerdict: + """Check one message. ``history`` is the conversation so far as ``(role, text)``, role user/assistant. + A judge call that errors or can't be parsed doesn't fail the message: the patterns' verdict stands.""" + reasons = pattern_reasons(candidate) if self.patterns else [] + verdict = GateVerdict(passed=not reasons, pattern_reasons=reasons) + if self.judge_model: + try: + judged = await self._judge(candidate, history, call_config) + except Exception as e: # the judge is a second opinion; never let it take the conversation down + logger.warning(f"user-sim gate judge call failed ({type(e).__name__}: {e}); using patterns only") + verdict.judge_error = f"{type(e).__name__}: {e}"[:300] + return verdict + if judged is None: + verdict.judge_error = "unparseable judge output" + return verdict + verdict.judge_in_role = judged["in_role"] + verdict.judge_reason = str(judged.get("reason", ""))[:500] + if not judged["in_role"] and (reasons or len(candidate) >= self.judge_min_chars): + verdict.passed = False + return verdict + + async def _judge(self, candidate: str, history: list[tuple[str, str]], call_config) -> Optional[dict]: + import litellm + + from agent_env.config.model import ModelParam + + prompt = _JUDGE_PROMPT.format( + history=render_history(history, self.context_turns, self.context_chars), + candidate=candidate[: self.context_chars * 2], + ) + response = await litellm.acompletion( + model=self.judge_model, + messages=[{"role": "user", "content": prompt}], + timeout=self.judge_timeout_seconds, + **{ModelParam.RESPONSE_FORMAT: {"type": "json_schema", + "json_schema": {"name": "user_sim_gate", "schema": _JUDGE_SCHEMA, + "strict": True}}}, + **(call_config.client_kwargs() if call_config is not None else {}), + ) + return parse_judge(response.choices[0].message.content or "") + + +@dataclass +class GateVerdict: + passed: bool + pattern_reasons: list[str] = field(default_factory=list) + judge_in_role: Optional[bool] = None + judge_reason: Optional[str] = None + judge_error: Optional[str] = None + + def reasons(self) -> list[str]: + out = list(self.pattern_reasons) + if self.judge_in_role is False and self.judge_reason: + out.append(self.judge_reason) + return out + + def to_dict(self) -> dict: + return {k: v for k, v in asdict(self).items() if v not in (None, [])} + + +def rewrite_request(verdict: GateVerdict) -> str: + """What the user-sim is told when its message was held back.""" + why = "; ".join(verdict.reasons()) or "it reads as the assistant's message" + return ( + "[Conversation system note, not from the assistant.] Your last message was not sent to the assistant, " + f"because it reads as the assistant's reply rather than yours: {why}. You are the user. The assistant does " + "the work and reports it; you ask, answer what you were asked, decide, push back, or end. Write your next " + "message to the assistant again, as yourself, in the same reply format as before." + ) diff --git a/tst/unit/task_step/test_prompt_agent_user_side.py b/tst/unit/task_step/test_prompt_agent_user_side.py index 0aba179a8..971bef3c8 100644 --- a/tst/unit/task_step/test_prompt_agent_user_side.py +++ b/tst/unit/task_step/test_prompt_agent_user_side.py @@ -479,3 +479,131 @@ async def test_with_no_user_agent_the_configured_human_peer_answers(monkeypatch) *exchange(HUMAN, text("one"), "", "", "task-2"), *exchange(AGENT, text("go on"), "", "", "task-3"), ] + + +# ---- the user-sim gate ---------------------------------------------------------------------------------------------- + +_SWAP = ("Done. Here's what I found when I checked the figures against the records:\n\n" + "| Metric | Reported | Actual |\n|---|---|---|\n| Installs | 1,400 | 901 |\n| Retention | 43% | 25% |\n\n" + "None of the three reported numbers hold up, and I've updated the summary page to say so. " * 2) + + +_TAKEOVER = "I'll handle the rest of it from here, so there's no need for you to keep going on this. " * 4 + + +def _gated(turns: int = 2, **gate) -> PromptAgentTaskStep: + return _step(turns=turns, user_agent_name="human_agent", user_sim_gate={"patterns": True, **gate}) + + +@pytest.mark.asyncio +async def test_the_gate_lets_a_user_message_through_and_records_its_verdict(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one"), text("draft two")], USER: [sim("make it shorter")]}) + ctx = _context(user_card=_USER_SIM_CARD) + + await _gated().execute(ctx) + + assert peers.requests(ctx)[-2] == send(AGENT, text("make it shorter"), "", "") + assert ctx.metadata["user_sim_gate"] == {"talk": [{"turn": 1, "attempt": 0, "chars": 15, "passed": True}]} + assert "user_sim_invalid" not in ctx.metadata + + +@pytest.mark.asyncio +async def test_a_message_that_reads_as_the_assistants_is_held_back_and_rewritten(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one"), text("draft two")], + USER: [sim(_SWAP), sim("do those numbers hold up?")]}) + ctx = _context(user_card=_USER_SIM_CARD) + + await _gated().execute(ctx) + + sent = [r for r in peers.requests(ctx) if r[2] and r[2].get("method") == "message/send"] + to_user = [r[2]["params"]["message"]["parts"][0]["text"] for r in sent if r[1].startswith(USER)] + to_agent = [r[2]["params"]["message"]["parts"] for r in sent if r[1].startswith(AGENT)] + assert to_user[0] == "draft one" + assert to_user[1].startswith("[Conversation system note, not from the assistant.] Your last message was not sent") + assert to_agent == [text("plan my week"), text("do those numbers hold up?")] # the swap never reached it + verdicts = ctx.metadata["user_sim_gate"]["talk"] + assert [(v["attempt"], v["passed"]) for v in verdicts] == [(0, False), (1, True)] + assert "it opens like a report of work done" in verdicts[0]["pattern_reasons"] + + +@pytest.mark.asyncio +async def test_a_message_that_still_fails_after_its_rewrites_ends_the_run_and_fails_the_step(monkeypatch): + peers = Peers(monkeypatch, {AGENT: [text("draft one")], USER: [sim(_SWAP), sim(_SWAP), sim(_SWAP)]}) + ctx = _context(user_card=_USER_SIM_CARD) + + with pytest.raises(RuntimeError, match=r"user_sim_gate: user-sim message at turn 1 still reads as the assistant's"): + await _gated(turns=3).execute(ctx) + + assert [v["passed"] for v in ctx.metadata["user_sim_gate"]["talk"]] == [False, False, False] + assert ctx.metadata["user_sim_invalid"]["talk"]["turn"] == 1 + assert _stored(ctx, peers)["status"] == "closed" + assert [m["role"] for m in _stored(ctx, peers)["messages"]] == ["user", "agent"] # nothing more was sent + assert ctx.prompt_responses[-1].response == "draft one" # the run kept its record + + +@pytest.mark.asyncio +async def test_on_fail_end_closes_the_conversation_without_failing_the_step(monkeypatch): + Peers(monkeypatch, {AGENT: [text("draft one")], USER: [sim(_SWAP), sim(_SWAP)]}) + ctx = _context(user_card=_USER_SIM_CARD) + + await _gated(turns=3, max_rewrites=1, on_fail="end").execute(ctx) + + assert ctx.metadata["user_sim_invalid"]["talk"]["turn"] == 1 + assert len(ctx.metadata["user_sim_gate"]["talk"]) == 2 + + +@pytest.mark.asyncio +async def test_the_judge_fails_a_message_the_patterns_pass_and_sees_the_conversation(monkeypatch): + from agent_env.task_step.task_steps import user_sim_gate + + seen = [] + + async def judge(self, candidate, history, call_config): + seen.append((candidate, list(history))) + return {"in_role": candidate != _TAKEOVER, "reason": "it offers to do the work"} + + monkeypatch.setattr(user_sim_gate.UserSimGate, "_judge", judge) + Peers(monkeypatch, {AGENT: [text("draft one"), text("draft two")], + USER: [sim(_TAKEOVER), sim("shorter, please")]}) + ctx = _context(user_card=_USER_SIM_CARD) + monkeypatch.setattr("agent_env.config.Config.resolve_model_call", lambda self, model, **kw: None, raising=False) + + await _gated(judge_model="judge-model").execute(ctx) + + verdicts = ctx.metadata["user_sim_gate"]["talk"] + assert [(v["passed"], v.get("judge_in_role")) for v in verdicts] == [(False, False), (True, True)] + assert verdicts[0]["judge_reason"] == "it offers to do the work" + assert seen[0][1] == [("user", "plan my week"), ("assistant", "draft one")] + + +@pytest.mark.asyncio +async def test_a_judge_that_errors_leaves_the_patterns_verdict(monkeypatch): + from agent_env.task_step.task_steps import user_sim_gate + + async def judge(self, candidate, history, call_config): + raise TimeoutError("judge timed out") + + monkeypatch.setattr(user_sim_gate.UserSimGate, "_judge", judge) + Peers(monkeypatch, {AGENT: [text("draft one"), text("draft two")], USER: [sim("shorter, please")]}) + ctx = _context(user_card=_USER_SIM_CARD) + monkeypatch.setattr("agent_env.config.Config.resolve_model_call", lambda self, model, **kw: None, raising=False) + + await _gated(judge_model="judge-model").execute(ctx) + + (verdict,) = ctx.metadata["user_sim_gate"]["talk"] + assert verdict["passed"] is True + assert verdict["judge_error"] == "TimeoutError: judge timed out" + + +def test_the_gate_round_trips_through_the_step_dict_and_rejects_bad_settings(): + step = _gated(judge_model="judge-model", max_rewrites=1) + again = PromptAgentTaskStep.from_dict(step.to_dict()) + assert again.user_sim_gate == step.user_sim_gate + assert again.user_sim_gate.max_rewrites == 1 + assert "user_sim_gate" not in _step().to_dict() + with pytest.raises(ValueError, match="unknown keys"): + _step(user_sim_gate={"judge": "x"}) + with pytest.raises(ValueError, match="on_fail"): + _step(user_sim_gate={"on_fail": "retry"}) + with pytest.raises(ValueError, match="needs patterns"): + _step(user_sim_gate={"patterns": False}) diff --git a/tst/unit/task_step/test_user_sim_gate.py b/tst/unit/task_step/test_user_sim_gate.py new file mode 100644 index 000000000..fca748ad4 --- /dev/null +++ b/tst/unit/task_step/test_user_sim_gate.py @@ -0,0 +1,71 @@ +"""The user-sim gate's own checks: the patterns, the judge's output parsing and the history it is shown.""" + +from agent_env.task_step.task_steps.user_sim_gate import ( + GateVerdict, parse_judge, pattern_reasons, render_history, rewrite_request) + +_FILLER = " The details are in the shared folder and the numbers come from last month's records." * 6 + + +def test_the_patterns_pass_what_users_write(): + for message in [ + "and the numbers she gave the board, do those hold up?", + "ok. who actually worked on it, and what does each of them cost per month?", + "Here's the email my manager sent, pasted so you have it:\n\n> Please hold the order until Friday." + _FILLER, + "No, go ahead and send it. Use the second draft.", + "I checked my inbox and there's nothing from them yet.", + ]: + assert pattern_reasons(message) == [], message + + +def test_the_patterns_catch_the_assistants_side(): + cases = { + "Done. Here's what I found when I went through the files." + _FILLER: "it opens like a report of work done", + "Here's where things stand after the changes." + _FILLER: "it opens like a report of work done", + "Understood. Standing by.": "it talks like the assistant", + "Want me to draft the reply to them now?": "it talks like the assistant", + "I'm ready to help! What would you like to work on today?": "it talks like the assistant", + "The page is now live, with the corrected totals." : "it says something is now live or updated", + "Thanks for those corrections, incorporated. The cleaner version:" + _FILLER: "it reports something delivered", + "| Item | Was | Is |\n|---|---|---|\n| A | 1 | 2 |\n" + _FILLER * 2: "it is a formatted deliverable", + "I've reviewed the contract and I've compared it with the invoice." + _FILLER: "it reports checking or building", + } + for message, expected in cases.items(): + reasons = pattern_reasons(message) + assert any(r.startswith(expected) for r in reasons), (message[:60], reasons) + + +def test_the_judge_output_is_read_through_surrounding_text_and_rejected_without_a_boolean(): + assert parse_judge('{"in_role": false, "reason": "reports work"}') == {"in_role": False, "reason": "reports work"} + assert parse_judge('Sure.\n{"in_role": true, "reason": "asks"} trailing') == {"in_role": True, "reason": "asks"} + assert parse_judge('{"in_role": "no"}') is None + assert parse_judge("no json here") is None + + +def test_the_history_shows_the_last_turns_each_cut_to_size(): + history = [("user", "first"), ("assistant", "x" * 50), ("user", "second"), ("assistant", "third")] + shown = render_history(history, turns=3, max_chars=10) + assert shown.startswith("ASSISTANT:\nxxxxxxxxxx\n[… 40 more characters]") + assert "first" not in shown and shown.endswith("ASSISTANT:\nthird") + assert render_history([], turns=4, max_chars=10) == "(nothing yet)" + + +def test_the_rewrite_request_names_why_and_says_it_is_not_from_the_assistant(): + note = rewrite_request(GateVerdict(passed=False, pattern_reasons=["it opens like a report of work done"], + judge_in_role=False, judge_reason="it answers its own question")) + assert note.startswith("[Conversation system note, not from the assistant.]") + assert "it opens like a report of work done; it answers its own question" in note + + +def test_the_judge_alone_fails_only_a_long_message(): + import asyncio + from agent_env.task_step.task_steps.user_sim_gate import UserSimGate + + class Gate(UserSimGate): + async def _judge(self, candidate, history, call_config): + return {"in_role": False, "reason": "reads as analysis"} + + gate = Gate(judge_model="judge-model") + short = asyncio.run(gate.check("Contracted physicians are required in two states.", [])) + long = asyncio.run(gate.check("The two states that need contracted physicians are these." + _FILLER, [])) + assert (short.passed, short.judge_in_role) == (True, False) # recorded, not failed + assert long.passed is False From fcec1e1184a2140bde141cb86888b29ef6aa3676 Mon Sep 17 00:00:00 2001 From: t Date: Tue, 10 Nov 2026 12:43:00 -0500 Subject: [PATCH 2/3] fix(prompt_agent): empty user_sim_gate settings turn the gate on; name its cutoffs `user_sim_gate={}` now enables the gate with its defaults; only None leaves it off. The patterns' length and count cutoffs, and how much of a judge's reason and error a verdict keeps, are named class constants. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/user_sim_gate.py | 38 +++++++++++++++---- .../task_step/test_prompt_agent_user_side.py | 2 + 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/src/agent_env/task_step/task_steps/user_sim_gate.py b/src/agent_env/task_step/task_steps/user_sim_gate.py index 93880ce55..a4807ac82 100644 --- a/src/agent_env/task_step/task_steps/user_sim_gate.py +++ b/src/agent_env/task_step/task_steps/user_sim_gate.py @@ -22,7 +22,7 @@ import logging import re from dataclasses import asdict, dataclass, field -from typing import Optional +from typing import ClassVar, Optional logger = logging.getLogger(__name__) @@ -68,22 +68,39 @@ _BOLD_BULLET = re.compile(r"^\s*(?:[-*•]|\d+\.)\s+\*\*", re.M) +class PatternLimits: + """How long or how formatted a message must be before a shape-based pattern counts. Phrases that only the + assistant says (offers, "standing by", "now live") count at any length; shapes a user can also write (an + opening like "Done.", a table, "I checked") count only on a message long enough to be a report.""" + + REPORT_OPENING_MIN_CHARS: ClassVar[int] = 300 + DELIVERED_MIN_CHARS: ClassVar[int] = 300 + FORMATTED_MIN_CHARS: ClassVar[int] = 600 + FORMATTED_MIN_HEADERS: ClassVar[int] = 2 + FORMATTED_MIN_BOLD_BULLETS: ClassVar[int] = 3 + WORK_REPORT_MIN_CHARS: ClassVar[int] = 500 + WORK_REPORT_MIN_PHRASES: ClassVar[int] = 2 + + def pattern_reasons(text: str) -> list[str]: """Why ``text`` reads as the assistant's message, one reason per pattern that matched; empty if none.""" + lim = PatternLimits reasons = [] n = len(text) if _AS_ASSISTANT.search(text): reasons.append("it talks like the assistant (offers to do the work, waits on the user, or speaks as the AI)") - if _REPORT_OPENING.search(text) and n > 300: + if _REPORT_OPENING.search(text) and n > lim.REPORT_OPENING_MIN_CHARS: reasons.append("it opens like a report of work done") if _NOW_LIVE.search(text): reasons.append("it says something is now live or updated") - if _DELIVERED.search(text) and n > 300: + if _DELIVERED.search(text) and n > lim.DELIVERED_MIN_CHARS: reasons.append("it reports something delivered or revised") - if (_TABLE.search(text) or len(_HEADER.findall(text)) >= 2 or len(_BOLD_BULLET.findall(text)) >= 3) and n > 600: + formatted = (_TABLE.search(text) or len(_HEADER.findall(text)) >= lim.FORMATTED_MIN_HEADERS + or len(_BOLD_BULLET.findall(text)) >= lim.FORMATTED_MIN_BOLD_BULLETS) + if formatted and n > lim.FORMATTED_MIN_CHARS: reasons.append("it is a formatted deliverable (a table, headings or bold bullet points)") work = _WORK_DONE.findall(text) - if work and (n > 500 or len(work) >= 2): + if work and (n > lim.WORK_REPORT_MIN_CHARS or len(work) >= lim.WORK_REPORT_MIN_PHRASES): reasons.append("it reports checking or building things itself") return reasons @@ -164,6 +181,10 @@ class UserSimGate: context_chars: int = 3000 judge_timeout_seconds: float = 60 + # How much of a judge error and of the judge's reason a verdict keeps. + RECORDED_ERROR_CHARS: ClassVar[int] = 300 + RECORDED_REASON_CHARS: ClassVar[int] = 500 + def __post_init__(self) -> None: if self.on_fail not in ON_FAIL: raise ValueError(f"user_sim_gate.on_fail must be one of {ON_FAIL}, not {self.on_fail!r}") @@ -174,7 +195,8 @@ def __post_init__(self) -> None: @classmethod def from_dict(cls, data: Optional[dict]) -> Optional[UserSimGate]: - if not data: + """None (no gate) for None; the defaults for {}, as for ``UserSimGate()``.""" + if data is None: return None known = set(cls.__dataclass_fields__) unknown = set(data) - known @@ -195,13 +217,13 @@ async def check(self, candidate: str, history: list[tuple[str, str]], *, call_co judged = await self._judge(candidate, history, call_config) except Exception as e: # the judge is a second opinion; never let it take the conversation down logger.warning(f"user-sim gate judge call failed ({type(e).__name__}: {e}); using patterns only") - verdict.judge_error = f"{type(e).__name__}: {e}"[:300] + verdict.judge_error = f"{type(e).__name__}: {e}"[: self.RECORDED_ERROR_CHARS] return verdict if judged is None: verdict.judge_error = "unparseable judge output" return verdict verdict.judge_in_role = judged["in_role"] - verdict.judge_reason = str(judged.get("reason", ""))[:500] + verdict.judge_reason = str(judged.get("reason", ""))[: self.RECORDED_REASON_CHARS] if not judged["in_role"] and (reasons or len(candidate) >= self.judge_min_chars): verdict.passed = False return verdict diff --git a/tst/unit/task_step/test_prompt_agent_user_side.py b/tst/unit/task_step/test_prompt_agent_user_side.py index 971bef3c8..a18ccb134 100644 --- a/tst/unit/task_step/test_prompt_agent_user_side.py +++ b/tst/unit/task_step/test_prompt_agent_user_side.py @@ -601,6 +601,8 @@ def test_the_gate_round_trips_through_the_step_dict_and_rejects_bad_settings(): assert again.user_sim_gate == step.user_sim_gate assert again.user_sim_gate.max_rewrites == 1 assert "user_sim_gate" not in _step().to_dict() + defaults = _step(user_sim_gate={}).user_sim_gate # empty settings turn the gate on with its defaults + assert defaults is not None and defaults.patterns and defaults.judge_model is None and defaults.max_rewrites == 2 with pytest.raises(ValueError, match="unknown keys"): _step(user_sim_gate={"judge": "x"}) with pytest.raises(ValueError, match="on_fail"): From e63d512fe64d116dd196b06c913aa07b40f51efd Mon Sep 17 00:00:00 2001 From: t Date: Tue, 10 Nov 2026 13:52:41 -0500 Subject: [PATCH 3/3] feat(prompt_agent): run_gate for any conversation loop; refuse the gate where it would be ignored The check-and-rewrite loop moves to user_sim_gate.run_gate, with a rewrite callback, so a step type that runs its own conversation can apply the gate; prompt_agent's user side now calls it. A PromptAgentTaskStep subclass that replaces prompt_agent's conversation refuses user_sim_gate at construction, instead of accepting and ignoring it, unless it sets USER_SIM_GATE_SUPPORTED = True. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 68 ++++++++++------- .../task_step/task_steps/user_sim_gate.py | 68 ++++++++++++++++- tst/unit/task_step/test_user_sim_gate.py | 73 +++++++++++++++++++ 3 files changed, 180 insertions(+), 29 deletions(-) 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 3bce8bae4..ed9a069f5 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -41,7 +41,7 @@ from agent_env.task_step.task_step import RetryConfig, TaskStep, TaskStepDependency from agent_env.task_step.task_steps.notification_delivery import delivered_parts, next_message, take_due 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.user_sim_gate import UserSimGate, rewrite_request +from agent_env.task_step.task_steps.user_sim_gate import UserSimGate, judge_call_config, run_gate from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import ( agent_error_text, fetch_container_logs, @@ -143,6 +143,16 @@ class PromptAgentTaskStep(TaskStep): FieldScope.USER)} DEFAULT_PROMPT_TIMEOUT_SECONDS: ClassVar[int] = 600 DEFAULT_POLL_INTERVAL_SECONDS: ClassVar[int] = 2 + # Whether this step type applies user_sim_gate. None: it does when it keeps prompt_agent's conversation; a + # subclass that runs its own loop sets True once that loop calls user_sim_gate.run_gate. + USER_SIM_GATE_SUPPORTED: ClassVar[Optional[bool]] = None + _CONVERSATION_METHODS: ClassVar[tuple[str, ...]] = ("_execute", "_execute_conversation", "_converse") + + @classmethod + def applies_user_sim_gate(cls) -> bool: + if cls.USER_SIM_GATE_SUPPORTED is not None: + return cls.USER_SIM_GATE_SUPPORTED + return all(getattr(cls, m) is getattr(PromptAgentTaskStep, m) for m in cls._CONVERSATION_METHODS) def __init__( self, @@ -229,6 +239,11 @@ def __init__( # too, as JSON task files and the kwargs form of put() carry it. self.user_sim_gate = (UserSimGate.from_dict(user_sim_gate) if isinstance(user_sim_gate, dict) else user_sim_gate) + if self.user_sim_gate is not None and not type(self).applies_user_sim_gate(): + raise ValueError( + f"{type(self).__name__} '{id}': user_sim_gate is applied by prompt_agent's own conversation, which " + f"this step type replaces; it would be ignored. Call user_sim_gate.run_gate from its loop and set " + f"USER_SIM_GATE_SUPPORTED = True, or leave user_sim_gate out") @staticmethod def _validate_parts(parts: list[dict]) -> None: @@ -1113,37 +1128,32 @@ async def _ask_peer(self, number: int, agent_parts: list[dict]) -> dict: return await self._gated(number, answer) async def _gated(self, number: int, answer: dict) -> dict: - """Check the user-sim's message; ask for a rewrite while it fails, up to the gate's limit.""" + """Check the user-sim's message; ask for a rewrite while it fails, up to the gate's limit (``run_gate``).""" gate = self._gate - if self._gate_call_config is None and gate.judge_model: - config = get_config() - overrides = self._context.metadata.get("user_overrides", {}) - self._gate_call_config = config.resolve_model_call( - gate.judge_model, - default_api_key=config.resolve_reference( - overrides.get("judge_litellm_api_key") or overrides.get("litellm_api_key")), - base_override=overrides.get("judge_litellm_base_url") or overrides.get("litellm_base_url"), - ) - for attempt in range(gate.max_rewrites + 1): - text = "\n".join(p.get("text", "") for p in answer["message"] if p.get("kind") == "text") - verdict = await gate.check(text, self._history, call_config=self._gate_call_config) + if self._gate_call_config is None: + self._gate_call_config = judge_call_config(gate, self._context.metadata.get("user_overrides", {})) + latest = {"answer": answer} + + async def rewrite(note: str) -> Optional[str]: + reply = await self._send_to_peer(number, [{"kind": "text", "text": note}]) + latest["answer"] = reply + return None if "end" in reply else _text_of(reply["message"]) + + def on_verdict(attempt: int, text: str, verdict) -> None: self._record("user_sim_gate", {"turn": number, "attempt": attempt, "chars": len(text), **verdict.to_dict()}) - if verdict.passed: - return answer - logger.warning(f"Turn {number}: user-sim message failed the gate (attempt {attempt + 1}/" - f"{gate.max_rewrites + 1}): {'; '.join(verdict.reasons())[:300]}") - if attempt == gate.max_rewrites: - break - answer = await self._send_to_peer(number, [{"kind": "text", "text": rewrite_request(verdict)}]) - if "end" in answer: - return answer - if gate.on_fail == "send": - return answer + if not verdict.passed: + logger.warning(f"Turn {number}: user-sim message failed the gate (attempt {attempt + 1}/" + f"{gate.max_rewrites + 1}): {'; '.join(verdict.reasons())[:300]}") + + outcome = await run_gate(gate, _text_of(answer["message"]), self._history, rewrite, + call_config=self._gate_call_config, on_verdict=on_verdict) + if not outcome.invalid: + return latest["answer"] self._context.metadata.setdefault("user_sim_invalid", {})[self._step.id] = { - "turn": number, "reasons": verdict.reasons()} + "turn": number, "reasons": outcome.reasons()} if gate.on_fail == "fail": self.gate_failure = (f"user-sim message at turn {number} still reads as the assistant's after " - f"{gate.max_rewrites} rewrite(s): {'; '.join(verdict.reasons())[:500]}") + f"{gate.max_rewrites} rewrite(s): {'; '.join(outcome.reasons())[:500]}") return {"end": "user_sim_invalid"} async def _send_to_peer(self, number: int, agent_parts: list[dict]) -> dict: @@ -1198,6 +1208,10 @@ def _record(self, key: str, entry: dict) -> None: self._context.metadata.setdefault(key, {}).setdefault(self._step.id, []).append(entry) +def _text_of(parts: list[dict]) -> str: + return "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text") + + def _trigger_answer(resp: dict) -> Optional[dict]: """An agent-trigger decide's answer: it ended the conversation, it speaks for the user, or None when none fired.""" if bool(resp.get("done")): diff --git a/src/agent_env/task_step/task_steps/user_sim_gate.py b/src/agent_env/task_step/task_steps/user_sim_gate.py index a4807ac82..8a127d42b 100644 --- a/src/agent_env/task_step/task_steps/user_sim_gate.py +++ b/src/agent_env/task_step/task_steps/user_sim_gate.py @@ -12,7 +12,8 @@ - ``judge_model``: one model call that reads the last turns and the candidate message and says whether the message is written as the user. -Opt in with ``user_sim_gate`` on a ``prompt_agent`` step that has a user-sim:: +Opt in with ``user_sim_gate`` on a ``prompt_agent`` step that has a user-sim (a step type that runs its own +conversation loop applies the gate with ``run_gate``, and sets ``USER_SIM_GATE_SUPPORTED``):: {"judge_model": "", "max_rewrites": 2, "on_fail": "fail"} """ @@ -22,7 +23,7 @@ import logging import re from dataclasses import asdict, dataclass, field -from typing import ClassVar, Optional +from typing import Awaitable, Callable, ClassVar, Optional logger = logging.getLogger(__name__) @@ -276,3 +277,66 @@ def rewrite_request(verdict: GateVerdict) -> str: "the work and reports it; you ask, answer what you were asked, decide, push back, or end. Write your next " "message to the assistant again, as yourself, in the same reply format as before." ) + + +@dataclass +class GateOutcome: + """How one user-sim message fared: the text to send (the last one checked), every verdict in order, and + ``invalid`` when it still failed after the rewrites and ``on_fail`` isn't "send". ``stopped`` is set when the + rewrite callback ended the exchange instead of returning a new message.""" + + text: Optional[str] + verdicts: list[GateVerdict] = field(default_factory=list) + invalid: bool = False + stopped: bool = False + + def reasons(self) -> list[str]: + return self.verdicts[-1].reasons() if self.verdicts else [] + + +async def run_gate( + gate: UserSimGate, + text: str, + history: list[tuple[str, str]], + rewrite: Callable[[str], Awaitable[Optional[str]]], + *, + call_config=None, + on_verdict: Optional[Callable[[int, str, GateVerdict], None]] = None, +) -> GateOutcome: + """Check ``text``; while it fails, send the user-sim ``rewrite_request`` through ``rewrite`` and check its new + message, up to ``gate.max_rewrites`` times. ``rewrite`` returns the new message, or None when the user side + ended instead. ``on_verdict(attempt, text, verdict)`` sees every check, for recording. Any conversation loop + can call this; ``prompt_agent`` does, between a user-sim's reply and the agent's next turn.""" + outcome = GateOutcome(text=text) + for attempt in range(gate.max_rewrites + 1): + verdict = await gate.check(text, history, call_config=call_config) + outcome.verdicts.append(verdict) + if on_verdict is not None: + on_verdict(attempt, text, verdict) + if verdict.passed: + return outcome + if attempt == gate.max_rewrites: + break + new_text = await rewrite(rewrite_request(verdict)) + if new_text is None: + outcome.stopped = True + return outcome + text = outcome.text = new_text + outcome.invalid = gate.on_fail != "send" + return outcome + + +def judge_call_config(gate: UserSimGate, user_overrides: dict): + """The model call settings for the gate's judge, honoring a run's judge or LiteLLM overrides the way the + rubrics verifier does; None without a judge.""" + if not gate.judge_model: + return None + from agent_env.store import get_config + + config = get_config() + return config.resolve_model_call( + gate.judge_model, + default_api_key=config.resolve_reference( + user_overrides.get("judge_litellm_api_key") or user_overrides.get("litellm_api_key")), + base_override=user_overrides.get("judge_litellm_base_url") or user_overrides.get("litellm_base_url"), + ) diff --git a/tst/unit/task_step/test_user_sim_gate.py b/tst/unit/task_step/test_user_sim_gate.py index fca748ad4..e305d47ee 100644 --- a/tst/unit/task_step/test_user_sim_gate.py +++ b/tst/unit/task_step/test_user_sim_gate.py @@ -69,3 +69,76 @@ async def _judge(self, candidate, history, call_config): long = asyncio.run(gate.check("The two states that need contracted physicians are these." + _FILLER, [])) assert (short.passed, short.judge_in_role) == (True, False) # recorded, not failed assert long.passed is False + + +# ---- run_gate, for any conversation loop ------------------------------------------------------------------------------ + +def _scripted_gate(verdicts, **settings): + from agent_env.task_step.task_steps.user_sim_gate import GateVerdict, UserSimGate + + class Gate(UserSimGate): + async def check(self, candidate, history, *, call_config=None): + return GateVerdict(passed=verdicts.pop(0), pattern_reasons=["it opens like a report of work done"]) + + return Gate(**settings) + + +def test_run_gate_rewrites_until_a_message_passes(): + import asyncio + from agent_env.task_step.task_steps.user_sim_gate import run_gate + + asked, seen = [], [] + + async def rewrite(note): + asked.append(note) + return f"rewrite {len(asked)}" + + outcome = asyncio.run(run_gate(_scripted_gate([False, False, True]), "draft", [], rewrite, + on_verdict=lambda attempt, text, v: seen.append((attempt, text, v.passed)))) + assert (outcome.text, outcome.invalid, outcome.stopped) == ("rewrite 2", False, False) + assert seen == [(0, "draft", False), (1, "rewrite 1", False), (2, "rewrite 2", True)] + assert asked[0].startswith("[Conversation system note, not from the assistant.]") + + +def test_run_gate_marks_a_message_that_never_passes_and_honors_send(): + import asyncio + from agent_env.task_step.task_steps.user_sim_gate import run_gate + + async def rewrite(note): + return "still the assistant" + + failed = asyncio.run(run_gate(_scripted_gate([False, False], max_rewrites=1), "draft", [], rewrite)) + sent = asyncio.run(run_gate(_scripted_gate([False, False], max_rewrites=1, on_fail="send"), "draft", [], rewrite)) + assert (failed.invalid, len(failed.verdicts)) == (True, 2) + assert (sent.invalid, sent.text) == (False, "still the assistant") + + +def test_run_gate_stops_when_the_user_side_ends_instead_of_rewriting(): + import asyncio + from agent_env.task_step.task_steps.user_sim_gate import run_gate + + async def rewrite(note): + return None + + outcome = asyncio.run(run_gate(_scripted_gate([False]), "draft", [], rewrite)) + assert (outcome.stopped, outcome.invalid, len(outcome.verdicts)) == (True, False, 1) + + +# ---- step types that run their own conversation ------------------------------------------------------------------ + +def test_a_step_type_that_replaces_the_conversation_refuses_the_gate_unless_it_applies_it(): + import pytest + from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep + + class OwnLoop(PromptAgentTaskStep): + async def _execute(self, context): + return context + + class OwnLoopWithGate(OwnLoop): + USER_SIM_GATE_SUPPORTED = True + + with pytest.raises(ValueError, match="OwnLoop 'talk': user_sim_gate is applied by prompt_agent's own conversation"): + OwnLoop(id="talk", version=None, prompt="hi", user_sim_gate={}) + assert OwnLoop(id="talk", version=None, prompt="hi").user_sim_gate is None + assert OwnLoopWithGate(id="talk", version=None, prompt="hi", user_sim_gate={}).user_sim_gate is not None + assert PromptAgentTaskStep.applies_user_sim_gate() and not OwnLoop.applies_user_sim_gate()