Channel API

Channels are independently installable transports for one persistent agent identity. See Channels and Listening for lifecycle, delivery semantics, and an extension example.

nagents.channels.Channel

Bases: ABC

A trusted connector. Installations remain explicit; imports do not connect.

open initializes resources, listen delivers events with backpressure, and close releases resources after producers/execution have stopped. Transport acknowledgements belong to the connector. Agent responses are emitted only through explicit send/action tools, never auto-forwarded. actions supplies connector operations through the existing channel_list, channel_send and channel_action tools. on_event is optional execution observation for status notices, independent of tools and typing activity.

Source code in src/nagents/channels/types.py

class Channel(ABC):
    """A trusted connector. Installations remain explicit; imports do not connect.

    ``open`` initializes resources, ``listen`` delivers events with backpressure,
    and ``close`` releases resources after producers/execution have stopped.
    Transport acknowledgements belong to the connector. Agent responses are
    emitted only through explicit send/action tools, never auto-forwarded.
    ``actions`` supplies connector operations through the existing channel_list,
    channel_send and channel_action tools. ``on_event`` is optional execution
    observation for status notices, independent of tools and typing activity.
    """

    name: str
    description: str = ""
    capabilities: tuple[str, ...] = ("receive", "send_text")
    content_capabilities: ChannelContentCapabilities | None = None
    actions: tuple[ChannelAction, ...] = ()

    async def open(self) -> None:
        return None

    @abstractmethod
    async def listen(self, receive: ChannelReceiver) -> None:
        """Deliver events until cancelled, or return when a finite source ends."""

    @abstractmethod
    async def send(self, message: ChannelSend) -> ChannelDelivery:
        """Send once; return confirmed IDs or raise a sanitized ChannelError."""

    async def action(self, name: str, arguments: dict[str, ChannelValue]) -> dict[str, ChannelValue]:
        raise ChannelError(f"Unsupported channel action: {name}")

    async def fetch_attachment(self, attachment: ChannelAttachment) -> tuple[bytes, str]:
        """Download a referenced attachment for model input.

        Only connectors that advertise ``fetch_attachment`` in ``capabilities``
        need to implement this. Return the raw bytes and the effective media type
        within the host's caps; raise a sanitized ``ChannelError`` otherwise. The
        host never treats the returned content as instructions.
        """
        raise ChannelError("This connector does not support attachment downloads")

    def command(self, message: ChannelMessage) -> ChannelCommand | None:
        """Recognize an explicit transport command without doing I/O."""
        return None

    def approval(self, message: ChannelMessage) -> ChannelApproval | None:
        """Recognize a decision for a rendered approval prompt, without doing I/O.

        Only connectors that render approval prompts and advertise ``approvals``
        implement this. Returning a value claims the interaction: the host either
        applies a fully validated decision or ignores it, and never enqueues it as
        model input. Returning ``None`` leaves the message to normal handling.
        """
        return None

    async def activity(self, event: ChannelActivity) -> None:
        """Start/stop an optional typing indicator; close() must stop keepalives."""
        return None

    async def on_event(self, event: ChannelExecutionEvent) -> None:
        """Optionally render a source-scoped status notice; default does nothing.

        Cooperate with cancellation and bound transport I/O. Hosts should await
        dispatch_channel_execution_event for failure isolation and joined,
        bounded dispatch. Never retry an uncertain notice send automatically.
        """
        return None

    async def close(self) -> None:
        return None

activity(event) async

Start/stop an optional typing indicator; close() must stop keepalives.

Source code in src/nagents/channels/types.py

async def activity(self, event: ChannelActivity) -> None:
    """Start/stop an optional typing indicator; close() must stop keepalives."""
    return None

approval(message)

Recognize a decision for a rendered approval prompt, without doing I/O.

Only connectors that render approval prompts and advertise approvals implement this. Returning a value claims the interaction: the host either applies a fully validated decision or ignores it, and never enqueues it as model input. Returning None leaves the message to normal handling.

Source code in src/nagents/channels/types.py

def approval(self, message: ChannelMessage) -> ChannelApproval | None:
    """Recognize a decision for a rendered approval prompt, without doing I/O.

    Only connectors that render approval prompts and advertise ``approvals``
    implement this. Returning a value claims the interaction: the host either
    applies a fully validated decision or ignores it, and never enqueues it as
    model input. Returning ``None`` leaves the message to normal handling.
    """
    return None

command(message)

Recognize an explicit transport command without doing I/O.

Source code in src/nagents/channels/types.py

def command(self, message: ChannelMessage) -> ChannelCommand | None:
    """Recognize an explicit transport command without doing I/O."""
    return None

fetch_attachment(attachment) async

Download a referenced attachment for model input.

Only connectors that advertise fetch_attachment in capabilities need to implement this. Return the raw bytes and the effective media type within the host's caps; raise a sanitized ChannelError otherwise. The host never treats the returned content as instructions.

Source code in src/nagents/channels/types.py

async def fetch_attachment(self, attachment: ChannelAttachment) -> tuple[bytes, str]:
    """Download a referenced attachment for model input.

    Only connectors that advertise ``fetch_attachment`` in ``capabilities``
    need to implement this. Return the raw bytes and the effective media type
    within the host's caps; raise a sanitized ``ChannelError`` otherwise. The
    host never treats the returned content as instructions.
    """
    raise ChannelError("This connector does not support attachment downloads")

listen(receive) abstractmethod async

Deliver events until cancelled, or return when a finite source ends.

Source code in src/nagents/channels/types.py

@abstractmethod
async def listen(self, receive: ChannelReceiver) -> None:
    """Deliver events until cancelled, or return when a finite source ends."""

on_event(event) async

Optionally render a source-scoped status notice; default does nothing.

Cooperate with cancellation and bound transport I/O. Hosts should await dispatch_channel_execution_event for failure isolation and joined, bounded dispatch. Never retry an uncertain notice send automatically.

Source code in src/nagents/channels/types.py

async def on_event(self, event: ChannelExecutionEvent) -> None:
    """Optionally render a source-scoped status notice; default does nothing.

    Cooperate with cancellation and bound transport I/O. Hosts should await
    dispatch_channel_execution_event for failure isolation and joined,
    bounded dispatch. Never retry an uncertain notice send automatically.
    """
    return None

send(message) abstractmethod async

Send once; return confirmed IDs or raise a sanitized ChannelError.

Source code in src/nagents/channels/types.py

@abstractmethod
async def send(self, message: ChannelSend) -> ChannelDelivery:
    """Send once; return confirmed IDs or raise a sanitized ChannelError."""

