Method _execute

tests/test_tool_pipeline_round3.py:70–88

(self, ctx: ToolContext, tool_use: _ToolUse, can_use_tool=None)

        return ctx

    def _execute(self, ctx: ToolContext, tool_use: _ToolUse, can_use_tool=None) -> list:
        # The services lane fails CLOSED without a permission handler, so
        # tests that want execution must say so explicitly.
        if can_use_tool is None:
            def can_use_tool(*_a, **_k):
                return {"behavior": "allow"}

        async def drive():
            updates = []
            async for update in run_tool_use(
                tool_use,
                AssistantMessage(content="using a tool"),
                can_use_tool,
                ctx,
            ):
                updates.append(update)
            return updates

        return _run(drive())

    @staticmethod
    def _result_blocks(updates: list) -> list[dict]: