Harness

octomate.capabilities.harness.agent

Agent: a Pydantic AI Agent subclass that streams events AND the typed output.

run_stream_events gives raw events but not validated output; stream_output gives validated partial output but not thinking/tool events. stream_events drives iter() ourselves and emits BOTH from each node:

• thinking + tool-call events pass through unchanged,
• text replies stream as Pydantic AI's native text events; every validated output type surfaces as Pydantic AI's own FinalResult[OutputT]; outputs that validate to list[MessageSegment] also stream one ResultSegmentEvent per segment as partial validation reveals them. Other structured outputs surface only at FinalResult.
• capability-injected events (e.g. todo events) flow through the capability wrapper that Pydantic AI applies to each node's stream,
• a terminal AgentRunResultEvent closes the stream (run-complete / deferred).

The two output_type overloads carry the output type into the event stream, so the final Agent[Deps, list[Row]].stream_events(...) value is a FinalResult[list[Row]].

Agent

Bases: Agent[AgentDepsT, OutputDataT]

A Pydantic AI agent that can stream events and the typed output together.

Source code in octomate/capabilities/harness/agent.py

class Agent(PydanticAgent[AgentDepsT, OutputDataT]):
    """A Pydantic AI agent that can stream events and the typed output together."""

    @overload
    def stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        output_type: None = None,
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        deps: AgentDepsT = None,
        model: Model | KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        output_retries: int | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        builtin_tools: Sequence[AgentNativeTool[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AsyncIterator[
        StreamEvents[OutputDataT] | AgentRunResultEvent[OutputDataT]
    ]: ...

    @overload
    def stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        output_type: OutputSpec[RunOutputDataT],
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        deps: AgentDepsT = None,
        model: Model | KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        output_retries: int | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        builtin_tools: Sequence[AgentNativeTool[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AsyncIterator[
        StreamEvents[RunOutputDataT] | AgentRunResultEvent[RunOutputDataT]
    ]: ...

    async def stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        output_type: OutputSpec[Any] | None = None,
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        deps: AgentDepsT = None,
        model: Model | KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        output_retries: int | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        builtin_tools: Sequence[AgentNativeTool[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AsyncIterator[StreamEvents[Any] | AgentRunResultEvent[Any]]:
        # builtin_tools and output_retries are deprecated run kwargs in pydantic-ai
        # 1.x: native tools register as NativeTool capabilities, and the output-retry
        # budget moves under retries={"output": ...}.
        retries = (
            AgentRetries(output=output_retries) if output_retries is not None else None
        )
        if builtin_tools:
            capabilities = [
                *(capabilities or []),
                *(NativeTool(tool) for tool in builtin_tools),
            ]
        async with self.iter(
            user_prompt,
            output_type=output_type,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            conversation_id=conversation_id,
            deps=deps,
            model=model,
            instructions=instructions,
            model_settings=model_settings,
            usage_limits=usage_limits or UsageLimits(request_limit=None),
            usage=usage,
            metadata=metadata,
            retries=retries,
            infer_name=infer_name,
            toolsets=toolsets,
            capabilities=capabilities,
            spec=spec,
        ) as run:
            # Driven by hand rather than `async for node in run`: that path is the
            # graph's own iteration, which `AgentRun.__anext__` documents as calling
            # none of the node hooks, so a capability built on them would be silently
            # inert here while working on a plain agent. The order below is the one
            # `run_stream_events` uses for the same reason we need it — `before_node_run`
            # may replace the node, so it has to fire before the node streams, and the
            # rest of the lifecycle can only run once the stream is drained.
            node = run.next_node
            while not isinstance(node, End):
                # A `wrap_run` hook can answer without the graph running at all.
                if run.result is not None:
                    break
                capability = run.ctx.deps.root_capability
                node = await capability.before_node_run(
                    build_run_context(run.ctx), node=node
                )
                if self.is_model_request_node(node):
                    # The model node streams thinking/pre-output events unchanged.
                    # Plain-text output (FinalResultEvent.tool_name is None) keeps
                    # streaming its native text events after the final result is
                    # recognized — that text *is* the answer. Tool-backed structured
                    # outputs instead surface segment-list elements as
                    # ResultSegmentEvent; every output type still reaches FinalResult.
                    async with node.stream(run.ctx) as stream:
                        final_event: FinalResultEvent | None = None
                        emitted_segments = 0
                        # The element whose text is streaming as deltas, and how much
                        # of it has already been emitted.
                        streamed_index: int | None = None
                        streamed_chars = 0
                        # Pydantic AI times the first chunk (the `chat` span's
                        # `gen_ai.client.operation.time_to_first_chunk`) but marks
                        # neither instant on the trace, and has no notion of a last
                        # one. These two bracket the window a channel timeline renders
                        # into, so a render that trails the model reads as a gap
                        # rather than being inferred from the request span.
                        first_token: float | None = None
                        streamed_events = 0

                        def settle_element(
                            segment: object,
                            index: int,
                            streamed_index: int | None,
                            streamed_chars: int,
                        ) -> ResultSegmentEvent | ResultTextDeltaEvent | None:
                            # A validated segment-list element is a union member; the
                            # Annotated discriminated union can't be
                            # isinstance-narrowed, so guard on the base then cast.
                            if not isinstance(segment, Segment):
                                return None
                            if index == streamed_index:
                                # Already rendered as deltas; only the text revealed
                                # since the last one is still owed.
                                remainder = str(segment)[streamed_chars:]
                                if not remainder:
                                    return None
                                return ResultTextDeltaEvent(delta=remainder)
                            return ResultSegmentEvent(
                                segment=cast(MessageSegment, segment)
                            )

                        # A capability injecting a non-AgentStreamEvent before the
                        # FinalResultEvent passes through here; injecting after it is
                        # not supported (todo events inject on the tools node).
                        async for event in stream:
                            if first_token is None:
                                first_token = time.perf_counter()
                                react_logfire.info("llm.first_token")
                            streamed_events += 1
                            if isinstance(event, FinalResultEvent):
                                final_event = event
                                continue
                            # Pre-final events pass through; once a plain-text final
                            # result is recognized its native text deltas keep flowing.
                            if final_event is None or final_event.tool_name is None:
                                yield event
                                continue
                            # After a tool-backed final result is recognized, outputs
                            # that validate into a growing list of message segments are
                            # surfaced one segment at a time. Other structured
                            # outputs are intentionally left for FinalResult.
                            try:
                                partial = await stream.validate_response_output(
                                    stream.response, allow_partial=True
                                )
                            except (ValidationError, ModelRetry):
                                continue
                            # TODO: emit all the items from a iterable structured output
                            # as they validate, not only lists of segments.
                            if isinstance(partial, list):
                                # Elements a later one has sealed are final — emit them
                                # whole, except the one already streaming as deltas,
                                # which settles with its remainder.
                                for index in range(emitted_segments, len(partial) - 1):
                                    settled = settle_element(
                                        partial[index],
                                        index,
                                        streamed_index,
                                        streamed_chars,
                                    )
                                    if settled is not None:
                                        yield settled
                                emitted_segments = max(
                                    emitted_segments, len(partial) - 1
                                )
                                # The trailing element may still be growing under
                                # partial validation. A text-bearing tail streams its
                                # growth immediately — the visible reply must not wait
                                # for the model to finish; other shapes wait to be
                                # sealed (or for the final validated output below).
                                tail_index = len(partial) - 1
                                if tail_index >= emitted_segments and isinstance(
                                    partial[tail_index], TextSegment | MarkdownSegment
                                ):
                                    if streamed_index != tail_index:
                                        streamed_index = tail_index
                                        streamed_chars = 0
                                    text = str(partial[tail_index])
                                    if len(text) > streamed_chars:
                                        delta = text[streamed_chars:]
                                        streamed_chars = len(text)
                                        yield ResultTextDeltaEvent(delta=delta)
                        if first_token is not None:
                            react_logfire.info(
                                "llm.last_token",
                                streamed=time.perf_counter() - first_token,
                                events=streamed_events,
                            )
                        if final_event is not None:
                            try:
                                final = await stream.validate_response_output(
                                    stream.response
                                )
                            except (ValidationError, ModelRetry):
                                final = None
                            if isinstance(final, list):
                                for index in range(emitted_segments, len(final)):
                                    settled = settle_element(
                                        final[index],
                                        index,
                                        streamed_index,
                                        streamed_chars,
                                    )
                                    if settled is not None:
                                        yield settled
                            yield FinalResult(
                                output=final,
                                tool_name=final_event.tool_name,
                                tool_call_id=final_event.tool_call_id,
                            )
                elif self.is_call_tools_node(node):
                    # The tools node streams tool call/result events. Wrapping with the
                    # run's capabilities lets capability-injected events (e.g. todo
                    # events stashed on a ToolReturn) reach the consumer.
                    async with node.stream(run.ctx) as tool_stream:
                        async for tool_event in tool_stream:
                            yield tool_event
                # `wrap_node_run` → `on_node_run_error` → `after_node_run`, around the
                # step that advances the graph. The context is rebuilt so those hooks
                # see the state the streaming above left behind (`run_step`), matching
                # what `run_stream_events` does at the same point.
                node = await run._wrap_and_advance(  # pyright: ignore[reportPrivateUsage]
                    build_run_context(run.ctx),
                    node,
                    run._advance_graph,  # pyright: ignore[reportPrivateUsage]
                )
            if run.result is not None:
                yield AgentRunResultEvent(run.result)

stream_events async

stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    output_type: None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    conversation_id: str | None = None,
    deps: AgentDepsT = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    model_settings: AgentModelSettings[AgentDepsT]
    | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    output_retries: int | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]]
    | None = None,
    builtin_tools: Sequence[AgentNativeTool[AgentDepsT]]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[
    StreamEvents[OutputDataT]
    | AgentRunResultEvent[OutputDataT]
]

stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    output_type: OutputSpec[RunOutputDataT],
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    conversation_id: str | None = None,
    deps: AgentDepsT = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    model_settings: AgentModelSettings[AgentDepsT]
    | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    output_retries: int | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]]
    | None = None,
    builtin_tools: Sequence[AgentNativeTool[AgentDepsT]]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[
    StreamEvents[RunOutputDataT]
    | AgentRunResultEvent[RunOutputDataT]
]

stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    conversation_id: str | None = None,
    deps: AgentDepsT = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    model_settings: AgentModelSettings[AgentDepsT]
    | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    output_retries: int | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]]
    | None = None,
    builtin_tools: Sequence[AgentNativeTool[AgentDepsT]]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[
    StreamEvents[Any] | AgentRunResultEvent[Any]
]

Source code in octomate/capabilities/harness/agent.py