nagents.channels.ChannelDispatcher

A bound catalog and tools, independent of execution and connector lifecycle.

Source code in src/nagents/channels/dispatcher.py

class ChannelDispatcher:
    """A bound catalog and tools, independent of execution and connector lifecycle."""

    def __init__(self, channels: tuple[Channel, ...] = (), *, workspace: Path | None = None) -> None:
        self.workspace = workspace.resolve() if workspace is not None else None
        self._bindings = tuple(_bind(channel) for channel in channels)
        self._routes = {binding.name: binding for binding in self._bindings}
        if len(self._routes) != len(self._bindings):
            raise ChannelError("Channel names must be unique")

    def catalog(self) -> list[dict[str, ChannelValue]]:
        """Return a fresh, detached snapshot, including for an empty dispatcher."""
        return [cast("dict[str, ChannelValue]", json.loads(binding.catalog)) for binding in self._bindings]

    def contains_channel(self, name: str, channel: Channel) -> bool:
        """Check bound name AND connector identity without trusting mutable names."""
        binding = self._routes.get(name)
        return binding is not None and binding.channel is channel

    def action_schema(self, channel: str, action: str) -> dict[str, ChannelValue]:
        """Return a detached advertised schema; reject unknown channels/actions."""
        binding = self._route(channel)
        _identifier(action, "action")
        schema = dict(binding.actions).get(action)
        if schema is None:
            raise ChannelError("Unknown channel action")
        return cast("dict[str, ChannelValue]", json.loads(schema))

    def _route(self, channel: str) -> _Binding:
        _identifier(channel, "channel")
        binding = self._routes.get(channel)
        if binding is None:
            raise ChannelError("Unknown channel")
        return binding

    @contextmanager
    def register_tools(self, registry: ToolRegistry) -> Iterator[tuple[ToolDefinition, ...]]:
        """Own temporary tool registration without replacing or restoring tools.

        Preflight all names before registration. On every exit remove only the
        exact definitions installed here, preserving any subsequent replacement.
        Connector lifecycle and execution policy remain the caller's concern.
        """
        if any(registry.get(name) is not None for name in _TOOL_NAMES):
            raise ChannelError("Channel tool name collision")
        owned: list[ToolDefinition] = []
        try:
            owned.append(registry.register(self.channel_list, name="channel_list"))
            owned.append(registry.register(self.channel_send, name="channel_send"))
            owned.append(
                registry.register(
                    self.channel_action,
                    name="channel_action",
                    parameters={
                        "type": "object",
                        "properties": {
                            "channel": {"type": "string"},
                            "action": {"type": "string"},
                            "arguments": {"type": "object"},
                        },
                        "required": ["channel", "action", "arguments"],
                        "additionalProperties": False,
                    },
                )
            )
            yield tuple(owned)
        finally:
            for tool in owned:
                if registry.get(tool.name) is tool:
                    registry.unregister(tool.name)

    @_tool_arguments
    async def channel_list(self) -> list[dict[str, ChannelValue]]:
        """Discover configured channels, descriptive capabilities, and action schemas."""
        try:
            return self.catalog()
        except Exception as error:
            raise _failure(error, "list") from None

    @_tool_arguments
    async def channel_send(
        self,
        channel: str,
        destination: str,
        text: str,
        thread_id: str = "",
        reply_to: str = "",
        attachments: list[str] | None = None,
    ) -> dict[str, ChannelValue]:
        """Send one explicit message to a chosen channel and destination, without retries.

        Args:
            attachments: Up to three workspace-relative file paths to upload with the message. Connectors may reject unsupported attachments; the text may be empty when files are present.
            reply_to: Remote transport message ID in the chosen channel/conversation, not an ingress message_id. Use the inbound reply_to when replying to that event; otherwise leave empty.
        """
        try:
            binding = self._route(channel)
            body = _text(text, "text", blank=True, limit=MAX_TEXT_LENGTH)
            files = _outbound_files(self.workspace, attachments)
            if not body and not files:
                raise ChannelError("A channel message needs text or an attachment")
            message = ChannelSend(
                destination=_text(destination, "destination"),
                text=body,
                thread_id=_text(thread_id, "thread_id", blank=True),
                reply_to=_text(reply_to, "reply_to", blank=True),
                files=files,
            )
            if binding.content:
                validate_content_send(_object_copy(json.loads(binding.content)), message)
        except Exception as error:
            raise _failure(error, "send") from None
        try:
            delivery = await binding.channel.send(message)
        except Exception as error:
            raise _failure(error, "send", side_effect=True) from None
        try:
            if not isinstance(delivery, ChannelDelivery) or not isinstance(delivery.message_ids, tuple):
                raise TypeError("Invalid delivery")
            ids = [_text(value, "delivery message ID") for value in delivery.message_ids]
            return _object_copy({"message_ids": ids, "metadata": delivery.metadata})
        except Exception:
            raise _failure(ChannelError("Invalid channel delivery response", outcome_unknown=True), "send") from None

    @_tool_arguments
    async def channel_action(
        self, channel: str, action: str, arguments: dict[str, ChannelValue]
    ) -> dict[str, ChannelValue]:
        """Perform one advertised connector action; consult channel_list for its argument schema."""
        try:
            binding = self._route(channel)
            schema = self.action_schema(channel, action)
            copied = _object_copy(arguments)
            _validate_schema(copied, schema)
        except Exception as error:
            raise _failure(error, "action") from None
        try:
            result = await binding.channel.action(action, copied)
        except Exception as error:
            raise _failure(error, "action", side_effect=True) from None
        try:
            return _object_copy(result)
        except Exception:
            raise _failure(ChannelError("Invalid channel action response", outcome_unknown=True), "action") from None

action_schema(channel, action)

Return a detached advertised schema; reject unknown channels/actions.

Source code in src/nagents/channels/dispatcher.py

def action_schema(self, channel: str, action: str) -> dict[str, ChannelValue]:
    """Return a detached advertised schema; reject unknown channels/actions."""
    binding = self._route(channel)
    _identifier(action, "action")
    schema = dict(binding.actions).get(action)
    if schema is None:
        raise ChannelError("Unknown channel action")
    return cast("dict[str, ChannelValue]", json.loads(schema))

catalog()

Return a fresh, detached snapshot, including for an empty dispatcher.

Source code in src/nagents/channels/dispatcher.py

def catalog(self) -> list[dict[str, ChannelValue]]:
    """Return a fresh, detached snapshot, including for an empty dispatcher."""
    return [cast("dict[str, ChannelValue]", json.loads(binding.catalog)) for binding in self._bindings]

channel_action(channel, action, arguments) async

Perform one advertised connector action; consult channel_list for its argument schema.

Source code in src/nagents/channels/dispatcher.py

@_tool_arguments
async def channel_action(
    self, channel: str, action: str, arguments: dict[str, ChannelValue]
) -> dict[str, ChannelValue]:
    """Perform one advertised connector action; consult channel_list for its argument schema."""
    try:
        binding = self._route(channel)
        schema = self.action_schema(channel, action)
        copied = _object_copy(arguments)
        _validate_schema(copied, schema)
    except Exception as error:
        raise _failure(error, "action") from None
    try:
        result = await binding.channel.action(action, copied)
    except Exception as error:
        raise _failure(error, "action", side_effect=True) from None
    try:
        return _object_copy(result)
    except Exception:
        raise _failure(ChannelError("Invalid channel action response", outcome_unknown=True), "action") from None

channel_list() async

Discover configured channels, descriptive capabilities, and action schemas.

Source code in src/nagents/channels/dispatcher.py

@_tool_arguments
async def channel_list(self) -> list[dict[str, ChannelValue]]:
    """Discover configured channels, descriptive capabilities, and action schemas."""
    try:
        return self.catalog()
    except Exception as error:
        raise _failure(error, "list") from None

channel_send(channel, destination, text, thread_id='', reply_to='', attachments=None) async

Send one explicit message to a chosen channel and destination, without retries.

Parameters:

Name	Type	Description	Default
attachments	list[str] | None	Up to three workspace-relative file paths to upload with the message. Connectors may reject unsupported attachments; the text may be empty when files are present.	None
reply_to	str	Remote transport message ID in the chosen channel/conversation, not an ingress message_id. Use the inbound reply_to when replying to that event; otherwise leave empty.	''

Source code in src/nagents/channels/dispatcher.py

@_tool_arguments
async def channel_send(
    self,
    channel: str,
    destination: str,
    text: str,
    thread_id: str = "",
    reply_to: str = "",
    attachments: list[str] | None = None,
) -> dict[str, ChannelValue]:
    """Send one explicit message to a chosen channel and destination, without retries.

    Args:
        attachments: Up to three workspace-relative file paths to upload with the message. Connectors may reject unsupported attachments; the text may be empty when files are present.
        reply_to: Remote transport message ID in the chosen channel/conversation, not an ingress message_id. Use the inbound reply_to when replying to that event; otherwise leave empty.
    """
    try:
        binding = self._route(channel)
        body = _text(text, "text", blank=True, limit=MAX_TEXT_LENGTH)
        files = _outbound_files(self.workspace, attachments)
        if not body and not files:
            raise ChannelError("A channel message needs text or an attachment")
        message = ChannelSend(
            destination=_text(destination, "destination"),
            text=body,
            thread_id=_text(thread_id, "thread_id", blank=True),
            reply_to=_text(reply_to, "reply_to", blank=True),
            files=files,
        )
        if binding.content:
            validate_content_send(_object_copy(json.loads(binding.content)), message)
    except Exception as error:
        raise _failure(error, "send") from None
    try:
        delivery = await binding.channel.send(message)
    except Exception as error:
        raise _failure(error, "send", side_effect=True) from None
    try:
        if not isinstance(delivery, ChannelDelivery) or not isinstance(delivery.message_ids, tuple):
            raise TypeError("Invalid delivery")
        ids = [_text(value, "delivery message ID") for value in delivery.message_ids]
        return _object_copy({"message_ids": ids, "metadata": delivery.metadata})
    except Exception:
        raise _failure(ChannelError("Invalid channel delivery response", outcome_unknown=True), "send") from None

contains_channel(name, channel)

Check bound name AND connector identity without trusting mutable names.

Source code in src/nagents/channels/dispatcher.py

def contains_channel(self, name: str, channel: Channel) -> bool:
    """Check bound name AND connector identity without trusting mutable names."""
    binding = self._routes.get(name)
    return binding is not None and binding.channel is channel

register_tools(registry)

Own temporary tool registration without replacing or restoring tools.

Preflight all names before registration. On every exit remove only the exact definitions installed here, preserving any subsequent replacement. Connector lifecycle and execution policy remain the caller's concern.

Source code in src/nagents/channels/dispatcher.py

@contextmanager
def register_tools(self, registry: ToolRegistry) -> Iterator[tuple[ToolDefinition, ...]]:
    """Own temporary tool registration without replacing or restoring tools.

    Preflight all names before registration. On every exit remove only the
    exact definitions installed here, preserving any subsequent replacement.
    Connector lifecycle and execution policy remain the caller's concern.
    """
    if any(registry.get(name) is not None for name in _TOOL_NAMES):
        raise ChannelError("Channel tool name collision")
    owned: list[ToolDefinition] = []
    try:
        owned.append(registry.register(self.channel_list, name="channel_list"))
        owned.append(registry.register(self.channel_send, name="channel_send"))
        owned.append(
            registry.register(
                self.channel_action,
                name="channel_action",
                parameters={
                    "type": "object",
                    "properties": {
                        "channel": {"type": "string"},
                        "action": {"type": "string"},
                        "arguments": {"type": "object"},
                    },
                    "required": ["channel", "action", "arguments"],
                    "additionalProperties": False,
                },
            )
        )
        yield tuple(owned)
    finally:
        for tool in owned:
            if registry.get(tool.name) is tool:
                registry.unregister(tool.name)

nagents.channels.ChannelContentCapabilities dataclass

Optional directional declarations, validated and snapshotted at binding.

An omitted direction is undeclared, not unsupported or unrestricted. An explicit direction with false text support and no file types supports neither. Existing transport Channel.capabilities retain their meanings.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelContentCapabilities:
    """Optional directional declarations, validated and snapshotted at binding.

    An omitted direction is undeclared, not unsupported or unrestricted. An
    explicit direction with false text support and no file types supports
    neither. Existing transport ``Channel.capabilities`` retain their meanings.
    """

    receive: ChannelReceiveCapabilities | None = None
    send: ChannelSendCapabilities | None = None
    render: ChannelRenderCapabilities | None = None

nagents.channels.ChannelReceiveCapabilities dataclass

Content admitted into agent input; via identifies the admission owner.

Host-managed input does not imply support for Channel.listen. File limits are optional declarations, independent of provider modality support.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelReceiveCapabilities:
    """Content admitted into agent input; ``via`` identifies the admission owner.

    Host-managed input does not imply support for ``Channel.listen``. File
    limits are optional declarations, independent of provider modality support.
    """

    via: Literal["host", "listen"] = "listen"
    text: bool = False
    file_media_types: tuple[str, ...] = ()
    max_files: int | None = None
    max_file_bytes: int | None = None
    max_total_bytes: int | None = None

