RFC-0018: Agent Simulation Framework¶

class SimulationBuffer:
    """Simple text buffer that accumulates all execution events as readable text.

    Sim Agent sees the accumulated context as plain text, not structured events.
    """

    def __init__(self, run_id: str):
        self.run_id = run_id
        self.lines: list[str] = []

    def add_text(self, text: str) -> None:
        """Add text fragment from PartDeltaEvent."""
        self.lines.append(text)

    def add_tool_call(self, name: str, args: dict) -> None:
        """Add tool call notification."""
        self.lines.append(f"\n[Calling tool: {name}]")
        if args:
            args_str = ", ".join(f"{k}={v}" for k, v in args.items())
            self.lines.append(f"  Arguments: {args_str}")

    def add_tool_result(self, name: str, result: Any) -> str:
        """Add tool result summary."""
        summary = self._summarize(result)
        self.lines.append(f"[Tool {name} result: {summary}]\n")

    def get_context(self) -> str:
        """Get accumulated context as single text block."""
        return "".join(self.lines)

    def _summarize(self, result: Any, max_len: int = 200) -> str:
        """Summarize tool result for display."""
        text = str(result)
        if len(text) > max_len:
            return text[:max_len] + "..."
        return text

class SimulationToolProvider(ResourceProvider):
    """Provides tools for Sim Agent to interact with Target Agent.

    Key design: Non-blocking initiation with background monitoring.
    All execution events accumulated as text, only 2 event types emitted.
    """

    def __init__(
        self,
        target_agent: Agent,
        input_provider: SimulationInputProvider,
        timeout: float = 300.0,
    ):
        self.target_agent = target_agent
        self.input_provider = input_provider
        self.timeout = timeout
        self._active_runs: dict[str, SimulationBuffer] = {}

    @tool
    async def talk_to_target(
        self,
        ctx: AgentContext,
        message: str,
    ) -> dict:
        """Send message to Target Agent and return immediately.

        Returns run_id immediately. Target runs in background,
        events monitored and accumulated as text.

        Returns:
            {"status": "started", "run_id": str}
        """
        run_id = str(uuid.uuid4())

        # Start background monitoring task
        asyncio.create_task(
            self._monitor_target_run(
                run_id=run_id,
                message=message,
                sim_ctx=ctx,
            )
        )

        return {
            "status": "started",
            "run_id": run_id,
            "message": f"Target run {run_id} started, monitoring in background",
        }

    async def _monitor_target_run(
        self,
        run_id: str,
        message: str,
        sim_ctx: AgentContext,
    ) -> None:
        """Background task: monitor Target execution, accumulate text, emit events.

        This runs independently of Sim Agent's main execution flow.
        """
        buffer = SimulationBuffer(run_id)
        self._active_runs[run_id] = buffer

        try:
            async for event in self.target_agent.run_stream(message):
                match event:
                    case PartDeltaEvent(content=text):
                        # Accumulate text
                        buffer.add_text(text)

                    case ToolCallStartEvent(tool_name=name, args=args):
                        # Add tool call to buffer
                        buffer.add_tool_call(name, args)

                    case ToolCallCompleteEvent(tool_name=name, result=result):
                        # Add tool result to buffer
                        buffer.add_tool_result(name, result)

                    case ElicitationEvent(question=q, request_id=rid):
                        # KEY POINT: Emit event with accumulated context
                        await sim_ctx.events.custom({
                            "run_id": run_id,
                            "type": "needs_input",
                            "progress_so_far": buffer.get_context(),
                            "question": q,
                            "request_id": rid,
                        })
                        # Target blocks here in InputProvider until answer_elicitation called

                    case StreamCompleteEvent():
                        # Run completed - emit final event
                        await sim_ctx.events.custom({
                            "run_id": run_id,
                            "type": "completed",
                            "full_conversation": buffer.get_context(),
                        })

        except Exception as e:
            # Error during monitoring
            await sim_ctx.events.custom({
                "run_id": run_id,
                "type": "error",
                "error": str(e),
                "partial_progress": buffer.get_context(),
            })

        finally:
            # Cleanup
            self._active_runs.pop(run_id, None)

    @tool
    async def answer_elicitation(
        self,
        ctx: AgentContext,
        request_id: str,
        answer: str,
        run_id: str | None = None,
    ) -> dict:
        """Answer a pending elicitation request from Target Agent.

        This unblocks the Target Agent's input request via InputProvider.

        Returns:
            {"status": "submitted" | "not_found"}
        """
        success = self.input_provider.submit_answer(request_id, answer)

        return {
            "status": "submitted" if success else "not_found",
            "request_id": request_id,
        }