async def stream_events(
    self,
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    conversation_id: str | None = None,
    deps: AgentDepsT = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    model_settings: AgentModelSettings[AgentDepsT] | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    output_retries: int | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
    builtin_tools: Sequence[AgentNativeTool[AgentDepsT]] | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[StreamEvents[Any] | AgentRunResultEvent[Any]]:
    # builtin_tools and output_retries are deprecated run kwargs in pydantic-ai
    # 1.x: native tools register as NativeTool capabilities, and the output-retry
    # budget moves under retries={"output": ...}.
    retries = (
        AgentRetries(output=output_retries) if output_retries is not None else None
    )
    if builtin_tools:
        capabilities = [
            *(capabilities or []),
            *(NativeTool(tool) for tool in builtin_tools),
        ]
    async with self.iter(
        user_prompt,
        output_type=output_type,
        message_history=message_history,
        deferred_tool_results=deferred_tool_results,
        conversation_id=conversation_id,
        deps=deps,
        model=model,
        instructions=instructions,
        model_settings=model_settings,
        usage_limits=usage_limits or UsageLimits(request_limit=None),
        usage=usage,
        metadata=metadata,
        retries=retries,
        infer_name=infer_name,
        toolsets=toolsets,
        capabilities=capabilities,
        spec=spec,
    ) as run:
        # Driven by hand rather than `async for node in run`: that path is the
        # graph's own iteration, which `AgentRun.__anext__` documents as calling
        # none of the node hooks, so a capability built on them would be silently
        # inert here while working on a plain agent. The order below is the one
        # `run_stream_events` uses for the same reason we need it — `before_node_run`
        # may replace the node, so it has to fire before the node streams, and the
        # rest of the lifecycle can only run once the stream is drained.
        node = run.next_node
        while not isinstance(node, End):
            # A `wrap_run` hook can answer without the graph running at all.
            if run.result is not None:
                break
            capability = run.ctx.deps.root_capability
            node = await capability.before_node_run(
                build_run_context(run.ctx), node=node
            )
            if self.is_model_request_node(node):
                # The model node streams thinking/pre-output events unchanged.
                # Plain-text output (FinalResultEvent.tool_name is None) keeps
                # streaming its native text events after the final result is
                # recognized — that text *is* the answer. Tool-backed structured
                # outputs instead surface segment-list elements as
                # ResultSegmentEvent; every output type still reaches FinalResult.
                async with node.stream(run.ctx) as stream:
                    final_event: FinalResultEvent | None = None
                    emitted_segments = 0
                    # The element whose text is streaming as deltas, and how much
                    # of it has already been emitted.
                    streamed_index: int | None = None
                    streamed_chars = 0
                    # Pydantic AI times the first chunk (the `chat` span's
                    # `gen_ai.client.operation.time_to_first_chunk`) but marks
                    # neither instant on the trace, and has no notion of a last
                    # one. These two bracket the window a channel timeline renders
                    # into, so a render that trails the model reads as a gap
                    # rather than being inferred from the request span.
                    first_token: float | None = None
                    streamed_events = 0

                    def settle_element(
                        segment: object,
                        index: int,
                        streamed_index: int | None,
                        streamed_chars: int,
                    ) -> ResultSegmentEvent | ResultTextDeltaEvent | None:
                        # A validated segment-list element is a union member; the
                        # Annotated discriminated union can't be
                        # isinstance-narrowed, so guard on the base then cast.
                        if not isinstance(segment, Segment):
                            return None
                        if index == streamed_index:
                            # Already rendered as deltas; only the text revealed
                            # since the last one is still owed.
                            remainder = str(segment)[streamed_chars:]
                            if not remainder:
                                return None
                            return ResultTextDeltaEvent(delta=remainder)
                        return ResultSegmentEvent(
                            segment=cast(MessageSegment, segment)
                        )

                    # A capability injecting a non-AgentStreamEvent before the
                    # FinalResultEvent passes through here; injecting after it is
                    # not supported (todo events inject on the tools node).
                    async for event in stream:
                        if first_token is None:
                            first_token = time.perf_counter()
                            react_logfire.info("llm.first_token")
                        streamed_events += 1
                        if isinstance(event, FinalResultEvent):
                            final_event = event
                            continue
                        # Pre-final events pass through; once a plain-text final
                        # result is recognized its native text deltas keep flowing.
                        if final_event is None or final_event.tool_name is None:
                            yield event
                            continue
                        # After a tool-backed final result is recognized, outputs
                        # that validate into a growing list of message segments are
                        # surfaced one segment at a time. Other structured
                        # outputs are intentionally left for FinalResult.
                        try:
                            partial = await stream.validate_response_output(
                                stream.response, allow_partial=True
                            )
                        except (ValidationError, ModelRetry):
                            continue
                        # TODO: emit all the items from a iterable structured output
                        # as they validate, not only lists of segments.
                        if isinstance(partial, list):
                            # Elements a later one has sealed are final — emit them
                            # whole, except the one already streaming as deltas,
                            # which settles with its remainder.
                            for index in range(emitted_segments, len(partial) - 1):
                                settled = settle_element(
                                    partial[index],
                                    index,
                                    streamed_index,
                                    streamed_chars,
                                )
                                if settled is not None:
                                    yield settled
                            emitted_segments = max(
                                emitted_segments, len(partial) - 1
                            )
                            # The trailing element may still be growing under
                            # partial validation. A text-bearing tail streams its
                            # growth immediately — the visible reply must not wait
                            # for the model to finish; other shapes wait to be
                            # sealed (or for the final validated output below).
                            tail_index = len(partial) - 1
                            if tail_index >= emitted_segments and isinstance(
                                partial[tail_index], TextSegment | MarkdownSegment
                            ):
                                if streamed_index != tail_index:
                                    streamed_index = tail_index
                                    streamed_chars = 0
                                text = str(partial[tail_index])
                                if len(text) > streamed_chars:
                                    delta = text[streamed_chars:]
                                    streamed_chars = len(text)
                                    yield ResultTextDeltaEvent(delta=delta)
                    if first_token is not None:
                        react_logfire.info(
                            "llm.last_token",
                            streamed=time.perf_counter() - first_token,
                            events=streamed_events,
                        )
                    if final_event is not None:
                        try:
                            final = await stream.validate_response_output(
                                stream.response
                            )
                        except (ValidationError, ModelRetry):
                            final = None
                        if isinstance(final, list):
                            for index in range(emitted_segments, len(final)):
                                settled = settle_element(
                                    final[index],
                                    index,
                                    streamed_index,
                                    streamed_chars,
                                )
                                if settled is not None:
                                    yield settled
                        yield FinalResult(
                            output=final,
                            tool_name=final_event.tool_name,
                            tool_call_id=final_event.tool_call_id,
                        )
            elif self.is_call_tools_node(node):
                # The tools node streams tool call/result events. Wrapping with the
                # run's capabilities lets capability-injected events (e.g. todo
                # events stashed on a ToolReturn) reach the consumer.
                async with node.stream(run.ctx) as tool_stream:
                    async for tool_event in tool_stream:
                        yield tool_event
            # `wrap_node_run` → `on_node_run_error` → `after_node_run`, around the
            # step that advances the graph. The context is rebuilt so those hooks
            # see the state the streaming above left behind (`run_step`), matching
            # what `run_stream_events` does at the same point.
            node = await run._wrap_and_advance(  # pyright: ignore[reportPrivateUsage]
                build_run_context(run.ctx),
                node,
                run._advance_graph,  # pyright: ignore[reportPrivateUsage]
            )
        if run.result is not None:
            yield AgentRunResultEvent(run.result)

octomate.capabilities.harness.events

Typed event stream the agent run produces and channels consume.

One stream, with these event families:

• Pydantic AI passthrough (AgentStreamEvent) — thinking + tool-call events.
• Output values — text replies stream as Pydantic AI's native TextPart events, every validated output type is emitted as Pydantic AI's own FinalResult[OutputT], and outputs that validate to list[MessageSegment] additionally stream one ResultSegmentEvent per completed segment as partial validation reveals it — except a trailing text segment, whose growth streams as ResultTextDeltaEvents instead. Rendering (typing text, sending segments, or handling any other structured value) is a consumer concern; the event just carries the typed value.
• Display events (DisplayEvent) — fire-and-forget, e.g. the granular todo events (TodoCreatedEvent/TodoUpdatedEvent/TodoStatusChangedEvent/ TodoCompletedEvent/TodoDeletedEvent), each carrying the affected Todo.
• Action batch (ActionBatchEvent) — a persisted batch of deferred actions (questions + approvals) presented as one unit; the run suspends until the user replies. batch_id correlates the reply through the deferred-action machinery.
• Thread metadata (ThreadEvent) — committed titles, project bindings, handoffs, ledger entries and command outcomes published by the thread manager.
• Subagent lifecycle (SubagentStartedEvent/SubagentSettledEvent) — an accomplice's run opening and finishing around its commission or whisper call.

Display, action and subagent events are emitted by capabilities (a capability bundles a tool + instructions + wrap_run_event_stream) or the suspender; the output events are emitted by Agent.stream_events (see octomate/capabilities/harness/agent.py).

The reflex graph also reports outside any run (GatewayEvent, RunErrorEvent), handing each to a channel's Feelers.present, and brackets what a kick does on the nerve: ReflexAwakeEvent and ReflexAsleepEvent around the kick, RunStartedEvent and RunEndedEvent around each agent run in it. Channels present every event and build none; the one consumer-made form is RunResultEvent, the wire shape of a run result.

The run-stream union itself stays generic (FinalResult[OutputT]), so it has no single serialized form — but the wire family is concrete, so WireEvent and its wire_event_adapter live here too: the run stream as a wire consumer sees it, with the generic/unserializable members replaced by their wire forms.

SubagentActivityKind module-attribute

SubagentActivityKind = Literal['commission', 'whisper']

SubagentActivityStatus module-attribute

SubagentActivityStatus = Literal[
    "completed", "failed", "timed_out", "cancelled"
]

wire_event_adapter module-attribute

wire_event_adapter: TypeAdapter[WireEvent] = TypeAdapter(
    WireEvent, config=ConfigDict(ser_json_bytes="base64")
)

ThreadEvent

ThreadEvent = (
    ThreadTitleChangedEvent
    | ThreadProjectBoundEvent
    | ThreadHandoffRecordedEvent
    | ThreadMessageRecordedEvent
    | ThreadCommandOutcomeRecordedEvent
)

TodoEvent

TodoEvent = (
    TodoCreatedEvent
    | TodoUpdatedEvent
    | TodoStatusChangedEvent
    | TodoCompletedEvent
    | TodoDeletedEvent
)

StreamEvents

StreamEvents = (
    AgentStreamEvent
    | ResultSegmentEvent
    | ResultTextDeltaEvent
    | FinalResult[OutputT]
    | TodoEvent
    | MessageSentEvent
    | OAuthAuthorizationEvent
    | ActionBatchEvent
    | ActionBatchAnsweredEvent
    | SubagentStartedEvent
    | SubagentSettledEvent
)

WireEvent

WireEvent = (
    Annotated[
        AgentStreamEvent,
        Field(json_schema_extra=omit_native_discriminator),
    ]
    | ResultSegmentEvent
    | ResultTextDeltaEvent
    | TodoEvent
    | MessageSentEvent
    | ArtifactAvailableEvent
    | OAuthDeviceAuthorizationEvent
    | OAuthAuthorizationEvent
    | ActionBatchEvent
    | ActionBatchAnsweredEvent
    | SubagentStartedEvent
    | SubagentSettledEvent
    | RunResultEvent
    | ReflexAwakeEvent
    | ReflexAsleepEvent
    | ThreadEvent
    | RunStartedEvent
    | RunEndedEvent
    | RunErrorEvent
    | GatewayEvent
)

SubagentStartedEvent pydantic-model

Bases: BaseModel

An accomplice's run opened, named by the call that started it; a channel draws it on a timeline of its own.

Fields:

• event_kind (Literal['subagent_started'])
• invocation_id (str)
• kind (SubagentActivityKind)
• name (str)

Source code in octomate/capabilities/harness/events.py

class SubagentStartedEvent(BaseModel):
    """An accomplice's run opened, named by the call that started it; a channel
    draws it on a timeline of its own."""

    event_kind: Literal["subagent_started"] = "subagent_started"
    invocation_id: str
    kind: SubagentActivityKind
    name: str

event_kind pydantic-field

event_kind: Literal["subagent_started"] = "subagent_started"

invocation_id pydantic-field

invocation_id: str

kind pydantic-field

kind: SubagentActivityKind

name pydantic-field

name: str

SubagentSettledEvent pydantic-model

Bases: BaseModel

The child run finished; response is its accumulated reply text.

Fields:

• event_kind (Literal['subagent_settled'])
• invocation_id (str)
• status (SubagentActivityStatus)
• detail (str | None)
• response (str)

Source code in octomate/capabilities/harness/events.py

class SubagentSettledEvent(BaseModel):
    """The child run finished; `response` is its accumulated reply text."""

    event_kind: Literal["subagent_settled"] = "subagent_settled"
    invocation_id: str
    status: SubagentActivityStatus
    detail: str | None = None
    response: str = ""

event_kind pydantic-field

event_kind: Literal["subagent_settled"] = "subagent_settled"

invocation_id pydantic-field

invocation_id: str

status pydantic-field

status: SubagentActivityStatus

detail pydantic-field

detail: str | None = None

response pydantic-field

response: str = ''

ReflexAwakeEvent pydantic-model

Bases: BaseModel

A kick's first word: the reflex woken, before anything it does.

Fields:

• event_kind (Literal['reflex_awake'])

Source code in octomate/capabilities/harness/events.py

class ReflexAwakeEvent(BaseModel):
    """A kick's first word: the reflex woken, before anything it does."""

    event_kind: Literal["reflex_awake"] = "reflex_awake"

event_kind pydantic-field

event_kind: Literal['reflex_awake'] = 'reflex_awake'

ReflexAsleepEvent pydantic-model

Bases: BaseModel

A kick's last word, the counterpart of ReflexAwakeEvent: the reflex asleep again, every run and move of the kick over, finished or failed. What a console waits for before it reads the thread again.

Fields:

• event_kind (Literal['reflex_asleep'])

Source code in octomate/capabilities/harness/events.py

class ReflexAsleepEvent(BaseModel):
    """A kick's last word, the counterpart of `ReflexAwakeEvent`: the reflex
    asleep again, every run and move of the kick over, finished or failed. What
    a console waits for before it reads the thread again."""

    event_kind: Literal["reflex_asleep"] = "reflex_asleep"

event_kind pydantic-field

event_kind: Literal['reflex_asleep'] = 'reflex_asleep'

ThreadTitleChangedEvent pydantic-model

Bases: BaseModel

A committed title change, including a title derived from an opening message.

Fields:

• event_kind (Literal['thread_title_changed'])
• thread_id (UUID7)
• title (str | None)
• updated_at (AwareDatetime)

Source code in octomate/capabilities/harness/events.py

class ThreadTitleChangedEvent(BaseModel):
    """A committed title change, including a title derived from an opening message."""

    event_kind: Literal["thread_title_changed"] = "thread_title_changed"
    thread_id: UUID7
    title: str | None
    updated_at: AwareDatetime

event_kind pydantic-field

event_kind: Literal["thread_title_changed"] = (
    "thread_title_changed"
)

thread_id pydantic-field

thread_id: UUID7

title pydantic-field

title: str | None

updated_at pydantic-field

updated_at: AwareDatetime

ThreadProjectBoundEvent pydantic-model

Bases: BaseModel

A thread committed its project binding.

Fields:

• event_kind (Literal['thread_project_bound'])
• thread_id (UUID7)
• project_id (UUID7)
• updated_at (AwareDatetime)

Source code in octomate/capabilities/harness/events.py

class ThreadProjectBoundEvent(BaseModel):
    """A thread committed its project binding."""

    event_kind: Literal["thread_project_bound"] = "thread_project_bound"
    thread_id: UUID7
    project_id: UUID7
    updated_at: AwareDatetime

event_kind pydantic-field

event_kind: Literal["thread_project_bound"] = (
    "thread_project_bound"
)

thread_id pydantic-field

thread_id: UUID7

project_id pydantic-field

project_id: UUID7

updated_at pydantic-field

updated_at: AwareDatetime

ThreadHandoffRecordedEvent pydantic-model

Bases: BaseModel

A committed handoff identifies the thread's new agent and model.

Fields:

• event_kind (Literal['thread_handoff_recorded'])
• thread_id (UUID7)
• handoff (Handoff)

Source code in octomate/capabilities/harness/events.py

class ThreadHandoffRecordedEvent(BaseModel):
    """A committed handoff identifies the thread's new agent and model."""

    event_kind: Literal["thread_handoff_recorded"] = "thread_handoff_recorded"
    thread_id: UUID7
    handoff: Handoff

event_kind pydantic-field

event_kind: Literal["thread_handoff_recorded"] = (
    "thread_handoff_recorded"
)

thread_id pydantic-field

thread_id: UUID7

handoff pydantic-field

handoff: Handoff

ThreadMessageRecordedEvent pydantic-model

Bases: BaseModel

A ledger entry committed, advancing the thread's listing timestamp.

Includes command receipts and fork notices. The first recorded entry makes the thread visible to its registered participants; an unknown thread can be loaded by its id. No message content is copied into a listing notification.

Fields:

• event_kind (Literal['thread_message_recorded'])
• thread_id (UUID7)
• message_id (UUID7)
• updated_at (AwareDatetime)

Source code in octomate/capabilities/harness/events.py

class ThreadMessageRecordedEvent(BaseModel):
    """A ledger entry committed, advancing the thread's listing timestamp.

    Includes command receipts and fork notices. The first recorded entry makes
    the thread visible to its registered participants; an unknown thread can be
    loaded by its id. No message content is copied into a listing notification.
    """

    event_kind: Literal["thread_message_recorded"] = "thread_message_recorded"
    thread_id: UUID7
    message_id: UUID7
    updated_at: AwareDatetime

event_kind pydantic-field

event_kind: Literal["thread_message_recorded"] = (
    "thread_message_recorded"
)

thread_id pydantic-field

thread_id: UUID7

message_id pydantic-field

message_id: UUID7

updated_at pydantic-field

updated_at: AwareDatetime

ThreadCommandOutcomeRecordedEvent pydantic-model

Bases: BaseModel

A command's terminal outcome committed to its receipt.

Fields:

• event_kind (Literal['thread_command_outcome_recorded'])
• thread_id (UUID7)
• receipt_id (UUID7)
• outcome (CommandOutcome)
• updated_at (AwareDatetime)

Source code in octomate/capabilities/harness/events.py

class ThreadCommandOutcomeRecordedEvent(BaseModel):
    """A command's terminal outcome committed to its receipt."""

    event_kind: Literal["thread_command_outcome_recorded"] = (
        "thread_command_outcome_recorded"
    )
    thread_id: UUID7
    receipt_id: UUID7
    outcome: CommandOutcome
    updated_at: AwareDatetime

event_kind pydantic-field

event_kind: Literal["thread_command_outcome_recorded"] = (
    "thread_command_outcome_recorded"
)

thread_id pydantic-field

thread_id: UUID7

receipt_id pydantic-field

receipt_id: UUID7

outcome pydantic-field

outcome: CommandOutcome

updated_at pydantic-field

updated_at: AwareDatetime

RunStartedEvent pydantic-model

Bases: BaseModel

An agent run beginning, before anything it streams: the thread it reports into — another than the one asked about when a move carried the turn there. A kick holds one run or several, one after another.

Fields:

• event_kind (Literal['run_started'])
• address (ChannelAddress)
• sender (UserProfile | None)

Source code in octomate/capabilities/harness/events.py

class RunStartedEvent(BaseModel):
    """An agent run beginning, before anything it streams: the thread it reports
    into — another than the one asked about when a move carried the turn there.
    A kick holds one run or several, one after another."""

    event_kind: Literal["run_started"] = "run_started"
    address: ChannelAddress
    sender: UserProfile | None = Field(
        default=None,
        description="The profile supplying a native transcript's prompts; None "
        "when the run does not have a single native sender.",
    )

event_kind pydantic-field

event_kind: Literal['run_started'] = 'run_started'

address pydantic-field

address: ChannelAddress

sender pydantic-field

sender: UserProfile | None = None

The profile supplying a native transcript's prompts; None when the run does not have a single native sender.

RunEndedEvent pydantic-model

Bases: BaseModel

An agent run over, finished or failed, after everything it streamed: the counterpart of RunStartedEvent.

Fields:

• event_kind (Literal['run_ended'])
• thread_message_id (UUID7 | None)

Source code in octomate/capabilities/harness/events.py

class RunEndedEvent(BaseModel):
    """An agent run over, finished or failed, after everything it streamed: the
    counterpart of `RunStartedEvent`."""

    event_kind: Literal["run_ended"] = "run_ended"
    thread_message_id: UUID7 | None = Field(
        default=None,
        description="The ledger row of the run's reply, for the channel to mark "
        "with the platform id it renders the reply as; None for a run that left "
        "no reply.",
    )

event_kind pydantic-field

event_kind: Literal['run_ended'] = 'run_ended'

thread_message_id pydantic-field

thread_message_id: UUID7 | None = None

The ledger row of the run's reply, for the channel to mark with the platform id it renders the reply as; None for a run that left no reply.

RunResultEvent pydantic-model

Bases: BaseModel

The serializable projection of AgentRunResultEvent, whose payload drags the run's private graph state and cannot go on a wire. Carries the output value so a reply that never streamed (a plain non-streaming model, a segment-less structured output) still reaches the consumer, and the usage a session display renders.

Fields:

• event_kind (Literal['run_result'])
• output (str | list[MessageSegment] | None)
• usage (RunUsage)

Source code in octomate/capabilities/harness/events.py

class RunResultEvent(BaseModel):
    """The serializable projection of `AgentRunResultEvent`, whose payload
    drags the run's private graph state and cannot go on a wire. Carries the
    output value so a reply that never streamed (a plain non-streaming model,
    a segment-less structured output) still reaches the consumer, and the
    usage a session display renders."""

    event_kind: Literal["run_result"] = "run_result"
    output: str | list[MessageSegment] | None = None
    usage: RunUsage

event_kind pydantic-field

event_kind: Literal['run_result'] = 'run_result'

output pydantic-field

output: str | list[MessageSegment] | None = None

usage pydantic-field

usage: RunUsage

RunErrorEvent pydantic-model

Bases: BaseModel

A turn the graph could not finish, reported to the channel it came from, or to the one it was operated from.

Fields:

• event_kind (Literal['run_error'])
• message (str)
• trace_id (str)

Source code in octomate/capabilities/harness/events.py

class RunErrorEvent(BaseModel):
    """A turn the graph could not finish, reported to the channel it came from, or
    to the one it was operated from."""

    event_kind: Literal["run_error"] = "run_error"
    message: str
    trace_id: str = Field(description="The trace holding the failure's detail.")

event_kind pydantic-field

event_kind: Literal['run_error'] = 'run_error'

message pydantic-field

message: str

trace_id pydantic-field

trace_id: str

The trace holding the failure's detail.

GatewayEvent pydantic-model

Bases: BaseModel

A spell carried the conversation: the console's own operation, or one the agent cast itself mid-turn.

Fields:

• event_kind (Literal['gateway'])
• action (GatewayAction)
• destination (ChannelAddress)
• announcement (str | None)
• thread_message_id (UUID7 | None)

Source code in octomate/capabilities/harness/events.py

class GatewayEvent(BaseModel):
    """A spell carried the conversation: the console's own operation, or one the
    agent cast itself mid-turn."""

    event_kind: Literal["gateway"] = "gateway"
    action: GatewayAction
    destination: ChannelAddress = Field(
        description="Where the conversation is now; a summon's own thread."
    )
    announcement: str | None = Field(
        default=None,
        description="The line the move leaves where the conversation was, or None "
        "when it leaves none there.",
    )
    thread_message_id: UUID7 | None = Field(
        default=None,
        description="The ledger row of that line, for the channel to mark with the "
        "platform id it posts the line as; None when the move recorded none.",
    )

event_kind pydantic-field

event_kind: Literal['gateway'] = 'gateway'

action pydantic-field

action: GatewayAction

destination pydantic-field

destination: ChannelAddress

Where the conversation is now; a summon's own thread.

announcement pydantic-field

announcement: str | None = None

The line the move leaves where the conversation was, or None when it leaves none there.

thread_message_id pydantic-field

thread_message_id: UUID7 | None = None

The ledger row of that line, for the channel to mark with the platform id it posts the line as; None when the move recorded none.

ResultSegmentEvent dataclass

One completed message segment from a list[MessageSegment] output.

This is a convenience streaming event for channel renderers. The full typed output, segment list or otherwise, still arrives as FinalResult[OutputT].

Source code in octomate/capabilities/harness/events.py

@dataclass
class ResultSegmentEvent:
    """One completed message segment from a `list[MessageSegment]` output.

    This is a convenience streaming event for channel renderers. The full typed
    output, segment list or otherwise, still arrives as `FinalResult[OutputT]`.
    """

    segment: MessageSegment
    event_kind: Literal["result_segment"] = "result_segment"

segment instance-attribute

segment: MessageSegment

event_kind class-attribute instance-attribute

event_kind: Literal['result_segment'] = 'result_segment'

ResultTextDeltaEvent dataclass

The trailing text segment of a list[MessageSegment] output, as it grows.

A tool-backed segment reply carries its visible text inside the output tool's streamed arguments, where no native text events exist; partial validation reveals the trailing segment's text as it grows, and each event carries the newly revealed piece so the channel renders the reply while the model is still writing it. A segment that streamed this way is settled by its last delta and never re-emitted as a ResultSegmentEvent.

Source code in octomate/capabilities/harness/events.py

@dataclass
class ResultTextDeltaEvent:
    """The trailing text segment of a `list[MessageSegment]` output, as it grows.

    A tool-backed segment reply carries its visible text inside the output
    tool's streamed arguments, where no native text events exist; partial
    validation reveals the trailing segment's text as it grows, and each event
    carries the newly revealed piece so the channel renders the reply while the
    model is still writing it. A segment that streamed this way is settled by
    its last delta and never re-emitted as a `ResultSegmentEvent`.
    """

    delta: str
    event_kind: Literal["result_text_delta"] = "result_text_delta"

delta instance-attribute

delta: str

event_kind class-attribute instance-attribute

event_kind: Literal["result_text_delta"] = (
    "result_text_delta"
)

DisplayEvent pydantic-model

Bases: BaseModel

Fire-and-forget outbound event; the run continues after it is emitted.

Source code in octomate/capabilities/harness/events.py

class DisplayEvent(BaseModel):
    """Fire-and-forget outbound event; the run continues after it is emitted."""

ArtifactAvailableEvent pydantic-model

Bases: DisplayEvent

An artifact and its ledger reference have been committed.

The message ID deduplicates this appearance across live events and history; the artifact ID identifies the content for an independently authorized read. No invocation data or resource contents travel in this event.

Fields:

• event_kind (Literal['artifact_available'])
• message_id (UUID7)
• artifact_id (UUID7)

Source code in octomate/capabilities/harness/events.py

class ArtifactAvailableEvent(DisplayEvent):
    """An artifact and its ledger reference have been committed.

    The message ID deduplicates this appearance across live events and history;
    the artifact ID identifies the content for an independently authorized read.
    No invocation data or resource contents travel in this event.
    """

    event_kind: Literal["artifact_available"] = "artifact_available"
    message_id: UUID7 = Field(
        description="Committed ledger row referencing the artifact."
    )
    artifact_id: UUID7 = Field(description="The artifact's independent identity.")

event_kind pydantic-field

event_kind: Literal["artifact_available"] = (
    "artifact_available"
)

message_id pydantic-field

message_id: UUID7

Committed ledger row referencing the artifact.

artifact_id pydantic-field

artifact_id: UUID7

The artifact's independent identity.

TodoCreatedEvent pydantic-model

Bases: DisplayEvent

A todo was created.

Fields:

• event_kind (Literal['todo_created'])
• todo (Todo)

Source code in octomate/capabilities/harness/events.py

class TodoCreatedEvent(DisplayEvent):
    """A todo was created."""

    event_kind: Literal["todo_created"] = "todo_created"
    todo: Todo

event_kind pydantic-field

event_kind: Literal['todo_created'] = 'todo_created'

todo pydantic-field

todo: Todo

TodoUpdatedEvent pydantic-model

Bases: DisplayEvent

A todo's content or placement changed.

Fields:

• event_kind (Literal['todo_updated'])
• todo (Todo)
• previous (Todo | None)

Source code in octomate/capabilities/harness/events.py

class TodoUpdatedEvent(DisplayEvent):
    """A todo's content or placement changed."""

    event_kind: Literal["todo_updated"] = "todo_updated"
    todo: Todo
    previous: Todo | None = None

event_kind pydantic-field

event_kind: Literal['todo_updated'] = 'todo_updated'

todo pydantic-field

todo: Todo

previous pydantic-field

previous: Todo | None = None

TodoStatusChangedEvent pydantic-model

Bases: DisplayEvent

A todo moved to another status.

Fields:

• event_kind (Literal['todo_status_changed'])
• todo (Todo)
• previous (Todo | None)

Source code in octomate/capabilities/harness/events.py

class TodoStatusChangedEvent(DisplayEvent):
    """A todo moved to another status."""

    event_kind: Literal["todo_status_changed"] = "todo_status_changed"
    todo: Todo
    previous: Todo | None = None

event_kind pydantic-field

event_kind: Literal["todo_status_changed"] = (
    "todo_status_changed"
)

todo pydantic-field

todo: Todo

previous pydantic-field

previous: Todo | None = None

TodoCompletedEvent pydantic-model

Bases: DisplayEvent

A todo was completed.

Fields:

• event_kind (Literal['todo_completed'])
• todo (Todo)

Source code in octomate/capabilities/harness/events.py

class TodoCompletedEvent(DisplayEvent):
    """A todo was completed."""

    event_kind: Literal["todo_completed"] = "todo_completed"
    todo: Todo

event_kind pydantic-field

event_kind: Literal['todo_completed'] = 'todo_completed'

todo pydantic-field

todo: Todo

TodoDeletedEvent pydantic-model

Bases: DisplayEvent

A todo was deleted.

Fields:

• event_kind (Literal['todo_deleted'])
• todo (Todo)

Source code in octomate/capabilities/harness/events.py

class TodoDeletedEvent(DisplayEvent):
    """A todo was deleted."""

    event_kind: Literal["todo_deleted"] = "todo_deleted"
    todo: Todo

event_kind pydantic-field

event_kind: Literal['todo_deleted'] = 'todo_deleted'

todo pydantic-field

todo: Todo

MessageSentEvent pydantic-model

Bases: DisplayEvent

The gate's send emitted segments to be delivered mid-run, without ending the turn. Emit-only: the tool touches no channel. Where it goes is settled by the time this is emitted — destination is None for the run's own conversation, and otherwise a surface the gate resolved from the identity registry.

Fields:

• event_kind (Literal['message_sent'])
• segments (list[MessageSegment])
• destination (ChannelAddress | None)
• thread_message_id (UUID7 | None)

Source code in octomate/capabilities/harness/events.py

class MessageSentEvent(DisplayEvent):
    """The gate's `send` emitted `segments` to be delivered mid-run, without ending
    the turn. Emit-only: the tool touches no channel. Where it goes is settled by
    the time this is emitted — `destination` is None for the run's own conversation,
    and otherwise a surface the gate resolved from the identity registry."""

    event_kind: Literal["message_sent"] = "message_sent"
    segments: list[MessageSegment]
    destination: ChannelAddress | None = Field(
        default=None,
        description="None for this conversation. Otherwise the address the gate "
        "already resolved from the target the model named, so the consumer delivers "
        "without re-deciding anything and a refused destination never reaches one.",
    )
    thread_message_id: UUID7 | None = Field(
        default=None,
        description="The ledger row the send left, for the channel to mark with the "
        "platform id it delivers it as; None for a send recorded nowhere.",
    )

event_kind pydantic-field

event_kind: Literal['message_sent'] = 'message_sent'

segments pydantic-field

segments: list[MessageSegment]

destination pydantic-field

destination: ChannelAddress | None = None

None for this conversation. Otherwise the address the gate already resolved from the target the model named, so the consumer delivers without re-deciding anything and a refused destination never reaches one.

thread_message_id pydantic-field

thread_message_id: UUID7 | None = None

The ledger row the send left, for the channel to mark with the platform id it delivers it as; None for a send recorded nowhere.

OAuthAuthorizationEvent pydantic-model

Bases: DisplayEvent

An integration is waiting for this user to authorize their own account.

Carries the authorization itself rather than a rendered message, so each channel presents it as well as it can — a link in a plain timeline, a card whose button opens the page where the platform has one. connector_id is the connection this authorization belongs to.

Emitted as it stands by an authorization-code flow, where opening the link and approving is the whole errand: nothing to copy, and nothing to come back for because the provider's callback finishes the connection.

Fields:

• event_kind (Literal['oauth_authorization', 'oauth_device_authorization'])
• connector_id (str)
• label (str)
• authorization_uri (str)

Source code in octomate/capabilities/harness/events.py

class OAuthAuthorizationEvent(DisplayEvent):
    """An integration is waiting for this user to authorize their own account.

    Carries the authorization itself rather than a rendered message, so each channel
    presents it as well as it can — a link in a plain timeline, a card whose button
    opens the page where the platform has one. `connector_id` is the connection this
    authorization belongs to.

    Emitted as it stands by an authorization-code flow, where opening the link and
    approving is the whole errand: nothing to copy, and nothing to come back for
    because the provider's callback finishes the connection.
    """

    event_kind: Literal["oauth_authorization", "oauth_device_authorization"] = (
        "oauth_authorization"
    )
    connector_id: str
    label: str
    authorization_uri: str = Field(
        description="The page the user opens. A device flow's verification page, or "
        "the UUID-only link an authorization-code flow staged its provider request "
        "behind."
    )

event_kind pydantic-field

event_kind: Literal[
    "oauth_authorization", "oauth_device_authorization"
] = "oauth_authorization"

connector_id pydantic-field

connector_id: str

label pydantic-field

label: str

authorization_uri pydantic-field

authorization_uri: str

The page the user opens. A device flow's verification page, or the UUID-only link an authorization-code flow staged its provider request behind.

OAuthDeviceAuthorizationEvent pydantic-model

Bases: OAuthAuthorizationEvent

A device flow, which asks the user for one thing more.

Finishing is a second errand they have to come back for, which is why the code travels with the link and why a presenter asks them to return. A presenter that does not distinguish the two renders this as its base and simply leaves the code out — wrong, but not misleading.

Fields:

• connector_id (str)
• label (str)
• authorization_uri (str)
• event_kind (Literal['oauth_authorization', 'oauth_device_authorization'])
• user_code (str)

Source code in octomate/capabilities/harness/events.py

class OAuthDeviceAuthorizationEvent(OAuthAuthorizationEvent):
    """A device flow, which asks the user for one thing more.

    Finishing is a second errand they have to come back for, which is why the code
    travels with the link and why a presenter asks them to return. A presenter that
    does not distinguish the two renders this as its base and simply leaves the code
    out — wrong, but not misleading.
    """

    # The declared type has to stay the base's, or the override is a narrowing of a
    # mutable field; only the default changes, which is what names this on the wire.
    event_kind: Literal["oauth_authorization", "oauth_device_authorization"] = (
        "oauth_device_authorization"
    )
    user_code: str = Field(description="The one-time code the user types on the page.")

event_kind pydantic-field

event_kind: Literal[
    "oauth_authorization", "oauth_device_authorization"
] = "oauth_device_authorization"

user_code pydantic-field

user_code: str

The one-time code the user types on the page.

LinkProfileAuthorizationEvent pydantic-model

Bases: DisplayEvent

A channel requests account authorization from the Octomate host.

Delivered directly through private feelers, not the agent's run stream. The host keeps the request; browser consent links its profile without returning an OAuth credential to the channel.

Fields:

• event_kind (Literal['link_profile_authorization'])
• host (str)
• authorization (LinkProfileAuthorization)

Source code in octomate/capabilities/harness/events.py

class LinkProfileAuthorizationEvent(DisplayEvent):
    """A channel requests account authorization from the Octomate host.

    Delivered directly through private feelers, not the agent's run stream. The
    host keeps the request; browser consent links its profile without returning
    an OAuth credential to the channel.
    """

    event_kind: Literal["link_profile_authorization"] = "link_profile_authorization"
    host: str = Field(description="The Octomate host asking for browser consent.")
    authorization: LinkProfileAuthorization

event_kind pydantic-field

event_kind: Literal["link_profile_authorization"] = (
    "link_profile_authorization"
)

host pydantic-field

host: str

The Octomate host asking for browser consent.

authorization pydantic-field

authorization: LinkProfileAuthorization

ActionBatchEvent pydantic-model

Bases: BaseModel

A persisted batch of deferred actions presented as one unit; the run suspends until the user replies.

Carries the batch's questions and approvals (the persisted actions) so the consumer renders + marks them together, and batch_id to correlate the reply back to the right run.

Fields:

• event_kind (Literal['action_batch'])
• batch_id (str)
• questions (list[DeferredQuestion])
• approvals (list[DeferredApproval])

Source code in octomate/capabilities/harness/events.py

class ActionBatchEvent(BaseModel):
    """A persisted batch of deferred actions presented as one unit; the run
    suspends until the user replies.

    Carries the batch's `questions` and `approvals` (the persisted actions) so the
    consumer renders + marks them together, and `batch_id` to correlate the reply
    back to the right run.
    """

    event_kind: Literal["action_batch"] = "action_batch"
    batch_id: str
    questions: list[DeferredQuestion] = Field(default_factory=list)
    approvals: list[DeferredApproval] = Field(default_factory=list)

    @classmethod
    def from_batch(cls, batch: DeferredActionBatch) -> ActionBatchEvent:
        return cls(
            batch_id=str(batch.id),
            questions=list(batch.questions),
            approvals=list(batch.approvals),
        )

event_kind pydantic-field

event_kind: Literal['action_batch'] = 'action_batch'

batch_id pydantic-field

batch_id: str

questions pydantic-field

questions: list[DeferredQuestion]

approvals pydantic-field

approvals: list[DeferredApproval]

from_batch classmethod

from_batch(batch: DeferredActionBatch) -> ActionBatchEvent

Source code in octomate/capabilities/harness/events.py

@classmethod
def from_batch(cls, batch: DeferredActionBatch) -> ActionBatchEvent:
    return cls(
        batch_id=str(batch.id),
        questions=list(batch.questions),
        approvals=list(batch.approvals),
    )

ActionBatchAnsweredEvent pydantic-model

Bases: BaseModel

A batch's actions answered, from whichever surface answered them: every reader still showing the batch's cards marks them, so a card answered in one console, or on another channel, stops waiting everywhere.

Fields:

• event_kind (Literal['action_batch_answered'])
• batch_id (str)
• approvals (dict[UUID7, bool])
• answers (dict[UUID7, QuestionAnswer])

Source code in octomate/capabilities/harness/events.py

class ActionBatchAnsweredEvent(BaseModel):
    """A batch's actions answered, from whichever surface answered them: every
    reader still showing the batch's cards marks them, so a card answered in one
    console, or on another channel, stops waiting everywhere."""

    event_kind: Literal["action_batch_answered"] = "action_batch_answered"
    batch_id: str
    approvals: dict[UUID7, bool] = Field(default_factory=dict)
    answers: dict[UUID7, QuestionAnswer] = Field(default_factory=dict)

    @classmethod
    def of(cls, response: DeferredActionBatchResponse) -> ActionBatchAnsweredEvent:
        return cls(
            batch_id=str(response.batch_id),
            approvals=response.approvals,
            answers=response.answers,
        )

event_kind pydantic-field

event_kind: Literal["action_batch_answered"] = (
    "action_batch_answered"
)

batch_id pydantic-field

batch_id: str

approvals pydantic-field

approvals: dict[UUID7, bool]

answers pydantic-field

answers: dict[UUID7, QuestionAnswer]

of classmethod

of(
    response: DeferredActionBatchResponse,
) -> ActionBatchAnsweredEvent

Source code in octomate/capabilities/harness/events.py

@classmethod
def of(cls, response: DeferredActionBatchResponse) -> ActionBatchAnsweredEvent:
    return cls(
        batch_id=str(response.batch_id),
        approvals=response.approvals,
        answers=response.answers,
    )

omit_native_discriminator

omit_native_discriminator(
    schema: dict[str, JsonValue],
) -> None

Keep native event variants without OpenAPI-invalid registry mappings.

Native registries emit inline discriminator targets, while OpenAPI requires references. The generated oneOf still describes every event variant.

Source code in octomate/capabilities/harness/events.py

def omit_native_discriminator(schema: dict[str, JsonValue]) -> None:
    """Keep native event variants without OpenAPI-invalid registry mappings.

    Native registries emit inline discriminator targets, while OpenAPI requires
    references. The generated oneOf still describes every event variant.
    """
    schema.pop("discriminator", None)

to_wire

to_wire(
    item: StreamEvents[OutputT]
    | AgentRunResultEvent[OutputT],
) -> WireEvent | None

One run stream item as its wire event, or None for the shapes the wire replaces (see the wire module docstring).

Source code in octomate/capabilities/harness/events.py

def to_wire[OutputT: str | Sequence[MessageSegment] | DeferredToolRequests | None](
    item: StreamEvents[OutputT] | AgentRunResultEvent[OutputT],
) -> WireEvent | None:
    """One run stream item as its wire event, or None for the shapes the wire
    replaces (see the wire module docstring)."""
    match item:
        case AgentRunResultEvent():
            match item.result.output:
                case str() | None:
                    output: str | list[MessageSegment] | None = item.result.output
                case DeferredToolRequests():
                    # The suspension already went out as an ActionBatchEvent.
                    output = None
                case segments:
                    output = list(segments)
            return RunResultEvent(output=output, usage=item.result.usage)
        case FinalResult():
            return None
        case _:
            return item

octomate.capabilities.harness.react

logger module-attribute

logger = logging.getLogger(__name__)

ReactStreamEvent

ReactStreamEvent = (
    StreamEvents[ReactOutputT]
    | AgentRunResultEvent[ReactOutputT]
)

ReactGraphInput

ReactGraphInput = (
    StartTurn[ReactOutputT, ReactDepsT]
    | ResumeTurn[ReactOutputT, ReactDepsT]
)

ReactState dataclass

The react graph holds no history of its own — only the identity needed to fetch the live conversation from the ConversationManager. Every node ensure()s the conversation and reads its messages.

Source code in octomate/capabilities/harness/react.py

@dataclass
class ReactState:
    """The react graph holds no history of its own — only the identity needed to
    fetch the live conversation from the ConversationManager. Every node
    `ensure()`s the conversation and reads its messages."""

    conversation_address: ChannelAddress
    agent_tentacle_id: str
    thread_id: UUID7
    # A pre-ensured conversation to run in, by id — the caller that spawned this
    # run chose the context (e.g. a commissioned accomplice's child conversation).
    # None resolves the agent's own (thread, agent) conversation as usual.
    conversation_id: UUID7 | None = None
    source_thread_address: ChannelAddress | None = None
    source_thread_message_ids: list[UUID7] = field(default_factory=list)

conversation_address instance-attribute

conversation_address: ChannelAddress

agent_tentacle_id instance-attribute

agent_tentacle_id: str

thread_id instance-attribute

thread_id: UUID7

conversation_id class-attribute instance-attribute

conversation_id: UUID7 | None = None

source_thread_address class-attribute instance-attribute

source_thread_address: ChannelAddress | None = None

source_thread_message_ids class-attribute instance-attribute

source_thread_message_ids: list[UUID7] = field(
    default_factory=list
)

RunPersistence dataclass

Source code in octomate/capabilities/harness/react.py

@dataclass
class RunPersistence:
    conversation_manager: ConversationManager
    thread_manager: ThreadManager | None
    conversation: Conversation
    state: ReactState
    run_name: str
    cwd: Path | None
    binds_prompt_sources: bool
    permission_mode: AgentPermissionMode | None

    async def record(
        self,
        run_id: str,
        messages: Sequence[PydanticModelMessage],
        *,
        model_name: str | None,
    ) -> PersistedAgentRun | None:
        recorded_run = await self.conversation_manager.record_agent_run(
            self.conversation,
            run_id=run_id,
            messages=messages,
            name=self.run_name,
            model_name=model_name,
            permission_mode=self.permission_mode,
            cwd=self.cwd,
        )
        if not self.state.source_thread_message_ids or not self.binds_prompt_sources:
            return recorded_run
        if recorded_run is None:
            raise RuntimeError("prompt-source bindings require a persisted agent run")
        prompt_request = next(
            (
                message
                for message in recorded_run.messages
                if isinstance(message, ModelRequest) and message.role == "user"
            ),
            None,
        )
        if prompt_request is None:
            raise RuntimeError(
                "prompt-source bindings require a persisted user ModelRequest"
            )
        if self.thread_manager is None:
            raise RuntimeError("prompt-source bindings require a ThreadManager")
        await self.thread_manager.bind_messages(
            self.state.source_thread_message_ids,
            prompt_request.id,
            kind="request_source",
            run_id=recorded_run.id,
        )
        source_thread = await self.thread_manager.ensure(
            self.state.source_thread_address or self.state.conversation_address
        )
        await self.thread_manager.advance_prompt_cursor(
            source_thread,
            self.state.source_thread_message_ids[-1],
        )
        return recorded_run

conversation_manager instance-attribute

conversation_manager: ConversationManager

thread_manager instance-attribute

thread_manager: ThreadManager | None

conversation instance-attribute

conversation: Conversation

state instance-attribute

state: ReactState

run_name instance-attribute

run_name: str

cwd instance-attribute

cwd: Path | None

binds_prompt_sources instance-attribute

binds_prompt_sources: bool

permission_mode instance-attribute

permission_mode: AgentPermissionMode | None

record async

record(
    run_id: str,
    messages: Sequence[ModelMessage],
    *,
    model_name: str | None,
) -> PersistedAgentRun | None

Source code in octomate/capabilities/harness/react.py

async def record(
    self,
    run_id: str,
    messages: Sequence[PydanticModelMessage],
    *,
    model_name: str | None,
) -> PersistedAgentRun | None:
    recorded_run = await self.conversation_manager.record_agent_run(
        self.conversation,
        run_id=run_id,
        messages=messages,
        name=self.run_name,
        model_name=model_name,
        permission_mode=self.permission_mode,
        cwd=self.cwd,
    )
    if not self.state.source_thread_message_ids or not self.binds_prompt_sources:
        return recorded_run
    if recorded_run is None:
        raise RuntimeError("prompt-source bindings require a persisted agent run")
    prompt_request = next(
        (
            message
            for message in recorded_run.messages
            if isinstance(message, ModelRequest) and message.role == "user"
        ),
        None,
    )
    if prompt_request is None:
        raise RuntimeError(
            "prompt-source bindings require a persisted user ModelRequest"
        )
    if self.thread_manager is None:
        raise RuntimeError("prompt-source bindings require a ThreadManager")
    await self.thread_manager.bind_messages(
        self.state.source_thread_message_ids,
        prompt_request.id,
        kind="request_source",
        run_id=recorded_run.id,
    )
    source_thread = await self.thread_manager.ensure(
        self.state.source_thread_address or self.state.conversation_address
    )
    await self.thread_manager.advance_prompt_cursor(
        source_thread,
        self.state.source_thread_message_ids[-1],
    )
    return recorded_run

PersistRunFailure dataclass

Bases: AbstractCapability[ReactDepsT]

Source code in octomate/capabilities/harness/react.py

@dataclass
class PersistRunFailure[ReactDepsT](AbstractCapability[ReactDepsT]):
    persistence: RunPersistence
    previous_message_count: int
    recorded: bool = False

    async def record_failure(self, ctx: RunContext[ReactDepsT]) -> None:
        if self.recorded:
            return
        messages = ctx.messages[self.previous_message_count :]
        if not messages:
            return
        if ctx.run_id is None:
            raise RuntimeError("failed agent run has no run_id")
        await self.persistence.record(
            ctx.run_id,
            messages,
            model_name=ctx.model.model_name,
        )
        self.recorded = True

    async def on_node_run_error(
        self,
        ctx: RunContext[ReactDepsT],
        *,
        node: AgentNode[ReactDepsT],
        error: Exception,
    ) -> NodeResult[ReactDepsT]:
        await self.record_failure(ctx)
        raise error

    async def on_model_request_error(
        self,
        ctx: RunContext[ReactDepsT],
        *,
        request_context: ModelRequestContext,
        error: Exception,
    ) -> ModelResponse:
        await self.record_failure(ctx)
        raise error

persistence instance-attribute

persistence: RunPersistence

previous_message_count instance-attribute

previous_message_count: int

recorded class-attribute instance-attribute

recorded: bool = False

record_failure async

record_failure(ctx: RunContext[ReactDepsT]) -> None

Source code in octomate/capabilities/harness/react.py

async def record_failure(self, ctx: RunContext[ReactDepsT]) -> None:
    if self.recorded:
        return
    messages = ctx.messages[self.previous_message_count :]
    if not messages:
        return
    if ctx.run_id is None:
        raise RuntimeError("failed agent run has no run_id")
    await self.persistence.record(
        ctx.run_id,
        messages,
        model_name=ctx.model.model_name,
    )
    self.recorded = True

on_node_run_error async

on_node_run_error(
    ctx: RunContext[ReactDepsT],
    *,
    node: AgentNode[ReactDepsT],
    error: Exception,
) -> NodeResult[ReactDepsT]

Source code in octomate/capabilities/harness/react.py

async def on_node_run_error(
    self,
    ctx: RunContext[ReactDepsT],
    *,
    node: AgentNode[ReactDepsT],
    error: Exception,
) -> NodeResult[ReactDepsT]:
    await self.record_failure(ctx)
    raise error

on_model_request_error async

on_model_request_error(
    ctx: RunContext[ReactDepsT],
    *,
    request_context: ModelRequestContext,
    error: Exception,
) -> ModelResponse

Source code in octomate/capabilities/harness/react.py

async def on_model_request_error(
    self,
    ctx: RunContext[ReactDepsT],
    *,
    request_context: ModelRequestContext,
    error: Exception,
) -> ModelResponse:
    await self.record_failure(ctx)
    raise error

PersistStreamRunFailure dataclass

Bases: PersistRunFailure[ReactDepsT]

Source code in octomate/capabilities/harness/react.py

@dataclass
class PersistStreamRunFailure[ReactDepsT](PersistRunFailure[ReactDepsT]):
    async def wrap_run_event_stream(
        self,
        ctx: RunContext[ReactDepsT],
        *,
        stream: AsyncIterable[AgentStreamEvent],
    ) -> AsyncIterable[AgentStreamEvent]:
        try:
            async for event in stream:
                yield event
        except Exception:
            await self.record_failure(ctx)
            raise

wrap_run_event_stream async

wrap_run_event_stream(
    ctx: RunContext[ReactDepsT],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]

Source code in octomate/capabilities/harness/react.py

async def wrap_run_event_stream(
    self,
    ctx: RunContext[ReactDepsT],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]:
    try:
        async for event in stream:
            yield event
    except Exception:
        await self.record_failure(ctx)
        raise

ReactDeps dataclass

Source code in octomate/capabilities/harness/react.py

@dataclass
class ReactDeps[ReactOutputT, ReactDepsT]:
    agent: Agent[ReactDepsT, ReactOutputT]
    conversation_manager: ConversationManager
    agent_deps: ReactDepsT
    thread_manager: ThreadManager | None = None
    event_send_stream: ObjectSendStream[ReactStreamEvent[ReactOutputT]] | None = None
    # Asked at each deferral, not held as one resolver for the run — see
    # `ResolverChoice`. The suspender is the other half: what an empty choice means.
    choose_resolvers: ResolverChoice | None = None
    suspender: DeferredSuspender | None = None
    output_type: OutputSpec[ReactOutputT] | None = None
    run_name: str = "react"
    # The directory this run happens in, recorded on every run it persists. None when
    # the run is in no project, since a react run has no directory of its own.
    cwd: DirectoryPath | None = None
    model: Model | KnownModelName | str | None = None
    # The preset used when the conversation has no explicit override.
    permission_mode: AgentPermissionMode | None = None
    instructions: AgentInstructions[ReactDepsT] = None
    model_settings: AgentModelSettings[ReactDepsT] | None = None
    usage_limits: UsageLimits | None = None
    usage: RunUsage | None = None
    metadata: AgentMetadata[ReactDepsT] | None = None
    output_retries: int | None = None
    infer_name: bool = True
    toolsets: Sequence[AbstractToolset[ReactDepsT]] | None = None
    builtin_tools: Sequence[AgentNativeTool[ReactDepsT]] | None = None
    capabilities: Sequence[AgentCapability[ReactDepsT]] | None = None
    spec: dict[str, Any] | AgentSpec | None = None

agent instance-attribute

agent: Agent[ReactDepsT, ReactOutputT]

conversation_manager instance-attribute

conversation_manager: ConversationManager

agent_deps instance-attribute

agent_deps: ReactDepsT

thread_manager class-attribute instance-attribute

thread_manager: ThreadManager | None = None

event_send_stream class-attribute instance-attribute

event_send_stream: (
    ObjectSendStream[ReactStreamEvent[ReactOutputT]] | None
) = None

choose_resolvers class-attribute instance-attribute

choose_resolvers: ResolverChoice | None = None

suspender class-attribute instance-attribute

suspender: DeferredSuspender | None = None

output_type class-attribute instance-attribute

output_type: OutputSpec[ReactOutputT] | None = None

run_name class-attribute instance-attribute

run_name: str = 'react'

cwd class-attribute instance-attribute

cwd: DirectoryPath | None = None

model class-attribute instance-attribute

model: Model | KnownModelName | str | None = None

permission_mode class-attribute instance-attribute

permission_mode: AgentPermissionMode | None = None

instructions class-attribute instance-attribute

instructions: AgentInstructions[ReactDepsT] = None

model_settings class-attribute instance-attribute

model_settings: AgentModelSettings[ReactDepsT] | None = None

usage_limits class-attribute instance-attribute

usage_limits: UsageLimits | None = None

usage class-attribute instance-attribute

usage: RunUsage | None = None

metadata class-attribute instance-attribute

metadata: AgentMetadata[ReactDepsT] | None = None

output_retries class-attribute instance-attribute

output_retries: int | None = None

infer_name class-attribute instance-attribute

infer_name: bool = True

toolsets class-attribute instance-attribute

toolsets: Sequence[AbstractToolset[ReactDepsT]] | None = (
    None
)

builtin_tools class-attribute instance-attribute

builtin_tools: (
    Sequence[AgentNativeTool[ReactDepsT]] | None
) = None

capabilities class-attribute instance-attribute

capabilities: (
    Sequence[AgentCapability[ReactDepsT]] | None
) = None

spec class-attribute instance-attribute

spec: dict[str, Any] | AgentSpec | None = None

StartTurn dataclass

Bases: BaseNode[ReactState, ReactDeps[ReactOutputT, ReactDepsT], AgentRunResult[ReactOutputT]]

Source code in octomate/capabilities/harness/react.py

@dataclass
class StartTurn[ReactOutputT, ReactDepsT](
    BaseNode[
        ReactState,
        ReactDeps[ReactOutputT, ReactDepsT],
        AgentRunResult[ReactOutputT],
    ],
):
    user_prompt: str | Sequence[UserContent] | None

    async def run(
        self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
    ) -> RunAgent[ReactOutputT, ReactDepsT]:
        if self.user_prompt is None:
            return RunAgent()
        conversation = await resolve_conversation(ctx)
        # A new prompt answers every call the last response still waits on, denied,
        # rather than leaving it unanswered: the history stays whole, so a fork
        # that shares the run can still resume the call.
        answered: set[str] = set()
        abandoned: list[str] = []
        for message in reversed(conversation.messages):
            if isinstance(message, ModelResponse):
                abandoned = [
                    part.tool_call_id
                    for part in message.parts
                    if isinstance(part, ToolCallPart)
                    and part.tool_call_id not in answered
                ]
                break
            answered.update(
                part.tool_call_id
                for part in message.parts
                if isinstance(part, ToolReturnPart | RetryPromptPart)
            )
        if not abandoned:
            return RunAgent(user_prompt=self.user_prompt)
        logger.info("StartTurn denies %d abandoned tool calls", len(abandoned))
        return RunAgent(
            user_prompt=self.user_prompt,
            deferred_results=DeferredToolResults(
                approvals={
                    tool_call_id: ToolDenied(
                        "Not carried out: a new message arrived before this call "
                        "was answered."
                    )
                    for tool_call_id in abandoned
                }
            ),
        )

user_prompt instance-attribute

user_prompt: str | Sequence[UserContent] | None

run async

run(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
) -> RunAgent[ReactOutputT, ReactDepsT]

Source code in octomate/capabilities/harness/react.py

async def run(
    self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
) -> RunAgent[ReactOutputT, ReactDepsT]:
    if self.user_prompt is None:
        return RunAgent()
    conversation = await resolve_conversation(ctx)
    # A new prompt answers every call the last response still waits on, denied,
    # rather than leaving it unanswered: the history stays whole, so a fork
    # that shares the run can still resume the call.
    answered: set[str] = set()
    abandoned: list[str] = []
    for message in reversed(conversation.messages):
        if isinstance(message, ModelResponse):
            abandoned = [
                part.tool_call_id
                for part in message.parts
                if isinstance(part, ToolCallPart)
                and part.tool_call_id not in answered
            ]
            break
        answered.update(
            part.tool_call_id
            for part in message.parts
            if isinstance(part, ToolReturnPart | RetryPromptPart)
        )
    if not abandoned:
        return RunAgent(user_prompt=self.user_prompt)
    logger.info("StartTurn denies %d abandoned tool calls", len(abandoned))
    return RunAgent(
        user_prompt=self.user_prompt,
        deferred_results=DeferredToolResults(
            approvals={
                tool_call_id: ToolDenied(
                    "Not carried out: a new message arrived before this call "
                    "was answered."
                )
                for tool_call_id in abandoned
            }
        ),
    )

ResumeTurn dataclass

Bases: BaseNode[ReactState, ReactDeps[ReactOutputT, ReactDepsT], AgentRunResult[ReactOutputT]]

Source code in octomate/capabilities/harness/react.py

@dataclass
class ResumeTurn[ReactOutputT, ReactDepsT](
    BaseNode[
        ReactState,
        ReactDeps[ReactOutputT, ReactDepsT],
        AgentRunResult[ReactOutputT],
    ],
):
    deferred_results: DeferredToolResults

    async def run(
        self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
    ) -> RunAgent[ReactOutputT, ReactDepsT]:
        if not self.deferred_results.calls and not self.deferred_results.approvals:
            raise ValueError(
                "ResumeTurn requires at least one resolved call or approval"
            )
        return RunAgent(deferred_results=self.deferred_results)

deferred_results instance-attribute

deferred_results: DeferredToolResults

run async

run(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
) -> RunAgent[ReactOutputT, ReactDepsT]

Source code in octomate/capabilities/harness/react.py

async def run(
    self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
) -> RunAgent[ReactOutputT, ReactDepsT]:
    if not self.deferred_results.calls and not self.deferred_results.approvals:
        raise ValueError(
            "ResumeTurn requires at least one resolved call or approval"
        )
    return RunAgent(deferred_results=self.deferred_results)

RunAgent dataclass

Bases: BaseNode[ReactState, ReactDeps[ReactOutputT, ReactDepsT], AgentRunResult[ReactOutputT]]

Source code in octomate/capabilities/harness/react.py

@dataclass
class RunAgent[ReactOutputT, ReactDepsT](
    BaseNode[
        ReactState,
        ReactDeps[ReactOutputT, ReactDepsT],
        AgentRunResult[ReactOutputT],
    ],
):
    user_prompt: str | Sequence[UserContent] | None = None
    deferred_results: DeferredToolResults | None = None

    async def run(
        self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
    ) -> ResolveDeferred[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
        with react_logfire.span(
            "react.agent_run",
            run_name=ctx.deps.run_name,
            streaming=ctx.deps.event_send_stream is not None,
            conversation_address=str(ctx.state.conversation_address),
            resumed=self.deferred_results is not None,
        ) as span:
            conversation = await resolve_conversation(ctx)
            persistence = RunPersistence(
                conversation_manager=ctx.deps.conversation_manager,
                thread_manager=ctx.deps.thread_manager,
                conversation=conversation,
                state=ctx.state,
                run_name=ctx.deps.run_name,
                cwd=ctx.deps.cwd,
                binds_prompt_sources=(
                    self.user_prompt is not None or self.deferred_results is None
                ),
                permission_mode=conversation.permission_mode
                or ctx.deps.permission_mode,
            )
            capabilities = [
                (
                    PersistRunFailure(
                        persistence=persistence,
                        previous_message_count=len(conversation.messages),
                    )
                    if ctx.deps.event_send_stream is None
                    else PersistStreamRunFailure(
                        persistence=persistence,
                        previous_message_count=len(conversation.messages),
                    )
                ),
                *(ctx.deps.capabilities or []),
            ]
            if ctx.deps.event_send_stream is None:
                # builtin_tools and output_retries are deprecated run kwargs in
                # pydantic-ai 1.x: native tools register as NativeTool capabilities,
                # and the output-retry budget moves under retries={"output": ...}.
                # (The stream_events path below still takes the deprecated-shaped
                # kwargs; Agent.stream_events translates them internally.)
                retries = (
                    AgentRetries(output=ctx.deps.output_retries)
                    if ctx.deps.output_retries is not None
                    else None
                )
                if ctx.deps.builtin_tools:
                    capabilities = [
                        *capabilities,
                        *(NativeTool(tool) for tool in ctx.deps.builtin_tools),
                    ]
                result = await ctx.deps.agent.run(
                    self.user_prompt,
                    output_type=ctx.deps.output_type,
                    message_history=conversation.messages or None,
                    deferred_tool_results=self.deferred_results,
                    conversation_id=str(conversation.id),
                    model=ctx.deps.model,
                    instructions=ctx.deps.instructions,
                    deps=ctx.deps.agent_deps,
                    model_settings=ctx.deps.model_settings,
                    usage_limits=ctx.deps.usage_limits,
                    usage=ctx.deps.usage,
                    metadata=ctx.deps.metadata,
                    retries=retries,
                    infer_name=ctx.deps.infer_name,
                    toolsets=ctx.deps.toolsets,
                    capabilities=capabilities,
                    spec=ctx.deps.spec,
                )
                return await self.next_node(ctx, result, persistence, span)

            result: AgentRunResult[ReactOutputT] | None = None
            # stream_events (the normalizer) instead of run_stream_events: thinking +
            # tool/text events pass through raw, while structured segment replies
            # surface as ResultSegmentEvent for the consumer to render.
            async for event in ctx.deps.agent.stream_events(
                self.user_prompt,
                output_type=ctx.deps.output_type,
                message_history=conversation.messages or None,
                deferred_tool_results=self.deferred_results,
                conversation_id=str(conversation.id),
                model=ctx.deps.model,
                instructions=ctx.deps.instructions,
                deps=ctx.deps.agent_deps,
                model_settings=ctx.deps.model_settings,
                usage_limits=ctx.deps.usage_limits,
                usage=ctx.deps.usage,
                metadata=ctx.deps.metadata,
                output_retries=ctx.deps.output_retries,
                infer_name=ctx.deps.infer_name,
                toolsets=ctx.deps.toolsets,
                builtin_tools=ctx.deps.builtin_tools,
                capabilities=capabilities,
                spec=ctx.deps.spec,
            ):
                if ctx.deps.event_send_stream is not None:
                    await ctx.deps.event_send_stream.send(event)
                if isinstance(event, AgentRunResultEvent):
                    result = event.result

            if result is None:
                raise RuntimeError(
                    "agent.stream_events did not yield AgentRunResultEvent"
                )
            return await self.next_node(ctx, result, persistence, span)

    async def next_node(
        self,
        ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]],
        result: AgentRunResult[ReactOutputT],
        persistence: RunPersistence,
        span: logfire.LogfireSpan,
    ) -> ResolveDeferred[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
        new_messages = result.new_messages()
        span.set_attribute("react.run_id", result.run_id)
        span.set_attribute(
            "react.deferred", isinstance(result.output, DeferredToolRequests)
        )
        span.set_attribute("react.new_messages", len(new_messages))
        if new_messages:
            # Recording persists the turn, so the next RunAgent's ensure() picks
            # it up from the manager — no copy in state. Only the prompt turn
            # binds source messages; deferred resumes carry no new user request.
            await persistence.record(
                result.run_id, new_messages, model_name=result.response.model_name
            )

        if isinstance(result.output, DeferredToolRequests) and (
            ctx.deps.choose_resolvers is not None or ctx.deps.suspender is not None
        ):
            react_logfire.info("react run deferred", run_id=result.run_id)
            return ResolveDeferred(requests=result.output, result=result)
        return End(result)

user_prompt class-attribute instance-attribute

user_prompt: str | Sequence[UserContent] | None = None

deferred_results class-attribute instance-attribute

deferred_results: DeferredToolResults | None = None

run async

run(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
) -> (
    ResolveDeferred[ReactOutputT, ReactDepsT]
    | End[AgentRunResult[ReactOutputT]]
)

Source code in octomate/capabilities/harness/react.py

async def run(
    self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
) -> ResolveDeferred[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
    with react_logfire.span(
        "react.agent_run",
        run_name=ctx.deps.run_name,
        streaming=ctx.deps.event_send_stream is not None,
        conversation_address=str(ctx.state.conversation_address),
        resumed=self.deferred_results is not None,
    ) as span:
        conversation = await resolve_conversation(ctx)
        persistence = RunPersistence(
            conversation_manager=ctx.deps.conversation_manager,
            thread_manager=ctx.deps.thread_manager,
            conversation=conversation,
            state=ctx.state,
            run_name=ctx.deps.run_name,
            cwd=ctx.deps.cwd,
            binds_prompt_sources=(
                self.user_prompt is not None or self.deferred_results is None
            ),
            permission_mode=conversation.permission_mode
            or ctx.deps.permission_mode,
        )
        capabilities = [
            (
                PersistRunFailure(
                    persistence=persistence,
                    previous_message_count=len(conversation.messages),
                )
                if ctx.deps.event_send_stream is None
                else PersistStreamRunFailure(
                    persistence=persistence,
                    previous_message_count=len(conversation.messages),
                )
            ),
            *(ctx.deps.capabilities or []),
        ]
        if ctx.deps.event_send_stream is None:
            # builtin_tools and output_retries are deprecated run kwargs in
            # pydantic-ai 1.x: native tools register as NativeTool capabilities,
            # and the output-retry budget moves under retries={"output": ...}.
            # (The stream_events path below still takes the deprecated-shaped
            # kwargs; Agent.stream_events translates them internally.)
            retries = (
                AgentRetries(output=ctx.deps.output_retries)
                if ctx.deps.output_retries is not None
                else None
            )
            if ctx.deps.builtin_tools:
                capabilities = [
                    *capabilities,
                    *(NativeTool(tool) for tool in ctx.deps.builtin_tools),
                ]
            result = await ctx.deps.agent.run(
                self.user_prompt,
                output_type=ctx.deps.output_type,
                message_history=conversation.messages or None,
                deferred_tool_results=self.deferred_results,
                conversation_id=str(conversation.id),
                model=ctx.deps.model,
                instructions=ctx.deps.instructions,
                deps=ctx.deps.agent_deps,
                model_settings=ctx.deps.model_settings,
                usage_limits=ctx.deps.usage_limits,
                usage=ctx.deps.usage,
                metadata=ctx.deps.metadata,
                retries=retries,
                infer_name=ctx.deps.infer_name,
                toolsets=ctx.deps.toolsets,
                capabilities=capabilities,
                spec=ctx.deps.spec,
            )
            return await self.next_node(ctx, result, persistence, span)

        result: AgentRunResult[ReactOutputT] | None = None
        # stream_events (the normalizer) instead of run_stream_events: thinking +
        # tool/text events pass through raw, while structured segment replies
        # surface as ResultSegmentEvent for the consumer to render.
        async for event in ctx.deps.agent.stream_events(
            self.user_prompt,
            output_type=ctx.deps.output_type,
            message_history=conversation.messages or None,
            deferred_tool_results=self.deferred_results,
            conversation_id=str(conversation.id),
            model=ctx.deps.model,
            instructions=ctx.deps.instructions,
            deps=ctx.deps.agent_deps,
            model_settings=ctx.deps.model_settings,
            usage_limits=ctx.deps.usage_limits,
            usage=ctx.deps.usage,
            metadata=ctx.deps.metadata,
            output_retries=ctx.deps.output_retries,
            infer_name=ctx.deps.infer_name,
            toolsets=ctx.deps.toolsets,
            builtin_tools=ctx.deps.builtin_tools,
            capabilities=capabilities,
            spec=ctx.deps.spec,
        ):
            if ctx.deps.event_send_stream is not None:
                await ctx.deps.event_send_stream.send(event)
            if isinstance(event, AgentRunResultEvent):
                result = event.result

        if result is None:
            raise RuntimeError(
                "agent.stream_events did not yield AgentRunResultEvent"
            )
        return await self.next_node(ctx, result, persistence, span)

next_node async

next_node(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
    result: AgentRunResult[ReactOutputT],
    persistence: RunPersistence,
    span: LogfireSpan,
) -> (
    ResolveDeferred[ReactOutputT, ReactDepsT]
    | End[AgentRunResult[ReactOutputT]]
)

Source code in octomate/capabilities/harness/react.py

async def next_node(
    self,
    ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]],
    result: AgentRunResult[ReactOutputT],
    persistence: RunPersistence,
    span: logfire.LogfireSpan,
) -> ResolveDeferred[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
    new_messages = result.new_messages()
    span.set_attribute("react.run_id", result.run_id)
    span.set_attribute(
        "react.deferred", isinstance(result.output, DeferredToolRequests)
    )
    span.set_attribute("react.new_messages", len(new_messages))
    if new_messages:
        # Recording persists the turn, so the next RunAgent's ensure() picks
        # it up from the manager — no copy in state. Only the prompt turn
        # binds source messages; deferred resumes carry no new user request.
        await persistence.record(
            result.run_id, new_messages, model_name=result.response.model_name
        )

    if isinstance(result.output, DeferredToolRequests) and (
        ctx.deps.choose_resolvers is not None or ctx.deps.suspender is not None
    ):
        react_logfire.info("react run deferred", run_id=result.run_id)
        return ResolveDeferred(requests=result.output, result=result)
    return End(result)

ResolveDeferred dataclass

Bases: BaseNode[ReactState, ReactDeps[ReactOutputT, ReactDepsT], AgentRunResult[ReactOutputT]]

Source code in octomate/capabilities/harness/react.py

@dataclass
class ResolveDeferred[ReactOutputT, ReactDepsT](
    BaseNode[
        ReactState,
        ReactDeps[ReactOutputT, ReactDepsT],
        AgentRunResult[ReactOutputT],
    ],
):
    requests: DeferredToolRequests
    result: AgentRunResult[ReactOutputT]

    async def run(
        self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
    ) -> RunAgent[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
        if ctx.deps.choose_resolvers is None and ctx.deps.suspender is None:
            raise RuntimeError(
                "ResolveDeferred requires a react graph resolver or suspender"
            )
        if ctx.deps.choose_resolvers is not None:
            # Asked here, at the deferral, rather than carried from the start of the
            # run: the conversation decides who answers this, and it can have been
            # told something new since the last time round.
            resolvers = ctx.deps.choose_resolvers(await resolve_conversation(ctx))
            results = DeferredToolResults()
            for resolver in resolvers:
                answered = await resolver.resolve(self.requests)
                # Earlier resolvers win — each answers what it speaks for and a later
                # one only fills what is still open, which is what puts a catch-all
                # last. Each sees the batch entire rather than the remainder, so one
                # that classifies by `metadata` keeps what it classifies by.
                for tool_call_id, value in answered.calls.items():
                    results.calls.setdefault(tool_call_id, value)
                for tool_call_id, verdict in answered.approvals.items():
                    results.approvals.setdefault(tool_call_id, verdict)
            # All of the batch or none of it: the runner refuses results covering only
            # some of what it deferred, so a batch nobody wholly answered is the
            # human's — along with the parts of it that were answerable.
            covered = all(
                call.tool_call_id in results.calls for call in self.requests.calls
            ) and all(
                approval.tool_call_id in results.approvals
                for approval in self.requests.approvals
            )
            if resolvers and covered:
                react_logfire.info(
                    "deferred resolved in-process, looping back to RunAgent"
                )
                return RunAgent(deferred_results=results)
        if ctx.deps.suspender is not None:
            react_logfire.info("deferred suspended, ending run")
            event = await ctx.deps.suspender.suspend(self.requests)
            if event is not None and ctx.deps.event_send_stream is not None:
                await ctx.deps.event_send_stream.send(event)
            return End(self.result)
        # Nothing in-process took it and there is no human to ask, so the requests go
        # back to the caller — the same end a graph carrying neither hook reaches.
        return End(self.result)

requests instance-attribute

requests: DeferredToolRequests

result instance-attribute

result: AgentRunResult[ReactOutputT]

run async

run(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
) -> (
    RunAgent[ReactOutputT, ReactDepsT]
    | End[AgentRunResult[ReactOutputT]]
)

Source code in octomate/capabilities/harness/react.py

async def run(
    self, ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]]
) -> RunAgent[ReactOutputT, ReactDepsT] | End[AgentRunResult[ReactOutputT]]:
    if ctx.deps.choose_resolvers is None and ctx.deps.suspender is None:
        raise RuntimeError(
            "ResolveDeferred requires a react graph resolver or suspender"
        )
    if ctx.deps.choose_resolvers is not None:
        # Asked here, at the deferral, rather than carried from the start of the
        # run: the conversation decides who answers this, and it can have been
        # told something new since the last time round.
        resolvers = ctx.deps.choose_resolvers(await resolve_conversation(ctx))
        results = DeferredToolResults()
        for resolver in resolvers:
            answered = await resolver.resolve(self.requests)
            # Earlier resolvers win — each answers what it speaks for and a later
            # one only fills what is still open, which is what puts a catch-all
            # last. Each sees the batch entire rather than the remainder, so one
            # that classifies by `metadata` keeps what it classifies by.
            for tool_call_id, value in answered.calls.items():
                results.calls.setdefault(tool_call_id, value)
            for tool_call_id, verdict in answered.approvals.items():
                results.approvals.setdefault(tool_call_id, verdict)
        # All of the batch or none of it: the runner refuses results covering only
        # some of what it deferred, so a batch nobody wholly answered is the
        # human's — along with the parts of it that were answerable.
        covered = all(
            call.tool_call_id in results.calls for call in self.requests.calls
        ) and all(
            approval.tool_call_id in results.approvals
            for approval in self.requests.approvals
        )
        if resolvers and covered:
            react_logfire.info(
                "deferred resolved in-process, looping back to RunAgent"
            )
            return RunAgent(deferred_results=results)
    if ctx.deps.suspender is not None:
        react_logfire.info("deferred suspended, ending run")
        event = await ctx.deps.suspender.suspend(self.requests)
        if event is not None and ctx.deps.event_send_stream is not None:
            await ctx.deps.event_send_stream.send(event)
        return End(self.result)
    # Nothing in-process took it and there is no human to ask, so the requests go
    # back to the caller — the same end a graph carrying neither hook reaches.
    return End(self.result)