nagents.channels.ChannelSendCapabilities dataclass

Accepted explicit outbound content. Text support does not permit text files.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelSendCapabilities:
    """Accepted explicit outbound content. Text support does not permit text files."""

    text: bool = False
    file_media_types: tuple[str, ...] = ()
    max_files: int | None = None
    max_file_bytes: int | None = None
    max_total_bytes: int | None = None

nagents.channels.ChannelRenderCapabilities dataclass

Recipient presentation support, independent of model input or authorization.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelRenderCapabilities:
    """Recipient presentation support, independent of model input or authorization."""

    text: bool = False
    file_media_types: tuple[str, ...] = ()
    playback: Literal["none", "browser_dependent"] = "none"

nagents.channels.ChannelMessage dataclass

An external event. Its identity is stable within the connector instance.

Conversation/thread IDs describe the source; they do not select an Agent session. All connectors attached to one listener share that listener's session. reply_to is a transport message ID suitable for replying to this event; it can differ from the ingress message_id used for deduplication. Metadata must be JSON-compatible and must not contain credentials.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelMessage:
    """An external event. Its identity is stable within the connector instance.

    Conversation/thread IDs describe the source; they do not select an Agent
    session. All connectors attached to one listener share that listener's session.
    ``reply_to`` is a transport message ID suitable for replying to this event;
    it can differ from the ingress ``message_id`` used for deduplication.
    Metadata must be JSON-compatible and must not contain credentials.
    """

    message_id: str
    conversation_id: str
    sender_id: str
    text: str = ""
    thread_id: str = ""
    reply_to: str = ""
    event_type: str = "message"
    attachments: tuple[ChannelAttachment, ...] = ()
    metadata: dict[str, ChannelValue] = field(default_factory=dict)
    # Optional generic presentation fields; untrusted transport data, never authority.
    sent_at: float = 0.0
    sender_name: str = ""
    sender_username: str = ""
    conversation_type: str = ""

nagents.channels.ChannelSend dataclass

An explicit outgoing message selected by the application or model.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelSend:
    """An explicit outgoing message selected by the application or model."""

    destination: str
    text: str
    thread_id: str = ""
    reply_to: str = ""
    attachments: tuple[ChannelAttachment, ...] = ()
    files: tuple[ChannelFile, ...] = ()
    metadata: dict[str, ChannelValue] = field(default_factory=dict)

nagents.channels.ChannelAttachment dataclass

A connector-owned reference, not automatically downloaded or executed.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelAttachment:
    """A connector-owned reference, not automatically downloaded or executed."""

    reference: str
    media_type: str = "application/octet-stream"
    filename: str = ""
    size: int = 0

nagents.channels.ChannelDelivery dataclass

Remote message identifiers returned after confirmed delivery.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelDelivery:
    """Remote message identifiers returned after confirmed delivery."""

    message_ids: tuple[str, ...]
    metadata: dict[str, ChannelValue] = field(default_factory=dict)

nagents.channels.ChannelAction dataclass

Advertise an integration-specific operation, such as editing a message.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelAction:
    """Advertise an integration-specific operation, such as editing a message."""

    name: str
    description: str
    parameters: JsonSchema

nagents.channels.ChannelPlugin dataclass

Callable entry point with a configuration schema for management clients.

JSON Schema properties marked writeOnly: true are credentials: clients collect them separately and never echo saved values. The host validates and persists configuration; the factory performs connector-specific validation. Ordinary callable factories remain supported by load_channel.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelPlugin:
    """Callable entry point with a configuration schema for management clients.

    JSON Schema properties marked ``writeOnly: true`` are credentials: clients
    collect them separately and never echo saved values. The host validates and
    persists configuration; the factory performs connector-specific validation.
    Ordinary callable factories remain supported by ``load_channel``.
    """

    name: str
    description: str
    config_schema: dict[str, ChannelValue]
    factory: ChannelFactory

    def __call__(self, config: dict[str, ChannelValue]) -> Channel:
        return self.factory(config)

nagents.channels.ChannelCommand dataclass

A connector-recognized host command; the host decides which commands exist.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelCommand:
    """A connector-recognized host command; the host decides which commands exist."""

    name: str
    arguments: str = ""

nagents.channels.ChannelActivity dataclass

Session activity for transport indicators, independent of model replies.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelActivity:
    """Session activity for transport indicators, independent of model replies."""

    conversation_id: str
    active: bool
    thread_id: str = ""
    session_id: str = ""

nagents.channels.ChannelError

Bases: Exception

A sanitized connector failure suitable for a tool result.

No implicit send retries are performed. outcome_unknown means a remote side effect may have happened. Never put credential-bearing URLs in errors.

Source code in src/nagents/channels/types.py

class ChannelError(Exception):
    """A sanitized connector failure suitable for a tool result.

    No implicit send retries are performed. ``outcome_unknown`` means a remote
    side effect may have happened. Never put credential-bearing URLs in errors.
    """

    def __init__(self, message: str, *, retry_after: float = 0, outcome_unknown: bool = False) -> None:
        super().__init__(message)
        self.retry_after = retry_after
        self.outcome_unknown = outcome_unknown

nagents.channels.ChannelEvent dataclass

Observe execution locally; observation never sends a channel reply.

Source code in src/nagents/channels/types.py

@dataclass(frozen=True)
class ChannelEvent:
    """Observe execution locally; observation never sends a channel reply."""

    session_id: str
    channel: str
    message_id: str
    event: Event

nagents.channels.load_channel(name, config)

Load one installed, explicitly selected factory; never auto-enable plugins.

A factory receives a JSON-compatible configuration dictionary and returns a Channel. Constructors/factories must defer network I/O until open.

Source code in src/nagents/channels/plugins.py

def load_channel(name: str, config: dict[str, ChannelValue]) -> Channel:
    """Load one installed, explicitly selected factory; never auto-enable plugins.

    A factory receives a JSON-compatible configuration dictionary and returns a
    Channel. Constructors/factories must defer network I/O until ``open``.
    """
    matches = tuple(entry_points(group="nagents.channels", name=name))
    if len(matches) != 1:
        raise ValueError(f"Expected one installed channel plugin named {name!r}; found {len(matches)}")
    factory = matches[0].load()
    if not callable(factory):
        raise TypeError("Channel entry point must be a callable factory")
    channel = cast("Callable[[dict[str, ChannelValue]], Channel]", factory)(config)
    if not isinstance(channel, Channel):
        raise TypeError("Channel factory must return a Channel")
    return channel

