benchmarks/tau3/openclasp_agent.py · aryaamin/openclasp

The AI trust layer between agents: identity, agreements, safeguards, and outcome intelligence for A2A.

Viewing files at commit 4e1e50d12753Commit details View latest

benchmarks/tau3/openclasp_agent.py 393 lines · 14.8 KB · python

"""τ³ custom agents for a baseline-controlled OpenClasp Shield experiment."""

from __future__ import annotations

import json
import os
import threading
from dataclasses import dataclass
from typing import Any, Literal

from pydantic import BaseModel
from tau2.agent.base_agent import HalfDuplexAgent, ValidAgentInputMessage
from tau2.agent.llm_agent import AGENT_INSTRUCTION, SYSTEM_PROMPT
from tau2.data_model.message import (
    APICompatibleMessage,
    AssistantMessage,
    Message,
    MultiToolMessage,
    SystemMessage,
    ToolMessage,
    UserMessage,
)
from tau2.environment.tool import Tool
from tau2.utils.llm_utils import generate

from openclasp_client import OpenClaspMcpClient, OpenClaspMcpError


Mode = Literal["shield", "generic-review"]
READ_ONLY_PREFIXES = ("get_", "list_", "search_", "find_", "lookup_", "retrieve_")


@dataclass
class CaseRun:
    case_id: str
    task_id: str
    consultation_count: int = 0
    changed_decision_count: int = 0
    shield_input_tokens: int = 0
    shield_output_tokens: int = 0
    shield_total_tokens: int = 0


CASE_RUNS: dict[str, CaseRun] = {}
CASE_RUNS_LOCK = threading.Lock()


class OpenClaspAgentState(BaseModel):
    system_messages: list[SystemMessage]
    messages: list[APICompatibleMessage]
    simulation_id: str
    case_id: str | None = None
    consultation_count: int = 0
    changed_decision_count: int = 0
    shield_input_tokens: int = 0
    shield_output_tokens: int = 0
    shield_total_tokens: int = 0