ReactEventStream

Deterministic-cleanup handle over a react event stream: entering the context yields the underlying generator, exiting closes it. Mirrors pydantic-ai's AgentEventStream, but typed over ReactStreamEvent — a react run also streams octomate output/display/action-batch events.

Source code in octomate/capabilities/harness/react.py

class ReactEventStream[ReactOutputCoT]:
    """Deterministic-cleanup handle over a react event stream: entering the
    context yields the underlying generator, exiting closes it. Mirrors
    pydantic-ai's ``AgentEventStream``, but typed over ``ReactStreamEvent`` — a
    react run also streams octomate output/display/action-batch events."""

    # Private so the covariant type parameter has no mutable public surface.
    def __init__(
        self, generator: AsyncGenerator[ReactStreamEvent[ReactOutputCoT], None]
    ) -> None:
        self._generator = generator

    async def __aenter__(
        self,
    ) -> AsyncGenerator[ReactStreamEvent[ReactOutputCoT], None]:
        return self._generator

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self._generator.aclose()

resolve_conversation async

resolve_conversation(
    ctx: GraphRunContext[
        ReactState, ReactDeps[ReactOutputT, ReactDepsT]
    ],
) -> Conversation

The node's conversation: the pre-ensured one when the run is addressed by id, otherwise the agent's own (thread, agent) conversation.