Local delivery metadata

These records describe durable local deliveries separately from model messages. An origin is correlation data, not authorization; host execution must supply and validate it. Tool receipts and notifications contain references rather than media bytes. Concrete interface adapters use the journal through their owned host.

nagents.channels.delivery_types.DeliveryOrigin dataclass

Durable correlation captured by the owned execution/assistant-write bridge.

Values are data, not execution authority. A host must validate its live invocation before using an origin to commit a local delivery.

Source code in src/nagents/channels/delivery_types.py

@dataclass(frozen=True)
class DeliveryOrigin:
    """Durable correlation captured by the owned execution/assistant-write bridge.

    Values are data, not execution authority. A host must validate its live
    invocation before using an origin to commit a local delivery.
    """

    root_session_id: str
    actor_session_id: str
    host_run_id: str
    turn_id: str
    task_id: str
    activation: int
    invocation_id: str
    anchor_message_id: int
    call_position: int
    call_id: str
    tool_name: str

nagents.channels.delivery_types.DeliveryAsset dataclass

An opaque attachment reference; no content bytes in presentation metadata.

Source code in src/nagents/channels/delivery_types.py

@dataclass(frozen=True)
class DeliveryAsset:
    """An opaque attachment reference; no content bytes in presentation metadata."""

    asset_id: str
    position: int
    filename: str
    media_type: str
    byte_length: int

nagents.channels.delivery_types.DeliveryReceipt dataclass

One committed local delivery, independently identifiable across reloads.

Source code in src/nagents/channels/delivery_types.py

@dataclass(frozen=True)
class DeliveryReceipt:
    """One committed local delivery, independently identifiable across reloads."""

    delivery_id: str
    sequence: int
    channel: str
    origin: DeliveryOrigin
    text: str
    created_at: str
    assets: tuple[DeliveryAsset, ...] = ()

    def channel_delivery(self) -> ChannelDelivery:
        """Return a compact tool receipt, without duplicating text or media bytes."""
        return ChannelDelivery(
            (self.delivery_id,),
            metadata={
                "delivery_id": self.delivery_id,
                "channel": self.channel,
                "destination": self.origin.root_session_id,
                "asset_ids": [asset.asset_id for asset in self.assets],
            },
        )

    def notification(self) -> dict[str, ChannelValue]:
        """Return a small invalidation notice; hosts retrieve content separately.

        Even a large text delivery must not exceed a bounded replay/subscription
        queue's budget. Repeated notices identify the same committed delivery.
        """
        return {
            "delivery_id": self.delivery_id,
            "session_id": self.origin.root_session_id,
            "channel": self.channel,
            "sequence": self.sequence,
        }

channel_delivery()

Return a compact tool receipt, without duplicating text or media bytes.

Source code in src/nagents/channels/delivery_types.py

def channel_delivery(self) -> ChannelDelivery:
    """Return a compact tool receipt, without duplicating text or media bytes."""
    return ChannelDelivery(
        (self.delivery_id,),
        metadata={
            "delivery_id": self.delivery_id,
            "channel": self.channel,
            "destination": self.origin.root_session_id,
            "asset_ids": [asset.asset_id for asset in self.assets],
        },
    )

notification()

Return a small invalidation notice; hosts retrieve content separately.

Even a large text delivery must not exceed a bounded replay/subscription queue's budget. Repeated notices identify the same committed delivery.

Source code in src/nagents/channels/delivery_types.py

def notification(self) -> dict[str, ChannelValue]:
    """Return a small invalidation notice; hosts retrieve content separately.

    Even a large text delivery must not exceed a bounded replay/subscription
    queue's budget. Repeated notices identify the same committed delivery.
    """
    return {
        "delivery_id": self.delivery_id,
        "session_id": self.origin.root_session_id,
        "channel": self.channel,
        "sequence": self.sequence,
    }

nagents.channels.delivery_types.DeliveryHistory dataclass

Stable anchor-based groups; rendering belongs to the owning interface.

Source code in src/nagents/channels/delivery_types.py

@dataclass(frozen=True)
class DeliveryHistory:
    """Stable anchor-based groups; rendering belongs to the owning interface."""

    boundary: int
    earlier: tuple[DeliveryReceipt, ...] = ()
    current: tuple[DeliveryReceipt, ...] = ()

nagents.channels.local_delivery.LocalDeliveryService

Connect an execution-bound authority to a journal without starting listeners.

Trusted hosts supply an authority backed by their approved invocation bridge, never by tool arguments or an inherited context variable alone. This service does not register channels, grant approval or infer an executing session.

Source code in src/nagents/channels/local_delivery.py

class LocalDeliveryService:
    """Connect an execution-bound authority to a journal without starting listeners.

    Trusted hosts supply an authority backed by their approved invocation bridge,
    never by tool arguments or an inherited context variable alone. This service
    does not register channels, grant approval or infer an executing session.
    """

    def __init__(self, journal: DeliveryJournal, authority: DeliveryAuthority) -> None:
        self.journal = journal
        self.authority = authority

    async def send(
        self,
        channel: str,
        message: ChannelSend,
        capabilities: ChannelSendCapabilities,
        *,
        notify: DeliveryNotifier = discard_notification,
    ) -> ChannelDelivery:
        """Commit once, then publish a small best-effort invalidation notice.

        Cancellation joins the owned commit. If commit succeeded its receipt and
        identity remain durable even when cancellation prevents a returned tool
        result. Notification failure never makes a known commit a failed send.
        Observers must cooperate with cancellation, like other channel hooks.
        """
        try:
            origin = self.authority(channel, message.destination)
            prepared = self.journal.prepare(origin, channel, message, capabilities)
            # No await separates these checks and handoff to the owned commit.
            # The journal rechecks durable state in its write transaction.
            if self.authority(channel, message.destination) != origin:
                raise ChannelError("Local delivery execution changed before commit")
        except ChannelError:
            raise
        except Exception:
            # The authority is a synchronous, side-effect-free host check. A
            # PermissionError here is a known rejection, not an uncertain send.
            raise ChannelError("Local delivery execution is unavailable") from None
        receipt = await prepared.commit()

        async def publish() -> None:
            try:
                async with asyncio.timeout(NOTIFICATION_TIMEOUT):
                    await notify(receipt.notification())
            except (asyncio.CancelledError, Exception):
                # This is an isolated observer task. Owner cancellation is joined
                # and propagated by join_owned; observer self-cancellation is not.
                pass

        await join_owned(asyncio.create_task(publish()))
        return receipt.channel_delivery()

