Backends

The backend protocols and the concrete LLM and agent backends. See the Backends overview.

Protocols

agentflow.backends.base

Backend protocols and permission policies.

A backend drives one underlying engine — an agent process (Kiro, Codex, Claude Code) or an LLM endpoint (Ollama, OpenAI-compatible). The graph engine depends only on :class:Backend; user code that needs agent- or LLM-specific features constructs the concrete subtype and gets the narrowed method.

This module imports only :mod:agentflow.events and :mod:agentflow.errors — never the graph engine. Backends are a leaf layer.

Backend

Bases: Protocol

Common surface for every backend.

invoke is an async generator: it yields :class:BackendEvents as they arrive and completes when the turn ends. A caller wanting only the final result drains it and keeps the terminal :class:~agentflow.events.TurnEnd; a streaming caller consumes events as they come.

Source code in agentflow/backends/base.py

@runtime_checkable
class Backend(Protocol):
    """Common surface for every backend.

    ``invoke`` is an async generator: it yields :class:`BackendEvent`s as they
    arrive and completes when the turn ends. A caller wanting only the final
    result drains it and keeps the terminal :class:`~agentflow.events.TurnEnd`;
    a streaming caller consumes events as they come.
    """

    async def start(self) -> None:
        """Bring the backend up (spawn a process, open a client)."""
        ...

    async def close(self) -> None:
        """Tear the backend down. Idempotent."""
        ...

    def invoke(
        self, request: BackendRequest, *, session: str | None = None
    ) -> AsyncIterator[BackendEvent]:
        """Run one turn for ``request``, yielding events until it ends."""
        ...

start async

start() -> None

Bring the backend up (spawn a process, open a client).

Source code in agentflow/backends/base.py

async def start(self) -> None:
    """Bring the backend up (spawn a process, open a client)."""
    ...

close async

close() -> None

Tear the backend down. Idempotent.

Source code in agentflow/backends/base.py

async def close(self) -> None:
    """Tear the backend down. Idempotent."""
    ...

invoke

invoke(request: BackendRequest, *, session: str | None = None) -> AsyncIterator[BackendEvent]

Run one turn for request, yielding events until it ends.

Source code in agentflow/backends/base.py

def invoke(
    self, request: BackendRequest, *, session: str | None = None
) -> AsyncIterator[BackendEvent]:
    """Run one turn for ``request``, yielding events until it ends."""
    ...

AgentBackend

Bases: Backend, Protocol

A session-based agent that runs its own tools and asks permission.

Accepts a :class:~agentflow.events.TextRequest. Exposes a :class:PermissionPolicy used to answer the agent's permission prompts.

Source code in agentflow/backends/base.py

@runtime_checkable
class AgentBackend(Backend, Protocol):
    """A session-based agent that runs its own tools and asks permission.

    Accepts a :class:`~agentflow.events.TextRequest`. Exposes a
    :class:`PermissionPolicy` used to answer the agent's permission prompts.
    """

    permission: PermissionPolicy

    def prompt(self, text: str, *, session: str | None = None) -> AsyncIterator[BackendEvent]:
        """Convenience wrapper over ``invoke(TextRequest(text))``."""
        ...

prompt

prompt(text: str, *, session: str | None = None) -> AsyncIterator[BackendEvent]

Convenience wrapper over invoke(TextRequest(text)).

Source code in agentflow/backends/base.py

def prompt(self, text: str, *, session: str | None = None) -> AsyncIterator[BackendEvent]:
    """Convenience wrapper over ``invoke(TextRequest(text))``."""
    ...

LLMBackend

Bases: Backend, Protocol

A stateless chat model: messages in, token stream out.

Accepts a :class:~agentflow.events.ChatRequest. It never executes tools itself — a :class:~agentflow.events.ToolCall it emits is a request the graph fulfils and feeds back on the next call.

Source code in agentflow/backends/base.py

@runtime_checkable
class LLMBackend(Backend, Protocol):
    """A stateless chat model: messages in, token stream out.

    Accepts a :class:`~agentflow.events.ChatRequest`. It never executes tools
    itself — a :class:`~agentflow.events.ToolCall` it emits is a request the
    graph fulfils and feeds back on the next call.
    """

    def chat(
        self,
        messages: list[Message],
        *,
        tools: list[ToolSpec] | None = None,
        options: dict[str, Any] | None = None,
    ) -> AsyncIterator[BackendEvent]:
        """Convenience wrapper over ``invoke(ChatRequest(...))``."""
        ...

chat

chat(messages: list[Message], *, tools: list[ToolSpec] | None = None, options: dict[str, Any] | None = None) -> AsyncIterator[BackendEvent]

Convenience wrapper over invoke(ChatRequest(...)).

Source code in agentflow/backends/base.py

def chat(
    self,
    messages: list[Message],
    *,
    tools: list[ToolSpec] | None = None,
    options: dict[str, Any] | None = None,
) -> AsyncIterator[BackendEvent]:
    """Convenience wrapper over ``invoke(ChatRequest(...))``."""
    ...