Source code in octomate/capabilities/harness/react.py

async def resolve_conversation[ReactOutputT, ReactDepsT](
    ctx: GraphRunContext[ReactState, ReactDeps[ReactOutputT, ReactDepsT]],
) -> Conversation:
    """The node's conversation: the pre-ensured one when the run is addressed
    by id, otherwise the agent's own (thread, agent) conversation."""
    if ctx.state.conversation_id is not None:
        conversation = await ctx.deps.conversation_manager.get(
            ctx.state.conversation_id
        )
        if (
            conversation.agent_tentacle_id != ctx.state.agent_tentacle_id
            or conversation.thread_id != ctx.state.thread_id
        ):
            raise ValueError(
                f"conversation {ctx.state.conversation_id} does not belong to "
                f"({ctx.state.agent_tentacle_id!r}, {ctx.state.thread_id})"
            )
        return conversation
    return await ctx.deps.conversation_manager.ensure(
        ctx.state.thread_id,
        agent_tentacle_id=ctx.state.agent_tentacle_id,
    )

build_react_graph

build_react_graph(
    start_node: ReactGraphInput[ReactOutputT, ReactDepsT],
) -> Graph[
    ReactState,
    ReactDeps[ReactOutputT, ReactDepsT],
    ReactGraphInput[ReactOutputT, ReactDepsT],
    AgentRunResult[ReactOutputT],
]