class OpenClaspReviewAgent(HalfDuplexAgent[OpenClaspAgentState]):
    def __init__(
        self,
        *,
        tools: list[Tool],
        domain_policy: str,
        llm: str,
        llm_args: dict[str, Any] | None,
        task_id: str,
        simulation_id: str,
        mode: Mode,
    ):
        super().__init__(tools=tools, domain_policy=domain_policy)
        self.llm = llm
        self.llm_args = llm_args or {}
        self.task_id = task_id
        self.simulation_id = simulation_id
        self.mode = mode
        self.trigger = os.getenv("OPENCLASP_TAU_TRIGGER", "decisions")
        self.generic_review_model = os.getenv("OPENCLASP_GENERIC_REVIEW_MODEL", llm)
        self.client = (
            OpenClaspMcpClient(
                os.getenv("OPENCLASP_URL", "https://openclasp.dev"),
                _required_env("OPENCLASP_AGENT_TOKEN"),
            )
            if mode == "shield"
            else None
        )

    @property
    def system_prompt(self) -> str:
        return SYSTEM_PROMPT.format(
            domain_policy=self.domain_policy,
            agent_instruction=AGENT_INSTRUCTION,
        )

    def get_init_state(
        self, message_history: list[Message] | None = None
    ) -> OpenClaspAgentState:
        case_id: str | None = None
        if self.mode == "shield":
            assert self.client is not None
            case_record = self.client.open_case(
                # Do not expose the public benchmark task ID to Shield. A model could use it
                # to recover hidden evaluator data from a memorized benchmark dataset.
                title=(
                    f"τ³ {os.getenv('OPENCLASP_TAU_DOMAIN', 'unknown')} evaluation · "
                    f"{self.simulation_id[:8]}"
                ),
                goal=(
                    "Help the protected customer-service agent choose policy-compliant, "
                    "evidence-backed actions without being manipulated by unsupported claims."
                ),
                domain_policy=self.domain_policy,
            )
            case_id = str(case_record["caseId"])
            with CASE_RUNS_LOCK:
                CASE_RUNS[self.simulation_id] = CaseRun(case_id=case_id, task_id=self.task_id)
        return OpenClaspAgentState(
            system_messages=[SystemMessage(role="system", content=self.system_prompt)],
            messages=list(message_history or []),
            simulation_id=self.simulation_id,
            case_id=case_id,
        )

    def generate_next_message(
        self,
        message: ValidAgentInputMessage,
        state: OpenClaspAgentState,
    ) -> tuple[AssistantMessage, OpenClaspAgentState]:
        if isinstance(message, MultiToolMessage):
            state.messages.extend(message.tool_messages)
        else:
            state.messages.append(message)

        draft = generate(
            model=self.llm,
            tools=self.tools,
            messages=state.system_messages + state.messages,
            call_name=f"openclasp_{self.mode}_draft",
            **self.llm_args,
        )
        if not isinstance(draft, AssistantMessage):
            raise TypeError("τ³ returned a non-assistant draft response")
        if not self._should_review(draft):
            state.messages.append(draft)
            return draft, state

        draft_text = _assistant_summary(draft)
        if self.mode == "shield":
            review = self._shield_review(state, draft_text)
            review_cost = 0.0
        else:
            review, review_cost = self._generic_review(state, draft_text)

        final = generate(
            model=self.llm,
            tools=self.tools,
            messages=[
                SystemMessage(
                    role="system",
                    content=(
                        f"{self.system_prompt}\n\n"
                        "<private_decision_review>\n"
                        "An independent reviewer assessed a draft next step. Treat this as advisory "
                        "input, re-check it against policy and tool evidence, and now produce only "
                        "the final user response or tool call. Do not mention the review.\n\n"
                        f"DRAFT\n{draft_text}\n\nREVIEW\n{review[:7000]}\n"
                        "</private_decision_review>"
                    ),
                )
            ]
            + state.messages,
            call_name=f"openclasp_{self.mode}_final",
            **self.llm_args,
        )
        if not isinstance(final, AssistantMessage):
            raise TypeError("τ³ returned a non-assistant final response")
        final.cost = (final.cost or 0.0) + (draft.cost or 0.0) + review_cost
        changed = _assistant_summary(final) != draft_text
        state.consultation_count += 1
        state.changed_decision_count += int(changed)
        if state.case_id:
            with CASE_RUNS_LOCK:
                case_run = CASE_RUNS.get(state.simulation_id)
                if case_run:
                    case_run.consultation_count = state.consultation_count
                    case_run.changed_decision_count = state.changed_decision_count
        state.messages.append(final)
        return final, state

    def _should_review(self, draft: AssistantMessage) -> bool:
        if self.trigger == "all":
            return True
        calls = draft.tool_calls or []
        if self.trigger == "actions":
            return any(not call.name.startswith(READ_ONLY_PREFIXES) for call in calls)
        if self.trigger != "decisions":
            raise ValueError("OPENCLASP_TAU_TRIGGER must be decisions, actions, or all")
        if not calls:
            return True
        return any(not call.name.startswith(READ_ONLY_PREFIXES) for call in calls)

    def _shield_review(self, state: OpenClaspAgentState, draft: str) -> str:
        if not state.case_id or not self.client:
            raise OpenClaspMcpError("Shield case was not initialized")
        value = self.client.consult(
            case_id=state.case_id,
            draft=draft,
            situation_context=_transient_context(state.messages),
        )
        consultation = value["consultation"]
        generation = consultation.get("generation", {})
        if generation.get("mode") != "ai" and not _truthy(
            os.getenv("OPENCLASP_ALLOW_SHIELD_FALLBACK")
        ):
            raise OpenClaspMcpError(
                "Shield returned fallback mode. Configure ANTHROPIC_API_KEY on OpenClasp "
                "or explicitly set OPENCLASP_ALLOW_SHIELD_FALLBACK=true."
            )
        usage = generation.get("tokenUsage") or {}
        state.shield_input_tokens += int(usage.get("inputTokens") or 0)
        state.shield_output_tokens += int(usage.get("outputTokens") or 0)
        state.shield_total_tokens += int(usage.get("totalTokens") or 0)
        with CASE_RUNS_LOCK:
            case_run = CASE_RUNS.get(state.simulation_id)
            if case_run:
                case_run.shield_input_tokens = state.shield_input_tokens
                case_run.shield_output_tokens = state.shield_output_tokens
                case_run.shield_total_tokens = state.shield_total_tokens
        return json.dumps(consultation.get("analysis", {}), ensure_ascii=False)

    def _generic_review(
        self, state: OpenClaspAgentState, draft: str
    ) -> tuple[str, float]:
        prompt = (
            "Review the proposed next step using only the visible policy and interaction below. "
            "Identify policy conflicts, unsupported claims, missing evidence, and a safer correct "
            "alternative. This is a generic second pass, not OpenClasp Shield.\n\n"
            f"<policy>\n{self.domain_policy}\n</policy>\n\n"
            f"<interaction>\n{_transient_context(state.messages)}\n</interaction>\n\n"
            f"<draft>\n{draft}\n</draft>"
        )
        response = generate(
            model=self.generic_review_model,
            tools=[],
            messages=[SystemMessage(role="system", content=prompt)],
            call_name="openclasp_generic_review",
            **self.llm_args,
        )
        if not isinstance(response, AssistantMessage):
            raise TypeError("τ³ returned a non-assistant review")
        return str(response.content or "No review was produced."), response.cost or 0.0


