Function _make_context

tests/test_streaming_executor_interruptible.py:50–70

(
    tools: list[Tool],
    captured: list[bool] | None = None,
    *,
    raise_setter: bool = False,
)

def _make_context(
    tools: list[Tool],
    captured: list[bool] | None = None,
    *,
    raise_setter: bool = False,
) -> ToolContext:
    ctx = ToolContext(
        workspace_root=Path("/tmp"),
        options=ToolUseOptions(tools=tools),
        abort_controller=AbortController(),
    )
    if captured is not None:
        if raise_setter:
            def _setter(v: bool) -> None:
                captured.append(v)
                raise RuntimeError("setter explodes")

            ctx.set_has_interruptible_tool_in_progress = _setter
        else:
            ctx.set_has_interruptible_tool_in_progress = lambda v: captured.append(v)
    return ctx


def _allow_all(_tool, tool_input, _ctx, _msg, _tool_use_id):