Wire the react nodes into a runnable graph for start_node to run through.

The node it will be run with is what binds the graph's type parameters — the output and deps types live only in annotations, so there is nothing else here to infer them from.

Source code in octomate/capabilities/harness/react.py

def build_react_graph[ReactOutputT, ReactDepsT](
    start_node: ReactGraphInput[ReactOutputT, ReactDepsT],
) -> Graph[
    ReactState,
    ReactDeps[ReactOutputT, ReactDepsT],
    ReactGraphInput[ReactOutputT, ReactDepsT],
    AgentRunResult[ReactOutputT],
]:
    """Wire the react nodes into a runnable graph for `start_node` to run through.

    The node it will be run with is what binds the graph's type parameters — the
    output and deps types live only in annotations, so there is nothing else here
    to infer them from.
    """
    builder = GraphBuilder(
        name="react",
        state_type=ReactState,
        deps_type=ReactDeps[ReactOutputT, ReactDepsT],
        input_type=TypeExpression[ReactGraphInput[ReactOutputT, ReactDepsT]],
        output_type=AgentRunResult[ReactOutputT],
    )
    # Every other edge comes from the nodes' own `run` return annotations, so a
    # transition stays declared where it is written. The one thing those cannot
    # express is where a run *starts* — a turn either begins fresh or resumes
    # deferred results — so the entry is a decision on the input's own type.
    entry = (
        builder.decision(note="a fresh turn, or one resuming deferred results")
        .branch(builder.match(StartTurn).to(StartTurn))
        .branch(builder.match(ResumeTurn).to(ResumeTurn))
    )
    builder.add(
        builder.edge_from(builder.start_node).to(entry),
        builder.node(StartTurn[ReactOutputT, ReactDepsT]),
        builder.node(ResumeTurn[ReactOutputT, ReactDepsT]),
        builder.node(RunAgent[ReactOutputT, ReactDepsT]),
        builder.node(ResolveDeferred[ReactOutputT, ReactDepsT]),
    )
    return builder.build()