send(channel, message, capabilities, *, notify=discard_notification) async

Commit once, then publish a small best-effort invalidation notice.

Cancellation joins the owned commit. If commit succeeded its receipt and identity remain durable even when cancellation prevents a returned tool result. Notification failure never makes a known commit a failed send. Observers must cooperate with cancellation, like other channel hooks.

Source code in src/nagents/channels/local_delivery.py

async def send(
    self,
    channel: str,
    message: ChannelSend,
    capabilities: ChannelSendCapabilities,
    *,
    notify: DeliveryNotifier = discard_notification,
) -> ChannelDelivery:
    """Commit once, then publish a small best-effort invalidation notice.

    Cancellation joins the owned commit. If commit succeeded its receipt and
    identity remain durable even when cancellation prevents a returned tool
    result. Notification failure never makes a known commit a failed send.
    Observers must cooperate with cancellation, like other channel hooks.
    """
    try:
        origin = self.authority(channel, message.destination)
        prepared = self.journal.prepare(origin, channel, message, capabilities)
        # No await separates these checks and handoff to the owned commit.
        # The journal rechecks durable state in its write transaction.
        if self.authority(channel, message.destination) != origin:
            raise ChannelError("Local delivery execution changed before commit")
    except ChannelError:
        raise
    except Exception:
        # The authority is a synchronous, side-effect-free host check. A
        # PermissionError here is a known rejection, not an uncertain send.
        raise ChannelError("Local delivery execution is unavailable") from None
    receipt = await prepared.commit()

    async def publish() -> None:
        try:
            async with asyncio.timeout(NOTIFICATION_TIMEOUT):
                await notify(receipt.notification())
        except (asyncio.CancelledError, Exception):
            # This is an isolated observer task. Owner cancellation is joined
            # and propagated by join_owned; observer self-cancellation is not.
            pass

    await join_owned(asyncio.create_task(publish()))
    return receipt.channel_delivery()

nagents.session.deliveries.DeliveryJournal

Source code in src/nagents/session/deliveries.py