BaseAgentBackend

Bases: _AsyncResource

Mixin implementing :meth:prompt on top of :meth:invoke.

Concrete agent backends inherit this and implement start, close, and invoke.

A permission policy is REQUIRED — there is no default. Auto-approving an agent's tool requests is a security decision the caller must make explicitly. Pass :class:AllowAll for a trusted local sandbox, :class:DenyAll to block tools, or :class:Interactive for human-in-the-loop approval.

NOTE on coverage: only agent backends that actually receive permission requests route them through this policy. Kiro (persistent ACP session) does. The one-shot CLI agents (Codex, Claude Code) manage their own permission model via CLI flags/sandbox and do NOT route prompts through this policy — the field still gates whether the SDK would auto-approve if a request surfaced, but the CLI's own controls apply first. See the capability matrix in the README.

Source code in agentflow/backends/base.py

class BaseAgentBackend(_AsyncResource):
    """Mixin implementing :meth:`prompt` on top of :meth:`invoke`.

    Concrete agent backends inherit this and implement ``start``, ``close``,
    and ``invoke``.

    A ``permission`` policy is REQUIRED — there is no default. Auto-approving
    an agent's tool requests is a security decision the caller must make
    explicitly. Pass :class:`AllowAll` for a trusted local sandbox,
    :class:`DenyAll` to block tools, or :class:`Interactive` for
    human-in-the-loop approval.

    NOTE on coverage: only agent backends that actually receive permission
    requests route them through this policy. Kiro (persistent ACP session)
    does. The one-shot CLI agents (Codex, Claude Code) manage their own
    permission model via CLI flags/sandbox and do NOT route prompts through
    this policy — the field still gates whether the SDK would auto-approve if
    a request surfaced, but the CLI's own controls apply first. See the
    capability matrix in the README.
    """

    permission: PermissionPolicy

    def __init__(self, permission: PermissionPolicy) -> None:
        if permission is None:
            raise TypeError(
                "an agent backend requires an explicit permission policy "
                "(no default). Pass AllowAll() for a trusted local sandbox, "
                "DenyAll(), or Interactive() for human-in-the-loop."
            )
        self.permission = permission

    def prompt(self, text: str, *, session: str | None = None) -> AsyncIterator[BackendEvent]:
        return self.invoke(TextRequest(text), session=session)  # type: ignore[attr-defined]

BaseLLMBackend

Bases: _AsyncResource

Mixin implementing :meth:chat on top of :meth:invoke.

Concrete LLM backends inherit this and implement start, close, and invoke.

Set max_concurrency (an int) to cap in-flight requests from this backend instance; :meth:concurrency_guard returns an async context to wrap the request with. This bounds pressure when many graph nodes hit the same provider at once.

Source code in agentflow/backends/base.py

class BaseLLMBackend(_AsyncResource):
    """Mixin implementing :meth:`chat` on top of :meth:`invoke`.

    Concrete LLM backends inherit this and implement ``start``, ``close``,
    and ``invoke``.

    Set ``max_concurrency`` (an int) to cap in-flight requests from this
    backend instance; :meth:`concurrency_guard` returns an async context to
    wrap the request with. This bounds pressure when many graph nodes hit the
    same provider at once.
    """

    max_concurrency: int | None = None

    def concurrency_guard(self):
        """Return an async context manager limiting concurrent requests.

        A no-op when ``max_concurrency`` is unset; otherwise a lazily-created
        :class:`asyncio.Semaphore` shared across this backend instance.
        """
        import asyncio
        import contextlib

        if not self.max_concurrency:
            return contextlib.nullcontext()
        sem = getattr(self, "_sem", None)
        if sem is None:
            sem = asyncio.Semaphore(self.max_concurrency)
            self._sem = sem
        return sem

    def chat(
        self,
        messages: list[Message],
        *,
        tools: list[ToolSpec] | None = None,
        options: dict[str, Any] | None = None,
    ) -> AsyncIterator[BackendEvent]:
        request = ChatRequest(
            messages=tuple(messages),
            tools=tuple(tools or ()),
            options=dict(options or {}),
        )
        return self.invoke(request)  # type: ignore[attr-defined]

concurrency_guard

concurrency_guard()

Return an async context manager limiting concurrent requests.

A no-op when max_concurrency is unset; otherwise a lazily-created :class:asyncio.Semaphore shared across this backend instance.

Source code in agentflow/backends/base.py

def concurrency_guard(self):
    """Return an async context manager limiting concurrent requests.

    A no-op when ``max_concurrency`` is unset; otherwise a lazily-created
    :class:`asyncio.Semaphore` shared across this backend instance.
    """
    import asyncio
    import contextlib

    if not self.max_concurrency:
        return contextlib.nullcontext()
    sem = getattr(self, "_sem", None)
    if sem is None:
        sem = asyncio.Semaphore(self.max_concurrency)
        self._sem = sem
    return sem

LLM backends

agentflow.backends.ollama.OllamaBackend

Bases: BaseLLMBackend

Talk to a local (or remote) Ollama server.

Source code in agentflow/backends/ollama.py

class OllamaBackend(BaseLLMBackend):
    """Talk to a local (or remote) Ollama server."""

    def __init__(
        self,
        model: str,
        *,
        host: str = "http://localhost:11434",
        options: dict[str, Any] | None = None,
        timeout: float = 120.0,
        retry: RetryPolicy | None = None,
        max_concurrency: int | None = None,
    ) -> None:
        self.model = model
        self.host = host.rstrip("/")
        self.options = dict(options or {})
        self.timeout = timeout
        self.retry = retry if retry is not None else RetryPolicy()
        self.max_concurrency = max_concurrency
        self._client: httpx.AsyncClient | None = None

    async def start(self) -> None:
        if self._client is None:
            self._client = httpx.AsyncClient(base_url=self.host, timeout=self.timeout)

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def invoke(
        self, request: BackendRequest, *, session: str | None = None
    ) -> AsyncIterator[BackendEvent]:
        if not isinstance(request, ChatRequest):
            raise BackendError(f"OllamaBackend accepts ChatRequest, got {type(request).__name__}")
        if self._client is None:
            raise BackendError("OllamaBackend.start() was not called")

        body = self._build_body(request)
        text_parts: list[str] = []
        tool_calls: list[ToolCallSpec] = []
        stop_reason: str | None = None
        usage: Usage | None = None

        async with (
            self.concurrency_guard(),
            open_stream(
                self._client,
                "POST",
                "/api/chat",
                json=body,
                policy=self.retry,
                label="ollama /api/chat",
            ) as resp,
        ):
            try:
                async for line in resp.aiter_lines():
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        chunk = json.loads(line)
                    except json.JSONDecodeError:
                        yield ErrorEvent(message=f"malformed stream line: {line!r}")
                        continue

                    for event in self._parse_chunk(chunk, text_parts, tool_calls):
                        yield event

                    if chunk.get("done"):
                        stop_reason = chunk.get("done_reason") or "stop"
                        # Ollama reports token counts on the final chunk:
                        # prompt_eval_count = input, eval_count = output.
                        pin = chunk.get("prompt_eval_count")
                        pout = chunk.get("eval_count")
                        if pin is not None or pout is not None:
                            total = (pin or 0) + (pout or 0) if (pin or pout) else None
                            usage = Usage(input_tokens=pin, output_tokens=pout, total_tokens=total)
            except httpx.HTTPError as exc:
                raise BackendTransportError(f"ollama stream failed: {exc}") from exc

        final_text = "".join(text_parts)
        message = Message(
            role="assistant",
            content=final_text,
            tool_calls=tuple(tool_calls),
        )
        yield TurnEnd(text=final_text, stop_reason=stop_reason, message=message, usage=usage)

    # ------------------------------------------------------------------

    def _build_body(self, request: ChatRequest) -> dict[str, Any]:
        body: dict[str, Any] = {
            "model": self.model,
            "messages": [_message_to_ollama(m) for m in request.messages],
            "stream": True,
        }
        merged = {**self.options, **request.options}
        if merged:
            body["options"] = merged
        if request.tools:
            body["tools"] = [_tool_to_ollama(t) for t in request.tools]
        return body

    def _parse_chunk(
        self,
        chunk: dict[str, Any],
        text_parts: list[str],
        tool_calls: list[ToolCallSpec],
    ) -> list[Any]:
        events: list[Any] = []
        message = chunk.get("message") or {}

        content = message.get("content")
        if content:
            text_parts.append(content)
            events.append(TextChunk(text=content))

        for i, call in enumerate(message.get("tool_calls") or []):
            fn = call.get("function") or {}
            args = fn.get("arguments")
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {"_raw": args}
            call_id = call.get("id") or f"call_{len(tool_calls) + i}"
            name = fn.get("name", "")
            spec = ToolCallSpec(id=call_id, name=name, args=args or {})
            tool_calls.append(spec)
            events.append(ToolCall(id=call_id, name=name, args=args or {}))

        return events

agentflow.backends.openai.OpenAIBackend

Bases: BaseLLMBackend

Talk to any OpenAI-compatible chat-completions endpoint.

Source code in agentflow/backends/openai.py

class OpenAIBackend(BaseLLMBackend):
    """Talk to any OpenAI-compatible chat-completions endpoint."""

    def __init__(
        self,
        model: str,
        *,
        base_url: str = "https://api.openai.com/v1",
        api_key: str | None = None,
        options: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        timeout: float = 120.0,
        retry: RetryPolicy | None = None,
        max_concurrency: int | None = None,
    ) -> None:
        self.model = model
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.options = dict(options or {})
        self.extra_headers = dict(headers or {})
        self.timeout = timeout
        self.retry = retry if retry is not None else RetryPolicy()
        self.max_concurrency = max_concurrency
        self._client: httpx.AsyncClient | None = None

    async def start(self) -> None:
        if self._client is None:
            headers = dict(self.extra_headers)
            if self.api_key:
                headers.setdefault("Authorization", f"Bearer {self.api_key}")
            self._client = httpx.AsyncClient(
                base_url=self.base_url, timeout=self.timeout, headers=headers
            )

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def invoke(
        self, request: BackendRequest, *, session: str | None = None
    ) -> AsyncIterator[BackendEvent]:
        if not isinstance(request, ChatRequest):
            raise BackendError(f"OpenAIBackend accepts ChatRequest, got {type(request).__name__}")
        if self._client is None:
            raise BackendError("OpenAIBackend.start() was not called")

        body = self._build_body(request)
        text_parts: list[str] = []
        # Assemble streamed tool-call fragments keyed by their choice index.
        tool_frags: dict[int, dict[str, Any]] = {}
        finish_reason: str | None = None

        async with (
            self.concurrency_guard(),
            open_stream(
                self._client,
                "POST",
                "/chat/completions",
                json=body,
                policy=self.retry,
                label="chat/completions",
            ) as resp,
        ):
            try:
                async for line in resp.aiter_lines():
                    line = line.strip()
                    if not line or not line.startswith("data:"):
                        continue
                    data = line[len("data:") :].strip()
                    if data == "[DONE]":
                        break
                    try:
                        chunk = json.loads(data)
                    except json.JSONDecodeError:
                        yield ErrorEvent(message=f"malformed SSE data: {data!r}")
                        continue
                    for event in self._parse_chunk(chunk, text_parts, tool_frags):
                        yield event
                    fr = _finish_reason(chunk)
                    if fr:
                        finish_reason = fr
            except httpx.HTTPError as exc:
                raise BackendTransportError(f"chat/completions stream failed: {exc}") from exc

        tool_calls = _assemble_tool_calls(tool_frags)
        final_text = "".join(text_parts)
        message = Message(role="assistant", content=final_text, tool_calls=tuple(tool_calls))
        yield TurnEnd(text=final_text, stop_reason=finish_reason or "stop", message=message)

    # ------------------------------------------------------------------

    def _build_body(self, request: ChatRequest) -> dict[str, Any]:
        body: dict[str, Any] = {
            "model": self.model,
            "messages": [_message_to_openai(m) for m in request.messages],
            "stream": True,
        }
        body.update(self.options)
        body.update(request.options)
        if request.tools:
            body["tools"] = [_tool_to_openai(t) for t in request.tools]
        return body

    def _parse_chunk(
        self,
        chunk: dict[str, Any],
        text_parts: list[str],
        tool_frags: dict[int, dict[str, Any]],
    ) -> list[BackendEvent]:
        events: list[BackendEvent] = []
        for choice in chunk.get("choices") or []:
            delta = choice.get("delta") or {}

            content = delta.get("content")
            if content:
                text_parts.append(content)
                events.append(TextChunk(text=content))

            for tc in delta.get("tool_calls") or []:
                index = tc.get("index", 0)
                frag = tool_frags.setdefault(index, {"id": None, "name": None, "args": ""})
                if tc.get("id"):
                    frag["id"] = tc["id"]
                fn = tc.get("function") or {}
                if fn.get("name"):
                    frag["name"] = fn["name"]
                if fn.get("arguments"):
                    frag["args"] += fn["arguments"]

        return events

agentflow.backends.anthropic.AnthropicBackend

Bases: BaseLLMBackend

Talk to the Anthropic Messages API.

Source code in agentflow/backends/anthropic.py

class AnthropicBackend(BaseLLMBackend):
    """Talk to the Anthropic Messages API."""

    def __init__(
        self,
        model: str,
        *,
        api_key: str | None = None,
        base_url: str = "https://api.anthropic.com",
        max_tokens: int = 1024,
        anthropic_version: str = _DEFAULT_VERSION,
        workspace_id: str | None = None,
        options: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        timeout: float = 120.0,
        retry: RetryPolicy | None = None,
        max_concurrency: int | None = None,
    ) -> None:
        self.model = model
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.max_tokens = max_tokens
        self.anthropic_version = anthropic_version
        # An org-level (unscoped) key needs the workspace id header; a
        # workspace-scoped key does not.
        self.workspace_id = workspace_id
        self.options = dict(options or {})
        self.extra_headers = dict(headers or {})
        self.timeout = timeout
        self.retry = retry if retry is not None else RetryPolicy()
        self.max_concurrency = max_concurrency
        self._client: httpx.AsyncClient | None = None

    async def start(self) -> None:
        if self._client is None:
            headers = {
                "anthropic-version": self.anthropic_version,
                "content-type": "application/json",
                **self.extra_headers,
            }
            if self.api_key:
                headers.setdefault("x-api-key", self.api_key)
            if self.workspace_id:
                headers.setdefault("anthropic-workspace-id", self.workspace_id)
            self._client = httpx.AsyncClient(
                base_url=self.base_url, timeout=self.timeout, headers=headers
            )

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def invoke(
        self, request: BackendRequest, *, session: str | None = None
    ) -> AsyncIterator[BackendEvent]:
        if not isinstance(request, ChatRequest):
            raise BackendError(
                f"AnthropicBackend accepts ChatRequest, got {type(request).__name__}"
            )
        if self._client is None:
            raise BackendError("AnthropicBackend.start() was not called")

        body = self._build_body(request)
        text_parts: list[str] = []
        # Content blocks assembled by index. Text blocks accumulate text;
        # tool_use blocks accumulate a partial_json string to parse at stop.
        blocks: dict[int, dict[str, Any]] = {}
        tool_calls: list[ToolCallSpec] = []
        stop_reason: str | None = None

        async with (
            self.concurrency_guard(),
            open_stream(
                self._client,
                "POST",
                "/v1/messages",
                json=body,
                policy=self.retry,
                label="/v1/messages",
            ) as resp,
        ):
            try:
                async for line in resp.aiter_lines():
                    line = line.strip()
                    if not line or not line.startswith("data:"):
                        continue
                    data = line[len("data:") :].strip()
                    try:
                        event = json.loads(data)
                    except json.JSONDecodeError:
                        yield ErrorEvent(message=f"malformed SSE data: {data!r}")
                        continue
                    for out in self._parse_event(event, text_parts, blocks, tool_calls):
                        yield out
                    sr = _stop_reason(event)
                    if sr:
                        stop_reason = sr
            except httpx.HTTPError as exc:
                raise BackendTransportError(f"/v1/messages stream failed: {exc}") from exc

        final_text = "".join(text_parts)
        message = Message(role="assistant", content=final_text, tool_calls=tuple(tool_calls))
        yield TurnEnd(text=final_text, stop_reason=stop_reason or "end_turn", message=message)

    # ------------------------------------------------------------------

    def _build_body(self, request: ChatRequest) -> dict[str, Any]:
        system, messages = _split_system(request.messages)
        body: dict[str, Any] = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": messages,
            "stream": True,
        }
        if system:
            body["system"] = system
        body.update(self.options)
        body.update(request.options)
        if request.tools:
            body["tools"] = [_tool_to_anthropic(t) for t in request.tools]
        return body

    def _parse_event(
        self,
        event: dict[str, Any],
        text_parts: list[str],
        blocks: dict[int, dict[str, Any]],
        tool_calls: list[ToolCallSpec],
    ) -> list[BackendEvent]:
        etype = event.get("type")

        if etype == "content_block_start":
            index = event.get("index", 0)
            block = event.get("content_block") or {}
            blocks[index] = {
                "type": block.get("type"),
                "id": block.get("id"),
                "name": block.get("name"),
                "json": "",
            }
            return []

        if etype == "content_block_delta":
            delta = event.get("delta") or {}
            dtype = delta.get("type")
            if dtype == "text_delta":
                text = delta.get("text") or ""
                if text:
                    text_parts.append(text)
                    return [TextChunk(text=text)]
            elif dtype == "input_json_delta":
                index = event.get("index", 0)
                blk = blocks.get(index)
                if blk is not None:
                    blk["json"] += delta.get("partial_json") or ""
            return []

        if etype == "content_block_stop":
            index = event.get("index", 0)
            blk = blocks.get(index)
            if blk and blk.get("type") == "tool_use":
                raw = blk.get("json") or ""
                try:
                    args = json.loads(raw) if raw else {}
                except json.JSONDecodeError:
                    args = {"_raw": raw}
                call_id = blk.get("id") or f"call_{index}"
                name = blk.get("name") or ""
                spec_args = args if isinstance(args, dict) else {"_value": args}
                tool_calls.append(ToolCallSpec(id=call_id, name=name, args=spec_args))
                event_args = args if isinstance(args, dict) else {}
                return [ToolCall(id=call_id, name=name, args=event_args)]
            return []

        return []

Agent backends

agentflow.backends.kiro.KiroBackend

Bases: BaseAgentBackend

Async ACP client for a Kiro agent session.

Source code in agentflow/backends/kiro.py