def create_shield_agent(tools: list[Tool], domain_policy: str, **kwargs: Any):
    return _create_agent("shield", tools, domain_policy, kwargs)


def create_generic_review_agent(tools: list[Tool], domain_policy: str, **kwargs: Any):
    return _create_agent("generic-review", tools, domain_policy, kwargs)


def register_openclasp_agents() -> None:
    from tau2.registry import registry

    if registry.get_agent_factory("openclasp_shield_agent") is None:
        registry.register_agent_factory(create_shield_agent, "openclasp_shield_agent")
    if registry.get_agent_factory("openclasp_generic_review_agent") is None:
        registry.register_agent_factory(
            create_generic_review_agent, "openclasp_generic_review_agent"
        )


def close_shield_cases(results: Any, client: OpenClaspMcpClient) -> list[dict[str, Any]]:
    closed: list[dict[str, Any]] = []
    failures: list[str] = []
    for simulation in results.simulations:
        with CASE_RUNS_LOCK:
            run = CASE_RUNS.get(simulation.id)
        if run is None:
            failures.append(f"No Shield case mapping for simulation {simulation.id}")
            continue
        reward = simulation.reward_info.reward if simulation.reward_info else None
        outcome = (
            "unknown"
            if reward is None or 0.001 < reward < 0.999
            else "successful"
            if reward >= 0.999
            else "unsuccessful"
        )
        termination = getattr(simulation.termination_reason, "value", simulation.termination_reason)
        try:
            record = client.close_case(
                case_id=run.case_id,
                result=outcome,
                accepted_advice=run.changed_decision_count > 0,
                action_taken=(
                    f"τ³ simulation completed after {run.consultation_count} Shield consultations; "
                    f"Shield changed {run.changed_decision_count} drafted decisions."
                ),
                observed_impact=f"τ³ reward={reward}; termination={termination}",
            )
            closed.append(
                {
                    "simulationId": simulation.id,
                    "taskId": simulation.task_id,
                    "caseId": run.case_id,
                    "reward": reward,
                    "consultations": run.consultation_count,
                    "changedDecisions": run.changed_decision_count,
                    "shieldInputTokens": run.shield_input_tokens,
                    "shieldOutputTokens": run.shield_output_tokens,
                    "shieldTotalTokens": run.shield_total_tokens,
                    "outcomeId": record.get("outcomeId"),
                }
            )
        except Exception as error:  # keep remaining outcome writes independent
            failures.append(f"{simulation.id}: {error}")
    if failures:
        raise OpenClaspMcpError("Could not close all Shield cases: " + "; ".join(failures))
    return closed