iter_react_graph_events async

iter_react_graph_events(
    start_node: StartTurn[ReactOutputT, ReactDepsT]
    | ResumeTurn[ReactOutputT, ReactDepsT],
    *,
    state: ReactState,
    deps: ReactDeps[ReactOutputT, ReactDepsT],
) -> AsyncGenerator[ReactStreamEvent[ReactOutputT], None]

Source code in octomate/capabilities/harness/react.py

async def iter_react_graph_events[ReactOutputT, ReactDepsT](
    start_node: StartTurn[ReactOutputT, ReactDepsT]
    | ResumeTurn[ReactOutputT, ReactDepsT],
    *,
    state: ReactState,
    deps: ReactDeps[ReactOutputT, ReactDepsT],
) -> AsyncGenerator[ReactStreamEvent[ReactOutputT], None]:
    send_stream, receive_stream = anyio.create_memory_object_stream[
        ReactStreamEvent[ReactOutputT]
    ](100)
    graph_deps = replace(deps, event_send_stream=send_stream)
    captured: list[Exception] = []

    async def run_graph() -> None:
        # Capture the graph error instead of letting it escape: a raising child
        # task trips the task group's cancel scope and cancels the consumer
        # mid-event (e.g. blocked on a channel send), which masks the real error
        # and leaves spans unclosed. Closing send_stream ends the consumer loop
        # cleanly; we re-raise below, in the consumer's own frame.
        graph = build_react_graph(start_node)
        try:
            async with send_stream:
                await graph.run(
                    inputs=start_node,
                    state=state,
                    deps=graph_deps,
                )
        except Exception as exc:
            captured.append(exc)

    async with receive_stream:
        async with anyio.create_task_group() as tg:
            tg.start_soon(run_graph)
            async for event in receive_stream:
                yield event
    for error in captured:
        raise error