class DeliveryJournal:
    def __init__(self, db_path: Path):
        self.db_path = Path(db_path)
        self._initialized = False
        self._lock = asyncio.Lock()

    async def initialize(self) -> None:
        async with self._lock:
            if not self._initialized:
                self.db_path.parent.mkdir(parents=True, exist_ok=True)
                await MigrationManager(self.db_path, db_name="sessions", migrations=list(migrations)).initialize()
                self._initialized = True

    def prepare(
        self, origin: DeliveryOrigin, channel: str, message: ChannelSend, capabilities: ChannelSendCapabilities
    ) -> PreparedDelivery:
        """Validate actual bytes and snapshot the entire payload before any await."""
        if (
            type(origin) is not DeliveryOrigin
            or type(message) is not ChannelSend
            or type(capabilities) is not ChannelSendCapabilities
        ):
            raise ChannelError("Invalid local delivery types")
        for name in (
            "root_session_id",
            "actor_session_id",
            "host_run_id",
            "turn_id",
            "task_id",
            "invocation_id",
            "call_id",
            "tool_name",
        ):
            value = getattr(origin, name)
            if type(value) is not str or len(value) > 512 or (value and not value.isprintable()):
                raise ChannelError("Invalid local delivery origin")
            if name != "task_id" and not value:
                raise ChannelError("Missing local delivery origin")
        for value, minimum in ((origin.activation, 0), (origin.anchor_message_id, 1), (origin.call_position, 0)):
            if type(value) is not int or not minimum <= value < 2**63:
                raise ChannelError("Invalid local delivery origin position")
        if origin.actor_session_id != origin.root_session_id or origin.task_id:
            raise ChannelError("Local deliveries currently require the root actor")
        if type(message.destination) is not str or message.destination != origin.root_session_id:
            raise ChannelError("Local delivery destination must match the root")
        if (
            type(message.thread_id) is not str
            or message.thread_id
            or type(message.reply_to) is not str
            or message.reply_to
        ):
            raise ChannelError("Local deliveries do not support thread or reply references")
        if type(message.metadata) is not dict or message.metadata:
            raise ChannelError("Local deliveries do not support metadata")
        if type(channel) is not str or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.-]{0,63}", channel):
            raise ChannelError("Invalid local delivery channel")
        if type(message.text) is not str or len(message.text) > MAX_TEXT_LENGTH or "\x00" in message.text:
            raise ChannelError("Invalid local delivery text")
        try:
            message.text.encode("utf-8")
        except UnicodeError as error:
            raise ChannelError("Invalid local delivery text") from error
        if (
            type(message.attachments) is not tuple
            or message.attachments
            or type(message.files) is not tuple
            or len(message.files) > MAX_OUTBOUND_FILES
        ):
            raise ChannelError("Invalid local delivery attachments")
        files: list[ChannelFile] = []
        total = 0
        for file in message.files:
            if type(file) is not ChannelFile or type(file.data) is not bytes:
                raise ChannelError("Local delivery files require bytes")
            if (
                type(file.filename) is not str
                or not file.filename
                or len(file.filename) > 255
                or not file.filename.isprintable()
                or any(c in file.filename for c in "/\\:")
                or file.filename in (".", "..")
            ):
                raise ChannelError("Invalid local delivery filename")
            if (
                type(file.media_type) is not str
                or len(file.media_type) > 127
                or not re.fullmatch(r"[a-zA-Z0-9!#$&^_.+-]+/[a-zA-Z0-9!#$&^_.+-]+", file.media_type)
            ):
                raise ChannelError("Invalid local delivery MIME type")
            total += len(file.data)
            if len(file.data) > MAX_OUTBOUND_FILE_BYTES or total > MAX_OUTBOUND_TOTAL_BYTES:
                raise ChannelError("Local delivery attachment size limit exceeded")
            files.append(replace(file))
        if not message.text.strip() and not files:
            raise ChannelError("Local delivery is empty")
        snapshot = ChannelSend(destination=message.destination, text=message.text, files=tuple(files))
        validate_content_send(content_capabilities_catalog(ChannelContentCapabilities(send=capabilities)), snapshot)
        return PreparedDelivery(self, replace(origin), channel, snapshot)

    def _commit(self, prepared: PreparedDelivery) -> DeliveryReceipt:
        try:
            db = self._connect_write()
        except Exception:
            prepared.outcome = "failed"
            raise ChannelError("Local delivery storage could not be opened") from None
        try:
            try:
                candidate = self._insert(db, prepared)
            except Exception as error:
                # No COMMIT has been attempted. Closing also rolls back if this
                # explicit rollback fails; this operation cannot have committed.
                with suppress(Exception):
                    db.rollback()
                prepared.outcome = "failed"
                if isinstance(error, ChannelError):
                    raise
                raise ChannelError("Local delivery transaction failed") from None
            try:
                db.commit()
            except Exception:
                # A still-open transaction proves COMMIT did not complete. Only
                # a successful rollback establishes that outcome conclusively.
                rolled_back = False
                try:
                    if db.in_transaction:
                        db.rollback()
                        rolled_back = True
                except Exception:
                    pass
                if rolled_back:
                    prepared.outcome = "failed"
                    raise ChannelError("Local delivery transaction rolled back") from None
                try:
                    recovered = self._recover_receipt(prepared.delivery_id)
                except Exception:
                    recovered = ()
                if recovered == (candidate,):
                    return prepared._confirmed(candidate)
                # Absence alone cannot prove rollback: a concurrent lifecycle
                # purge could have removed a committed delivery before recovery.
                prepared.outcome = "unknown"
                raise ChannelError("Local delivery outcome could not be established", outcome_unknown=True) from None
            # Publish confirmation before close or any other fallible cleanup.
            return prepared._confirmed(candidate)
        finally:
            with suppress(Exception):
                db.close()

    def _connect_write(self) -> sqlite3.Connection:
        return sqlite3.connect(self.db_path)

    def _recover_receipt(self, delivery_id: str) -> tuple[DeliveryReceipt, ...]:
        """Resolve acknowledgement loss by identity, never by reissuing a send.

        Recovery ignores current membership: trash or ownership changes cannot
        change whether this exact operation committed. Read-only open avoids
        accidentally creating an empty database when storage is unavailable.
        """
        uri = self.db_path.resolve().as_uri() + "?mode=ro"
        db = sqlite3.connect(uri, uri=True)
        try:
            db.execute("BEGIN")
            row = db.execute(
                f"SELECT delivery_id, sequence, channel, {_ORIGIN_COLUMNS}, text, created_at "
                "FROM ngn_local_deliveries WHERE delivery_id = ?",
                (delivery_id,),
            ).fetchone()
            if row is None:
                return ()
            assets = tuple(
                DeliveryAsset(*asset)
                for asset in db.execute(
                    "SELECT asset_id, position, filename, media_type, byte_length "
                    "FROM ngn_local_delivery_assets WHERE delivery_id = ? ORDER BY position",
                    (delivery_id,),
                )
            )
            return (DeliveryReceipt(row[0], row[1], row[2], DeliveryOrigin(*row[3:14]), row[14], row[15], assets),)
        finally:
            # A completed read establishes status even if reader cleanup fails.
            with suppress(Exception):
                db.close()

    def _insert(self, db: sqlite3.Connection, prepared: PreparedDelivery) -> DeliveryReceipt:
        origin = prepared._origin
        created_at = datetime.now(UTC).isoformat()
        assets = tuple(
            DeliveryAsset(uuid4().hex, i, f.filename, f.media_type, len(f.data))
            for i, f in enumerate(prepared._message.files)
        )
        db.execute("BEGIN IMMEDIATE")
        _active(db, origin.root_session_id)
        _anchor(db, origin)
        cursor = db.execute(
            f"INSERT INTO ngn_local_deliveries (delivery_id, channel, {_ORIGIN_COLUMNS}, text, created_at) "
            f"VALUES ({','.join('?' for _ in range(15))})",
            (prepared.delivery_id, prepared._channel, *astuple(origin), prepared._message.text, created_at),
        )
        sequence = cursor.lastrowid
        assert sequence is not None
        for asset, file in zip(assets, prepared._message.files, strict=True):
            db.execute(
                "INSERT INTO ngn_local_delivery_assets VALUES (?, ?, ?, ?, ?, ?, ?)",
                (
                    asset.asset_id,
                    prepared.delivery_id,
                    asset.position,
                    asset.filename,
                    asset.media_type,
                    asset.byte_length,
                    file.data,
                ),
            )
        return DeliveryReceipt(
            prepared.delivery_id, sequence, prepared._channel, origin, prepared._message.text, created_at, assets
        )

    async def history(self, root_session_id: str) -> DeliveryHistory:
        await self.initialize()
        return await join_owned(asyncio.create_task(asyncio.to_thread(self._history, root_session_id)))

    def _history(self, root: str) -> DeliveryHistory:
        with closing(sqlite3.connect(self.db_path)) as db, db:
            db.execute("BEGIN")
            return self._history_in(db, root)

    def _history_in(self, db: sqlite3.Connection, root: str) -> DeliveryHistory:
        """Read metadata in an owned caller's transcript snapshot transaction."""
        boundary = _active(db, root)
        earlier: list[DeliveryReceipt] = []
        current: list[DeliveryReceipt] = []
        rows = db.execute(
            f"SELECT delivery_id, sequence, channel, {_ORIGIN_COLUMNS}, text, created_at "
            "FROM ngn_local_deliveries WHERE root_session_id = ? "
            "ORDER BY anchor_message_id, call_position, sequence",
            (root,),
        ).fetchall()
        for row in rows:
            origin = DeliveryOrigin(*row[3:14])
            _anchor(db, origin)
            assets = tuple(
                DeliveryAsset(*asset)
                for asset in db.execute(
                    "SELECT asset_id, position, filename, media_type, byte_length "
                    "FROM ngn_local_delivery_assets WHERE delivery_id = ? ORDER BY position",
                    (row[0],),
                )
            )
            receipt = DeliveryReceipt(row[0], row[1], row[2], origin, row[14], row[15], assets)
            (earlier if origin.anchor_message_id < boundary else current).append(receipt)
        return DeliveryHistory(boundary, tuple(earlier), tuple(current))

    async def read_asset(self, root_session_id: str, delivery_id: str, asset_id: str) -> bytes:
        await self.initialize()
        return await join_owned(
            asyncio.create_task(asyncio.to_thread(self._read_asset, root_session_id, delivery_id, asset_id))
        )

    def _read_asset(self, root: str, delivery_id: str, asset_id: str) -> bytes:
        with closing(sqlite3.connect(self.db_path)) as db, db:
            db.execute("BEGIN")
            _active(db, root)
            row = db.execute(
                "SELECT a.byte_length, length(a.data), typeof(a.data) FROM ngn_local_delivery_assets a "
                "JOIN ngn_local_deliveries d ON d.delivery_id = a.delivery_id "
                "WHERE d.root_session_id = ? AND d.delivery_id = ? AND a.asset_id = ?",
                (root, delivery_id, asset_id),
            ).fetchone()
            if row is None or row[2] != "blob" or not 0 <= row[0] == row[1] <= MAX_OUTBOUND_FILE_BYTES:
                raise ChannelError("Local delivery asset is missing or invalid")
            data = db.execute("SELECT data FROM ngn_local_delivery_assets WHERE asset_id = ?", (asset_id,)).fetchone()[
                0
            ]
            return bytes(data)