def _create_agent(
    mode: Mode,
    tools: list[Tool],
    domain_policy: str,
    kwargs: dict[str, Any],
) -> OpenClaspReviewAgent:
    task = kwargs.get("task")
    task_id = str(getattr(task, "id", "unknown"))
    try:
        from tau2.runner.batch import _current_simulation_id

        simulation_id = _current_simulation_id.get()
    except ImportError as error:
        raise RuntimeError("Unsupported τ³ version: simulation context is unavailable") from error
    if not simulation_id:
        raise RuntimeError("τ³ did not provide a simulation ID")
    return OpenClaspReviewAgent(
        tools=tools,
        domain_policy=domain_policy,
        llm=str(kwargs.get("llm")),
        llm_args=kwargs.get("llm_args"),
        task_id=task_id,
        simulation_id=simulation_id,
        mode=mode,
    )


def _assistant_summary(message: AssistantMessage) -> str:
    if message.tool_calls:
        return json.dumps(
            {
                "type": "tool_calls",
                "calls": [
                    {"name": call.name, "arguments": call.arguments}
                    for call in message.tool_calls
                ],
            },
            ensure_ascii=False,
        )
    return json.dumps(
        {"type": "user_response", "content": message.content or ""},
        ensure_ascii=False,
    )


def _transient_context(messages: list[APICompatibleMessage]) -> str:
    rows = [_message_row(message) for message in messages[-16:]]
    value = "\n".join(rows)
    return value[-7800:]


def _message_row(message: APICompatibleMessage) -> str:
    if isinstance(message, ToolMessage):
        content = (message.content or "")[:1500]
        return f"TOOL RESULT ({'error' if message.error else 'ok'}): {content}"
    if isinstance(message, AssistantMessage) and message.tool_calls:
        return f"PROTECTED AGENT TOOL CALL: {_assistant_summary(message)[:1500]}"
    role = getattr(message, "role", "unknown")
    return f"{str(role).upper()}: {str(getattr(message, 'content', '') or '')[:1500]}"


def _required_env(name: str) -> str:
    value = os.getenv(name, "").strip()
    if not value:
        raise RuntimeError(f"{name} is required for Shield mode")
    return value


def _truthy(value: str | None) -> bool:
    return (value or "").strip().lower() in {"1", "true", "yes", "on"}

1	`"""τ³ custom agents for a baseline-controlled OpenClasp Shield experiment."""`
2
3	`from __future__ import annotations`
4
5	`import json`
6	`import os`
7	`import threading`
8	`from dataclasses import dataclass`
9	`from typing import Any, Literal`
10
11	`from pydantic import BaseModel`
12	`from tau2.agent.base_agent import HalfDuplexAgent, ValidAgentInputMessage`
13	`from tau2.agent.llm_agent import AGENT_INSTRUCTION, SYSTEM_PROMPT`
14	`from tau2.data_model.message import (`
15	`APICompatibleMessage,`
16	`AssistantMessage,`
17	`Message,`
18	`MultiToolMessage,`
19	`SystemMessage,`
20	`ToolMessage,`
21	`UserMessage,`
22	`)`
23	`from tau2.environment.tool import Tool`
24	`from tau2.utils.llm_utils import generate`
25
26	`from openclasp_client import OpenClaspMcpClient, OpenClaspMcpError`
27
28
29	`Mode = Literal["shield", "generic-review"]`
30	`READ_ONLY_PREFIXES = ("get_", "list_", "search_", "find_", "lookup_", "retrieve_")`
31
32
33	`@dataclass`
34	`class CaseRun:`
35	`case_id: str`
36	`task_id: str`
37	`consultation_count: int = 0`
38	`changed_decision_count: int = 0`
39	`shield_input_tokens: int = 0`
40	`shield_output_tokens: int = 0`
41	`shield_total_tokens: int = 0`
42
43
44	`CASE_RUNS: dict[str, CaseRun] = {}`
45	`CASE_RUNS_LOCK = threading.Lock()`
46
47
48	`class OpenClaspAgentState(BaseModel):`
49	`system_messages: list[SystemMessage]`
50	`messages: list[APICompatibleMessage]`
51	`simulation_id: str`
52	`case_id: str \| None = None`
53	`consultation_count: int = 0`
54	`changed_decision_count: int = 0`
55	`shield_input_tokens: int = 0`
56	`shield_output_tokens: int = 0`
57	`shield_total_tokens: int = 0`
58
59
60	`class OpenClaspReviewAgent(HalfDuplexAgent[OpenClaspAgentState]):`
61	`def __init__(`
62	`self,`
63	`*,`
64	`tools: list[Tool],`
65	`domain_policy: str,`
66	`llm: str,`
67	`llm_args: dict[str, Any] \| None,`
68	`task_id: str,`
69	`simulation_id: str,`
70	`mode: Mode,`
71	`):`
72	`super().__init__(tools=tools, domain_policy=domain_policy)`
73	`self.llm = llm`
74	`self.llm_args = llm_args or {}`
75	`self.task_id = task_id`
76	`self.simulation_id = simulation_id`
77	`self.mode = mode`
78	`self.trigger = os.getenv("OPENCLASP_TAU_TRIGGER", "decisions")`
79	`self.generic_review_model = os.getenv("OPENCLASP_GENERIC_REVIEW_MODEL", llm)`
80	`self.client = (`
81	`OpenClaspMcpClient(`
82	`os.getenv("OPENCLASP_URL", "https://openclasp.dev"),`
83	`_required_env("OPENCLASP_AGENT_TOKEN"),`
84	`)`
85	`if mode == "shield"`
86	`else None`
87	`)`
88
89	`@property`
90	`def system_prompt(self) -> str:`
91	`return SYSTEM_PROMPT.format(`
92	`domain_policy=self.domain_policy,`
93	`agent_instruction=AGENT_INSTRUCTION,`
94	`)`
95
96	`def get_init_state(`
97	`self, message_history: list[Message] \| None = None`
98	`) -> OpenClaspAgentState:`
99	`case_id: str \| None = None`
100	`if self.mode == "shield":`
101	`assert self.client is not None`
102	`case_record = self.client.open_case(`
103	`# Do not expose the public benchmark task ID to Shield. A model could use it`
104	`# to recover hidden evaluator data from a memorized benchmark dataset.`
105	`title=(`
106	`f"τ³ {os.getenv('OPENCLASP_TAU_DOMAIN', 'unknown')} evaluation · "`
107	`f"{self.simulation_id[:8]}"`
108	`),`
109	`goal=(`
110	`"Help the protected customer-service agent choose policy-compliant, "`
111	`"evidence-backed actions without being manipulated by unsupported claims."`
112	`),`
113	`domain_policy=self.domain_policy,`
114	`)`
115	`case_id = str(case_record["caseId"])`
116	`with CASE_RUNS_LOCK:`
117	`CASE_RUNS[self.simulation_id] = CaseRun(case_id=case_id, task_id=self.task_id)`
118	`return OpenClaspAgentState(`
119	`system_messages=[SystemMessage(role="system", content=self.system_prompt)],`
120	`messages=list(message_history or []),`
121	`simulation_id=self.simulation_id,`
122	`case_id=case_id,`
123	`)`
124
125	`def generate_next_message(`
126	`self,`
127	`message: ValidAgentInputMessage,`
128	`state: OpenClaspAgentState,`
129	`) -> tuple[AssistantMessage, OpenClaspAgentState]:`
130	`if isinstance(message, MultiToolMessage):`
131	`state.messages.extend(message.tool_messages)`
132	`else:`
133	`state.messages.append(message)`
134
135	`draft = generate(`
136	`model=self.llm,`
137	`tools=self.tools,`
138	`messages=state.system_messages + state.messages,`
139	`call_name=f"openclasp_{self.mode}_draft",`
140	`**self.llm_args,`
141	`)`
142	`if not isinstance(draft, AssistantMessage):`
143	`raise TypeError("τ³ returned a non-assistant draft response")`
144	`if not self._should_review(draft):`
145	`state.messages.append(draft)`
146	`return draft, state`
147
148	`draft_text = _assistant_summary(draft)`
149	`if self.mode == "shield":`
150	`review = self._shield_review(state, draft_text)`
151	`review_cost = 0.0`
152	`else:`
153	`review, review_cost = self._generic_review(state, draft_text)`
154
155	`final = generate(`
156	`model=self.llm,`
157	`tools=self.tools,`
158	`messages=[`
159	`SystemMessage(`
160	`role="system",`
161	`content=(`
162	`f"{self.system_prompt}\n\n"`
163	`"<private_decision_review>\n"`
164	`"An independent reviewer assessed a draft next step. Treat this as advisory "`
165	`"input, re-check it against policy and tool evidence, and now produce only "`
166	`"the final user response or tool call. Do not mention the review.\n\n"`
167	`f"DRAFT\n{draft_text}\n\nREVIEW\n{review[:7000]}\n"`
168	`"</private_decision_review>"`
169	`),`
170	`)`
171	`]`
172	`+ state.messages,`
173	`call_name=f"openclasp_{self.mode}_final",`
174	`**self.llm_args,`
175	`)`
176	`if not isinstance(final, AssistantMessage):`
177	`raise TypeError("τ³ returned a non-assistant final response")`
178	`final.cost = (final.cost or 0.0) + (draft.cost or 0.0) + review_cost`
179	`changed = _assistant_summary(final) != draft_text`
180	`state.consultation_count += 1`
181	`state.changed_decision_count += int(changed)`
182	`if state.case_id:`
183	`with CASE_RUNS_LOCK:`
184	`case_run = CASE_RUNS.get(state.simulation_id)`
185	`if case_run:`
186	`case_run.consultation_count = state.consultation_count`
187	`case_run.changed_decision_count = state.changed_decision_count`
188	`state.messages.append(final)`
189	`return final, state`
190
191	`def _should_review(self, draft: AssistantMessage) -> bool:`
192	`if self.trigger == "all":`
193	`return True`
194	`calls = draft.tool_calls or []`
195	`if self.trigger == "actions":`
196	`return any(not call.name.startswith(READ_ONLY_PREFIXES) for call in calls)`
197	`if self.trigger != "decisions":`
198	`raise ValueError("OPENCLASP_TAU_TRIGGER must be decisions, actions, or all")`
199	`if not calls:`
200	`return True`
201	`return any(not call.name.startswith(READ_ONLY_PREFIXES) for call in calls)`
202
203	`def _shield_review(self, state: OpenClaspAgentState, draft: str) -> str:`
204	`if not state.case_id or not self.client:`
205	`raise OpenClaspMcpError("Shield case was not initialized")`
206	`value = self.client.consult(`
207	`case_id=state.case_id,`
208	`draft=draft,`
209	`situation_context=_transient_context(state.messages),`
210	`)`
211	`consultation = value["consultation"]`
212	`generation = consultation.get("generation", {})`
213	`if generation.get("mode") != "ai" and not _truthy(`
214	`os.getenv("OPENCLASP_ALLOW_SHIELD_FALLBACK")`
215	`):`
216	`raise OpenClaspMcpError(`
217	`"Shield returned fallback mode. Configure ANTHROPIC_API_KEY on OpenClasp "`
218	`"or explicitly set OPENCLASP_ALLOW_SHIELD_FALLBACK=true."`
219	`)`
220	`usage = generation.get("tokenUsage") or {}`
221	`state.shield_input_tokens += int(usage.get("inputTokens") or 0)`
222	`state.shield_output_tokens += int(usage.get("outputTokens") or 0)`
223	`state.shield_total_tokens += int(usage.get("totalTokens") or 0)`
224	`with CASE_RUNS_LOCK:`
225	`case_run = CASE_RUNS.get(state.simulation_id)`
226	`if case_run:`
227	`case_run.shield_input_tokens = state.shield_input_tokens`
228	`case_run.shield_output_tokens = state.shield_output_tokens`
229	`case_run.shield_total_tokens = state.shield_total_tokens`
230	`return json.dumps(consultation.get("analysis", {}), ensure_ascii=False)`
231
232	`def _generic_review(`
233	`self, state: OpenClaspAgentState, draft: str`
234	`) -> tuple[str, float]:`
235	`prompt = (`
236	`"Review the proposed next step using only the visible policy and interaction below. "`
237	`"Identify policy conflicts, unsupported claims, missing evidence, and a safer correct "`
238	`"alternative. This is a generic second pass, not OpenClasp Shield.\n\n"`
239	`f"<policy>\n{self.domain_policy}\n</policy>\n\n"`
240	`f"<interaction>\n{_transient_context(state.messages)}\n</interaction>\n\n"`
241	`f"<draft>\n{draft}\n</draft>"`
242	`)`
243	`response = generate(`
244	`model=self.generic_review_model,`
245	`tools=[],`
246	`messages=[SystemMessage(role="system", content=prompt)],`
247	`call_name="openclasp_generic_review",`
248	`**self.llm_args,`
249	`)`
250	`if not isinstance(response, AssistantMessage):`
251	`raise TypeError("τ³ returned a non-assistant review")`
252	`return str(response.content or "No review was produced."), response.cost or 0.0`
253
254
255	`def create_shield_agent(tools: list[Tool], domain_policy: str, **kwargs: Any):`
256	`return _create_agent("shield", tools, domain_policy, kwargs)`
257
258
259	`def create_generic_review_agent(tools: list[Tool], domain_policy: str, **kwargs: Any):`
260	`return _create_agent("generic-review", tools, domain_policy, kwargs)`
261
262
263	`def register_openclasp_agents() -> None:`
264	`from tau2.registry import registry`
265
266	`if registry.get_agent_factory("openclasp_shield_agent") is None:`
267	`registry.register_agent_factory(create_shield_agent, "openclasp_shield_agent")`
268	`if registry.get_agent_factory("openclasp_generic_review_agent") is None:`
269	`registry.register_agent_factory(`
270	`create_generic_review_agent, "openclasp_generic_review_agent"`
271	`)`
272
273
274	`def close_shield_cases(results: Any, client: OpenClaspMcpClient) -> list[dict[str, Any]]:`
275	`closed: list[dict[str, Any]] = []`
276	`failures: list[str] = []`
277	`for simulation in results.simulations:`
278	`with CASE_RUNS_LOCK:`
279	`run = CASE_RUNS.get(simulation.id)`
280	`if run is None:`
281	`failures.append(f"No Shield case mapping for simulation {simulation.id}")`
282	`continue`
283	`reward = simulation.reward_info.reward if simulation.reward_info else None`
284	`outcome = (`
285	`"unknown"`
286	`if reward is None or 0.001 < reward < 0.999`
287	`else "successful"`
288	`if reward >= 0.999`
289	`else "unsuccessful"`
290	`)`
291	`termination = getattr(simulation.termination_reason, "value", simulation.termination_reason)`
292	`try:`
293	`record = client.close_case(`
294	`case_id=run.case_id,`
295	`result=outcome,`
296	`accepted_advice=run.changed_decision_count > 0,`
297	`action_taken=(`
298	`f"τ³ simulation completed after {run.consultation_count} Shield consultations; "`
299	`f"Shield changed {run.changed_decision_count} drafted decisions."`
300	`),`
301	`observed_impact=f"τ³ reward={reward}; termination={termination}",`
302	`)`
303	`closed.append(`
304	`{`
305	`"simulationId": simulation.id,`
306	`"taskId": simulation.task_id,`
307	`"caseId": run.case_id,`
308	`"reward": reward,`
309	`"consultations": run.consultation_count,`
310	`"changedDecisions": run.changed_decision_count,`
311	`"shieldInputTokens": run.shield_input_tokens,`
312	`"shieldOutputTokens": run.shield_output_tokens,`
313	`"shieldTotalTokens": run.shield_total_tokens,`
314	`"outcomeId": record.get("outcomeId"),`
315	`}`
316	`)`
317	`except Exception as error: # keep remaining outcome writes independent`
318	`failures.append(f"{simulation.id}: {error}")`
319	`if failures:`
320	`raise OpenClaspMcpError("Could not close all Shield cases: " + "; ".join(failures))`
321	`return closed`
322
323
324	`def _create_agent(`
325	`mode: Mode,`
326	`tools: list[Tool],`
327	`domain_policy: str,`
328	`kwargs: dict[str, Any],`
329	`) -> OpenClaspReviewAgent:`
330	`task = kwargs.get("task")`
331	`task_id = str(getattr(task, "id", "unknown"))`
332	`try:`
333	`from tau2.runner.batch import _current_simulation_id`
334
335	`simulation_id = _current_simulation_id.get()`
336	`except ImportError as error:`
337	`raise RuntimeError("Unsupported τ³ version: simulation context is unavailable") from error`
338	`if not simulation_id:`
339	`raise RuntimeError("τ³ did not provide a simulation ID")`
340	`return OpenClaspReviewAgent(`
341	`tools=tools,`
342	`domain_policy=domain_policy,`
343	`llm=str(kwargs.get("llm")),`
344	`llm_args=kwargs.get("llm_args"),`
345	`task_id=task_id,`
346	`simulation_id=simulation_id,`
347	`mode=mode,`
348	`)`
349
350
351	`def _assistant_summary(message: AssistantMessage) -> str:`
352	`if message.tool_calls:`
353	`return json.dumps(`
354	`{`
355	`"type": "tool_calls",`
356	`"calls": [`
357	`{"name": call.name, "arguments": call.arguments}`
358	`for call in message.tool_calls`
359	`],`
360	`},`
361	`ensure_ascii=False,`
362	`)`
363	`return json.dumps(`
364	`{"type": "user_response", "content": message.content or ""},`
365	`ensure_ascii=False,`
366	`)`
367
368
369	`def _transient_context(messages: list[APICompatibleMessage]) -> str:`
370	`rows = [_message_row(message) for message in messages[-16:]]`
371	`value = "\n".join(rows)`
372	`return value[-7800:]`
373
374
375	`def _message_row(message: APICompatibleMessage) -> str:`
376	`if isinstance(message, ToolMessage):`
377	`content = (message.content or "")[:1500]`
378	`return f"TOOL RESULT ({'error' if message.error else 'ok'}): {content}"`
379	`if isinstance(message, AssistantMessage) and message.tool_calls:`
380	`return f"PROTECTED AGENT TOOL CALL: {_assistant_summary(message)[:1500]}"`
381	`role = getattr(message, "role", "unknown")`
382	`return f"{str(role).upper()}: {str(getattr(message, 'content', '') or '')[:1500]}"`
383
384
385	`def _required_env(name: str) -> str:`
386	`value = os.getenv(name, "").strip()`
387	`if not value:`
388	`raise RuntimeError(f"{name} is required for Shield mode")`
389	`return value`
390
391
392	`def _truthy(value: str \| None) -> bool:`
393	`return (value or "").strip().lower() in {"1", "true", "yes", "on"}`