octomate.capabilities.harness.deferred

Deferred-tool resolution protocols — harness interfaces for the react loop.

Two complementary hooks (keep both):

• DeferredResolver: resolve deferred calls in-process and let the loop continue.
• DeferredSuspender: persist + present the requests out-of-band; the run ends and is resumed later by feeding the collected results back through a fresh run.

Which of the two a deferral gets is ResolverChoice's answer, and the loop asks it anew every time round rather than once per run.

The concrete, channel-coupled suspender (ReflexSuspender) lives in the triage layer (octomate.reflex.suspender) so the harness never imports channels.

Interjection

Interjection = ActionBatchEvent | ActionBatchAnsweredEvent

DeferredResolver

Bases: Protocol

Resolve deferred tool calls in-process so the react loop can continue.

A resolver answers what it speaks for and leaves the rest alone — a posture has an opinion about approvals and questions, and none at all about a teleport. It is not the resolver's job to notice that it covered only part of the batch: ResolveDeferred is where that is noticed, once, for whatever ran. The runner refuses results covering only some of the calls it deferred, so a batch nobody wholly answered goes to the suspender entire.

Source code in octomate/capabilities/harness/deferred.py

class DeferredResolver(Protocol):
    """Resolve deferred tool calls in-process so the react loop can continue.

    A resolver answers what it speaks for and leaves the rest alone — a posture has
    an opinion about approvals and questions, and none at all about a `teleport`. It
    is not the resolver's job to notice that it covered only part of the batch:
    `ResolveDeferred` is where that is noticed, once, for whatever ran. The runner
    refuses results covering only some of the calls it deferred, so a batch nobody
    wholly answered goes to the suspender entire.
    """

    async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults: ...

