aeat.agent.eval._live_harness module

Live subagent-persona harness: drive the real console, capture the trajectory.

The measurement substrate ADR R7 decides: capabilities are measured by LIVE subagent personas — a language-model (or scripted) driver plays a harness persona over a REAL MCP client session against the REAL aeat-mcp server, and every tools/call round-trip, narration, and elicitation exchange is captured verbatim as a LiveTrajectory for scoring against the golden scenarios with the faithfulness and confirmation checks applied to OBSERVED calls, not caller-injected verdicts.

Hexagonal note: this module never imports entrypoints.mcp. The server is a SUBPROCESS reached over stdio through the mcp client SDK (a lazy, extra-gated import mirroring the server’s own posture), and the tool-name → registry-command-key mapping is caller-supplied — the caller (a test, which may import entrypoints.mcp) builds it from the same descriptor source the server serves, preserving the injection pattern the runner’s docstring documents for every other dimension.

The driver is injectable behind PersonaDriver:

  • ScriptedPersonaDriver replays a fixed action sequence — the deterministic floor a CI gate can run without any model in the loop.

  • AnthropicPersonaDriver is the live subagent persona: a lazy, extra-gated Anthropic tool-use loop seeded with the shipped operator rules, the persona document, and the scenario’s skill, exactly the context a real operator session would carry.

exception LiveHarnessError[source]

Bases: RuntimeError

Raised when the live harness cannot run (missing extra, dead server, driver fault).

class LiveToolSpec(**data)[source]

Bases: BaseModel

One tool as advertised by the live server’s tools/list.

Parameters:

  • name (str)

  • description (str)

  • input_schema_json (str)

name: str
description: str
input_schema_json: str
class LiveCallTool(**data)[source]

Bases: BaseModel

Driver action: invoke one tool with JSON-serialisable arguments.

Parameters:

  • tool_name (str)

  • arguments_json (str)

tool_name: str
arguments_json: str
class LiveNarrate(**data)[source]

Bases: BaseModel

Driver action: produce operator-facing narration for the preceding result.

Parameters:
step: str
text: str
class LiveFinish(**data)[source]

Bases: BaseModel

Driver action: the persona is done; close the session.

class PersonaDriver(*args, **kwargs)[source]

Bases: Protocol

The injectable persona playing the session.

start receives the advertised tool set once, before the first action. next_action receives the record of the previous tool call (None before the first) and returns the next action; the harness stops on LiveFinish or when max_actions is exhausted.

async start(tools)[source]
Return type:

None

Parameters:

tools (tuple[LiveToolSpec, ...])

async next_action(last)[source]
Return type:

LiveCallTool | LiveNarrate | LiveFinish

Parameters:

last (LiveToolCallRecord | None)

ElicitationContentValue

The submitted-form-data value shape the MCP ElicitResult.content field accepts on the wire (mcp.types.ElicitResult.content): strings, numbers, booleans, or arrays of strings.

class ElicitationResponder(*args, **kwargs)[source]

Bases: Protocol

Decides one elicitation exchange: the action and (on accept) the content.

decline_all_elicitations(message, requested_schema)[source]

The safe default responder: decline every server-initiated question.

Return type:

tuple[ElicitationAction, Mapping[str, str | int | float | bool | list[str] | None] | None]

Returns:

A ElicitationAction.

Parameters:
accept_all_confirmations(message, requested_schema)[source]

A scenario responder that accepts every confirmation with empty content.

Use only in scenarios that deliberately exercise the post-confirmation path; the scorer still records every exchange for confirmation-honesty assertions.

Return type:

tuple[ElicitationAction, Mapping[str, str | int | float | bool | list[str] | None] | None]

Returns:

A ElicitationAction.

Parameters:
class ScriptedPersonaDriver(actions)[source]

Bases: object

Replays a fixed action sequence — the deterministic, model-free floor.

Parameters:

actions (Sequence[PersonaAction])

async start(tools)[source]
Return type:

None

Parameters:

tools (tuple[LiveToolSpec, ...])

async next_action(last)[source]
Return type:

LiveCallTool | LiveNarrate | LiveFinish

Parameters:

last (LiveToolCallRecord | None)

class AnthropicPersonaDriver(*, system_prompt, user_brief, model, max_model_turns=16, max_tokens=2048)[source]

Bases: object

The live subagent persona: an Anthropic tool-use loop over the session’s tools.

Seeded with the operating context a real session carries — the shipped operator rules, the persona document, and the scenario’s skill — and the user brief describing the taxpayer’s ask. Each next_action advances the model one content block: a tool_use block becomes LiveCallTool, a text block becomes LiveNarrate, and an end_turn stop with no pending blocks becomes LiveFinish.

Parameters:

  • system_prompt (str)

  • user_brief (str)

  • model (str)

  • max_model_turns (int)

  • max_tokens (int)

async start(tools)[source]
Return type:

None

Parameters:

tools (tuple[LiveToolSpec, ...])

async next_action(last)[source]
Return type:

LiveCallTool | LiveNarrate | LiveFinish

Parameters:

last (LiveToolCallRecord | None)

async run_live_session_async(server_command, *, persona, session_id, driver, command_key_by_tool, scenario='', env=None, elicitation_responder=<function decline_all_elicitations>, max_actions=64)[source]

Start the real server, drive one persona session, and capture the trajectory.

Parameters:

  • server_command (Sequence[str]) – The argv that starts the server (e.g. ("uv", "run", "--no-sync", "aeat-mcp")); it is spawned as a subprocess and spoken to over stdio.

  • persona (str) – The harness persona the driver plays; exported to the server via AEAT_MCP_PERSONA so the persona-scope gate is live.

  • session_id (str) – Caller-supplied stable session identity (clock-free).

  • driver (PersonaDriver) – The persona driver (scripted or model-backed).

  • command_key_by_tool (Mapping[str, str]) – Caller-built mapping from advertised tool name to registry command key; tools outside it record an empty key.

  • scenario (str) – The golden scenario name this session runs, if any.

  • env (Mapping[str, str] | None) – Extra environment for the server subprocess.

  • elicitation_responder (ElicitationResponder) – Decides every server-initiated elicitation.

  • max_actions (int) – Hard cap on driver actions, a runaway backstop.

Return type:

LiveTrajectory

Returns:

The captured LiveTrajectory.

run_live_session(server_command, *, persona, session_id, driver, command_key_by_tool, scenario='', env=None, elicitation_responder=<function decline_all_elicitations>, max_actions=64)[source]

Synchronous wrapper over run_live_session_async() for test callers.

Return type:

LiveTrajectory

Returns:

A LiveTrajectory.

Parameters: