Base

octomate.tentacles.base

The tentacle base class and the console log formatting that tints each tentacle's lines.

LEVEL_STYLES module-attribute

LEVEL_STYLES: dict[int, Style] = {
    logging.DEBUG: Style(dim=True),
    logging.INFO: Style(color="green"),
    logging.WARNING: Style(color="yellow"),
    logging.ERROR: Style(color="red"),
    logging.CRITICAL: Style(color="red", bold=True),
}

Tentacle dataclass

Base for external integrations managed by the Octomate host.

A tentacle is a lifecycle component, not a message dispatcher. Channel tentacles receive platform events and call Octomate.awake; agent tentacles expose pydantic-ai-style run methods.

A tentacle is an async context manager: the host enters it to start the tentacle's long-lived resources and exits it to tear them down. The base implementation is a no-op; subclasses override __aenter__/__aexit__.

Source code in octomate/tentacles/base.py

@dataclass
class Tentacle[TentaclePrimaryT, TentacleSecondaryT]:
    """Base for external integrations managed by the Octomate host.

    A tentacle is a lifecycle component, not a message dispatcher. Channel
    tentacles receive platform events and call Octomate.awake; agent tentacles
    expose pydantic-ai-style run methods.

    A tentacle is an async context manager: the host enters it to start the
    tentacle's long-lived resources and exits it to tear them down. The base
    implementation is a no-op; subclasses override `__aenter__`/`__aexit__`.
    """

    id: str
    octomate: Octomate = field(repr=False)
    # The console log color for this tentacle's lines, dispatched by the Octomate
    # host when the tentacle is connected (see `Octomate.connect`).
    log_color: Style | None = field(default=None, init=False, repr=False)
    # A tentacle fronting something with a color of its own claims it here, and the
    # host dispatches that instead of generating a hue from the connection order.
    brand_color: ClassVar[Style | None] = None

    @property
    def log_names(self) -> tuple[str, ...]:
        """Logger-name prefixes whose records belong to this tentacle — its own
        module tree by default. Channels with a chatty SDK extend this to claim
        the SDK's loggers too, so those lines share the tentacle's color."""
        return (type(self).__module__.rsplit(".", 1)[0],)

    def routers(self) -> tuple[APIRouter, ...]:
        """FastAPI routers the host mounts when this tentacle is connected — empty by
        default. A tentacle that exposes an HTTP surface (e.g. an inbound hook
        endpoint) overrides this. Mounting happens at connect time, off the tentacle's
        async lifecycle: the routes are owned by the Octomate app, not the tentacle."""
        return ()

    async def __aenter__(self) -> Self:
        """Start any long-lived platform resources owned by the tentacle."""
        return self

    async def __aexit__(self, *exc: object) -> None:
        """Stop any long-lived platform resources owned by the tentacle."""

id instance-attribute

id: str

octomate class-attribute instance-attribute

octomate: Octomate = field(repr=False)

log_color class-attribute instance-attribute

log_color: Style | None = field(
    default=None, init=False, repr=False
)

brand_color class-attribute

brand_color: Style | None = None

log_names property

log_names: tuple[str, ...]

Logger-name prefixes whose records belong to this tentacle — its own module tree by default. Channels with a chatty SDK extend this to claim the SDK's loggers too, so those lines share the tentacle's color.

routers

routers() -> tuple[APIRouter, ...]

FastAPI routers the host mounts when this tentacle is connected — empty by default. A tentacle that exposes an HTTP surface (e.g. an inbound hook endpoint) overrides this. Mounting happens at connect time, off the tentacle's async lifecycle: the routes are owned by the Octomate app, not the tentacle.

Source code in octomate/tentacles/base.py

def routers(self) -> tuple[APIRouter, ...]:
    """FastAPI routers the host mounts when this tentacle is connected — empty by
    default. A tentacle that exposes an HTTP surface (e.g. an inbound hook
    endpoint) overrides this. Mounting happens at connect time, off the tentacle's
    async lifecycle: the routes are owned by the Octomate app, not the tentacle."""
    return ()

TentacleLogFormatter

Bases: Formatter

Renders LEVEL timestamp [logger] message — the level tinted by severity and the [logger] header tinted with the color the host dispatched to the owning tentacle. colorize is off for non-TTY streams so redirected logs stay plain. Lives with the tentacle base since it renders tentacle identity; the host resolves each logger to its tentacle via log_tag.

Source code in octomate/tentacles/base.py

class TentacleLogFormatter(logging.Formatter):
    """Renders `LEVEL     timestamp [logger] message` — the level tinted by
    severity and the `[logger]` header tinted with the color the host dispatched
    to the owning tentacle. `colorize` is off for non-TTY streams so redirected
    logs stay plain. Lives with the tentacle base since it renders tentacle
    identity; the host resolves each logger to its tentacle via `log_tag`."""

    def __init__(self, host: Octomate, *, colorize: bool) -> None:
        super().__init__()
        self.host = host
        self.colorize = colorize

    def formatMessage(self, record: logging.LogRecord) -> str:
        tag, style = self.host.log_tag(record.name)
        level = f"{record.levelname:<8}"
        header = f"[{tag}]"
        if self.colorize:
            level = LEVEL_STYLES.get(record.levelno, Style()).render(level)
            if style is not None:
                header = style.render(header)
        line = f"{level} {self.formatTime(record)} {header} {record.getMessage()}"
        return line

host instance-attribute

host = host

colorize instance-attribute

colorize = colorize

formatMessage

formatMessage(record: LogRecord) -> str

Source code in octomate/tentacles/base.py

def formatMessage(self, record: logging.LogRecord) -> str:
    tag, style = self.host.log_tag(record.name)
    level = f"{record.levelname:<8}"
    header = f"[{tag}]"
    if self.colorize:
        level = LEVEL_STYLES.get(record.levelno, Style()).render(level)
        if style is not None:
            header = style.render(header)
    line = f"{level} {self.formatTime(record)} {header} {record.getMessage()}"
    return line

TentacleLogHandler

Bases: RichHandler

Preserve tentacle ANSI colors while Rich renders each exception once.

Source code in octomate/tentacles/base.py

class TentacleLogHandler(RichHandler):
    """Preserve tentacle ANSI colors while Rich renders each exception once."""

    def render_message(self, record: logging.LogRecord, message: str) -> Text:
        return Text.from_ansi(message)

render_message

render_message(record: LogRecord, message: str) -> Text

Source code in octomate/tentacles/base.py

def render_message(self, record: logging.LogRecord, message: str) -> Text:
    return Text.from_ansi(message)

octomate.tentacles.agent

The agent tentacle base: pydantic-ai-style run entrypoints, subagent runs, and the session bookkeeping every runtime shares.

AgentOutputT module-attribute

AgentOutputT = TypeVar('AgentOutputT')

AgentDepsT module-attribute

AgentDepsT = TypeVar('AgentDepsT')

AgentSpecInput

AgentSpecInput = JsonObject | AgentSpec

AgentTentacle

Bases: Tentacle[AgentOutputT, AgentDepsT], ABC

Base class for Octomate agents wrapping pydantic-ai run entrypoints.

Source code in octomate/tentacles/agent.py

class AgentTentacle(Tentacle[AgentOutputT, AgentDepsT], ABC):
    """Base class for Octomate agents wrapping pydantic-ai run entrypoints."""

    commands: CommandManager
    projects: ProjectManager
    threads: ThreadManager
    files: FileManager
    conversations: ConversationManager
    workspaces: WorkspaceManager

    def __init__(
        self,
        id: str,
        octomate: Octomate,
        *,
        commands: CommandManager,
        projects: ProjectManager,
        threads: ThreadManager,
        files: FileManager,
    ) -> None:
        super().__init__(id=id, octomate=octomate)
        self.commands = commands
        self.projects = projects
        self.threads = threads
        self.files = files

    # Capability blurb the triage agent reads when routing to a reception agent.
    # Subclasses refine this default; overridable at init.
    description: str = "General-purpose agent for handling user requests."

    # Routing metadata supplied by the harness, or config when it is unavailable.
    claims: Mapping[AgentRouteModelName, Claim] = MappingProxyType({})

    # Whether this agent's driven turns offer the gateway spells; the only switch,
    # read by `ReflexDeps.gateway_agents`. Subclasses assign it from their config
    # in `__init__`.
    gateway: bool = True

    # Native hook/stream identity for this runtime, shared by its configured agents.
    native_id: ClassVar[str | None] = None

    @property
    def info(self) -> AgentInfo:
        return AgentInfo(
            id=self.id,
            description=self.description,
            gateway=self.gateway,
            default_model=self.default_model,
            permission_modes=self.permission_modes,
            default_permission_mode=self.default_permission_mode,
            routes=self.routes,
            driven_sessions=sum(self.driven_sessions.values()),
            native_sessions=sum(self.native_sessions.values()),
        )

    @cached_property
    def driven_sessions(self) -> Counter[str]:
        """Live runs holding each external runtime session on this agent."""
        return Counter()

    @cached_property
    def native_sessions(self) -> Counter[str]:
        """Accepted native transcript streams currently attached to this agent."""
        return Counter()

    @cached_property
    def run_tasks(self) -> set[asyncio.Task[None]]:
        """Collectors registered by agent harnesses that opt into `observe_run`.

        Created on first access. Participating harnesses drain these tasks
        during shutdown; other harnesses do not need this registry.
        """
        return set()

    async def observe_run(
        self, events: AsyncGenerator[ReactStreamEvent[RunOutputDataT], None]
    ) -> AsyncGenerator[ReactStreamEvent[RunOutputDataT], None]:
        """Collect a run to completion even when its observer closes or cancels.

        Agent harnesses opt in by wrapping their event generators with this
        method and draining `run_tasks` during shutdown. Inheriting it alone
        does not change a harness's lifecycle.

        Joining before leaving keeps the caller's gateway and other enclosing
        resources available to the agent until its final notifications are recorded.
        """
        send, receive = anyio.create_memory_object_stream[
            ReactStreamEvent[RunOutputDataT]
        ](100)
        errors: list[Exception | asyncio.CancelledError] = []

        async def collect() -> None:
            try:
                async with aclosing(events):
                    async for event in events:
                        try:
                            await send.send(event)
                        except anyio.BrokenResourceError:
                            # Only the observer is gone; keep ingesting the run.
                            pass
            except (Exception, asyncio.CancelledError) as error:
                errors.append(error)
            finally:
                send.close()

        task = asyncio.create_task(collect())
        self.run_tasks.add(task)
        observed_end = False
        try:
            async with receive:
                async for event in receive:
                    yield event
            observed_end = True
            for error in errors:
                raise error
        finally:
            receive.close()
            cancelled = await drain_task(task)
            self.run_tasks.discard(task)
            if not observed_end:
                for error in errors:
                    logging.getLogger(type(self).__module__).error(
                        "Agent %s run failed after observer detached",
                        self.id,
                        exc_info=error,
                    )
            if cancelled:
                raise asyncio.CancelledError

    @cached_property
    def routes(self) -> list[AgentRoute]:
        """Every served model, using discovered or configured routing metadata."""
        return self.build_routes()

    @asynccontextmanager
    async def driving(
        self, session_id: str, *, native: bool = False
    ) -> AsyncGenerator[None]:
        """Count a driven runtime session or an accepted native stream.

        Driven runs hold the count through cleanup, including interruption or
        cancellation. Native streams hold it from their accepted handshake
        through stream cleanup.
        """
        sessions = self.native_sessions if native else self.driven_sessions
        sessions[session_id] += 1
        try:
            yield
        finally:
            sessions[session_id] -= 1
            if sessions[session_id] == 0:
                del sessions[session_id]

    def build_routes(self) -> list[AgentRoute]:
        return [
            AgentRoute(
                agent_id=self.id,
                model=model,
                claim=self.claims.get(model) or Claim(self.description, efforts=()),
            )
            for model in self.models
        ]

    async def __aenter__(self) -> Self:
        """Enter after the subclass has prepared its models and claims."""
        self.routes = self.build_routes()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None = None,
        exc_value: BaseException | None = None,
        traceback: TracebackType | None = None,
    ) -> None:
        self.routes = []

    @cached_property
    def default_model(self) -> str | None:
        """Configured first model, replaced by discovery with the harness's default."""
        return next(iter(self.models), None) if self.native_id is None else None

    def resolve_model(
        self, model: AgentRouteModelName | None = None
    ) -> AgentRouteModelName | None:
        """Resolve a default, exact model or unambiguous provider-less name."""
        if not self.models:
            raise ValueError(f"agent {self.id!r} has no available model catalog")
        if model is None:
            # Native harnesses retain their settings and resumed model selection.
            return self.default_model if self.native_id is None else None
        served = self.served_model(model)
        if served is not None:
            return served

        raise ValueError(
            f"agent {self.id!r} does not serve model {model!r} with an unambiguous provider"
        )

    def served_model(self, name: str) -> str | None:
        """The catalog entry a stored model name means: the entry itself, or, for
        a name saved without its provider, the one entry it can be. None when this
        agent serves no such model."""
        if name in self.models:
            return name
        if name and ":" not in name:
            matches = [key for key in self.models if key.partition(":")[2] == name]
            if len(matches) == 1:
                return matches[0]
        return None

    # Whether the agent keeps a live in-process run that can park on a human
    # deferral (approval/question) and resume by delivering the response to its
    # waiter, instead of resuming durably through the triage graph. In-process
    # agents populate `pending` (batch id -> waiter); the rest never touch it.
    in_process: ClassVar[bool] = False
    pendings: dict[UUID7, asyncio.Future[DeferredActionBatchResponse]]

    permission_modes: tuple[PermissionMode, ...] = ()

    def check_permission_mode(self, mode: str) -> None:
        """Reject selections absent from this tentacle's current catalog."""
        if not any(option.value == mode for option in self.permission_modes):
            raise ValueError(f"{mode!r} is not one of {self.id}'s modes")

    def check_effort(self, model: AgentRouteModelName | None, effort: str) -> None:
        """Reject a level the route for `model` does not claim."""
        model = self.resolve_model(model or self.default_model)
        if model is None:
            raise ValueError(
                "The runtime's model is unknown; select a model to set effort"
            )
        route = next((route for route in self.routes if route.model == model), None)
        efforts = route.claim.efforts if route is not None else ()
        if effort not in efforts:
            raise ValueError(
                f"{self.id} ({model}) does not take effort {effort!r}; it claims "
                f"{'/'.join(efforts) or 'none'}"
            )

    async def resolve_effort(
        self,
        conversation: Conversation,
        *,
        model: AgentRouteModelName | None,
        effort: str | None = None,
    ) -> str | None:
        """Validate the run's override or stored effort against its selected model.

        With no explicit model, use the conversation's last reported model, then
        the harness's catalog default for a conversation that has not run yet.
        """
        selected = effort if effort is not None else conversation.effort
        if selected is None:
            return None
        effective_model = model or await self.conversations.latest_model(
            conversation.id
        )
        self.check_effort(effective_model, selected)
        return selected

    async def set_permission_mode(
        self, conversation: Conversation, mode: str | None
    ) -> Conversation:
        """Validate and save a selection, updating a live runtime when supported.

        None restores the configured default. Native update failures propagate
        before the selection is saved. Runtimes without live updates read the
        saved selection on their next run; existing approvals remain separate.
        """
        if conversation.agent_tentacle_id != self.id:
            raise ValueError("conversation belongs to another agent")
        effective = mode if mode is not None else self.default_permission_mode
        if effective is not None:
            self.check_permission_mode(effective)
            await self.apply_permission_mode(conversation, effective)
        updated = await self.conversations.set_permission_mode(conversation, mode)
        self.commands.invalidate(agent_id=self.id, conversation_id=conversation.id)
        return updated

    async def apply_permission_mode(
        self, conversation: Conversation, mode: str
    ) -> None:
        """Update an active native runtime when it supports permission changes.

        The default leaves in-flight work alone. Codex applies both permission
        axes at the next turn start; Inkling reads storage at each deferral.
        """

    @property
    def default_permission_mode(self) -> str | None:
        """The posture this agent's conversations run under when they declare none —
        its configured default, in its own vocabulary.

        A row's NULL is not "no posture", it is "nothing said here"; this is what
        decides instead, and it is what the console reports for such a row rather than
        showing a blank. None only for an agent that answers to no posture at all."""
        return None

    models: dict[AgentRouteModelName, Model | str]

    async def discover_models(self) -> None:
        """Refresh models and claims once the harness connection is ready.

        Harnesses call this during entry and install results with set_model_catalog.
        Agents with config-supplied catalogs keep their existing models and claims.
        """

    async def discover_commands(
        self,
        context: CommandContext,
        *,
        refresh: bool = False,
        prefix: str = "",
    ) -> CommandCatalog:
        """Read this agent's cached catalog for an already authorized context.

        Callers resolve and authorize context before every lookup, including cache
        hits. Prefix matching filters a copy by command name; it does not imply
        argument-value completion. Runtime adapters implement `probe_commands`.
        """
        catalog = await self.commands.discover(self, context, refresh=refresh)
        catalog.descriptors = {
            descriptor
            for descriptor in catalog.descriptors
            if descriptor.name.casefold().startswith(prefix.casefold())
        }
        return catalog

    async def probe_commands(self, context: CommandContext) -> CommandCatalog:
        """Read runtime entries in the supplied user/session/workspace context.

        Discovery must not send a model prompt or create a visible turn. If a
        runtime needs a session before it can discover entries, report unavailable
        and explain that prerequisite; do not silently create one. Adapters with
        no discovery API retain this explicit unsupported result.
        """
        return CommandCatalog(
            context=context,
            status="unsupported",
            message=f"{self.id} does not expose command discovery.",
        )

    async def execute_command(
        self,
        context: CommandContext,
        invocation: CommandInvocation,
        *,
        deferred_suspender: DeferredSuspender | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    ) -> AsyncGenerator[CommandOutcome | ReactStreamEvent[AgentOutputT], None]:
        """Invoke a revalidated runtime entry without passing through chat triage.

        Direct controls yield one CommandOutcome without invoking `run`. Entries
        that run the agent yield the same events as `run_stream_events`, ending
        with an AgentRunResultEvent. Consuming this generator drives one native
        invocation; closing it must release that invocation's resources. The
        caller must close it even when abandoning the stream and hold the
        conversation's active-turn guard through cleanup. Forward the suspender
        and capabilities for agent runs; direct controls do not consume them.

        Resolve the opaque invocation id using this tentacle's current catalog.
        Execution behavior belongs to the invocation, not a fixed descriptor tag:
        a command may return a control result or run the agent depending on input.
        If its descriptor permits execution without a conversation, it must yield
        only a direct outcome in that context, without creating a session or run.

        Preserve invocation arguments verbatim. Never turn an unsupported or
        unknown invocation into a normal model prompt. Authorization, current
        catalog validation and delivery deduplication belong to the host caller;
        these adapter hooks do not establish any of them.
        """
        yield CommandError(
            status="unsupported",
            message=f"{self.id} does not support explicit command execution.",
        )

    def set_model_catalog(
        self, models: dict[str, Model | str], claims: dict[str, Claim]
    ) -> None:
        self.models = models
        self.claims = claims
        self.routes = self.build_routes()

    async def user_capabilities(
        self,
        profile: UserProfile,
    ) -> list[AgentCapability[AgentDepsT]]:
        """Build capabilities whose credentials belong to this run's user."""
        return []

    def resumed_prompt(self, results: DeferredToolResults) -> str:
        """What a resumed run opens with, for a runtime that takes no tool result
        back: what the graph resolved the deferral with, spoken as its next prompt
        — the sentence of a deferral the graph performed itself, or a person's
        answers and verdicts on a batch they came back to."""
        spoken: list[str] = []
        for value in results.calls.values():
            spoken.append(
                "\n".join(str(one) for one in value)
                if isinstance(value, list)
                else str(value)
            )
        for verdict in results.approvals.values():
            if isinstance(verdict, ToolDenied):
                spoken.append(f"Denied: {verdict.message}")
            else:
                spoken.append("Approved." if verdict else "Denied.")
        return "\n\n".join(spoken)

    async def fork(
        self,
        source: Conversation,
        destination: ThreadKey,
        *,
        sender: UserProfile,
        model: AgentRouteModelName | None = None,
    ) -> Thread:
        """Fork owned native or driven history onto an independent surface.

        Native imports use `fork_transcript`. Driven forks use `fork_session` and
        copy the history, settings and project workspace, retaining the source
        session. The caller opens the destination and holds the runtime's turn
        guard while forking a driven conversation.
        """
        owner_id = sender.user_id
        if owner_id is None:
            raise ValueError("A fork requires a registered owner")
        source = await self.conversations.get(source.id)
        if (
            source.agent_tentacle_id not in {self.id, self.native_id}
            or source.subagent_id
        ):
            raise ValueError("Only a root conversation of this runtime can be forked")
        driven = source.agent_tentacle_id == self.id
        threads = self.threads
        source_thread = await threads.get(source.thread_id, with_messages=False)
        if source_thread is None:
            raise FileNotFoundError("No conversation")
        surface = await threads.get(
            source_thread.parent_thread_id or source_thread.id,
            with_messages=False,
            user_id=owner_id,
        )
        if surface is None:
            raise FileNotFoundError("No conversation")
        if not driven:
            if source.transcript_file_id is None:
                raise ValueError("The session transcript has not been uploaded yet")
            await self.files.get(source.transcript_file_id, owner_id=owner_id)
        project = await self.projects.of(source_thread)
        if (
            not driven
            and project is not None
            and not await anyio.Path(project.root).is_dir()
        ):
            project = None
        thread = await threads.ensure(destination, project=project)
        if thread.id == source_thread.id:
            raise ValueError("A fork requires an independent destination thread")
        target = await self.conversations.ensure(thread.id, agent_tentacle_id=self.id)
        workspace = self.workspaces.open(thread.id, project)
        if driven:
            await self.workspaces.carry(source_thread, thread)
            async with workspace:
                external_id = await self.fork_session(source, cwd=workspace.path)
            if source.external_id is not None and (
                external_id is None or external_id == source.external_id
            ):
                raise ValueError("The runtime did not create an independent session.")
            await self.conversations.fork(source, target, external_id=external_id)
            await threads.record_handoff(
                thread,
                to_agent_tentacle_id=self.id,
                to_model=model,
                source_conversation_id=source.id,
                target_conversation_id=target.id,
            )
        else:
            await self.fork_transcript(
                source, target, owner_id=owner_id, cwd=workspace.path
            )
        return await threads.record_fork(
            source,
            thread,
            sender=sender,
            title=threads.fork_title(source, source_thread),
        )

    async def validate_fork(self, source: Conversation, *, sender: UserProfile) -> None:
        """Refuse unusable native history before a destination is created,
        including a session last run on a model this agent does not serve."""
        if sender.user_id is None:
            raise ValueError("A native fork requires a registered owner")
        _, completed = await self.read_fork_transcript(source, owner_id=sender.user_id)
        model = completed.model_name
        if model is not None and self.served_model(model) is None:
            raise ValueError(
                f"This session last ran {model!r}, which {self.id!r} does not offer "
                "here; the server's runtime needs updating to continue it."
            )

    async def read_fork_transcript(
        self, source: Conversation, *, owner_id: UUID7
    ) -> tuple[bytes, ExternalAgentRun]:
        """An owner's uploaded native history up to its latest whole turn, and that
        turn — what a fork of it carries."""
        raise NotImplementedError(
            f"Agent {self.id!r} does not support conversation forking"
        )

    async def fork_transcript(
        self,
        source: Conversation,
        target: Conversation,
        *,
        owner_id: UUID7,
        cwd: Path,
    ) -> Conversation:
        """Import an owner's native history into the empty `target`, as a session
        this runtime resumes in `cwd`."""
        raise NotImplementedError(
            f"Agent {self.id!r} does not support conversation forking"
        )

    @property
    def supports_session_fork(self) -> bool:
        """Whether this harness implements independent runtime session creation."""
        return type(self).fork_session is not AgentTentacle.fork_session

    async def fork_session(
        self, conversation: Conversation, *, cwd: Path
    ) -> str | None:
        """Fork the runtime session, or stop if this agent cannot fork safely."""
        raise NotImplementedError(f"Agent {self.id!r} does not support session forking")

    async def relocate(self, conversation: Conversation, *, cwd: Path) -> None:
        """Relocate the runtime session behind `conversation` to `cwd`, where its next
        run resumes. The graph's call, made once a teleport has settled where the
        agent lands; the tentacle only knows how to move one.

        Nothing to move for a runtime that resumes a session from anywhere. Claude
        files one under the directory it ran in, and every teleport changes that
        directory — a new thread's workspace, or the project's — so a driven Claude
        session needs this on each. A native session's transcript is on its own
        machine, so a native teleport has nothing here to move: bringing it over is
        a rebuild from the ledger, still to come, and this is where it lands."""

    async def run_project(self, thread_id: UUID7) -> Project | None:
        """The project a run in this thread is in, or None when it is in none.

        The thread is where a project is bound, and every conversation belongs to
        one, so this is the whole of "which project is this run in" — asked of the
        thread each run rather than copied onto the conversation. Reading the
        thread is what this adds; judging what it names is the registry's, and is
        `ProjectManager.of`.
        """
        if not self.projects.roots:
            # Nothing registered: no thread can be in a project, so a run is what it
            # was before there were projects, down to not reading the thread.
            return None
        thread = await self.threads.get(thread_id)
        if thread is None:
            raise ValueError(f"unknown thread {thread_id}")
        return await self.projects.of(thread)

    @overload
    async def run(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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,
        event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: AgentSpecInput | None = None,
    ) -> AgentRunResult[AgentOutputT]: ...

    @overload
    async def run(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: OutputSpec[RunOutputDataT],
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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,
        event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: AgentSpecInput | None = None,
    ) -> AgentRunResult[RunOutputDataT]: ...

    @abstractmethod
    async def run(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: OutputSpec[RunOutputDataT] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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,
        event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: AgentSpecInput | None = None,
    ) -> AgentRunResult[AgentOutputT | RunOutputDataT]:
        """Run the agent for an Octomate conversation."""

    async def subagent_run(
        self: AgentTentacle[AgentOutputT, None],
        user_prompt: str,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7,
        conversation_id: UUID7,
        run_name: str | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        instructions: str | None = None,
        capabilities: Sequence[AgentCapability[None]] | None = None,
    ) -> AgentRunResult[AgentOutputT]:
        """Run as a subagent: the subagent contract in one place, apart from
        `run()`'s general surface.

        A subagent run is addressed at a pre-ensured conversation its spawner
        owns (`conversation_id` is required, not optional), and it is fully
        non-interactive — every human interaction declines at once. What
        `capabilities` mounts is the runtime's to decide: inkling adds the
        spawner's set to its own, claude reads it only for a `GatewayCapability`,
        and codex ignores it. The spawner passes its framing as `instructions`
        and stamps the run tree after the report returns.
        """
        return await self.run(
            user_prompt,
            conversation_address=conversation_address,
            thread_id=thread_id,
            conversation_id=conversation_id,
            interactive=False,
            run_name=run_name,
            model=model,
            effort=effort,
            instructions=instructions,
            capabilities=capabilities,
        )

    @overload
    def run_stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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: AgentSpecInput | None = None,
    ) -> ReactEventStream[AgentOutputT]: ...

    @overload
    def run_stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: OutputSpec[RunOutputDataT],
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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: AgentSpecInput | None = None,
    ) -> ReactEventStream[RunOutputDataT]: ...

    @abstractmethod
    def run_stream_events(
        self,
        user_prompt: str | Sequence[UserContent] | None = None,
        *,
        conversation_address: ChannelAddress,
        thread_id: UUID7 | None = None,
        source_thread_address: ChannelAddress | None = None,
        source_thread_message_ids: Sequence[UUID7] | None = None,
        run_name: str | None = None,
        output_type: OutputSpec[RunOutputDataT] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        deferred_suspender: DeferredSuspender | None = None,
        model: Model | KnownModelName | str | None = None,
        effort: str | None = None,
        conversation_id: UUID7 | None = None,
        interactive: bool = True,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: 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: AgentSpecInput | None = None,
    ) -> ReactEventStream[AgentOutputT | RunOutputDataT]:
        """Stream raw agent events for an Octomate conversation."""

conversations instance-attribute

conversations: ConversationManager

workspaces instance-attribute

workspaces: WorkspaceManager

commands instance-attribute

commands: CommandManager = commands

projects instance-attribute

projects: ProjectManager = projects

threads instance-attribute

threads: ThreadManager = threads

files instance-attribute

files: FileManager = files

description class-attribute instance-attribute

description: str = (
    "General-purpose agent for handling user requests."
)

claims class-attribute instance-attribute

claims: Mapping[AgentRouteModelName, Claim] = (
    MappingProxyType({})
)

gateway class-attribute instance-attribute

gateway: bool = True

native_id class-attribute

native_id: str | None = None

info property

info: AgentInfo

driven_sessions cached property

driven_sessions: Counter[str]

Live runs holding each external runtime session on this agent.

native_sessions cached property

native_sessions: Counter[str]

Accepted native transcript streams currently attached to this agent.

run_tasks cached property

run_tasks: set[Task[None]]

Collectors registered by agent harnesses that opt into observe_run.

Created on first access. Participating harnesses drain these tasks during shutdown; other harnesses do not need this registry.

routes cached property

routes: list[AgentRoute]

Every served model, using discovered or configured routing metadata.

default_model cached property

default_model: str | None

Configured first model, replaced by discovery with the harness's default.

in_process class-attribute

in_process: bool = False

pendings instance-attribute

pendings: dict[UUID7, Future[DeferredActionBatchResponse]]

permission_modes class-attribute instance-attribute

permission_modes: tuple[PermissionMode, ...] = ()

default_permission_mode property

default_permission_mode: str | None

The posture this agent's conversations run under when they declare none — its configured default, in its own vocabulary.

A row's NULL is not "no posture", it is "nothing said here"; this is what decides instead, and it is what the console reports for such a row rather than showing a blank. None only for an agent that answers to no posture at all.

models instance-attribute

models: dict[AgentRouteModelName, Model | str]

supports_session_fork property

supports_session_fork: bool

Whether this harness implements independent runtime session creation.

observe_run async

observe_run(
    events: AsyncGenerator[
        ReactStreamEvent[RunOutputDataT], None
    ],
) -> AsyncGenerator[ReactStreamEvent[RunOutputDataT], None]

Collect a run to completion even when its observer closes or cancels.

Agent harnesses opt in by wrapping their event generators with this method and draining run_tasks during shutdown. Inheriting it alone does not change a harness's lifecycle.

Joining before leaving keeps the caller's gateway and other enclosing resources available to the agent until its final notifications are recorded.

Source code in octomate/tentacles/agent.py

async def observe_run(
    self, events: AsyncGenerator[ReactStreamEvent[RunOutputDataT], None]
) -> AsyncGenerator[ReactStreamEvent[RunOutputDataT], None]:
    """Collect a run to completion even when its observer closes or cancels.

    Agent harnesses opt in by wrapping their event generators with this
    method and draining `run_tasks` during shutdown. Inheriting it alone
    does not change a harness's lifecycle.

    Joining before leaving keeps the caller's gateway and other enclosing
    resources available to the agent until its final notifications are recorded.
    """
    send, receive = anyio.create_memory_object_stream[
        ReactStreamEvent[RunOutputDataT]
    ](100)
    errors: list[Exception | asyncio.CancelledError] = []

    async def collect() -> None:
        try:
            async with aclosing(events):
                async for event in events:
                    try:
                        await send.send(event)
                    except anyio.BrokenResourceError:
                        # Only the observer is gone; keep ingesting the run.
                        pass
        except (Exception, asyncio.CancelledError) as error:
            errors.append(error)
        finally:
            send.close()

    task = asyncio.create_task(collect())
    self.run_tasks.add(task)
    observed_end = False
    try:
        async with receive:
            async for event in receive:
                yield event
        observed_end = True
        for error in errors:
            raise error
    finally:
        receive.close()
        cancelled = await drain_task(task)
        self.run_tasks.discard(task)
        if not observed_end:
            for error in errors:
                logging.getLogger(type(self).__module__).error(
                    "Agent %s run failed after observer detached",
                    self.id,
                    exc_info=error,
                )
        if cancelled:
            raise asyncio.CancelledError

driving async

driving(
    session_id: str, *, native: bool = False
) -> AsyncGenerator[None]

Count a driven runtime session or an accepted native stream.

Driven runs hold the count through cleanup, including interruption or cancellation. Native streams hold it from their accepted handshake through stream cleanup.

Source code in octomate/tentacles/agent.py

@asynccontextmanager
async def driving(
    self, session_id: str, *, native: bool = False
) -> AsyncGenerator[None]:
    """Count a driven runtime session or an accepted native stream.

    Driven runs hold the count through cleanup, including interruption or
    cancellation. Native streams hold it from their accepted handshake
    through stream cleanup.
    """
    sessions = self.native_sessions if native else self.driven_sessions
    sessions[session_id] += 1
    try:
        yield
    finally:
        sessions[session_id] -= 1
        if sessions[session_id] == 0:
            del sessions[session_id]

build_routes

build_routes() -> list[AgentRoute]

Source code in octomate/tentacles/agent.py

def build_routes(self) -> list[AgentRoute]:
    return [
        AgentRoute(
            agent_id=self.id,
            model=model,
            claim=self.claims.get(model) or Claim(self.description, efforts=()),
        )
        for model in self.models
    ]

resolve_model

resolve_model(
    model: AgentRouteModelName | None = None,
) -> AgentRouteModelName | None

Resolve a default, exact model or unambiguous provider-less name.

Source code in octomate/tentacles/agent.py

def resolve_model(
    self, model: AgentRouteModelName | None = None
) -> AgentRouteModelName | None:
    """Resolve a default, exact model or unambiguous provider-less name."""
    if not self.models:
        raise ValueError(f"agent {self.id!r} has no available model catalog")
    if model is None:
        # Native harnesses retain their settings and resumed model selection.
        return self.default_model if self.native_id is None else None
    served = self.served_model(model)
    if served is not None:
        return served

    raise ValueError(
        f"agent {self.id!r} does not serve model {model!r} with an unambiguous provider"
    )

served_model

served_model(name: str) -> str | None

The catalog entry a stored model name means: the entry itself, or, for a name saved without its provider, the one entry it can be. None when this agent serves no such model.

Source code in octomate/tentacles/agent.py

def served_model(self, name: str) -> str | None:
    """The catalog entry a stored model name means: the entry itself, or, for
    a name saved without its provider, the one entry it can be. None when this
    agent serves no such model."""
    if name in self.models:
        return name
    if name and ":" not in name:
        matches = [key for key in self.models if key.partition(":")[2] == name]
        if len(matches) == 1:
            return matches[0]
    return None

check_permission_mode

check_permission_mode(mode: str) -> None

Reject selections absent from this tentacle's current catalog.

Source code in octomate/tentacles/agent.py

def check_permission_mode(self, mode: str) -> None:
    """Reject selections absent from this tentacle's current catalog."""
    if not any(option.value == mode for option in self.permission_modes):
        raise ValueError(f"{mode!r} is not one of {self.id}'s modes")

check_effort

check_effort(
    model: AgentRouteModelName | None, effort: str
) -> None

Reject a level the route for model does not claim.

Source code in octomate/tentacles/agent.py

def check_effort(self, model: AgentRouteModelName | None, effort: str) -> None:
    """Reject a level the route for `model` does not claim."""
    model = self.resolve_model(model or self.default_model)
    if model is None:
        raise ValueError(
            "The runtime's model is unknown; select a model to set effort"
        )
    route = next((route for route in self.routes if route.model == model), None)
    efforts = route.claim.efforts if route is not None else ()
    if effort not in efforts:
        raise ValueError(
            f"{self.id} ({model}) does not take effort {effort!r}; it claims "
            f"{'/'.join(efforts) or 'none'}"
        )

resolve_effort async

resolve_effort(
    conversation: Conversation,
    *,
    model: AgentRouteModelName | None,
    effort: str | None = None,
) -> str | None

Validate the run's override or stored effort against its selected model.

With no explicit model, use the conversation's last reported model, then the harness's catalog default for a conversation that has not run yet.

Source code in octomate/tentacles/agent.py

async def resolve_effort(
    self,
    conversation: Conversation,
    *,
    model: AgentRouteModelName | None,
    effort: str | None = None,
) -> str | None:
    """Validate the run's override or stored effort against its selected model.

    With no explicit model, use the conversation's last reported model, then
    the harness's catalog default for a conversation that has not run yet.
    """
    selected = effort if effort is not None else conversation.effort
    if selected is None:
        return None
    effective_model = model or await self.conversations.latest_model(
        conversation.id
    )
    self.check_effort(effective_model, selected)
    return selected

set_permission_mode async

set_permission_mode(
    conversation: Conversation, mode: str | None
) -> Conversation

Validate and save a selection, updating a live runtime when supported.

None restores the configured default. Native update failures propagate before the selection is saved. Runtimes without live updates read the saved selection on their next run; existing approvals remain separate.

Source code in octomate/tentacles/agent.py

async def set_permission_mode(
    self, conversation: Conversation, mode: str | None
) -> Conversation:
    """Validate and save a selection, updating a live runtime when supported.

    None restores the configured default. Native update failures propagate
    before the selection is saved. Runtimes without live updates read the
    saved selection on their next run; existing approvals remain separate.
    """
    if conversation.agent_tentacle_id != self.id:
        raise ValueError("conversation belongs to another agent")
    effective = mode if mode is not None else self.default_permission_mode
    if effective is not None:
        self.check_permission_mode(effective)
        await self.apply_permission_mode(conversation, effective)
    updated = await self.conversations.set_permission_mode(conversation, mode)
    self.commands.invalidate(agent_id=self.id, conversation_id=conversation.id)
    return updated

apply_permission_mode async

apply_permission_mode(
    conversation: Conversation, mode: str
) -> None

Update an active native runtime when it supports permission changes.

The default leaves in-flight work alone. Codex applies both permission axes at the next turn start; Inkling reads storage at each deferral.

Source code in octomate/tentacles/agent.py

async def apply_permission_mode(
    self, conversation: Conversation, mode: str
) -> None:
    """Update an active native runtime when it supports permission changes.

    The default leaves in-flight work alone. Codex applies both permission
    axes at the next turn start; Inkling reads storage at each deferral.
    """

discover_models async

discover_models() -> None

Refresh models and claims once the harness connection is ready.

Harnesses call this during entry and install results with set_model_catalog. Agents with config-supplied catalogs keep their existing models and claims.

Source code in octomate/tentacles/agent.py

async def discover_models(self) -> None:
    """Refresh models and claims once the harness connection is ready.

    Harnesses call this during entry and install results with set_model_catalog.
    Agents with config-supplied catalogs keep their existing models and claims.
    """

discover_commands async

discover_commands(
    context: CommandContext,
    *,
    refresh: bool = False,
    prefix: str = "",
) -> CommandCatalog

Read this agent's cached catalog for an already authorized context.

Callers resolve and authorize context before every lookup, including cache hits. Prefix matching filters a copy by command name; it does not imply argument-value completion. Runtime adapters implement probe_commands.

Source code in octomate/tentacles/agent.py

async def discover_commands(
    self,
    context: CommandContext,
    *,
    refresh: bool = False,
    prefix: str = "",
) -> CommandCatalog:
    """Read this agent's cached catalog for an already authorized context.

    Callers resolve and authorize context before every lookup, including cache
    hits. Prefix matching filters a copy by command name; it does not imply
    argument-value completion. Runtime adapters implement `probe_commands`.
    """
    catalog = await self.commands.discover(self, context, refresh=refresh)
    catalog.descriptors = {
        descriptor
        for descriptor in catalog.descriptors
        if descriptor.name.casefold().startswith(prefix.casefold())
    }
    return catalog

probe_commands async

probe_commands(context: CommandContext) -> CommandCatalog

Read runtime entries in the supplied user/session/workspace context.

Discovery must not send a model prompt or create a visible turn. If a runtime needs a session before it can discover entries, report unavailable and explain that prerequisite; do not silently create one. Adapters with no discovery API retain this explicit unsupported result.

Source code in octomate/tentacles/agent.py

async def probe_commands(self, context: CommandContext) -> CommandCatalog:
    """Read runtime entries in the supplied user/session/workspace context.

    Discovery must not send a model prompt or create a visible turn. If a
    runtime needs a session before it can discover entries, report unavailable
    and explain that prerequisite; do not silently create one. Adapters with
    no discovery API retain this explicit unsupported result.
    """
    return CommandCatalog(
        context=context,
        status="unsupported",
        message=f"{self.id} does not expose command discovery.",
    )

execute_command async

execute_command(
    context: CommandContext,
    invocation: CommandInvocation,
    *,
    deferred_suspender: DeferredSuspender | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
) -> AsyncGenerator[
    CommandOutcome | ReactStreamEvent[AgentOutputT], None
]

Invoke a revalidated runtime entry without passing through chat triage.

Direct controls yield one CommandOutcome without invoking run. Entries that run the agent yield the same events as run_stream_events, ending with an AgentRunResultEvent. Consuming this generator drives one native invocation; closing it must release that invocation's resources. The caller must close it even when abandoning the stream and hold the conversation's active-turn guard through cleanup. Forward the suspender and capabilities for agent runs; direct controls do not consume them.

Resolve the opaque invocation id using this tentacle's current catalog. Execution behavior belongs to the invocation, not a fixed descriptor tag: a command may return a control result or run the agent depending on input. If its descriptor permits execution without a conversation, it must yield only a direct outcome in that context, without creating a session or run.