resolve async

resolve(
    requests: DeferredToolRequests,
) -> DeferredToolResults

Source code in octomate/capabilities/harness/deferred.py

async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults: ...

ResolverChoice

Bases: Protocol

Choose the resolvers for one deferral, from the conversation as it stands at that moment.

Asked at every deferral rather than once per run, because what may be resolved without a human is the conversation's approval posture — and that is switched while runs are going. Composed once, a switch would only reach the run after the one it was made during, which is the turn nobody was watching.

Several, in order, because one batch can hold deferrals of different kinds and no single policy speaks for all of them: each resolver answers its own and earlier ones win, so a catch-all goes last. An empty sequence means nothing answers this batch in-process — it goes to the suspender, and back to the caller when there is none.

Source code in octomate/capabilities/harness/deferred.py

class ResolverChoice(Protocol):
    """Choose the resolvers for one deferral, from the conversation as it stands at
    that moment.

    Asked at every deferral rather than once per run, because what may be resolved
    without a human is the conversation's approval posture — and that is switched
    while runs are going. Composed once, a switch would only reach the run after the
    one it was made during, which is the turn nobody was watching.

    Several, in order, because one batch can hold deferrals of different kinds and no
    single policy speaks for all of them: each resolver answers its own and earlier
    ones win, so a catch-all goes last. An empty sequence means nothing answers this
    batch in-process — it goes to the suspender, and back to the caller when there is
    none.
    """

    # Positional-only, so an implementation is free to name the conversation
    # whatever reads best — a callable protocol otherwise pins the parameter's name.
    def __call__(self, conversation: Conversation, /) -> Sequence[DeferredResolver]: ...

DeclineResolver dataclass

Resolves every deferred action by declining it immediately — the catch-all that closes a non-interactive run's chain (e.g. a commissioned accomplice). There is no human to ask, so approvals are denied and asks are answered with the decline, and the react loop continues to a final answer in-process instead of parking anything for review.

Last in the chain and answering everything, which is what guarantees a batch is covered: a run with no human has no suspender either, so anything left open would end the run holding requests nobody is coming to answer. Whatever ran before it has already had its say — earlier answers stand.

Source code in octomate/capabilities/harness/deferred.py

@dataclass
class DeclineResolver:
    """Resolves every deferred action by declining it immediately — the catch-all
    that closes a non-interactive run's chain (e.g. a commissioned accomplice). There
    is no human to ask, so approvals are denied and asks are answered with the
    decline, and the react loop continues to a final answer in-process instead
    of parking anything for review.

    Last in the chain and answering everything, which is what guarantees a batch is
    covered: a run with no human has no suspender either, so anything left open would
    end the run holding requests nobody is coming to answer. Whatever ran before it
    has already had its say — earlier answers stand."""

    message: str = (
        "Declined: this run has no user to ask. Proceed on your best judgment "
        "and state the assumption in your report."
    )

    async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults:
        results = DeferredToolResults()
        for approval in requests.approvals:
            results.approvals[approval.tool_call_id] = ToolDenied(self.message)
        for call in requests.calls:
            results.calls[call.tool_call_id] = self.message
        return results

message class-attribute instance-attribute

message: str = "Declined: this run has no user to ask. Proceed on your best judgment and state the assumption in your report."

resolve async

resolve(
    requests: DeferredToolRequests,
) -> DeferredToolResults

Source code in octomate/capabilities/harness/deferred.py

async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults:
    results = DeferredToolResults()
    for approval in requests.approvals:
        results.approvals[approval.tool_call_id] = ToolDenied(self.message)
    for call in requests.calls:
        results.calls[call.tool_call_id] = self.message
    return results

PostureResolver dataclass

Answers the two things an approval posture speaks about, and nothing else: the approvals it gates, and the questions it says not to put to a human.

What it walks past is the point. A teleport is deferred the same way a question is, but it is neither an approval nor a question — no posture has an opinion on it, and the suspender is what performs one. Left unanswered here, the batch is not wholly covered and all of it goes to the human instead. A deferral arriving with a kind this does not know goes the same way, so an unrecognized one reaches a human rather than being answered on their behalf.

Source code in octomate/capabilities/harness/deferred.py

@dataclass(frozen=True)
class PostureResolver:
    """Answers the two things an approval posture speaks about, and nothing else:
    the approvals it gates, and the questions it says not to put to a human.

    What it walks past is the point. A `teleport` is deferred the same way a question
    is, but it is neither an approval nor a question — no posture has an opinion on
    it, and the suspender is what performs one. Left unanswered here, the batch is
    not wholly covered and all of it goes to the human instead. A deferral arriving
    with a kind this does not know goes the same way, so an unrecognized one reaches
    a human rather than being answered on their behalf.
    """

    # Whether an approval is granted outright, or denied with `message` as the reason.
    approve: bool
    # What the agent is told in place of the answer it will not be given — a denial's
    # reason, and an unasked question's return.
    message: str

    async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults:
        results = DeferredToolResults()
        for call in requests.calls:
            if (
                requests.metadata.get(call.tool_call_id, {}).get("kind")
                == ASK_DEFERR_KIND
            ):
                results.calls[call.tool_call_id] = self.message
        for approval in requests.approvals:
            results.approvals[approval.tool_call_id] = (
                True if self.approve else ToolDenied(self.message)
            )
        return results

approve instance-attribute

approve: bool

message instance-attribute

message: str

resolve async

resolve(
    requests: DeferredToolRequests,
) -> DeferredToolResults

Source code in octomate/capabilities/harness/deferred.py

async def resolve(self, requests: DeferredToolRequests) -> DeferredToolResults:
    results = DeferredToolResults()
    for call in requests.calls:
        if (
            requests.metadata.get(call.tool_call_id, {}).get("kind")
            == ASK_DEFERR_KIND
        ):
            results.calls[call.tool_call_id] = self.message
    for approval in requests.approvals:
        results.approvals[approval.tool_call_id] = (
            True if self.approve else ToolDenied(self.message)
        )
    return results

DeferredSuspender

Bases: Protocol

Human-in-the-loop hook for deferred tool calls the agent cannot resolve in process. Unlike DeferredResolver, which returns results and lets the react loop continue, a suspender persists + presents the requests for an out-of-band response and the run ends; it is resumed later by feeding the collected DeferredToolResults back through a fresh agent run.

Returning an ActionBatchEvent asks the react loop to put the batch on the stream, for whoever draws the run to present; returning None means there is nothing to present, as for a deferral the graph performs itself.

pause is the same for a run that stays live while a human answers, as a runtime's in-process approval does: the event of the batch it waits on, for the run's own stream. The caller names the batch, so it is already waiting on that id when the cards go up and a quick reply finds it.

Source code in octomate/capabilities/harness/deferred.py

class DeferredSuspender(Protocol):
    """Human-in-the-loop hook for deferred tool calls the agent cannot resolve
    in process. Unlike `DeferredResolver`, which returns results and lets the
    react loop continue, a suspender persists + presents the requests for an
    out-of-band response and the run ends; it is resumed later by feeding the
    collected `DeferredToolResults` back through a fresh agent run.

    Returning an `ActionBatchEvent` asks the react loop to put the batch *on the
    stream*, for whoever draws the run to present; returning `None` means there
    is nothing to present, as for a deferral the graph performs itself.

    `pause` is the same for a run that stays live while a human answers, as a
    runtime's in-process approval does: the event of the batch it waits on, for
    the run's own stream. The caller names the batch, so it is already waiting on
    that id when the cards go up and a quick reply finds it.
    """

    async def suspend(
        self, requests: DeferredToolRequests
    ) -> ActionBatchEvent | None: ...

    async def pause(
        self, requests: DeferredToolRequests, *, batch_id: UUID7
    ) -> ActionBatchEvent: ...

suspend async

suspend(
    requests: DeferredToolRequests,
) -> ActionBatchEvent | None

Source code in octomate/capabilities/harness/deferred.py

async def suspend(
    self, requests: DeferredToolRequests
) -> ActionBatchEvent | None: ...

pause async

pause(
    requests: DeferredToolRequests, *, batch_id: UUID7
) -> ActionBatchEvent

Source code in octomate/capabilities/harness/deferred.py

async def pause(
    self, requests: DeferredToolRequests, *, batch_id: UUID7
) -> ActionBatchEvent: ...

Read dataclass

One item of a runtime's own stream.

Source code in octomate/capabilities/harness/deferred.py

@dataclass(frozen=True)
class Read[T]:
    """One item of a runtime's own stream."""

    item: T

item instance-attribute

item: T

Ended dataclass

A runtime's own stream is over, or failed with error.

Source code in octomate/capabilities/harness/deferred.py

@dataclass(frozen=True)
class Ended:
    """A runtime's own stream is over, or failed with `error`."""

    error: Exception | None = None

error class-attribute instance-attribute

error: Exception | None = None

Interjections

What a live run's callbacks put on its stream while it waits on the runtime: the batch an approval pauses the run on, and the answer it resumes on, for whoever draws the run.

The runtime's own stream is read in a task of its own, so a batch gets through while the runtime is still waiting for its answer.

Source code in octomate/capabilities/harness/deferred.py

class Interjections[T]:
    """What a live run's callbacks put on its stream while it waits on the runtime:
    the batch an approval pauses the run on, and the answer it resumes on, for
    whoever draws the run.

    The runtime's own stream is read in a task of its own, so a batch gets through
    while the runtime is still waiting for its answer."""

    def __init__(self) -> None:
        self.queue: asyncio.Queue[Read[T] | Interjection | Ended] = asyncio.Queue()

    def interject(self, event: Interjection) -> None:
        self.queue.put_nowait(event)

    async def around(
        self, source: AsyncIterable[T]
    ) -> AsyncGenerator[T | Interjection]:
        """`source`'s items in order, with each interjection as it is made."""
        reading = asyncio.create_task(self.read(source))
        try:
            while not isinstance(entry := await self.queue.get(), Ended):
                yield entry.item if isinstance(entry, Read) else entry
            if entry.error is not None:
                raise entry.error
        finally:
            reading.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await reading

    async def read(self, source: AsyncIterable[T]) -> None:
        try:
            async for item in source:
                self.queue.put_nowait(Read(item))
        except Exception as error:
            self.queue.put_nowait(Ended(error))
        else:
            self.queue.put_nowait(Ended())

queue instance-attribute

queue: Queue[Read[T] | Interjection | Ended] = (
    asyncio.Queue()
)

interject

interject(event: Interjection) -> None

Source code in octomate/capabilities/harness/deferred.py

def interject(self, event: Interjection) -> None:
    self.queue.put_nowait(event)

around async

around(
    source: AsyncIterable[T],
) -> AsyncGenerator[T | Interjection]

source's items in order, with each interjection as it is made.

Source code in octomate/capabilities/harness/deferred.py

async def around(
    self, source: AsyncIterable[T]
) -> AsyncGenerator[T | Interjection]:
    """`source`'s items in order, with each interjection as it is made."""
    reading = asyncio.create_task(self.read(source))
    try:
        while not isinstance(entry := await self.queue.get(), Ended):
            yield entry.item if isinstance(entry, Read) else entry
        if entry.error is not None:
            raise entry.error
    finally:
        reading.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await reading

read async

read(source: AsyncIterable[T]) -> None

Source code in octomate/capabilities/harness/deferred.py

async def read(self, source: AsyncIterable[T]) -> None:
    try:
        async for item in source:
            self.queue.put_nowait(Read(item))
    except Exception as error:
        self.queue.put_nowait(Ended(error))
    else:
        self.queue.put_nowait(Ended())