prepare(origin, channel, message, capabilities)

Validate actual bytes and snapshot the entire payload before any await.

Source code in src/nagents/session/deliveries.py

def prepare(
    self, origin: DeliveryOrigin, channel: str, message: ChannelSend, capabilities: ChannelSendCapabilities
) -> PreparedDelivery:
    """Validate actual bytes and snapshot the entire payload before any await."""
    if (
        type(origin) is not DeliveryOrigin
        or type(message) is not ChannelSend
        or type(capabilities) is not ChannelSendCapabilities
    ):
        raise ChannelError("Invalid local delivery types")
    for name in (
        "root_session_id",
        "actor_session_id",
        "host_run_id",
        "turn_id",
        "task_id",
        "invocation_id",
        "call_id",
        "tool_name",
    ):
        value = getattr(origin, name)
        if type(value) is not str or len(value) > 512 or (value and not value.isprintable()):
            raise ChannelError("Invalid local delivery origin")
        if name != "task_id" and not value:
            raise ChannelError("Missing local delivery origin")
    for value, minimum in ((origin.activation, 0), (origin.anchor_message_id, 1), (origin.call_position, 0)):
        if type(value) is not int or not minimum <= value < 2**63:
            raise ChannelError("Invalid local delivery origin position")
    if origin.actor_session_id != origin.root_session_id or origin.task_id:
        raise ChannelError("Local deliveries currently require the root actor")
    if type(message.destination) is not str or message.destination != origin.root_session_id:
        raise ChannelError("Local delivery destination must match the root")
    if (
        type(message.thread_id) is not str
        or message.thread_id
        or type(message.reply_to) is not str
        or message.reply_to
    ):
        raise ChannelError("Local deliveries do not support thread or reply references")
    if type(message.metadata) is not dict or message.metadata:
        raise ChannelError("Local deliveries do not support metadata")
    if type(channel) is not str or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.-]{0,63}", channel):
        raise ChannelError("Invalid local delivery channel")
    if type(message.text) is not str or len(message.text) > MAX_TEXT_LENGTH or "\x00" in message.text:
        raise ChannelError("Invalid local delivery text")
    try:
        message.text.encode("utf-8")
    except UnicodeError as error:
        raise ChannelError("Invalid local delivery text") from error
    if (
        type(message.attachments) is not tuple
        or message.attachments
        or type(message.files) is not tuple
        or len(message.files) > MAX_OUTBOUND_FILES
    ):
        raise ChannelError("Invalid local delivery attachments")
    files: list[ChannelFile] = []
    total = 0
    for file in message.files:
        if type(file) is not ChannelFile or type(file.data) is not bytes:
            raise ChannelError("Local delivery files require bytes")
        if (
            type(file.filename) is not str
            or not file.filename
            or len(file.filename) > 255
            or not file.filename.isprintable()
            or any(c in file.filename for c in "/\\:")
            or file.filename in (".", "..")
        ):
            raise ChannelError("Invalid local delivery filename")
        if (
            type(file.media_type) is not str
            or len(file.media_type) > 127
            or not re.fullmatch(r"[a-zA-Z0-9!#$&^_.+-]+/[a-zA-Z0-9!#$&^_.+-]+", file.media_type)
        ):
            raise ChannelError("Invalid local delivery MIME type")
        total += len(file.data)
        if len(file.data) > MAX_OUTBOUND_FILE_BYTES or total > MAX_OUTBOUND_TOTAL_BYTES:
            raise ChannelError("Local delivery attachment size limit exceeded")
        files.append(replace(file))
    if not message.text.strip() and not files:
        raise ChannelError("Local delivery is empty")
    snapshot = ChannelSend(destination=message.destination, text=message.text, files=tuple(files))
    validate_content_send(content_capabilities_catalog(ChannelContentCapabilities(send=capabilities)), snapshot)
    return PreparedDelivery(self, replace(origin), channel, snapshot)

nagents.session.deliveries.PreparedDelivery

Single-use write. Cancellation joins storage; inspect receipt/outcome afterward.

receipt is a zero-or-one tuple, including after caller cancellation. outcome is prepared, pending, committed, failed, or unknown. A second commit raises RuntimeError, even after failure; separate prepares always allocate fresh IDs. Failed means no commit or established rollback; unknown raises a sanitized ChannelError with outcome_unknown=True. Recovery only reads the allocated ID.

Source code in src/nagents/session/deliveries.py

class PreparedDelivery:
    """Single-use write. Cancellation joins storage; inspect receipt/outcome afterward.

    ``receipt`` is a zero-or-one tuple, including after caller cancellation.
    ``outcome`` is prepared, pending, committed, failed, or unknown. A second commit raises
    RuntimeError, even after failure; separate prepares always allocate fresh IDs.
    Failed means no commit or established rollback; unknown raises a sanitized
    ChannelError with outcome_unknown=True. Recovery only reads the allocated ID.
    """

    def __init__(self, journal: DeliveryJournal, origin: DeliveryOrigin, channel: str, message: ChannelSend):
        self.delivery_id = uuid4().hex
        self.receipt: tuple[DeliveryReceipt, ...] = ()
        self.outcome: Literal["prepared", "pending", "committed", "failed", "unknown"] = "prepared"
        self._journal = journal
        self._origin = origin
        self._channel = channel
        self._message = message

    def _confirmed(self, receipt: DeliveryReceipt) -> DeliveryReceipt:
        self.receipt = (receipt,)
        self.outcome = "committed"
        return receipt

    async def commit(self) -> DeliveryReceipt:
        if self.outcome != "prepared":
            raise RuntimeError("Prepared delivery commit is single-use")
        self.outcome = "pending"

        async def owned() -> DeliveryReceipt:
            try:
                await self._journal.initialize()
            except Exception:
                self.outcome = "failed"
                raise ChannelError("Local delivery storage initialization failed") from None
            try:
                return await asyncio.to_thread(self._journal._commit, self)
            except Exception:
                if self.receipt:
                    return self.receipt[0]
                if self.outcome == "pending":
                    self.outcome = "unknown"
                    raise ChannelError(
                        "Local delivery outcome could not be established", outcome_unknown=True
                    ) from None
                raise

        return await join_owned(asyncio.create_task(owned()))