Preserve invocation arguments verbatim. Never turn an unsupported or unknown invocation into a normal model prompt. Authorization, current catalog validation and delivery deduplication belong to the host caller; these adapter hooks do not establish any of them.

Source code in octomate/tentacles/agent.py

async def execute_command(
    self,
    context: CommandContext,
    invocation: CommandInvocation,
    *,
    deferred_suspender: DeferredSuspender | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
) -> AsyncGenerator[CommandOutcome | ReactStreamEvent[AgentOutputT], None]:
    """Invoke a revalidated runtime entry without passing through chat triage.

    Direct controls yield one CommandOutcome without invoking `run`. Entries
    that run the agent yield the same events as `run_stream_events`, ending
    with an AgentRunResultEvent. Consuming this generator drives one native
    invocation; closing it must release that invocation's resources. The
    caller must close it even when abandoning the stream and hold the
    conversation's active-turn guard through cleanup. Forward the suspender
    and capabilities for agent runs; direct controls do not consume them.

    Resolve the opaque invocation id using this tentacle's current catalog.
    Execution behavior belongs to the invocation, not a fixed descriptor tag:
    a command may return a control result or run the agent depending on input.
    If its descriptor permits execution without a conversation, it must yield
    only a direct outcome in that context, without creating a session or run.

    Preserve invocation arguments verbatim. Never turn an unsupported or
    unknown invocation into a normal model prompt. Authorization, current
    catalog validation and delivery deduplication belong to the host caller;
    these adapter hooks do not establish any of them.
    """
    yield CommandError(
        status="unsupported",
        message=f"{self.id} does not support explicit command execution.",
    )

set_model_catalog

set_model_catalog(
    models: dict[str, Model | str], claims: dict[str, Claim]
) -> None

Source code in octomate/tentacles/agent.py

def set_model_catalog(
    self, models: dict[str, Model | str], claims: dict[str, Claim]
) -> None:
    self.models = models
    self.claims = claims
    self.routes = self.build_routes()

user_capabilities async

user_capabilities(
    profile: UserProfile,
) -> list[AgentCapability[AgentDepsT]]

Build capabilities whose credentials belong to this run's user.

Source code in octomate/tentacles/agent.py

async def user_capabilities(
    self,
    profile: UserProfile,
) -> list[AgentCapability[AgentDepsT]]:
    """Build capabilities whose credentials belong to this run's user."""
    return []

resumed_prompt

resumed_prompt(results: DeferredToolResults) -> str

What a resumed run opens with, for a runtime that takes no tool result back: what the graph resolved the deferral with, spoken as its next prompt — the sentence of a deferral the graph performed itself, or a person's answers and verdicts on a batch they came back to.

Source code in octomate/tentacles/agent.py

def resumed_prompt(self, results: DeferredToolResults) -> str:
    """What a resumed run opens with, for a runtime that takes no tool result
    back: what the graph resolved the deferral with, spoken as its next prompt
    — the sentence of a deferral the graph performed itself, or a person's
    answers and verdicts on a batch they came back to."""
    spoken: list[str] = []
    for value in results.calls.values():
        spoken.append(
            "\n".join(str(one) for one in value)
            if isinstance(value, list)
            else str(value)
        )
    for verdict in results.approvals.values():
        if isinstance(verdict, ToolDenied):
            spoken.append(f"Denied: {verdict.message}")
        else:
            spoken.append("Approved." if verdict else "Denied.")
    return "\n\n".join(spoken)

fork async

fork(
    source: Conversation,
    destination: ThreadKey,
    *,
    sender: UserProfile,
    model: AgentRouteModelName | None = None,
) -> Thread

Fork owned native or driven history onto an independent surface.

Native imports use fork_transcript. Driven forks use fork_session and copy the history, settings and project workspace, retaining the source session. The caller opens the destination and holds the runtime's turn guard while forking a driven conversation.

Source code in octomate/tentacles/agent.py

async def fork(
    self,
    source: Conversation,
    destination: ThreadKey,
    *,
    sender: UserProfile,
    model: AgentRouteModelName | None = None,
) -> Thread:
    """Fork owned native or driven history onto an independent surface.

    Native imports use `fork_transcript`. Driven forks use `fork_session` and
    copy the history, settings and project workspace, retaining the source
    session. The caller opens the destination and holds the runtime's turn
    guard while forking a driven conversation.
    """
    owner_id = sender.user_id
    if owner_id is None:
        raise ValueError("A fork requires a registered owner")
    source = await self.conversations.get(source.id)
    if (
        source.agent_tentacle_id not in {self.id, self.native_id}
        or source.subagent_id
    ):
        raise ValueError("Only a root conversation of this runtime can be forked")
    driven = source.agent_tentacle_id == self.id
    threads = self.threads
    source_thread = await threads.get(source.thread_id, with_messages=False)
    if source_thread is None:
        raise FileNotFoundError("No conversation")
    surface = await threads.get(
        source_thread.parent_thread_id or source_thread.id,
        with_messages=False,
        user_id=owner_id,
    )
    if surface is None:
        raise FileNotFoundError("No conversation")
    if not driven:
        if source.transcript_file_id is None:
            raise ValueError("The session transcript has not been uploaded yet")
        await self.files.get(source.transcript_file_id, owner_id=owner_id)
    project = await self.projects.of(source_thread)
    if (
        not driven
        and project is not None
        and not await anyio.Path(project.root).is_dir()
    ):
        project = None
    thread = await threads.ensure(destination, project=project)
    if thread.id == source_thread.id:
        raise ValueError("A fork requires an independent destination thread")
    target = await self.conversations.ensure(thread.id, agent_tentacle_id=self.id)
    workspace = self.workspaces.open(thread.id, project)
    if driven:
        await self.workspaces.carry(source_thread, thread)
        async with workspace:
            external_id = await self.fork_session(source, cwd=workspace.path)
        if source.external_id is not None and (
            external_id is None or external_id == source.external_id
        ):
            raise ValueError("The runtime did not create an independent session.")
        await self.conversations.fork(source, target, external_id=external_id)
        await threads.record_handoff(
            thread,
            to_agent_tentacle_id=self.id,
            to_model=model,
            source_conversation_id=source.id,
            target_conversation_id=target.id,
        )
    else:
        await self.fork_transcript(
            source, target, owner_id=owner_id, cwd=workspace.path
        )
    return await threads.record_fork(
        source,
        thread,
        sender=sender,
        title=threads.fork_title(source, source_thread),
    )

validate_fork async

validate_fork(
    source: Conversation, *, sender: UserProfile
) -> None

Refuse unusable native history before a destination is created, including a session last run on a model this agent does not serve.

Source code in octomate/tentacles/agent.py

async def validate_fork(self, source: Conversation, *, sender: UserProfile) -> None:
    """Refuse unusable native history before a destination is created,
    including a session last run on a model this agent does not serve."""
    if sender.user_id is None:
        raise ValueError("A native fork requires a registered owner")
    _, completed = await self.read_fork_transcript(source, owner_id=sender.user_id)
    model = completed.model_name
    if model is not None and self.served_model(model) is None:
        raise ValueError(
            f"This session last ran {model!r}, which {self.id!r} does not offer "
            "here; the server's runtime needs updating to continue it."
        )

read_fork_transcript async

read_fork_transcript(
    source: Conversation, *, owner_id: UUID7
) -> tuple[bytes, ExternalAgentRun]

An owner's uploaded native history up to its latest whole turn, and that turn — what a fork of it carries.

Source code in octomate/tentacles/agent.py

async def read_fork_transcript(
    self, source: Conversation, *, owner_id: UUID7
) -> tuple[bytes, ExternalAgentRun]:
    """An owner's uploaded native history up to its latest whole turn, and that
    turn — what a fork of it carries."""
    raise NotImplementedError(
        f"Agent {self.id!r} does not support conversation forking"
    )

fork_transcript async

fork_transcript(
    source: Conversation,
    target: Conversation,
    *,
    owner_id: UUID7,
    cwd: Path,
) -> Conversation

Import an owner's native history into the empty target, as a session this runtime resumes in cwd.

Source code in octomate/tentacles/agent.py

async def fork_transcript(
    self,
    source: Conversation,
    target: Conversation,
    *,
    owner_id: UUID7,
    cwd: Path,
) -> Conversation:
    """Import an owner's native history into the empty `target`, as a session
    this runtime resumes in `cwd`."""
    raise NotImplementedError(
        f"Agent {self.id!r} does not support conversation forking"
    )

fork_session async

fork_session(
    conversation: Conversation, *, cwd: Path
) -> str | None

Fork the runtime session, or stop if this agent cannot fork safely.

Source code in octomate/tentacles/agent.py

async def fork_session(
    self, conversation: Conversation, *, cwd: Path
) -> str | None:
    """Fork the runtime session, or stop if this agent cannot fork safely."""
    raise NotImplementedError(f"Agent {self.id!r} does not support session forking")

relocate async

relocate(conversation: Conversation, *, cwd: Path) -> None

Relocate the runtime session behind conversation to cwd, where its next run resumes. The graph's call, made once a teleport has settled where the agent lands; the tentacle only knows how to move one.

Nothing to move for a runtime that resumes a session from anywhere. Claude files one under the directory it ran in, and every teleport changes that directory — a new thread's workspace, or the project's — so a driven Claude session needs this on each. A native session's transcript is on its own machine, so a native teleport has nothing here to move: bringing it over is a rebuild from the ledger, still to come, and this is where it lands.

Source code in octomate/tentacles/agent.py

async def relocate(self, conversation: Conversation, *, cwd: Path) -> None:
    """Relocate the runtime session behind `conversation` to `cwd`, where its next
    run resumes. The graph's call, made once a teleport has settled where the
    agent lands; the tentacle only knows how to move one.

    Nothing to move for a runtime that resumes a session from anywhere. Claude
    files one under the directory it ran in, and every teleport changes that
    directory — a new thread's workspace, or the project's — so a driven Claude
    session needs this on each. A native session's transcript is on its own
    machine, so a native teleport has nothing here to move: bringing it over is
    a rebuild from the ledger, still to come, and this is where it lands."""

run_project async

run_project(thread_id: UUID7) -> Project | None

The project a run in this thread is in, or None when it is in none.

The thread is where a project is bound, and every conversation belongs to one, so this is the whole of "which project is this run in" — asked of the thread each run rather than copied onto the conversation. Reading the thread is what this adds; judging what it names is the registry's, and is ProjectManager.of.

Source code in octomate/tentacles/agent.py

async def run_project(self, thread_id: UUID7) -> Project | None:
    """The project a run in this thread is in, or None when it is in none.

    The thread is where a project is bound, and every conversation belongs to
    one, so this is the whole of "which project is this run in" — asked of the
    thread each run rather than copied onto the conversation. Reading the
    thread is what this adds; judging what it names is the registry's, and is
    `ProjectManager.of`.
    """
    if not self.projects.roots:
        # Nothing registered: no thread can be in a project, so a run is what it
        # was before there were projects, down to not reading the thread.
        return None
    thread = await self.threads.get(thread_id)
    if thread is None:
        raise ValueError(f"unknown thread {thread_id}")
    return await self.projects.of(thread)

run abstractmethod async

run(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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,
    event_stream_handler: EventStreamHandler[AgentDepsT]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: AgentSpecInput | None = None,
) -> AgentRunResult[AgentOutputT]

run(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT],
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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,
    event_stream_handler: EventStreamHandler[AgentDepsT]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: AgentSpecInput | None = None,
) -> AgentRunResult[RunOutputDataT]

run(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT] | None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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,
    event_stream_handler: EventStreamHandler[AgentDepsT]
    | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]]
    | None = None,
    spec: AgentSpecInput | None = None,
) -> AgentRunResult[AgentOutputT | RunOutputDataT]

Run the agent for an Octomate conversation.

Source code in octomate/tentacles/agent.py

@abstractmethod
async def run(
    self,
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7] | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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,
    event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
    capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    spec: AgentSpecInput | None = None,
) -> AgentRunResult[AgentOutputT | RunOutputDataT]:
    """Run the agent for an Octomate conversation."""

subagent_run async

subagent_run(
    user_prompt: str,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7,
    conversation_id: UUID7,
    run_name: str | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    instructions: str | None = None,
    capabilities: Sequence[AgentCapability[None]]
    | None = None,
) -> AgentRunResult[AgentOutputT]

Run as a subagent: the subagent contract in one place, apart from run()'s general surface.

A subagent run is addressed at a pre-ensured conversation its spawner owns (conversation_id is required, not optional), and it is fully non-interactive — every human interaction declines at once. What capabilities mounts is the runtime's to decide: inkling adds the spawner's set to its own, claude reads it only for a GatewayCapability, and codex ignores it. The spawner passes its framing as instructions and stamps the run tree after the report returns.

Source code in octomate/tentacles/agent.py

async def subagent_run(
    self: AgentTentacle[AgentOutputT, None],
    user_prompt: str,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7,
    conversation_id: UUID7,
    run_name: str | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    instructions: str | None = None,
    capabilities: Sequence[AgentCapability[None]] | None = None,
) -> AgentRunResult[AgentOutputT]:
    """Run as a subagent: the subagent contract in one place, apart from
    `run()`'s general surface.

    A subagent run is addressed at a pre-ensured conversation its spawner
    owns (`conversation_id` is required, not optional), and it is fully
    non-interactive — every human interaction declines at once. What
    `capabilities` mounts is the runtime's to decide: inkling adds the
    spawner's set to its own, claude reads it only for a `GatewayCapability`,
    and codex ignores it. The spawner passes its framing as `instructions`
    and stamps the run tree after the report returns.
    """
    return await self.run(
        user_prompt,
        conversation_address=conversation_address,
        thread_id=thread_id,
        conversation_id=conversation_id,
        interactive=False,
        run_name=run_name,
        model=model,
        effort=effort,
        instructions=instructions,
        capabilities=capabilities,
    )

run_stream_events abstractmethod

run_stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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: AgentSpecInput | None = None,
) -> ReactEventStream[AgentOutputT]

run_stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT],
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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: AgentSpecInput | None = None,
) -> ReactEventStream[RunOutputDataT]

run_stream_events(
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7]
    | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT] | None = None,
    deferred_tool_results: DeferredToolResults
    | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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: AgentSpecInput | None = None,
) -> ReactEventStream[AgentOutputT | RunOutputDataT]

Stream raw agent events for an Octomate conversation.

Source code in octomate/tentacles/agent.py

@abstractmethod
def run_stream_events(
    self,
    user_prompt: str | Sequence[UserContent] | None = None,
    *,
    conversation_address: ChannelAddress,
    thread_id: UUID7 | None = None,
    source_thread_address: ChannelAddress | None = None,
    source_thread_message_ids: Sequence[UUID7] | None = None,
    run_name: str | None = None,
    output_type: OutputSpec[RunOutputDataT] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    deferred_suspender: DeferredSuspender | None = None,
    model: Model | KnownModelName | str | None = None,
    effort: str | None = None,
    conversation_id: UUID7 | None = None,
    interactive: bool = True,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: 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: AgentSpecInput | None = None,
) -> ReactEventStream[AgentOutputT | RunOutputDataT]:
    """Stream raw agent events for an Octomate conversation."""

octomate.tentacles.channel

Shared channel primitives.

Type variables: - RawT: native inbound platform payload accepted by a channel and its chromo. - MessageT: native outbound platform message payload built and sent by an ink.

logger module-attribute

logger = logging.getLogger(__name__)

ThreadStrategy module-attribute

ThreadStrategy = Literal['main_only', 'flat_thread']

MessageT module-attribute

MessageT = TypeVar('MessageT')

RawT module-attribute

RawT = TypeVar('RawT')

ChannelOutput

ChannelOutput = (
    str
    | Sequence[MessageSegment]
    | DeferredToolRequests
    | None
)

ChannelSurfaces dataclass

Which places this channel can open, as opposed to how it routes them.

thread_strategy says how an inbound threaded message is routed; these say what the platform can be asked to create. They are separate because a channel can want thread routing without having threads: the dev UI declares flat_thread to skip triage and can open nothing.

Declared per channel class rather than probed — a spell has to know before it runs, and asking the platform every turn would spend a round-trip to answer a constant.

Source code in octomate/tentacles/channel.py

@dataclass(frozen=True)
class ChannelSurfaces:
    """Which places this channel can open, as opposed to how it routes them.

    `thread_strategy` says how an inbound *threaded* message is routed; these say what
    the platform can be asked to create. They are separate because a channel can want
    thread routing without having threads: the dev UI declares `flat_thread` to skip
    triage and can open nothing.

    Declared per channel class rather than probed — a spell has to know before it runs,
    and asking the platform every turn would spend a round-trip to answer a constant.
    """

    sub_thread: bool = False
    direct_message: bool = False

sub_thread class-attribute instance-attribute

sub_thread: bool = False

direct_message class-attribute instance-attribute

direct_message: bool = False

DownloadedImage dataclass

An inbound image an ink fetched from the platform: its bytes and file name, with the content type and source URL when known.

Source code in octomate/tentacles/channel.py

@dataclass(frozen=True)
class DownloadedImage:
    """An inbound image an ink fetched from the platform: its bytes and file name,
    with the content type and source URL when known."""

    data: bytes
    file_name: str
    content_type: str = ""
    url: str | None = None

data instance-attribute

data: bytes

file_name instance-attribute

file_name: str

content_type class-attribute instance-attribute

content_type: str = ''

url class-attribute instance-attribute

url: str | None = None

Chromo

Bases: ABC

Two-way translation between platform-native wire data and core schemas.

sip converts inbound payloads to a MessageEvent; outbound_markdown converts an outbound markdown reply to platform-native message payloads. Translation only — the actual send/receive is the ink's job.

Source code in octomate/tentacles/channel.py

class Chromo[RawT, MessageT](
    ABC,
):
    """Two-way translation between platform-native wire data and core schemas.

    `sip` converts inbound payloads to a `MessageEvent`; `outbound_markdown`
    converts an outbound markdown reply to platform-native message payloads.
    Translation only — the actual send/receive is the ink's job.
    """

    @abstractmethod
    async def sip(self, raw: RawT) -> MessageEvent | None: ...

    @abstractmethod
    def outbound_markdown(self, text: str) -> list[MessageT]:
        """Encode markdown text as platform-native outbound message payloads."""

    async def outbound_segments(self, segments: list[MessageSegment]) -> list[MessageT]:
        """Encode output segments as platform-native outbound message payloads. The
        default flattens them to a single markdown body via their `str` form;
        channels with native transport override to ship media inline or to encode a
        mention token a user actually gets pinged by."""
        return self.outbound_markdown("\n\n".join(str(seg) for seg in segments))

sip abstractmethod async

sip(raw: RawT) -> MessageEvent | None

Source code in octomate/tentacles/channel.py

@abstractmethod
async def sip(self, raw: RawT) -> MessageEvent | None: ...

outbound_markdown abstractmethod

outbound_markdown(text: str) -> list[MessageT]

Encode markdown text as platform-native outbound message payloads.

Source code in octomate/tentacles/channel.py

@abstractmethod
def outbound_markdown(self, text: str) -> list[MessageT]:
    """Encode markdown text as platform-native outbound message payloads."""

outbound_segments async

outbound_segments(
    segments: list[MessageSegment],
) -> list[MessageT]

Encode output segments as platform-native outbound message payloads. The default flattens them to a single markdown body via their str form; channels with native transport override to ship media inline or to encode a mention token a user actually gets pinged by.

Source code in octomate/tentacles/channel.py

async def outbound_segments(self, segments: list[MessageSegment]) -> list[MessageT]:
    """Encode output segments as platform-native outbound message payloads. The
    default flattens them to a single markdown body via their `str` form;
    channels with native transport override to ship media inline or to encode a
    mention token a user actually gets pinged by."""
    return self.outbound_markdown("\n\n".join(str(seg) for seg in segments))

Ink

Bases: ABC

Base class for platform API clients (transport only).

An ink is an async context manager: the owning tentacle enters it as part of its own lifecycle. The base implementation holds no resources; inks whose platform calls can share a pooled connection override __aenter__/__aexit__ to open and close it.

Source code in octomate/tentacles/channel.py

class Ink[MessageT](ABC):
    """Base class for platform API clients (transport only).

    An ink is an async context manager: the owning tentacle enters it as part
    of its own lifecycle. The base implementation holds no resources; inks
    whose platform calls can share a pooled connection override
    `__aenter__`/`__aexit__` to open and close it.
    """

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, *exc: object) -> None:
        return None

    @abstractmethod
    async def inspect(self) -> UserProfile: ...

    @abstractmethod
    async def get_user_profile(self, user_id: str) -> UserProfile: ...

    async def get_group_name(self, chat_id: str) -> str | None:
        """The platform's group name, or None when this ink does not supply one."""
        return None

    @abstractmethod
    async def upload_media(self, data: bytes) -> str | None:
        """Upload media bytes and return a platform address or URL."""

    @abstractmethod
    async def download_image(
        self,
        seg: ImageSegment,
        message_id: str,
    ) -> DownloadedImage | None:
        """Download an inbound image segment."""

    @abstractmethod
    async def send_message(
        self,
        chat_id: str,
        chat_type: str,
        messages: list[MessageT],
        *,
        channel_thread_id: str,
        reply_to: str | None = None,
        reply_in_thread: bool = False,
    ) -> IMMessageID | None:
        """Send platform-native message payloads to a conversation destination.

        `channel_thread_id` is the platform's external destination id: the thread
        id on a thread surface, otherwise the chat id. `reply_to` references a
        message within that destination. Platforms where thread and reply are the
        same native field collapse them in their ink implementation.
        """

    async def suggest_addresses(
        self, address: ChannelAddress, source_address: ChannelAddress | None = None
    ) -> list[ChannelAddress]:
        """Suggest thread parents for this identity, without creating or posting."""
        return []

    async def list_addresses(
        self,
        address: ChannelAddress,
        inside: str | None = None,
        *,
        private: bool = False,
    ) -> list[ChannelAddress]:
        """One level of this identity's places, fetched when it is opened: the top
        level, or what a listed address's `inside` holds. Nothing is created. A
        `private` landing is listed as the platform would open it for a
        conversation only its user can read."""
        raise ValueError("This channel cannot be browsed for a destination.")

    async def prepare_address(
        self,
        address: ChannelAddress,
        source_address: ChannelAddress | None = None,
        *,
        private: bool = False,
    ) -> ChannelAddress:
        """Validate one parent and resolve platform details without creating a thread."""
        raise ValueError("This channel cannot prepare a thread at this address.")

    async def open_dm(self, user_id: str, opener: str | None = None) -> str | None:
        """The chat id of this bot's 1:1 with `user_id`, opening it if needed.

        `None` when the platform offers nowhere private to reach them — no DM
        surface at all, or an open that failed. Both leave a caller that needs
        privacy with no answer, which is the only distinction it can act on.
        Opening is idempotent wherever it is a call at all.

        `opener` is the first thing the person will read there, for a platform that
        cannot open a private chat without saying something. `None` from a caller
        with nothing to say — and ignored by every platform that opens one silently.
        """
        return None

inspect abstractmethod async

inspect() -> UserProfile

Source code in octomate/tentacles/channel.py

@abstractmethod
async def inspect(self) -> UserProfile: ...

get_user_profile abstractmethod async

get_user_profile(user_id: str) -> UserProfile

Source code in octomate/tentacles/channel.py

@abstractmethod
async def get_user_profile(self, user_id: str) -> UserProfile: ...

get_group_name async

get_group_name(chat_id: str) -> str | None

The platform's group name, or None when this ink does not supply one.

Source code in octomate/tentacles/channel.py

async def get_group_name(self, chat_id: str) -> str | None:
    """The platform's group name, or None when this ink does not supply one."""
    return None

upload_media abstractmethod async

upload_media(data: bytes) -> str | None

Upload media bytes and return a platform address or URL.

Source code in octomate/tentacles/channel.py

@abstractmethod
async def upload_media(self, data: bytes) -> str | None:
    """Upload media bytes and return a platform address or URL."""

download_image abstractmethod async

download_image(
    seg: ImageSegment, message_id: str
) -> DownloadedImage | None

Download an inbound image segment.

Source code in octomate/tentacles/channel.py

@abstractmethod
async def download_image(
    self,
    seg: ImageSegment,
    message_id: str,
) -> DownloadedImage | None:
    """Download an inbound image segment."""

send_message abstractmethod async

send_message(
    chat_id: str,
    chat_type: str,
    messages: list[MessageT],
    *,
    channel_thread_id: str,
    reply_to: str | None = None,
    reply_in_thread: bool = False,
) -> IMMessageID | None

Send platform-native message payloads to a conversation destination.

channel_thread_id is the platform's external destination id: the thread id on a thread surface, otherwise the chat id. reply_to references a message within that destination. Platforms where thread and reply are the same native field collapse them in their ink implementation.

Source code in octomate/tentacles/channel.py

@abstractmethod
async def send_message(
    self,
    chat_id: str,
    chat_type: str,
    messages: list[MessageT],
    *,
    channel_thread_id: str,
    reply_to: str | None = None,
    reply_in_thread: bool = False,
) -> IMMessageID | None:
    """Send platform-native message payloads to a conversation destination.

    `channel_thread_id` is the platform's external destination id: the thread
    id on a thread surface, otherwise the chat id. `reply_to` references a
    message within that destination. Platforms where thread and reply are the
    same native field collapse them in their ink implementation.
    """

suggest_addresses async

suggest_addresses(
    address: ChannelAddress,
    source_address: ChannelAddress | None = None,
) -> list[ChannelAddress]

Suggest thread parents for this identity, without creating or posting.

Source code in octomate/tentacles/channel.py

async def suggest_addresses(
    self, address: ChannelAddress, source_address: ChannelAddress | None = None
) -> list[ChannelAddress]:
    """Suggest thread parents for this identity, without creating or posting."""
    return []

list_addresses async

list_addresses(
    address: ChannelAddress,
    inside: str | None = None,
    *,
    private: bool = False,
) -> list[ChannelAddress]

One level of this identity's places, fetched when it is opened: the top level, or what a listed address's inside holds. Nothing is created. A private landing is listed as the platform would open it for a conversation only its user can read.

Source code in octomate/tentacles/channel.py

async def list_addresses(
    self,
    address: ChannelAddress,
    inside: str | None = None,
    *,
    private: bool = False,
) -> list[ChannelAddress]:
    """One level of this identity's places, fetched when it is opened: the top
    level, or what a listed address's `inside` holds. Nothing is created. A
    `private` landing is listed as the platform would open it for a
    conversation only its user can read."""
    raise ValueError("This channel cannot be browsed for a destination.")

prepare_address async

prepare_address(
    address: ChannelAddress,
    source_address: ChannelAddress | None = None,
    *,
    private: bool = False,
) -> ChannelAddress

Validate one parent and resolve platform details without creating a thread.

Source code in octomate/tentacles/channel.py

async def prepare_address(
    self,
    address: ChannelAddress,
    source_address: ChannelAddress | None = None,
    *,
    private: bool = False,
) -> ChannelAddress:
    """Validate one parent and resolve platform details without creating a thread."""
    raise ValueError("This channel cannot prepare a thread at this address.")

open_dm async

open_dm(
    user_id: str, opener: str | None = None
) -> str | None

The chat id of this bot's 1:1 with user_id, opening it if needed.

None when the platform offers nowhere private to reach them — no DM surface at all, or an open that failed. Both leave a caller that needs privacy with no answer, which is the only distinction it can act on. Opening is idempotent wherever it is a call at all.

opener is the first thing the person will read there, for a platform that cannot open a private chat without saying something. None from a caller with nothing to say — and ignored by every platform that opens one silently.

Source code in octomate/tentacles/channel.py