class SimulationInputProvider(InputProvider):
    """Intercepts Target Agent elicitation and blocks until answer provided.

    Used by Target Agent (not Sim Agent). Coordinates with 
    answer_elicitation tool via Future-based synchronization.
    """

    def __init__(self, timeout: float = 60.0):
        self.timeout = timeout
        self._pending: dict[str, asyncio.Future[str]] = {}
        self._lock = asyncio.Lock()

    async def prompt(
        self,
        message: str,
        request_id: str | None = None,
    ) -> str:
        """Called by Target Agent when it needs user input.

        Creates Future, blocks until Sim Agent submits answer via tool.
        """
        request_id = request_id or str(uuid.uuid4())
        future = asyncio.Future()

        async with self._lock:
            self._pending[request_id] = future

        try:
            # Wait for answer (submitted by answer_elicitation tool)
            return await asyncio.wait_for(future, timeout=self.timeout)
        finally:
            async with self._lock:
                self._pending.pop(request_id, None)

    def submit_answer(self, request_id: str, answer: str) -> bool:
        """Called by answer_elicitation tool to unblock Target."""
        future = self._pending.get(request_id)
        if future and not future.done():
            future.set_result(answer)
            return True
        return False

async def on_simulation_event(ctx: AgentContext, event: CustomEvent) -> None:
    """Handle simulation events - Sim Agent sees formatted text progress.

    This is the only event handler Sim Agent needs. It receives:
    - needs_input: when Target asks a question (with full context)
    - completed: when Target finishes (with full conversation)
    """
    data = event.event_data
    event_type = data.get("type")

    if event_type == "needs_input":
        # Sim Agent sees progress as formatted text
        progress = data["progress_so_far"]
        question = data["question"]
        request_id = data["request_id"]
        run_id = data["run_id"]

        # Use LLM to decide answer (based on Sim Agent's system prompt/goal)
        # Or direct the Sim Agent to use answer_elicitation tool
        prompt = f"""Current Target Agent progress:
{progress}

Target is asking: {question}

Please provide a realistic answer as a test user would."""

        # In actual implementation, this would trigger the Sim Agent
        # to call answer_elicitation tool with the answer
        # The calling code would handle this via agent.run() or tool call

    elif event_type == "completed":
        # Target finished - full conversation available
        conversation = data["full_conversation"]
        run_id = data["run_id"]

        # Sim Agent can now start a new conversation or evaluate results

    elif event_type == "error":
        # Handle error
        error_msg = data.get("error", "Unknown error")
        partial = data.get("partial_progress", "")

# Event handler registration (code-only, not YAML)
async def on_elicitation_request(event: ElicitationRequestEvent) -> None:
    """Handle elicitation request from Target Agent.

    This is where Sim Agent logic decides how to answer.
    In a real implementation, this might:
    - Queue the question for manual review
    - Use LLM to generate answer based on scenario
    - Look up answer in predefined test data
    """
    if event.event_type != ElicitationRequestEvent.event_type:
        return

    # Get Sim Agent instance (from pool or context)
    sim_agent = get_sim_agent_for_target(event.target_agent_id)

    # Generate or retrieve answer
    answer = await generate_answer(event.question)

    # Submit answer via tool
    await sim_agent.run(
        f"Use answer_elicitation with request_id='{event.request_id}' "
        f"and answer='{answer}'"
    )


# Register handler at AgentPool initialization
async def setup_simulation_framework(pool: AgentPool) -> None:
    """Configure simulation framework with event handlers."""
    pool.register_event_handler(
        ElicitationRequestEvent.event_type,
        on_elicitation_request,
    )

# File: src/agentpool/simulation/input_provider.py
class SimulationInputProvider(InputProvider):
    """Step 1 implementation - core blocking mechanism."""

    def __init__(self, timeout: float = 60.0):
        self.timeout = timeout
        self._pending: dict[str, asyncio.Future] = {}
        self._lock = asyncio.Lock()

    async def prompt(self, message: str, request_id: str | None = None) -> str:
        request_id = request_id or str(uuid.uuid4())
        future = asyncio.Future()

        async with self._lock:
            self._pending[request_id] = future

        try:
            # TODO: Emit event (Step 2)
            return await asyncio.wait_for(future, timeout=self.timeout)
        finally:
            async with self._lock:
                self._pending.pop(request_id, None)

    def submit_answer(self, request_id: str, answer: str) -> bool:
        future = self._pending.get(request_id)
        if future and not future.done():
            future.set_result(answer)
            return True
        return False

# File: src/agentpool/simulation/events.py
from dataclasses import dataclass
from agentpool.models import CustomEvent

@dataclass(frozen=True)
class ElicitationRequestEvent(CustomEvent):
    """Event to notify Sim Agent of pending elicitation."""
    event_type: ClassVar[str] = "simulation.elicitation_request"
    request_id: str
    target_agent_id: str
    question: str
    timestamp: float

# File: src/agentpool/simulation/tool_provider.py
class SimulationToolProvider(ResourceProvider):
    """Basic tools for Sim Agent."""

    def __init__(self, target_agent: Agent, input_provider: SimulationInputProvider):
        self.target_agent = target_agent
        self.input_provider = input_provider

    @tool
    async def talk_to_target(self, ctx: AgentContext, message: str) -> dict:
        """Send message to Target Agent."""
        result = await self.target_agent.run(message)
        return {"status": "completed", "response": result.data}

    @tool  
    async def answer_elicitation(self, ctx: AgentContext, request_id: str, answer: str) -> dict:
        """Answer Target Agent's pending question."""
        success = self.input_provider.submit_answer(request_id, answer)
        return {"status": "submitted" if success else "not_found"}

# Example: Setting up a complete simulation
async def demo_simulation():
    # 1. Create Target Agent with SimulationInputProvider
    input_provider = SimulationInputProvider(timeout=60.0)

    target_agent = Agent(
        name="diagnosis_target",
        model="openai:gpt-4o",
        input_provider=input_provider,
        tools=[question_tool],  # Elicitation tool
    )

    # 2. Create Sim Agent with SimulationToolProvider
    sim_provider = SimulationToolProvider(target_agent, input_provider)

    sim_agent = Agent(
        name="sim_agent",
        model="claude-sonnet-4",
        tools=sim_provider.get_tools(),
    )

    # 3. Register event handler (code-only)
    event_bus.register(ElicitationRequestEvent.event_type, my_handler)

    # 4. Run simulation
    async with AgentPool() as pool:
        pool.register_agent(target_agent)
        pool.register_agent(sim_agent)

        result = await sim_agent.run("Machine is making noise")

class SimpleSimulationToolProvider(ResourceProvider):
    """Simpler version that detects elicitation via tool call monitoring."""

    async def talk_to_target(self, ctx: AgentContext, message: str) -> dict:
        """Send message and detect if Target asks questions.

        Returns:
            {
                "status": "completed" | "elicitation",
                "response": str | None,
                "questions": list | None,
            }
        """
        questions = []
        response_parts = []

        async for event in self.target.run_stream(message):
            if isinstance(event, PartDeltaEvent):
                response_parts.append(event.delta)

            # Detect elicitation via tool call
            if isinstance(event, ToolCallStartEvent):
                if event.tool_name in self.elicitation_tools:
                    questions.append({
                        "id": event.tool_call_id,
                        "text": event.args.get("prompt", ""),
                    })

        if questions:
            return {
                "status": "elicitation",
                "questions": questions,
                "partial_response": "".join(response_parts),
            }

        return {
            "status": "completed",
            "response": "".join(response_parts),
        }

class ElicitationInterrupt(Exception):
    def __init__(self, questions: list, partial_response: str):
        self.questions = questions
        self.partial_response = partial_response

# In Provider
try:
    async for event in target.run_stream(message):
        if is_elicitation(event):
            raise ElicitationInterrupt(...)
except ElicitationInterrupt as e:
    return RunResult(status="elicitation", questions=e.questions)

# Both agents run concurrently, exchanging messages via queue
async def cooperative_simulation(sim_agent, target_agent):
    queue = asyncio.Queue()

    async def run_target():
        async for event in target_agent.run_stream():
            if is_elicitation(event):
                await queue.put(("elicitation", event))
            else:
                await queue.put(("output", event))

    async def run_sim():
        while True:
            msg_type, data = await queue.get()
            if msg_type == "elicitation":
                answer = await sim_agent.generate_answer(data)
                target_agent.provide_input(answer)

    await asyncio.gather(run_target(), run_sim())