class KiroBackend(BaseAgentBackend):
    """Async ACP client for a Kiro agent session."""

    def __init__(
        self,
        agent: str,
        *,
        model: str | None = None,
        engine: str = "v3",
        cwd: Path | str = _DEFAULT_CWD,
        permission: PermissionPolicy,
        close_grace: float = 0.5,
    ) -> None:
        super().__init__(permission=permission)
        self.agent = agent
        self.model = model
        self.engine = engine
        self.cwd = Path(cwd)
        # Per-phase grace on close(): wait this long for a clean exit, then
        # terminate, then (after another grace) kill. Kept short because the
        # v3 engine typically will not exit on stdin close, and we should not
        # block a caller's shutdown for seconds waiting on it.
        self._close_grace = close_grace

        self._proc: asyncio.subprocess.Process | None = None
        self._session_id: str | None = None
        self._request_id = 0

        # Correlates outstanding JSON-RPC requests to their response futures.
        self._pending: dict[int, asyncio.Future[dict[str, Any]]] = {}
        # The active turn's event queue, or None when no turn is in flight.
        self._turn_queue: asyncio.Queue[BackendEvent | _Sentinel] | None = None
        self._turn_text = ""
        self._reader_task: asyncio.Task[None] | None = None
        self._closing = False
        # Permission decisions awaited by the reader, resolved by the generator
        # frame (where the policy — and any engine interrupt — must run).
        self._perm_answers: dict[str, asyncio.Future[PermissionDecision]] = {}

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def start(self) -> None:
        cmd = ["kiro-cli", "acp"]
        if self.engine:
            cmd += ["--agent-engine", self.engine]
        if self.engine == "v3":
            # v3 selects the agent as a session mode and resolves auth from the
            # CLI credential store; it takes neither --agent nor --model.
            cmd += ["--auth-method", "cli"]
        else:
            cmd += ["--agent", self.agent]

        try:
            self._proc = await asyncio.create_subprocess_exec(
                *cmd,
                cwd=str(self.cwd),
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=None,  # inherit parent's stderr
                # ACP responses (e.g. session/new listing every available mode)
                # can exceed asyncio's 64KiB default readline buffer; raise it.
                limit=8 * 1024 * 1024,
                # Put kiro-cli in its own process group so close() can signal
                # the whole tree. The v3 CLI spawns a child (kiro-cli-chat)
                # that outlives a kill of the parent alone, stranding the
                # terminal; killing the group reaps it too. POSIX-only —
                # ignored on platforms without it.
                start_new_session=True,
            )
        except OSError as exc:
            raise BackendTransportError(f"could not start kiro-cli: {exc}") from exc

        self._reader_task = asyncio.create_task(self._read_loop())

        await self._initialize()
        await self._create_session()
        if self.model and self.engine != "v3":
            await self._set_model(self.model)

    async def close(self) -> None:
        self._closing = True
        proc = self._proc
        if proc is None:
            return
        try:
            if proc.stdin and not proc.stdin.is_closing():
                proc.stdin.close()
            # The v3 engine often does not exit on stdin close (it waits on its
            # own idle timeout), so don't block for seconds hoping it will.
            # Give it a short grace to exit cleanly, then signal the whole
            # process group (SIGTERM, then SIGKILL) so the CLI's child process
            # (kiro-cli-chat) is reaped too, not just the parent. Each phase
            # has a brief window so close() returns promptly.
            try:
                await asyncio.wait_for(proc.wait(), timeout=self._close_grace)
            except TimeoutError:
                self._signal_group(proc, signal.SIGTERM)
                try:
                    await asyncio.wait_for(proc.wait(), timeout=self._close_grace)
                except TimeoutError:
                    self._signal_group(proc, signal.SIGKILL)
        finally:
            if self._reader_task:
                self._reader_task.cancel()
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await self._reader_task
            self._proc = None
            self._fail_pending(BackendTransportError("backend closed"))

    @staticmethod
    def _signal_group(proc: asyncio.subprocess.Process, sig: int) -> None:
        """Send ``sig`` to the child's whole process group, if possible.

        The child was started with ``start_new_session=True`` so it leads its
        own group; signalling ``-pgid`` reaches its descendants too. Falls back
        to signalling just the process where process groups aren't available
        (e.g. Windows) or the group has already gone. Never raises.
        """
        with contextlib.suppress(ProcessLookupError, PermissionError, OSError, AttributeError):
            pgid = os.getpgid(proc.pid)
            os.killpg(pgid, sig)
            return
        # Fallback: signal the process directly.
        with contextlib.suppress(ProcessLookupError, OSError):
            if sig == signal.SIGKILL:
                proc.kill()
            else:
                proc.terminate()

    # ------------------------------------------------------------------
    # Public API — one turn as an event stream
    # ------------------------------------------------------------------

    async def invoke(
        self, request: BackendRequest, *, session: str | None = None
    ) -> AsyncIterator[BackendEvent]:
        if not isinstance(request, TextRequest):
            raise BackendError(f"KiroBackend accepts TextRequest, got {type(request).__name__}")
        if self._session_id is None:
            raise BackendError("KiroBackend.start() was not called")
        if self._turn_queue is not None:
            raise BackendError("a turn is already in progress on this backend")

        queue: asyncio.Queue[BackendEvent | _Sentinel] = asyncio.Queue()
        self._turn_text = ""
        self._turn_queue = queue

        # Fire the prompt request; its response resolves the turn.
        response_future = self._send_request(
            "session/prompt",
            {
                "sessionId": self._session_id,
                "prompt": [{"type": "text", "text": request.text}],
            },
        )

        async def _finish() -> None:
            # Await the RPC response, then enqueue the terminal TurnEnd.
            try:
                result = await response_future
                stop = result.get("stopReason") if isinstance(result, dict) else None
                await queue.put(TurnEnd(text=self._turn_text, stop_reason=stop))
            except Exception as exc:  # transport / RPC failure ends the turn
                await queue.put(ErrorEvent(message=str(exc)))
                await queue.put(TurnEnd(text=self._turn_text, stop_reason="error"))
            finally:
                await queue.put(_SENTINEL)

        finisher = asyncio.create_task(_finish())

        try:
            while True:
                item = await queue.get()
                if isinstance(item, _Sentinel):
                    break
                if isinstance(item, PermissionRequest):
                    # Resolve here, in the node's frame, so an Interactive
                    # policy's ctx.interrupt (and its InterruptError) propagate
                    # out through this generator to the runtime.
                    await self._resolve_permission(item)
                yield item
        finally:
            self._turn_queue = None
            # If we exit the loop early (e.g. an interrupt raised out of the
            # generator), unblock the reader so it doesn't hang on an answer.
            self._cancel_pending_permissions()
            await self._drain_finisher(finisher)

    async def _resolve_permission(self, req: PermissionRequest) -> None:
        future = self._perm_answers.get(req.id)
        if future is None or future.done():
            return
        decision = await self.permission.decide(req)
        if not future.done():
            future.set_result(decision)

    def _cancel_pending_permissions(self) -> None:
        for future in self._perm_answers.values():
            if not future.done():
                future.cancel()

    @staticmethod
    async def _drain_finisher(finisher: asyncio.Task[None]) -> None:
        if finisher.done():
            # Surface any exception the finisher captured.
            finisher.result()
            return
        finisher.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await finisher

    # ------------------------------------------------------------------
    # Session setup
    # ------------------------------------------------------------------

    async def _initialize(self) -> None:
        await self._request(
            "initialize",
            {
                "protocolVersion": 1,
                "clientCapabilities": {},
                "clientInfo": {"name": "agentflow", "version": "0.1.0"},
            },
        )

    async def _create_session(self) -> None:
        result = await self._request("session/new", {"cwd": str(self.cwd), "mcpServers": []})
        session_id = result.get("sessionId")
        if not session_id:
            raise BackendTransportError(f"session/new returned no sessionId: {result}")
        self._session_id = session_id
        if self.engine == "v3":
            await self._select_agent_mode(result)

    async def _select_agent_mode(self, session_result: dict[str, Any]) -> None:
        modes = session_result.get("modes", {})
        if modes.get("currentModeId") == self.agent:
            return
        available = {m.get("id") for m in modes.get("availableModes", [])}
        if self.agent not in available:
            raise BackendError(f"agent {self.agent!r} not in available modes: {sorted(available)}")
        await self._request(
            "session/set_mode", {"sessionId": self._session_id, "modeId": self.agent}
        )

    async def _set_model(self, model: str) -> None:
        await self._request("session/set_model", {"sessionId": self._session_id, "modelId": model})

    # ------------------------------------------------------------------
    # JSON-RPC transport (async)
    # ------------------------------------------------------------------

    def _next_request_id(self) -> int:
        self._request_id += 1
        return self._request_id

    def _send_request(self, method: str, params: dict[str, Any]) -> asyncio.Future[dict[str, Any]]:
        """Send a request and return a future for its response."""
        request_id = self._next_request_id()
        future: asyncio.Future[dict[str, Any]] = asyncio.get_running_loop().create_future()
        self._pending[request_id] = future
        self._write({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params})
        return future

    async def _request(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        """Send a request and await its response (used for setup RPCs)."""
        return await self._send_request(method, params)

    def _write(self, message: dict[str, Any]) -> None:
        proc = self._proc
        if proc is None or proc.stdin is None:
            raise BackendTransportError("kiro-cli process is not running")
        data = json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n"
        proc.stdin.write(data.encode("utf-8"))
        # Apply write backpressure: schedule a drain so a slow child stdin
        # cannot let our write buffer grow unbounded. Best-effort — a closing
        # pipe during shutdown must not raise here.
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            return
        loop.create_task(self._drain_stdin())

    async def _drain_stdin(self) -> None:
        proc = self._proc
        if proc is None or proc.stdin is None:
            return
        with contextlib.suppress(ConnectionResetError, BrokenPipeError, RuntimeError):
            await proc.stdin.drain()

    async def _read_loop(self) -> None:
        """Continuously read newline-delimited JSON and dispatch it."""
        proc = self._proc
        assert proc is not None and proc.stdout is not None
        try:
            while True:
                line = await proc.stdout.readline()
                if line == b"":
                    if not self._closing:
                        self._fail_pending(
                            BackendTransportError(
                                f"kiro-cli terminated unexpectedly (exit {proc.returncode})"
                            )
                        )
                    return
                text = line.strip()
                if not text:
                    continue
                try:
                    message = json.loads(text)
                except json.JSONDecodeError:
                    # A malformed line is fatal to the protocol.
                    self._fail_pending(
                        BackendTransportError(f"invalid JSON from kiro-cli: {text!r}")
                    )
                    return
                await self._dispatch(message)
        except asyncio.CancelledError:
            raise
        except Exception as exc:  # unexpected reader failure
            self._fail_pending(BackendTransportError(f"reader loop failed: {exc}"))

    async def _dispatch(self, message: dict[str, Any]) -> None:
        # Agent -> client request (has both method and id): e.g. permission.
        if "method" in message and "id" in message:
            await self._handle_agent_request(message)
            return
        # Notification / streaming (method, no id).
        if "method" in message and "id" not in message:
            self._handle_notification(message)
            return
        # Response to one of our requests.
        request_id = message.get("id")
        if not isinstance(request_id, int):
            return
        future = self._pending.pop(request_id, None)
        if future is None or future.done():
            return
        if "error" in message:
            future.set_exception(
                BackendTransportError(
                    f"RPC error: {json.dumps(message['error'], ensure_ascii=False)}"
                )
            )
            return
        result = message.get("result")
        if result is None:
            future.set_result({})
        elif isinstance(result, dict):
            future.set_result(result)
        else:
            future.set_result({"value": result})

    def _fail_pending(self, exc: Exception) -> None:
        for future in self._pending.values():
            if not future.done():
                future.set_exception(exc)
        self._pending.clear()

    # ------------------------------------------------------------------
    # Agent -> client requests (permission)
    # ------------------------------------------------------------------

    async def _handle_agent_request(self, message: dict[str, Any]) -> None:
        method = message.get("method")
        request_id = message.get("id")
        params = message.get("params", {})

        if method == "session/request_permission":
            await self._handle_permission(request_id, params)
            return

        self._write(
            {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {"code": -32601, "message": f"Unsupported client method: {method}"},
            }
        )

    async def _handle_permission(self, request_id: Any, params: dict[str, Any]) -> None:
        raw_options = params.get("options", [])
        options = tuple(
            PermissionOption(
                option_id=o.get("optionId", ""),
                name=o.get("name", ""),
                kind=o.get("kind", ""),
            )
            for o in raw_options
        )
        req = PermissionRequest(
            id=str(request_id),
            tool=_extract_tool_name(params),
            options=options,
            detail=params,
        )

        # The policy must run in the node's frame (so an Interactive policy can
        # drive an engine interrupt), not here in the reader task. Publish the
        # request to the turn stream and wait for the generator to resolve it.
        loop = asyncio.get_running_loop()
        answer: asyncio.Future[PermissionDecision] = loop.create_future()
        self._perm_answers[req.id] = answer
        self._emit(req)

        try:
            decision = await answer
        except asyncio.CancelledError:
            # The generator tore down before answering (e.g. the turn was
            # suspended by an engine interrupt). Leave the RPC unanswered; the
            # agent process is abandoned with the turn. Do not kill the reader.
            self._perm_answers.pop(req.id, None)
            return
        finally:
            self._perm_answers.pop(req.id, None)

        if isinstance(decision, Allow):
            option_id = decision.option_id or self._default_allow_option(raw_options)
            if option_id is None:
                self._answer_permission(request_id, {"outcome": {"outcome": "cancelled"}})
                return
            self._answer_permission(
                request_id,
                {"outcome": {"outcome": "selected", "optionId": option_id}},
            )
        else:  # Deny
            self._answer_permission(request_id, {"outcome": {"outcome": "cancelled"}})

    @staticmethod
    def _default_allow_option(raw_options: list[dict[str, Any]]) -> str | None:
        for kind in ("allow_always", "allow_once"):
            for o in raw_options:
                if o.get("kind") == kind:
                    return o.get("optionId")
        return None

    def _answer_permission(self, request_id: Any, result: dict[str, Any]) -> None:
        self._write({"jsonrpc": "2.0", "id": request_id, "result": result})

    # ------------------------------------------------------------------
    # Notifications / streaming -> events on the active turn queue
    # ------------------------------------------------------------------

    def _handle_notification(self, message: dict[str, Any]) -> None:
        method = message.get("method")
        if method not in ("session/update", "session/notification"):
            return
        params = message.get("params", {})
        update = params.get("update") or params.get("notification") or params
        if not isinstance(update, dict):
            return

        update_type = update.get("sessionUpdate") or update.get("type") or update.get("kind")
        event = self._notification_to_event(update_type, update)
        if event is not None:
            self._emit(event)

    def _notification_to_event(
        self, update_type: Any, update: dict[str, Any]
    ) -> BackendEvent | None:
        if update_type in ("agent_message_chunk", "AgentMessageChunk"):
            content = update.get("content") or update.get("chunk")
            text = content.get("text") if isinstance(content, dict) else content
            if text:
                self._turn_text += text
                return TextChunk(text=text)
            return None
        if update_type in ("tool_call", "ToolCall"):
            return ToolCall(
                id=str(update.get("toolCallId") or update.get("id") or ""),
                name=update.get("name") or update.get("toolName") or "tool",
                args=update.get("args") or update.get("input") or {},
                title=update.get("title"),
            )
        if update_type in ("tool_call_update", "ToolCallUpdate"):
            return ToolResult(
                id=str(update.get("toolCallId") or update.get("id") or ""),
                status=_map_status(update.get("status")),
                content=update.get("content"),
            )
        return None

    def _emit(self, event: BackendEvent) -> None:
        queue = self._turn_queue
        if queue is not None:
            queue.put_nowait(event)