async def open_dm(self, user_id: str, opener: str | None = None) -> str | None:
    """The chat id of this bot's 1:1 with `user_id`, opening it if needed.

    `None` when the platform offers nowhere private to reach them — no DM
    surface at all, or an open that failed. Both leave a caller that needs
    privacy with no answer, which is the only distinction it can act on.
    Opening is idempotent wherever it is a call at all.

    `opener` is the first thing the person will read there, for a platform that
    cannot open a private chat without saying something. `None` from a caller
    with nothing to say — and ignored by every platform that opens one silently.
    """
    return None

ChannelTentacle

Bases: Tentacle[RawT, MessageT]

Base class for IM channels.

A channel receives native platform events, converts them to MessageEvents, and asks Octomate to dispatch the turn to the assigned agent. Outbound messages are encoded by Chromo and sent by the configured ink implementation.

Source code in octomate/tentacles/channel.py

class ChannelTentacle(
    Tentacle[RawT, MessageT],
):
    """Base class for IM channels.

    A channel receives native platform events, converts them to MessageEvents,
    and asks Octomate to dispatch the turn to the assigned agent. Outbound
    messages are encoded by Chromo and sent by the configured ink implementation.
    """

    FILES_ROOT: ClassVar[Path] = Path(".octomate/files")
    # How an inbound *threaded* message is routed, not what the bot can open:
    # `flat_thread` means a message carrying a thread_id continues that thread without
    # triage (reflex.Route). Vercel declares it and can open nothing — see `surfaces`.
    thread_strategy: ClassVar[ThreadStrategy] = "main_only"
    surfaces: ClassVar[ChannelSurfaces] = ChannelSurfaces()

    # The platform account this tentacle is logged in as (the bot's own
    # identity on the channel) — probe() fills it from ink.inspect().
    self_profile: UserProfile
    feelers: Feelers
    # Where what the nerve carries to this channel leaves for the platform.
    siphon: Siphon
    ink: Ink[MessageT]
    chromo: Chromo[RawT, MessageT]
    config: ChannelConfig

    def __init__(
        self,
        id: str,
        octomate: Octomate,
        *,
        ink: Ink[MessageT],
        chromo: Chromo[RawT, MessageT],
        config: ChannelConfig,
    ) -> None:
        super().__init__(id=id, octomate=octomate)
        self.config = config
        self.ink = ink
        self.chromo = chromo
        self.user_profiles: dict[str, UserProfile] = {}

        markdown_feeler = DefaultMarkdownFeeler(ink=self.ink, chromo=self.chromo)
        approvals_feeler = PlainTextApprovalFeeler(markdown_feeler)
        questions_feeler = PlainTextAskQuestionFeeler(markdown_feeler)
        oauth_feeler = PlainTextOAuthFeeler(self.ink, markdown_feeler)
        segments_feeler = DefaultSegmentsFeeler[RawT, MessageT](
            ink=self.ink, chromo=self.chromo
        )

        self.feelers = Feelers(
            markdown=markdown_feeler,
            timeline=DefaultTimelineFeeler[RawT, MessageT](
                ink=self.ink,
                chromo=self.chromo,
                ask_questions=questions_feeler,
                approvals=approvals_feeler,
                oauth=oauth_feeler,
                deferred_actions=self.octomate.deferred_actions,
            ),
            segments=segments_feeler,
            approvals=approvals_feeler,
            ask_questions=questions_feeler,
            oauth=oauth_feeler,
            artifacts=PlainTextArtifactFeeler(
                artifacts=self.octomate.artifacts,
                users=self.octomate.users,
                segments=segments_feeler,
            ),
        )
        self.siphon = Siphon(self)

    @property
    def agent_ids(self) -> list[str]:
        """Agent tentacles this channel exposes, in entry order."""
        return self.config.agents

    async def probe(self) -> None:
        """Resolve the channel's own identity from the platform. Awaited by the
        host before the channel is served, so `self.self_profile` is set before any
        inbound event is ingested."""
        self.self_profile = await self.ink.inspect()
        logger.info(
            "Channel %s: probed as %s (%s)",
            self.id,
            self.self_profile.channel_user_id,
            self.self_profile.name,
        )

    async def __aenter__(self) -> Self:
        # The ink first, so the probe already rides its pooled connection; then
        # resolve identity before any inbound event is ingested. Channels with a
        # persistent connection override this to open it (and `__aexit__` to tear
        # it down); HTTP-driven channels (e.g. the dev UI) need only the probe.
        await self.ink.__aenter__()
        await self.probe()
        # Before any inbound event, so no run this channel draws is missed.
        self.siphon.start()
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.siphon.stop()
        await self.ink.__aexit__(*exc)

    @property
    def name(self) -> str:
        return self.self_profile.name

    @channel_logfire.instrument("ChannelTentacle {self.id} ingest")
    async def ingest(self, raw: RawT) -> None:
        """Inbound pipeline: decode, enrich sender, resolve media, dispatch.

        Before ensuring a group surface, refresh its title from the event's chat
        name, or ask the ink when the event supplies no name. DM surfaces use the
        bot's name from its inspected profile.
        """
        try:
            event = await self.chromo.sip(raw)
            if event is None:
                # Not necessarily a message: a chromo also returns None for a frame
                # that was never one — an API echo, a heartbeat. Only it can tell that
                # from a decode that failed, and it warns for the second already.
                return
            event.tentacle_id = self.id
            event.self_id = self.self_profile.channel_user_id
            event.sender = await self.get_user_profile(event.user_id)
            await self.submerge(event)
            address = ChannelAddress(
                channel_tentacle_id=self.id,
                chat_type=event.chat_type,
                chat_id=event.chat_id,
                user_id=event.user_id,
                channel_thread_id=event.channel_thread_id,
                shared=event.shared,
            )
            channel_logfire.info(
                "ingest decoded message",
                channel_id=self.id,
                conversation_address=str(address),
                message_id=str(event.message_id),
                # What arrived beside what we made of it. The span's own `raw` is the
                # platform payload where it serializes and the class name where it
                # does not (Lark hands us an SDK object), so the judgement is recorded
                # next to it rather than left to be re-derived from a decoded address.
                raw_type=type(raw).__name__,
                chat_type=event.chat_type,
                shared=event.shared,
            )
            title = None
            if event.chat_type == "group":
                title = event.chat_name or await self.ink.get_group_name(event.chat_id)
            elif event.chat_type == "dm":
                title = self.self_profile.name or None
            thread = await self.octomate.threads.ensure(address, title=title)
            if event.message_id and await self.octomate.threads.find_message(
                thread.id, event.message_id, "inbound"
            ):
                # The platform sent this one before, so it is already answered or
                # being answered now. Warn rather than debug: a channel doing this
                # often is a channel whose acks are not landing.
                logger.warning(
                    "Channel %s: ignored a re-delivery of message %s in %s",
                    self.id,
                    event.message_id,
                    address,
                )
                return
            thread_message = await self.octomate.threads.record_inbound(event)
            if self.config.mention_only and event.shared:
                # Only a surface others can read has to be addressed, and a thread an
                # agent already owns counts as addressed — its next turn continues
                # work that is already this agent's. A group main pins no owner, so
                # there it stays the mention.
                addressed = (
                    event.is_at(self.self_profile.channel_user_id)
                    or event.replies_to(self.self_profile.channel_user_id)
                    or thread.active_agent_tentacle_id is not None
                )
                if not addressed:
                    # Warn, not debug: this is a message nobody will answer, and the
                    # level it used to sit at was below the configured one, so the
                    # drop left nothing behind on the console or in Logfire.
                    logger.warning(
                        "Channel %s: ignored an unaddressed %s message %s",
                        self.id,
                        event.chat_type,
                        address,
                    )
                    return
            try:
                await self.octomate.kick(
                    UserMessageSignal(
                        [event],
                        trigger_thread_message_id=thread_message.id,
                    )
                )
            except StillAwake:
                # The message is in the ledger, and the next turn there answers it.
                logger.info(
                    "Channel %s: message %s waits for the turn running in %s",
                    self.id,
                    event.message_id,
                    address,
                )
        except Exception:
            # The active `ingest` span carries the full error: one from before the turn
            # ran, which only this log reports.
            trace_id = format(
                trace.get_current_span().get_span_context().trace_id, "032x"
            )
            logger.exception(
                "Channel %s: error in ingest [trace_id=%s]", self.id, trace_id
            )

    async def open_dm(
        self, user_id: str, opener: str | None = None
    ) -> ChannelAddress | None:
        """The 1:1 conversation with `user_id`, opening it if the platform needs to.

        `None` when this channel has nowhere private to reach them; a channel opts
        in by declaring `surfaces.direct_message` and giving its ink an `open_dm`.
        The returned address may already carry a history, which is why a `teleport`
        there lands in a fresh thread or stays put rather than forking onto it.

        `opener` is the first thing the person will read there, passed by a caller
        that is about to move a whole turn in — and empty from one that only wants
        somewhere to deliver a message. A channel that can only be *run* inside a
        thread starts one from it; most have nothing to do with it, since the
        conversation they hand back can be posted to as it is.
        """
        if not user_id:
            return None
        chat_id = await self.ink.open_dm(user_id, opener)
        if chat_id is None:
            return None
        return ChannelAddress(
            channel_tentacle_id=self.id,
            chat_type="dm",
            chat_id=chat_id,
            user_id=user_id,
            channel_thread_id=None,
        )

    def thread_user_id(self, profile: UserProfile) -> str | None:
        """The linked identity whose thread locations this channel may expose."""
        if profile.channel_tentacle_id != self.id:
            return None
        return profile.channel_user_id or None

    def is_shared(self, address: ChannelAddress) -> bool:
        """Whether anyone besides its user can read this surface, told from the
        address alone, for a caller that holds a thread's row and no message from
        it. A thread is taken for shared unless its channel can say otherwise.
        It answers what the chromo says of a message on the same surface; each
        channel's chromo tests hold the two to that."""
        return address.chat_type != "dm"

    async def suggest_addresses(
        self, profile: UserProfile, source_address: ChannelAddress | None = None
    ) -> list[ChannelAddress]:
        """Suggest addresses using the requesting user's linked channel identity."""
        user_id = self.thread_user_id(profile)
        if user_id is None:
            return []
        if source_address is not None and source_address.channel_tentacle_id != self.id:
            source_address = None
        return [
            address
            for address in await self.ink.suggest_addresses(
                ChannelAddress(self.id, "dm", "", user_id), source_address
            )
            if address.chat_type == "thread" or self.accepts_sub_thread(address)
        ]

    async def list_addresses(
        self, profile: UserProfile, inside: str | None = None, *, private: bool = False
    ) -> list[ChannelAddress]:
        """List one level using the requesting user's linked channel identity, as a
        landing would open for a conversation that is `private`."""
        user_id = self.thread_user_id(profile)
        if user_id is None:
            raise ValueError("The profile is not linked to this channel.")
        return await self.ink.list_addresses(
            ChannelAddress(self.id, "dm", "", user_id), inside, private=private
        )

    async def prepare_address(
        self,
        address: ChannelAddress,
        source_address: ChannelAddress | None = None,
        *,
        private: bool = False,
    ) -> ChannelAddress:
        """Validate the requested parent before any thread or history is created.
        A `private` conversation's landing opens so only its user can read it,
        where the platform allows, and the prepared `shared` says which it got."""
        if address.channel_tentacle_id != self.id:
            raise ValueError("The address belongs to another channel.")
        if address.channel_thread_id:
            raise ValueError("A thread cannot contain another thread.")
        if source_address is not None and source_address.channel_tentacle_id != self.id:
            source_address = None
        prepared = await self.ink.prepare_address(
            address, source_address, private=private
        )
        if prepared.chat_type != "thread" and not self.accepts_sub_thread(prepared):
            raise ValueError("This address cannot contain a thread.")
        return prepared

    async def start_thread(self, address: ChannelAddress, hint: str) -> ChannelAddress:
        """Open an isolated destination, refusing to reuse an existing chat."""
        if not self.accepts_sub_thread(address):
            raise ValueError("This destination cannot create a thread.")
        parent = address
        if address.chat_type == "dm" and not address.chat_id:
            if not address.user_id:
                raise ValueError("The destination could not open direct messages.")
            # A tentacle's open_dm may also start a thread; only resolve the DM here.
            chat_id = await self.ink.open_dm(address.user_id, hint)
            if chat_id is None:
                raise ValueError("The destination could not open direct messages.")
            parent = replace(address, chat_id=chat_id)
        opened = await self.start_sub_thread(parent, hint)
        if opened == parent or not opened.channel_thread_id:
            raise ValueError("The destination could not create a thread.")
        return opened

    @property
    def landing_unavailable(self) -> str | None:
        """Why no conversation can be moved or handed onto this channel, or None
        where a thread can be opened for one."""
        if self.surfaces.sub_thread:
            return None
        return "This channel has no threads, so no conversation can land here."

    def accepts_sub_thread(self, address: ChannelAddress) -> bool:
        """Whether the address belongs to this channel and can contain a new thread."""
        return (
            address.channel_tentacle_id == self.id
            and self.surfaces.sub_thread
            and not address.channel_thread_id
        )

    async def start_sub_thread(
        self,
        address: ChannelAddress,
        hint_text: str,
    ) -> ChannelAddress:
        logger.warning(
            "Channel %s does not support sub-thread startup; using main target",
            self.id,
        )
        await self.feelers.markdown.present(address, hint_text)
        return address

    async def get_user_profile(self, user_id: str) -> UserProfile:
        cached = self.user_profiles.get(user_id)
        if cached is not None:
            return cached
        profile = await self.ink.get_user_profile(user_id)
        self.user_profiles[user_id] = profile
        return profile

    async def submerge(self, event: MessageEvent) -> None:
        pending = [seg for seg in event.segments if isinstance(seg, ImageSegment)]
        if not pending:
            return
        save = self.den(event)
        message_id = str(event.message_id)
        await anyio.Path(save).mkdir(parents=True, exist_ok=True)
        async with asyncio.TaskGroup() as tg:
            for seg in pending:
                tg.create_task(self.absorb(seg, save, message_id))

    async def absorb(self, seg: ImageSegment, save_dir: Path, message_id: str) -> None:
        """Download one inbound image to save_dir and rewrite seg.data.file."""
        try:
            image = await self.ink.download_image(seg, message_id)
            if image is None:
                return
            ext = guess_image_ext(image.content_type, image.file_name)
            file_path = save_dir / f"{uuid7().hex}{ext}"
            await anyio.Path(file_path).write_bytes(image.data)
            seg.data.file = str(file_path.resolve())
            if image.url is not None:
                seg.data.url = image.url
        except Exception:
            logger.warning(
                "Channel %s: failed to download image", self.id, exc_info=True
            )

    def den(self, event: MessageEvent) -> Path:
        subdir = event.user_id if event.chat_type == "dm" else event.chat_id
        return self.FILES_ROOT / self.id / subdir

FILES_ROOT class-attribute

FILES_ROOT: Path = Path('.octomate/files')

thread_strategy class-attribute

thread_strategy: ThreadStrategy = 'main_only'

surfaces class-attribute

surfaces: ChannelSurfaces = ChannelSurfaces()

self_profile instance-attribute

self_profile: UserProfile

config instance-attribute

config: ChannelConfig = config

ink instance-attribute

ink: Ink[MessageT] = ink

chromo instance-attribute

chromo: Chromo[RawT, MessageT] = chromo

user_profiles instance-attribute

user_profiles: dict[str, UserProfile] = {}

feelers instance-attribute

feelers: Feelers = Feelers(
    markdown=markdown_feeler,
    timeline=DefaultTimelineFeeler[RawT, MessageT](
        ink=self.ink,
        chromo=self.chromo,
        ask_questions=questions_feeler,
        approvals=approvals_feeler,
        oauth=oauth_feeler,
        deferred_actions=self.octomate.deferred_actions,
    ),
    segments=segments_feeler,
    approvals=approvals_feeler,
    ask_questions=questions_feeler,
    oauth=oauth_feeler,
    artifacts=PlainTextArtifactFeeler(
        artifacts=self.octomate.artifacts,
        users=self.octomate.users,
        segments=segments_feeler,
    ),
)

siphon instance-attribute

siphon: Siphon = Siphon(self)

agent_ids property

agent_ids: list[str]

Agent tentacles this channel exposes, in entry order.

name property

name: str

landing_unavailable property

landing_unavailable: str | None

Why no conversation can be moved or handed onto this channel, or None where a thread can be opened for one.

probe async

probe() -> None

Resolve the channel's own identity from the platform. Awaited by the host before the channel is served, so self.self_profile is set before any inbound event is ingested.

Source code in octomate/tentacles/channel.py

async def probe(self) -> None:
    """Resolve the channel's own identity from the platform. Awaited by the
    host before the channel is served, so `self.self_profile` is set before any
    inbound event is ingested."""
    self.self_profile = await self.ink.inspect()
    logger.info(
        "Channel %s: probed as %s (%s)",
        self.id,
        self.self_profile.channel_user_id,
        self.self_profile.name,
    )

ingest async

ingest(raw: RawT) -> None

Inbound pipeline: decode, enrich sender, resolve media, dispatch.

Before ensuring a group surface, refresh its title from the event's chat name, or ask the ink when the event supplies no name. DM surfaces use the bot's name from its inspected profile.

Source code in octomate/tentacles/channel.py

@channel_logfire.instrument("ChannelTentacle {self.id} ingest")
async def ingest(self, raw: RawT) -> None:
    """Inbound pipeline: decode, enrich sender, resolve media, dispatch.

    Before ensuring a group surface, refresh its title from the event's chat
    name, or ask the ink when the event supplies no name. DM surfaces use the
    bot's name from its inspected profile.
    """
    try:
        event = await self.chromo.sip(raw)
        if event is None:
            # Not necessarily a message: a chromo also returns None for a frame
            # that was never one — an API echo, a heartbeat. Only it can tell that
            # from a decode that failed, and it warns for the second already.
            return
        event.tentacle_id = self.id
        event.self_id = self.self_profile.channel_user_id
        event.sender = await self.get_user_profile(event.user_id)
        await self.submerge(event)
        address = ChannelAddress(
            channel_tentacle_id=self.id,
            chat_type=event.chat_type,
            chat_id=event.chat_id,
            user_id=event.user_id,
            channel_thread_id=event.channel_thread_id,
            shared=event.shared,
        )
        channel_logfire.info(
            "ingest decoded message",
            channel_id=self.id,
            conversation_address=str(address),
            message_id=str(event.message_id),
            # What arrived beside what we made of it. The span's own `raw` is the
            # platform payload where it serializes and the class name where it
            # does not (Lark hands us an SDK object), so the judgement is recorded
            # next to it rather than left to be re-derived from a decoded address.
            raw_type=type(raw).__name__,
            chat_type=event.chat_type,
            shared=event.shared,
        )
        title = None
        if event.chat_type == "group":
            title = event.chat_name or await self.ink.get_group_name(event.chat_id)
        elif event.chat_type == "dm":
            title = self.self_profile.name or None
        thread = await self.octomate.threads.ensure(address, title=title)
        if event.message_id and await self.octomate.threads.find_message(
            thread.id, event.message_id, "inbound"
        ):
            # The platform sent this one before, so it is already answered or
            # being answered now. Warn rather than debug: a channel doing this
            # often is a channel whose acks are not landing.
            logger.warning(
                "Channel %s: ignored a re-delivery of message %s in %s",
                self.id,
                event.message_id,
                address,
            )
            return
        thread_message = await self.octomate.threads.record_inbound(event)
        if self.config.mention_only and event.shared:
            # Only a surface others can read has to be addressed, and a thread an
            # agent already owns counts as addressed — its next turn continues
            # work that is already this agent's. A group main pins no owner, so
            # there it stays the mention.
            addressed = (
                event.is_at(self.self_profile.channel_user_id)
                or event.replies_to(self.self_profile.channel_user_id)
                or thread.active_agent_tentacle_id is not None
            )
            if not addressed:
                # Warn, not debug: this is a message nobody will answer, and the
                # level it used to sit at was below the configured one, so the
                # drop left nothing behind on the console or in Logfire.
                logger.warning(
                    "Channel %s: ignored an unaddressed %s message %s",
                    self.id,
                    event.chat_type,
                    address,
                )
                return
        try:
            await self.octomate.kick(
                UserMessageSignal(
                    [event],
                    trigger_thread_message_id=thread_message.id,
                )
            )
        except StillAwake:
            # The message is in the ledger, and the next turn there answers it.
            logger.info(
                "Channel %s: message %s waits for the turn running in %s",
                self.id,
                event.message_id,
                address,
            )
    except Exception:
        # The active `ingest` span carries the full error: one from before the turn
        # ran, which only this log reports.
        trace_id = format(
            trace.get_current_span().get_span_context().trace_id, "032x"
        )
        logger.exception(
            "Channel %s: error in ingest [trace_id=%s]", self.id, trace_id
        )

open_dm async

open_dm(
    user_id: str, opener: str | None = None
) -> ChannelAddress | None

The 1:1 conversation with user_id, opening it if the platform needs to.

None when this channel has nowhere private to reach them; a channel opts in by declaring surfaces.direct_message and giving its ink an open_dm. The returned address may already carry a history, which is why a teleport there lands in a fresh thread or stays put rather than forking onto it.

opener is the first thing the person will read there, passed by a caller that is about to move a whole turn in — and empty from one that only wants somewhere to deliver a message. A channel that can only be run inside a thread starts one from it; most have nothing to do with it, since the conversation they hand back can be posted to as it is.

Source code in octomate/tentacles/channel.py

async def open_dm(
    self, user_id: str, opener: str | None = None
) -> ChannelAddress | None:
    """The 1:1 conversation with `user_id`, opening it if the platform needs to.

    `None` when this channel has nowhere private to reach them; a channel opts
    in by declaring `surfaces.direct_message` and giving its ink an `open_dm`.
    The returned address may already carry a history, which is why a `teleport`
    there lands in a fresh thread or stays put rather than forking onto it.

    `opener` is the first thing the person will read there, passed by a caller
    that is about to move a whole turn in — and empty from one that only wants
    somewhere to deliver a message. A channel that can only be *run* inside a
    thread starts one from it; most have nothing to do with it, since the
    conversation they hand back can be posted to as it is.
    """
    if not user_id:
        return None
    chat_id = await self.ink.open_dm(user_id, opener)
    if chat_id is None:
        return None
    return ChannelAddress(
        channel_tentacle_id=self.id,
        chat_type="dm",
        chat_id=chat_id,
        user_id=user_id,
        channel_thread_id=None,
    )

thread_user_id

thread_user_id(profile: UserProfile) -> str | None

The linked identity whose thread locations this channel may expose.

Source code in octomate/tentacles/channel.py

def thread_user_id(self, profile: UserProfile) -> str | None:
    """The linked identity whose thread locations this channel may expose."""
    if profile.channel_tentacle_id != self.id:
        return None
    return profile.channel_user_id or None

is_shared

is_shared(address: ChannelAddress) -> bool

Whether anyone besides its user can read this surface, told from the address alone, for a caller that holds a thread's row and no message from it. A thread is taken for shared unless its channel can say otherwise. It answers what the chromo says of a message on the same surface; each channel's chromo tests hold the two to that.

Source code in octomate/tentacles/channel.py

def is_shared(self, address: ChannelAddress) -> bool:
    """Whether anyone besides its user can read this surface, told from the
    address alone, for a caller that holds a thread's row and no message from
    it. A thread is taken for shared unless its channel can say otherwise.
    It answers what the chromo says of a message on the same surface; each
    channel's chromo tests hold the two to that."""
    return address.chat_type != "dm"

suggest_addresses async

suggest_addresses(
    profile: UserProfile,
    source_address: ChannelAddress | None = None,
) -> list[ChannelAddress]

Suggest addresses using the requesting user's linked channel identity.

Source code in octomate/tentacles/channel.py

async def suggest_addresses(
    self, profile: UserProfile, source_address: ChannelAddress | None = None
) -> list[ChannelAddress]:
    """Suggest addresses using the requesting user's linked channel identity."""
    user_id = self.thread_user_id(profile)
    if user_id is None:
        return []
    if source_address is not None and source_address.channel_tentacle_id != self.id:
        source_address = None
    return [
        address
        for address in await self.ink.suggest_addresses(
            ChannelAddress(self.id, "dm", "", user_id), source_address
        )
        if address.chat_type == "thread" or self.accepts_sub_thread(address)
    ]

list_addresses async

list_addresses(
    profile: UserProfile,
    inside: str | None = None,
    *,
    private: bool = False,
) -> list[ChannelAddress]

List one level using the requesting user's linked channel identity, as a landing would open for a conversation that is private.

Source code in octomate/tentacles/channel.py

async def list_addresses(
    self, profile: UserProfile, inside: str | None = None, *, private: bool = False
) -> list[ChannelAddress]:
    """List one level using the requesting user's linked channel identity, as a
    landing would open for a conversation that is `private`."""
    user_id = self.thread_user_id(profile)
    if user_id is None:
        raise ValueError("The profile is not linked to this channel.")
    return await self.ink.list_addresses(
        ChannelAddress(self.id, "dm", "", user_id), inside, private=private
    )

prepare_address async

prepare_address(
    address: ChannelAddress,
    source_address: ChannelAddress | None = None,
    *,
    private: bool = False,
) -> ChannelAddress

Validate the requested parent before any thread or history is created. A private conversation's landing opens so only its user can read it, where the platform allows, and the prepared shared says which it got.

Source code in octomate/tentacles/channel.py

async def prepare_address(
    self,
    address: ChannelAddress,
    source_address: ChannelAddress | None = None,
    *,
    private: bool = False,
) -> ChannelAddress:
    """Validate the requested parent before any thread or history is created.
    A `private` conversation's landing opens so only its user can read it,
    where the platform allows, and the prepared `shared` says which it got."""
    if address.channel_tentacle_id != self.id:
        raise ValueError("The address belongs to another channel.")
    if address.channel_thread_id:
        raise ValueError("A thread cannot contain another thread.")
    if source_address is not None and source_address.channel_tentacle_id != self.id:
        source_address = None
    prepared = await self.ink.prepare_address(
        address, source_address, private=private
    )
    if prepared.chat_type != "thread" and not self.accepts_sub_thread(prepared):
        raise ValueError("This address cannot contain a thread.")
    return prepared

start_thread async

start_thread(
    address: ChannelAddress, hint: str
) -> ChannelAddress

Open an isolated destination, refusing to reuse an existing chat.

Source code in octomate/tentacles/channel.py

async def start_thread(self, address: ChannelAddress, hint: str) -> ChannelAddress:
    """Open an isolated destination, refusing to reuse an existing chat."""
    if not self.accepts_sub_thread(address):
        raise ValueError("This destination cannot create a thread.")
    parent = address
    if address.chat_type == "dm" and not address.chat_id:
        if not address.user_id:
            raise ValueError("The destination could not open direct messages.")
        # A tentacle's open_dm may also start a thread; only resolve the DM here.
        chat_id = await self.ink.open_dm(address.user_id, hint)
        if chat_id is None:
            raise ValueError("The destination could not open direct messages.")
        parent = replace(address, chat_id=chat_id)
    opened = await self.start_sub_thread(parent, hint)
    if opened == parent or not opened.channel_thread_id:
        raise ValueError("The destination could not create a thread.")
    return opened

accepts_sub_thread

accepts_sub_thread(address: ChannelAddress) -> bool

Whether the address belongs to this channel and can contain a new thread.

Source code in octomate/tentacles/channel.py

def accepts_sub_thread(self, address: ChannelAddress) -> bool:
    """Whether the address belongs to this channel and can contain a new thread."""
    return (
        address.channel_tentacle_id == self.id
        and self.surfaces.sub_thread
        and not address.channel_thread_id
    )

start_sub_thread async

start_sub_thread(
    address: ChannelAddress, hint_text: str
) -> ChannelAddress

Source code in octomate/tentacles/channel.py

async def start_sub_thread(
    self,
    address: ChannelAddress,
    hint_text: str,
) -> ChannelAddress:
    logger.warning(
        "Channel %s does not support sub-thread startup; using main target",
        self.id,
    )
    await self.feelers.markdown.present(address, hint_text)
    return address

get_user_profile async

get_user_profile(user_id: str) -> UserProfile

Source code in octomate/tentacles/channel.py

async def get_user_profile(self, user_id: str) -> UserProfile:
    cached = self.user_profiles.get(user_id)
    if cached is not None:
        return cached
    profile = await self.ink.get_user_profile(user_id)
    self.user_profiles[user_id] = profile
    return profile

submerge async

submerge(event: MessageEvent) -> None

Source code in octomate/tentacles/channel.py

async def submerge(self, event: MessageEvent) -> None:
    pending = [seg for seg in event.segments if isinstance(seg, ImageSegment)]
    if not pending:
        return
    save = self.den(event)
    message_id = str(event.message_id)
    await anyio.Path(save).mkdir(parents=True, exist_ok=True)
    async with asyncio.TaskGroup() as tg:
        for seg in pending:
            tg.create_task(self.absorb(seg, save, message_id))

absorb async

absorb(
    seg: ImageSegment, save_dir: Path, message_id: str
) -> None

Download one inbound image to save_dir and rewrite seg.data.file.

Source code in octomate/tentacles/channel.py

async def absorb(self, seg: ImageSegment, save_dir: Path, message_id: str) -> None:
    """Download one inbound image to save_dir and rewrite seg.data.file."""
    try:
        image = await self.ink.download_image(seg, message_id)
        if image is None:
            return
        ext = guess_image_ext(image.content_type, image.file_name)
        file_path = save_dir / f"{uuid7().hex}{ext}"
        await anyio.Path(file_path).write_bytes(image.data)
        seg.data.file = str(file_path.resolve())
        if image.url is not None:
            seg.data.url = image.url
    except Exception:
        logger.warning(
            "Channel %s: failed to download image", self.id, exc_info=True
        )

den

den(event: MessageEvent) -> Path

Source code in octomate/tentacles/channel.py

def den(self, event: MessageEvent) -> Path:
    subdir = event.user_id if event.chat_type == "dm" else event.chat_id
    return self.FILES_ROOT / self.id / subdir

build_channel

build_channel(
    id: str,
    config: ChannelConfigVariant,
    octomate: Octomate,
) -> ChannelTentacle

Compose one configured channel into its tentacle.

The one place a type: becomes a platform, mirroring build_mcp. The configured key is the tentacle id throughout, which is what lets one platform be mounted more than once — two Lark apps are two keys, and nothing below here learns that they share a class.

Imported inside the function because each platform module pulls in its vendor SDK, and importing this package must not cost all four.

Source code in octomate/tentacles/channel.py

def build_channel(
    id: str,
    config: ChannelConfigVariant,
    octomate: Octomate,
) -> ChannelTentacle:
    """Compose one configured channel into its tentacle.

    The one place a `type:` becomes a platform, mirroring `build_mcp`. The
    configured key is the tentacle id throughout, which is what lets one platform be
    mounted more than once — two Lark apps are two keys, and nothing below here
    learns that they share a class.

    Imported inside the function because each platform module pulls in its vendor
    SDK, and importing this package must not cost all four.
    """
    match config:
        case SlackChannelConfig():
            from octomate.tentacles.slack import SlackTentacle

            return SlackTentacle(id, octomate, config=config)
        case LarkChannelConfig():
            from octomate.tentacles.lark import LarkTentacle

            return LarkTentacle(id, octomate, config=config)
        case DiscordChannelConfig():
            from octomate.tentacles.discord import DiscordTentacle

            return DiscordTentacle(id, octomate, config=config)
        case NapcatChannelConfig():
            from octomate.tentacles.napcat import NapcatTentacle

            return NapcatTentacle(id, octomate, config=config)
        case TrunklineChannelConfig():
            from octomate.tentacles.trunkline import TrunklineTentacle

            return TrunklineTentacle(id, octomate, config=config)

octomate.tentacles.mcp

Configured MCP declarations from which users create their own MCPs.

McpTentacle dataclass

Bases: Tentacle

Connection settings and tool instructions offered for installation.

Source code in octomate/tentacles/mcp.py

class McpTentacle(Tentacle):
    """Connection settings and tool instructions offered for installation."""

    auth_kind: Literal["none", "bearer", "oauth"] = "none"
    token: SecretStr | None = None
    label: str
    upstream: str
    instructions: str

    @property
    def log_names(self) -> tuple[str, ...]:
        return (type(self).__module__.removesuffix(".base"),)

    @cached_property
    def info(self) -> McpTentacleInfo:
        return McpTentacleInfo(
            id=self.id,
            name=self.label,
            url=self.upstream,
            auth_kind=self.auth_kind,
        )

    @property
    def serving(self) -> bool:
        """Whether this tentacle's tools are served at all; a channel may say no."""
        return True

auth_kind class-attribute instance-attribute

auth_kind: Literal['none', 'bearer', 'oauth'] = 'none'

token class-attribute instance-attribute

token: SecretStr | None = None

label instance-attribute

label: str

upstream instance-attribute

upstream: str

instructions instance-attribute

instructions: str

log_names property

log_names: tuple[str, ...]

info cached property

info: McpTentacleInfo

serving property

serving: bool

Whether this tentacle's tools are served at all; a channel may say no.

OAuthMcpTentacle dataclass

Bases: McpTentacle

An MCP backed by per-user OAuth authorization.

Source code in octomate/tentacles/mcp.py

class OAuthMcpTentacle(McpTentacle):
    """An MCP backed by per-user OAuth authorization."""

    auth_kind = "oauth"

auth_kind class-attribute instance-attribute

auth_kind = 'oauth'

BareMcpTentacle

Bases: McpTentacle

An MCP offered with an optional operator bearer token.

Source code in octomate/tentacles/mcp.py

class BareMcpTentacle(McpTentacle):
    """An MCP offered with an optional operator bearer token."""

    def __init__(
        self,
        id: str,
        octomate: Octomate,
        *,
        url: str,
        token: SecretStr | None = None,
    ) -> None:
        super().__init__(id=id, octomate=octomate)
        self.label = id
        self.upstream = url
        self.instructions = ""
        self.token = token
        self.auth_kind = "bearer" if token is not None else "none"

label instance-attribute

label = id

upstream instance-attribute

upstream = url

instructions instance-attribute

instructions = ''

token instance-attribute

token = token

auth_kind instance-attribute

auth_kind = 'bearer' if token is not None else 'none'

build_mcp

build_mcp(
    id: str, config: McpConfigVariant, octomate: Octomate
) -> McpTentacle

Build a tentacle and register any configured OAuth application.

Source code in octomate/tentacles/mcp.py

def build_mcp(id: str, config: McpConfigVariant, octomate: Octomate) -> McpTentacle:
    """Build a tentacle and register any configured OAuth application."""
    match config:
        case BareMcpConfig():
            return BareMcpTentacle(id, octomate, url=config.url, token=config.token)
        case DiscoveredOAuthMcpConfig():
            if octomate.oauth.callback_base_uri is None:
                raise ValueError(
                    "Configure oauth.callback_base_uri for OAuth discovery"
                )
            octomate.oauth.register(
                OAuthConnector(
                    id=id,
                    mcp_url=config.url,
                    flows=[
                        AuthorizationCodeFlow(
                            tokens=McpOAuthDiscovery(
                                url=config.url,
                                httpx_client_factory=octomate.oauth.httpx_client_factory,
                                client_metadata_url=octomate.oauth.client_metadata_url,
                            ),
                            authorization_lifetime=octomate.oauth.authorization_lifetime,
                        )
                    ],
                    callback_transport=DirectHttpOAuthCallbackTransport(
                        octomate.oauth.callback_base_uri
                    ),
                )
            )
        case OAuthMcpConfig():
            flows: list[DeviceAuthorizationFlow | AuthorizationCodeFlow] = []
            callback = None
            for flow_config in config.flows:
                tokens = OAuthTokenExchange(
                    authorization_endpoint=flow_config.authorization_endpoint
                    if isinstance(flow_config, AuthorizationCodeFlowConfig)
                    else None,
                    token_endpoint=flow_config.token_endpoint,
                    client=OAuthClientInformationFull(
                        client_id=config.client_id,
                        client_secret=config.client_secret.get_secret_value()
                        if config.client_secret is not None
                        else None,
                        token_endpoint_auth_method=flow_config.token_endpoint_auth_method,
                    ),
                    scopes=config.scopes,
                    scope_separator=config.scope_separator,
                    invalid_credentials_errors=config.invalid_credentials_errors,
                    httpx_client_factory=octomate.oauth.httpx_client_factory,
                )
                match flow_config:
                    case DeviceFlowConfig():
                        flow = DeviceAuthorizationFlow(
                            device_authorization_endpoint=flow_config.device_authorization_endpoint,
                            tokens=tokens,
                        )
                    case AuthorizationCodeFlowConfig():
                        if octomate.oauth.callback_base_uri is None:
                            raise ValueError(
                                "Configure oauth.callback_base_uri for authorization-code MCPs"
                            )
                        flow = AuthorizationCodeFlow(
                            tokens=tokens,
                            authorization_lifetime=octomate.oauth.authorization_lifetime,
                        )
                        callback = DirectHttpOAuthCallbackTransport(
                            octomate.oauth.callback_base_uri
                        )
                flows.append(flow)
            octomate.oauth.register(
                OAuthConnector(
                    id=id, mcp_url=config.url, flows=flows, callback_transport=callback
                )
            )
    tentacle = OAuthMcpTentacle(id=id, octomate=octomate)
    tentacle.label = id
    tentacle.upstream = str(config.url)
    tentacle.instructions = ""
    return tentacle

Codex status

octomate.tentacles.codex.base.CodexTentacle.token_usage_status async

token_usage_status(
    conversation: Conversation | None,
) -> str

Render the latest recorded native usage snapshot for this Codex session.

Native totals count all requests, including tool loops. The last request and context capacity are separate from those cumulative totals. Copied events from another native session do not describe this session's usage.

Source code in octomate/tentacles/codex/base.py

async def token_usage_status(self, conversation: Conversation | None) -> str:
    """Render the latest recorded native usage snapshot for this Codex session.

    Native totals count all requests, including tool loops. The last request
    and context capacity are separate from those cumulative totals. Copied
    events from another native session do not describe this session's usage.
    """
    if conversation is None or conversation.external_id is None:
        return "Token usage: not reported yet."
    conversation = await self.conversations.get(conversation.id)
    events = (
        event
        for message in reversed(conversation.messages)
        if (metadata := message.metadata) is not None
        and metadata.get("source") == CODEX_METADATA_SOURCE
        and isinstance(recorded := metadata.get("events"), list)
        for event in reversed(recorded)
        if isinstance(event, dict)
        and event.get("method") == "thread/tokenUsage/updated"
    )
    for event in events:
        snapshot = ThreadTokenUsageUpdatedNotification.model_validate(
            event["payload"]
        )
        if snapshot.thread_id != conversation.external_id:
            continue
        usage = snapshot.token_usage
        total = usage.total
        context = (
            f"{usage.model_context_window:,} tokens"
            if usage.model_context_window is not None
            else "not reported"
        )
        return (
            f"Token usage: {total.total_tokens:,} total "
            f"({total.input_tokens:,} input, {total.output_tokens:,} output; "
            f"{total.cached_input_tokens:,} cached input, "
            f"{total.reasoning_output_tokens:,} reasoning output)\n"
            f"Last request: {usage.last.total_tokens:,} tokens\n"
            f"Context window: {context}"
        )
    return "Token usage: not reported yet."

octomate.tentacles.codex.ink.CodexInk.account_status async

account_status() -> str

Read ChatGPT quota windows without creating a thread or refreshing auth.

Each bucket reports its own remaining percentage and UTC reset time. Other authentication modes have no ChatGPT quota; RPC failures propagate. The SDK leaves the multi-bucket map untyped, so validate it before use.

Source code in octomate/tentacles/codex/ink.py

async def account_status(self) -> str:
    """Read ChatGPT quota windows without creating a thread or refreshing auth.

    Each bucket reports its own remaining percentage and UTC reset time.
    Other authentication modes have no ChatGPT quota; RPC failures propagate.
    The SDK leaves the multi-bucket map untyped, so validate it before use.
    """
    account = await self.client._client.request(
        "account/read", {"refreshToken": False}, response_model=GetAccountResponse
    )
    if account.account is None or not isinstance(
        account.account.root, ChatgptAccount
    ):
        return "Account limits: unavailable (not signed in with ChatGPT)."
    response = await self.client._client.request(
        "account/rateLimits/read", None, response_model=GetAccountRateLimitsResponse
    )
    buckets = (
        TypeAdapter(dict[str, RateLimitSnapshot]).validate_python(
            response.rate_limits_by_limit_id
        )
        if response.rate_limits_by_limit_id is not None
        else {response.rate_limits.limit_id or "codex": response.rate_limits}
    )
    lines = [f"Account plan: {account.account.root.plan_type.value}"]
    for bucket_id, bucket in buckets.items():
        name = bucket.limit_name or bucket_id
        for label, window in (
            ("primary", bucket.primary),
            ("secondary", bucket.secondary),
        ):
            if window is None:
                continue
            minutes = window.window_duration_mins
            duration = (
                (
                    f"{minutes // 1440}d"
                    if minutes % 1440 == 0
                    else f"{minutes // 60}h"
                    if minutes % 60 == 0
                    else f"{minutes}m"
                )
                if minutes
                else label
            )
            reset = (
                datetime.fromtimestamp(window.resets_at, UTC).strftime(
                    "%Y-%m-%d %H:%M UTC"
                )
                if window.resets_at is not None
                else "not reported"
            )
            lines.append(
                f"{name} {duration} limit: {100 - window.used_percent}% left; resets {reset}"
            )
    if len(lines) == 1:
        lines.append("Account limits: not reported.")
    return "\n".join(lines)