Command discovery

octomate.managers.commands

Command discovery and guarded execution for host-authorized contexts.

logger module-attribute

logger = logging.getLogger(__name__)

CommandCatalogKey

Bases: NamedTuple

One user's agent catalog, before or after a conversation exists.

Source code in octomate/managers/commands.py

class CommandCatalogKey(NamedTuple):
    """One user's agent catalog, before or after a conversation exists."""

    user_id: uuid.UUID
    agent_id: str
    conversation_id: uuid.UUID | None

user_id instance-attribute

user_id: UUID

agent_id instance-attribute

agent_id: str

conversation_id instance-attribute

conversation_id: UUID | None

ValidatedCommand

Bases: NamedTuple

Authorized command inputs, valid while the validation scope remains open.

Source code in octomate/managers/commands.py

class ValidatedCommand(NamedTuple):
    """Authorized command inputs, valid while the validation scope remains open."""

    surface: Thread | None
    profile: UserProfile | None
    descriptor: CommandDescriptor

surface instance-attribute

surface: Thread | None

profile instance-attribute

profile: UserProfile | None

descriptor instance-attribute

descriptor: CommandDescriptor

CommandManager

Bases: Manager

Keep one LRU catalog per user, agent and conversation until invalidated.

Callers must resolve and authorize a fresh context before every lookup, even a cache hit. Adapters must call invalidate on runtime reconnects and catalog-change events. Context changes replace the existing catalog and discard its pending result. Discovery errors are explicit, never empty lists.

Source code in octomate/managers/commands.py

class CommandManager(Manager):
    """Keep one LRU catalog per user, agent and conversation until invalidated.

    Callers must resolve and authorize a fresh context before every lookup, even
    a cache hit. Adapters must call `invalidate` on runtime reconnects and
    catalog-change events. Context changes replace the existing catalog and
    discard its pending result. Discovery errors are explicit, never empty lists.
    """

    capacity: ClassVar[int] = 256
    timeout: ClassVar[float] = 10.0

    catalogs: LRUCache[CommandCatalogKey, CommandCatalog]
    probes: dict[CommandCatalogKey, asyncio.Task[CommandCatalog]]
    tentacles: dict[str, Tentacle]
    users: UserManager
    conversations: ConversationManager
    threads: ThreadManager
    workspaces: WorkspaceManager
    gateway: GatewayManager

    def __init__(
        self,
        *,
        tentacles: dict[str, Tentacle],
        users: UserManager,
        conversations: ConversationManager,
        threads: ThreadManager,
        workspaces: WorkspaceManager,
        gateway: GatewayManager,
    ) -> None:
        self.tentacles = tentacles
        self.users = users
        self.conversations = conversations
        self.threads = threads
        self.workspaces = workspaces
        self.gateway = gateway
        self.catalogs = LRUCache(maxsize=self.capacity)
        self.probes = {}

    async def discover(
        self,
        agent: AgentTentacle,
        context: CommandContext,
        *,
        refresh: bool = False,
    ) -> CommandCatalog:
        """Read or refresh a catalog, sharing active discovery in the same context.

        Loading and failed results are reprobed on the next lookup. Canceling one
        caller leaves shared discovery alive under its timeout. New composers
        share a catalog per user and agent while their contexts match.
        """
        conversation = context.conversation
        if context.agent_id != agent.id or (
            conversation is not None and conversation.agent_tentacle_id != agent.id
        ):
            raise ValueError("command context belongs to another agent")

        key = CommandCatalogKey(
            user_id=context.user_id,
            agent_id=context.agent_id,
            conversation_id=conversation.id if conversation is not None else None,
        )
        while True:
            catalog = self.catalogs.get(key)
            task = self.probes.get(key)
            if catalog is not None and catalog.context.matches(context):
                if task is not None and not task.done():
                    return (await asyncio.shield(task)).snapshot()
                if not refresh and catalog.status not in {"loading", "failed"}:
                    return catalog.snapshot()
            if task is None or task.done():
                break
            self.catalogs.pop(key, None)
            if not task.cancelling():
                task.cancel()
            await asyncio.shield(asyncio.gather(task, return_exceptions=True))
        self.catalogs.pop(key, None)
        catalog = CommandCatalog(context=context, status="loading").snapshot()
        task = asyncio.create_task(self.probe(agent, catalog))
        self.probes[key] = task
        self.catalogs[key] = catalog

        def complete(task: asyncio.Task[CommandCatalog]) -> None:
            if self.probes.get(key) is task:
                del self.probes[key]

        task.add_done_callback(complete)
        return (await asyncio.shield(task)).snapshot()

    async def probe(
        self, agent: AgentTentacle, catalog: CommandCatalog
    ) -> CommandCatalog:
        """Replace a loading catalog only while it still represents this discovery."""
        context = catalog.context
        key = CommandCatalogKey(
            user_id=context.user_id,
            agent_id=context.agent_id,
            conversation_id=context.conversation.id
            if context.conversation is not None
            else None,
        )
        try:
            async with asyncio.timeout(self.timeout):
                result = await agent.probe_commands(context)
                if not result.context.matches(context):
                    raise ValueError("command catalog belongs to another context")
                result = result.snapshot()
                descriptors = {
                    entry
                    for entry in result.descriptors
                    if entry.name.casefold() != "goal"
                }
                if descriptors != result.descriptors:
                    result.descriptors = descriptors
                    result.limitations.append(
                        "Goal commands are excluded until automatic continuation is supported."
                    )
        except asyncio.CancelledError:
            if self.catalogs.get(key) is not catalog:
                return CommandCatalog(context=context, status="loading")
            raise
        except Exception:
            logger.exception("Command discovery failed for agent %s", agent.id)
            result = CommandCatalog(
                context=context,
                status="failed",
                message="Command discovery failed; refresh to retry.",
            )
        if self.catalogs.get(key) is not catalog:
            return CommandCatalog(context=context, status="loading")
        self.catalogs[key] = result
        return result

    @asynccontextmanager
    async def validate(
        self,
        agent: AgentTentacle,
        context: CommandContext,
        invocation: CommandInvocation,
        *,
        delivery_id: str,
        session: OctomateSession | None = None,
    ) -> AsyncGenerator[ValidatedCommand | CommandOutcome]:
        """Hold the turn guard while validating a delivery and its execution.

        Refresh the catalog, then check access and current context once. Refusals
        and matching retries yield their outcome. Otherwise yield a ValidatedCommand
        with the authorized surface, profile and descriptor. The surface and
        profile may be absent before a conversation exists. Prepare available user
        capabilities and call `execute` inside this scope; validation never dispatches
        a command. The guard remains held through the caller's runtime cleanup.
        """
        if not delivery_id:
            raise ValueError("command deliveries require a delivery_id")
        conversation_id = context.conversation.id if context.conversation else None
        if session is not None and (
            session.conversation_id != conversation_id
            or session.current_agent_id != agent.id
        ):
            raise ValueError("gateway session belongs to another command context")
        if conversation_id is not None and conversation_id in self.gateway.sessions:
            yield CommandError(
                status="busy", message="This conversation already has an active turn."
            )
            return
        async with self.gateway.driving(session, conversation_id=conversation_id):
            catalog = await self.discover(agent, context, refresh=True)
            channel = self.tentacles.get(context.address.channel_tentacle_id)
            if (
                not isinstance(channel, ChannelTentacle)
                or self.tentacles.get(context.agent_id) is not agent
                or agent.id not in channel.agent_ids
            ):
                yield CommandError(
                    status="unavailable",
                    message="The command agent is unavailable on this channel.",
                )
                return
            profile = await self.users.profile(channel.id, context.address.user_id)
            new_composer = (
                conversation_id is None
                and isinstance(channel.config, TrunklineChannelConfig)
                and context.address.user_id
                == context.address.chat_id
                == str(context.user_id)
                and context.address.channel_thread_id is None
            )
            if (profile is None and not new_composer) or (
                profile is not None and profile.user_id != context.user_id
            ):
                yield CommandError(
                    status="unavailable", message="The command surface is unavailable."
                )
                return
            surface = None
            current = context
            route_matches = True
            if conversation_id is not None:
                surface = await self.threads.get(
                    ThreadKey.from_address(context.address),
                    with_messages=False,
                    user_id=context.user_id,
                )
                try:
                    conversation = await self.conversations.get(
                        conversation_id, with_history=False
                    )
                except ValueError:
                    conversation = None
                thread = surface
                if (
                    surface is not None
                    and conversation is not None
                    and conversation.thread_id != surface.id
                ):
                    thread = await self.threads.get(
                        conversation.thread_id, with_messages=False
                    )
                if (
                    conversation is None
                    or thread is None
                    or surface is None
                    or (thread.parent_thread_id or thread.id) != surface.id
                ):
                    yield CommandError(
                        status="unavailable",
                        message="The command conversation is unavailable.",
                    )
                    return
                route_matches = (
                    conversation.agent_tentacle_id == agent.id
                    and not conversation.subagent_id
                    and surface.active_agent_tentacle_id == agent.id
                )
                current = replace(
                    context,
                    conversation=conversation,
                    cwd=self.workspaces.open(thread.id, await thread.project).path,
                    model=surface.active_model,
                    permission_mode=conversation.permission_mode
                    or agent.default_permission_mode,
                )
            try:
                current = replace(current, model=agent.resolve_model(current.model))
                if current.permission_mode is not None:
                    agent.check_permission_mode(current.permission_mode)
            except ValueError:
                current = None
            if current is None or not route_matches or not context.matches(current):
                yield CommandError(
                    status="stale",
                    message="The command context changed; discover commands again.",
                )
                return
            recorded = (
                await self.threads.find_message(surface.id, delivery_id, "inbound")
                if surface is not None
                else None
            )
            if recorded is not None:
                if (
                    not isinstance(recorded, ThreadCommand)
                    or profile is None
                    or recorded.sender_id != profile.id
                    or recorded.conversation_id != conversation_id
                    or recorded.agent_tentacle_id != agent.id
                    or recorded.invocation != invocation
                ):
                    yield CommandError(
                        status="failed",
                        message="This delivery ID already identifies another request.",
                    )
                    return
                yield recorded.outcome or CommandError(
                    status="failed",
                    message="This command delivery was already accepted, but no outcome "
                    "was recorded; its effects may already have occurred.",
                )
                return
            descriptors = {
                descriptor.id: descriptor for descriptor in catalog.descriptors
            }
            descriptor = descriptors.get(invocation.command_id)
            if catalog.status != "ready":
                yield CommandError(
                    status="stale" if catalog.status == "loading" else catalog.status,
                    message=catalog.message
                    or "The command catalog changed; discover it again.",
                )
            elif descriptor is None:
                yield CommandError(
                    status="unknown", message="This command is no longer available."
                )
            elif descriptor.unavailable_reason is not None:
                yield CommandError(
                    status="unavailable", message=descriptor.unavailable_reason
                )
            elif descriptor.requires_conversation and conversation_id is None:
                yield CommandError(
                    status="unavailable",
                    message="This command requires an existing conversation.",
                )
            elif invocation.attachments and descriptor.accepts_attachments is not True:
                yield CommandError(
                    status="unsupported",
                    message="This command does not declare attachment support.",
                )
            else:
                yield ValidatedCommand(surface, profile, descriptor)

    @asynccontextmanager
    async def execute[OutputT, DepsT](
        self,
        agent: AgentTentacle[OutputT, DepsT],
        context: CommandContext,
        invocation: CommandInvocation,
        validated: ValidatedCommand,
        *,
        delivery_id: str,
        deferred_suspender: DeferredSuspender | None = None,
        capabilities: list[AgentCapability[DepsT]] | None = None,
    ) -> AsyncGenerator[
        CommandOutcome | AsyncGenerator[ReactStreamEvent[OutputT], None]
    ]:
        """Record and execute a delivery inside its successful `validate` scope.

        Validation and user preparation belong to the caller. Persist the receipt
        before native dispatch, preserve arguments, and invalidate the catalog on
        exit. Consume streamed events inside this context so runtime cleanup and
        outcome recording finish before the validation scope releases its guard.
        Without a conversation, dispatch produces only a direct outcome: no
        receipt, run or thread is created, and retries are not deduplicated.
        Direct adapter failures become failed outcomes; stream errors and
        cancellation propagate to the caller.
        """
        conversation = context.conversation
        surface = validated.surface
        profile = validated.profile
        async with AsyncExitStack() as stack:
            receipt = None
            if conversation is not None:
                assert surface is not None
                assert profile is not None
                text = f"/{validated.descriptor.name}"
                if invocation.arguments:
                    text += f" {invocation.arguments}"
                intent = invocation.model_copy(deep=True)
                receipt = ThreadCommand(
                    thread_id=surface.id,
                    platform_message_id=delivery_id,
                    direction="inbound",
                    actor_kind="human",
                    user_id=context.address.user_id,
                    sender_id=profile.id,
                    agent_tentacle_id=agent.id,
                    conversation_id=conversation.id,
                    invocation=intent,
                    segments=[TextSegment(data={"text": text}), *intent.attachments],
                    message_text=text,
                )
                await self.threads.store_message(receipt, surface)
                stack.callback(self.invalidate, conversation_id=conversation.id)
            else:
                stack.callback(self.invalidate, agent_id=agent.id)
            interrupted = CommandError(
                status="failed",
                message="Command execution did not finish; its effects may already have occurred.",
            )
            try:
                stream = await stack.enter_async_context(
                    aclosing(
                        agent.execute_command(
                            context,
                            invocation,
                            deferred_suspender=deferred_suspender,
                            capabilities=capabilities,
                        )
                    )
                )
                result = await anext(stream)
                if isinstance(result, CommandResult | CommandError):
                    await stream.aclose()
                elif receipt is None:
                    await stream.aclose()
                    result = CommandError(
                        status="failed",
                        message="The command adapter emitted a run without a conversation.",
                    )
            except asyncio.CancelledError:
                if receipt is not None:
                    await self.threads.record_command_outcome(receipt.id, interrupted)
                raise
            except Exception:
                logger.exception("Command execution failed for agent %s", agent.id)
                result = CommandError(
                    status="failed",
                    message="Command execution failed; its effects may already have occurred.",
                )
            if isinstance(result, CommandResult | CommandError):
                if receipt is not None:
                    await self.threads.record_command_outcome(receipt.id, result)
                yield result
            else:
                assert receipt is not None
                yield await stack.enter_async_context(
                    self.record_stream(stream, result, receipt.id, interrupted)
                )

    @asynccontextmanager
    async def record_stream[OutputT](
        self,
        stream: AsyncGenerator[CommandOutcome | ReactStreamEvent[OutputT], None],
        first_event: ReactStreamEvent[OutputT],
        receipt_id: uuid.UUID,
        interrupted: CommandError,
    ) -> AsyncGenerator[AsyncGenerator[ReactStreamEvent[OutputT], None]]:
        """Record completion only after consumption and runtime cleanup both finish.

        Events remain owned by the normal stream consumer; a duplicate delivery
        receives completion status rather than replayed events. An abandoned or
        failed stream keeps the failure outcome prepared before dispatch.
        """
        completed = asyncio.Event()
        outcome: CommandOutcome = interrupted

        async def forward() -> AsyncGenerator[ReactStreamEvent[OutputT], None]:
            yield first_event
            async for event in stream:
                if isinstance(event, CommandResult | CommandError):
                    raise RuntimeError(
                        "An agent command stream yielded a direct outcome"
                    )
                yield event
            completed.set()

        try:
            async with aclosing(stream), aclosing(forward()) as forwarded:
                yield forwarded
            if completed.is_set():
                outcome = CommandResult()
        finally:
            await self.threads.record_command_outcome(receipt_id, outcome)

    def invalidate(
        self,
        *,
        agent_id: str | None = None,
        conversation_id: uuid.UUID | None = None,
    ) -> None:
        """Forget matching catalogs; pending results cannot repopulate them.

        No filters clears all. Invalidated waiters receive loading and must
        resolve context again before requesting another catalog.
        """
        for key in list(self.catalogs):
            if agent_id is not None and key.agent_id != agent_id:
                continue
            if conversation_id is not None and key.conversation_id != conversation_id:
                continue
            del self.catalogs[key]

    async def close(self) -> None:
        """Cancel and drain discovery before the host shuts down its agents."""
        probes = list(self.probes.values())
        for task in probes:
            task.cancel()
        await asyncio.gather(*probes, return_exceptions=True)
        self.catalogs.clear()

capacity class-attribute

capacity: int = 256

timeout class-attribute

timeout: float = 10.0

tentacles instance-attribute

tentacles: dict[str, Tentacle] = tentacles

users instance-attribute

users: UserManager = users

conversations instance-attribute

conversations: ConversationManager = conversations

threads instance-attribute

threads: ThreadManager = threads

workspaces instance-attribute

workspaces: WorkspaceManager = workspaces

gateway instance-attribute

gateway: GatewayManager = gateway

catalogs instance-attribute

catalogs: LRUCache[CommandCatalogKey, CommandCatalog] = (
    LRUCache(maxsize=self.capacity)
)

probes instance-attribute

probes: dict[CommandCatalogKey, Task[CommandCatalog]] = {}

discover async

discover(
    agent: AgentTentacle,
    context: CommandContext,
    *,
    refresh: bool = False,
) -> CommandCatalog

Read or refresh a catalog, sharing active discovery in the same context.

Loading and failed results are reprobed on the next lookup. Canceling one caller leaves shared discovery alive under its timeout. New composers share a catalog per user and agent while their contexts match.

Source code in octomate/managers/commands.py

async def discover(
    self,
    agent: AgentTentacle,
    context: CommandContext,
    *,
    refresh: bool = False,
) -> CommandCatalog:
    """Read or refresh a catalog, sharing active discovery in the same context.

    Loading and failed results are reprobed on the next lookup. Canceling one
    caller leaves shared discovery alive under its timeout. New composers
    share a catalog per user and agent while their contexts match.
    """
    conversation = context.conversation
    if context.agent_id != agent.id or (
        conversation is not None and conversation.agent_tentacle_id != agent.id
    ):
        raise ValueError("command context belongs to another agent")

    key = CommandCatalogKey(
        user_id=context.user_id,
        agent_id=context.agent_id,
        conversation_id=conversation.id if conversation is not None else None,
    )
    while True:
        catalog = self.catalogs.get(key)
        task = self.probes.get(key)
        if catalog is not None and catalog.context.matches(context):
            if task is not None and not task.done():
                return (await asyncio.shield(task)).snapshot()
            if not refresh and catalog.status not in {"loading", "failed"}:
                return catalog.snapshot()
        if task is None or task.done():
            break
        self.catalogs.pop(key, None)
        if not task.cancelling():
            task.cancel()
        await asyncio.shield(asyncio.gather(task, return_exceptions=True))
    self.catalogs.pop(key, None)
    catalog = CommandCatalog(context=context, status="loading").snapshot()
    task = asyncio.create_task(self.probe(agent, catalog))
    self.probes[key] = task
    self.catalogs[key] = catalog

    def complete(task: asyncio.Task[CommandCatalog]) -> None:
        if self.probes.get(key) is task:
            del self.probes[key]

    task.add_done_callback(complete)
    return (await asyncio.shield(task)).snapshot()

probe async

probe(
    agent: AgentTentacle, catalog: CommandCatalog
) -> CommandCatalog

Replace a loading catalog only while it still represents this discovery.

Source code in octomate/managers/commands.py

async def probe(
    self, agent: AgentTentacle, catalog: CommandCatalog
) -> CommandCatalog:
    """Replace a loading catalog only while it still represents this discovery."""
    context = catalog.context
    key = CommandCatalogKey(
        user_id=context.user_id,
        agent_id=context.agent_id,
        conversation_id=context.conversation.id
        if context.conversation is not None
        else None,
    )
    try:
        async with asyncio.timeout(self.timeout):
            result = await agent.probe_commands(context)
            if not result.context.matches(context):
                raise ValueError("command catalog belongs to another context")
            result = result.snapshot()
            descriptors = {
                entry
                for entry in result.descriptors
                if entry.name.casefold() != "goal"
            }
            if descriptors != result.descriptors:
                result.descriptors = descriptors
                result.limitations.append(
                    "Goal commands are excluded until automatic continuation is supported."
                )
    except asyncio.CancelledError:
        if self.catalogs.get(key) is not catalog:
            return CommandCatalog(context=context, status="loading")
        raise
    except Exception:
        logger.exception("Command discovery failed for agent %s", agent.id)
        result = CommandCatalog(
            context=context,
            status="failed",
            message="Command discovery failed; refresh to retry.",
        )
    if self.catalogs.get(key) is not catalog:
        return CommandCatalog(context=context, status="loading")
    self.catalogs[key] = result
    return result

validate async

validate(
    agent: AgentTentacle,
    context: CommandContext,
    invocation: CommandInvocation,
    *,
    delivery_id: str,
    session: OctomateSession | None = None,
) -> AsyncGenerator[ValidatedCommand | CommandOutcome]

Hold the turn guard while validating a delivery and its execution.

Refresh the catalog, then check access and current context once. Refusals and matching retries yield their outcome. Otherwise yield a ValidatedCommand with the authorized surface, profile and descriptor. The surface and profile may be absent before a conversation exists. Prepare available user capabilities and call execute inside this scope; validation never dispatches a command. The guard remains held through the caller's runtime cleanup.

Source code in octomate/managers/commands.py

@asynccontextmanager
async def validate(
    self,
    agent: AgentTentacle,
    context: CommandContext,
    invocation: CommandInvocation,
    *,
    delivery_id: str,
    session: OctomateSession | None = None,
) -> AsyncGenerator[ValidatedCommand | CommandOutcome]:
    """Hold the turn guard while validating a delivery and its execution.

    Refresh the catalog, then check access and current context once. Refusals
    and matching retries yield their outcome. Otherwise yield a ValidatedCommand
    with the authorized surface, profile and descriptor. The surface and
    profile may be absent before a conversation exists. Prepare available user
    capabilities and call `execute` inside this scope; validation never dispatches
    a command. The guard remains held through the caller's runtime cleanup.
    """
    if not delivery_id:
        raise ValueError("command deliveries require a delivery_id")
    conversation_id = context.conversation.id if context.conversation else None
    if session is not None and (
        session.conversation_id != conversation_id
        or session.current_agent_id != agent.id
    ):
        raise ValueError("gateway session belongs to another command context")
    if conversation_id is not None and conversation_id in self.gateway.sessions:
        yield CommandError(
            status="busy", message="This conversation already has an active turn."
        )
        return
    async with self.gateway.driving(session, conversation_id=conversation_id):
        catalog = await self.discover(agent, context, refresh=True)
        channel = self.tentacles.get(context.address.channel_tentacle_id)
        if (
            not isinstance(channel, ChannelTentacle)
            or self.tentacles.get(context.agent_id) is not agent
            or agent.id not in channel.agent_ids
        ):
            yield CommandError(
                status="unavailable",
                message="The command agent is unavailable on this channel.",
            )
            return
        profile = await self.users.profile(channel.id, context.address.user_id)
        new_composer = (
            conversation_id is None
            and isinstance(channel.config, TrunklineChannelConfig)
            and context.address.user_id
            == context.address.chat_id
            == str(context.user_id)
            and context.address.channel_thread_id is None
        )
        if (profile is None and not new_composer) or (
            profile is not None and profile.user_id != context.user_id
        ):
            yield CommandError(
                status="unavailable", message="The command surface is unavailable."
            )
            return
        surface = None
        current = context
        route_matches = True
        if conversation_id is not None:
            surface = await self.threads.get(
                ThreadKey.from_address(context.address),
                with_messages=False,
                user_id=context.user_id,
            )
            try:
                conversation = await self.conversations.get(
                    conversation_id, with_history=False
                )
            except ValueError:
                conversation = None
            thread = surface
            if (
                surface is not None
                and conversation is not None
                and conversation.thread_id != surface.id
            ):
                thread = await self.threads.get(
                    conversation.thread_id, with_messages=False
                )
            if (
                conversation is None
                or thread is None
                or surface is None
                or (thread.parent_thread_id or thread.id) != surface.id
            ):
                yield CommandError(
                    status="unavailable",
                    message="The command conversation is unavailable.",
                )
                return
            route_matches = (
                conversation.agent_tentacle_id == agent.id
                and not conversation.subagent_id
                and surface.active_agent_tentacle_id == agent.id
            )
            current = replace(
                context,
                conversation=conversation,
                cwd=self.workspaces.open(thread.id, await thread.project).path,
                model=surface.active_model,
                permission_mode=conversation.permission_mode
                or agent.default_permission_mode,
            )
        try:
            current = replace(current, model=agent.resolve_model(current.model))
            if current.permission_mode is not None:
                agent.check_permission_mode(current.permission_mode)
        except ValueError:
            current = None
        if current is None or not route_matches or not context.matches(current):
            yield CommandError(
                status="stale",
                message="The command context changed; discover commands again.",
            )
            return
        recorded = (
            await self.threads.find_message(surface.id, delivery_id, "inbound")
            if surface is not None
            else None
        )
        if recorded is not None:
            if (
                not isinstance(recorded, ThreadCommand)
                or profile is None
                or recorded.sender_id != profile.id
                or recorded.conversation_id != conversation_id
                or recorded.agent_tentacle_id != agent.id
                or recorded.invocation != invocation
            ):
                yield CommandError(
                    status="failed",
                    message="This delivery ID already identifies another request.",
                )
                return
            yield recorded.outcome or CommandError(
                status="failed",
                message="This command delivery was already accepted, but no outcome "
                "was recorded; its effects may already have occurred.",
            )
            return
        descriptors = {
            descriptor.id: descriptor for descriptor in catalog.descriptors
        }
        descriptor = descriptors.get(invocation.command_id)
        if catalog.status != "ready":
            yield CommandError(
                status="stale" if catalog.status == "loading" else catalog.status,
                message=catalog.message
                or "The command catalog changed; discover it again.",
            )
        elif descriptor is None:
            yield CommandError(
                status="unknown", message="This command is no longer available."
            )
        elif descriptor.unavailable_reason is not None:
            yield CommandError(
                status="unavailable", message=descriptor.unavailable_reason
            )
        elif descriptor.requires_conversation and conversation_id is None:
            yield CommandError(
                status="unavailable",
                message="This command requires an existing conversation.",
            )
        elif invocation.attachments and descriptor.accepts_attachments is not True:
            yield CommandError(
                status="unsupported",
                message="This command does not declare attachment support.",
            )
        else:
            yield ValidatedCommand(surface, profile, descriptor)

execute async

execute(
    agent: AgentTentacle[OutputT, DepsT],
    context: CommandContext,
    invocation: CommandInvocation,
    validated: ValidatedCommand,
    *,
    delivery_id: str,
    deferred_suspender: DeferredSuspender | None = None,
    capabilities: list[AgentCapability[DepsT]]
    | None = None,
) -> AsyncGenerator[
    CommandOutcome
    | AsyncGenerator[ReactStreamEvent[OutputT], None]
]

Record and execute a delivery inside its successful validate scope.

Validation and user preparation belong to the caller. Persist the receipt before native dispatch, preserve arguments, and invalidate the catalog on exit. Consume streamed events inside this context so runtime cleanup and outcome recording finish before the validation scope releases its guard. Without a conversation, dispatch produces only a direct outcome: no receipt, run or thread is created, and retries are not deduplicated. Direct adapter failures become failed outcomes; stream errors and cancellation propagate to the caller.

Source code in octomate/managers/commands.py

@asynccontextmanager
async def execute[OutputT, DepsT](
    self,
    agent: AgentTentacle[OutputT, DepsT],
    context: CommandContext,
    invocation: CommandInvocation,
    validated: ValidatedCommand,
    *,
    delivery_id: str,
    deferred_suspender: DeferredSuspender | None = None,
    capabilities: list[AgentCapability[DepsT]] | None = None,
) -> AsyncGenerator[
    CommandOutcome | AsyncGenerator[ReactStreamEvent[OutputT], None]
]:
    """Record and execute a delivery inside its successful `validate` scope.

    Validation and user preparation belong to the caller. Persist the receipt
    before native dispatch, preserve arguments, and invalidate the catalog on
    exit. Consume streamed events inside this context so runtime cleanup and
    outcome recording finish before the validation scope releases its guard.
    Without a conversation, dispatch produces only a direct outcome: no
    receipt, run or thread is created, and retries are not deduplicated.
    Direct adapter failures become failed outcomes; stream errors and
    cancellation propagate to the caller.
    """
    conversation = context.conversation
    surface = validated.surface
    profile = validated.profile
    async with AsyncExitStack() as stack:
        receipt = None
        if conversation is not None:
            assert surface is not None
            assert profile is not None
            text = f"/{validated.descriptor.name}"
            if invocation.arguments:
                text += f" {invocation.arguments}"
            intent = invocation.model_copy(deep=True)
            receipt = ThreadCommand(
                thread_id=surface.id,
                platform_message_id=delivery_id,
                direction="inbound",
                actor_kind="human",
                user_id=context.address.user_id,
                sender_id=profile.id,
                agent_tentacle_id=agent.id,
                conversation_id=conversation.id,
                invocation=intent,
                segments=[TextSegment(data={"text": text}), *intent.attachments],
                message_text=text,
            )
            await self.threads.store_message(receipt, surface)
            stack.callback(self.invalidate, conversation_id=conversation.id)
        else:
            stack.callback(self.invalidate, agent_id=agent.id)
        interrupted = CommandError(
            status="failed",
            message="Command execution did not finish; its effects may already have occurred.",
        )
        try:
            stream = await stack.enter_async_context(
                aclosing(
                    agent.execute_command(
                        context,
                        invocation,
                        deferred_suspender=deferred_suspender,
                        capabilities=capabilities,
                    )
                )
            )
            result = await anext(stream)
            if isinstance(result, CommandResult | CommandError):
                await stream.aclose()
            elif receipt is None:
                await stream.aclose()
                result = CommandError(
                    status="failed",
                    message="The command adapter emitted a run without a conversation.",
                )
        except asyncio.CancelledError:
            if receipt is not None:
                await self.threads.record_command_outcome(receipt.id, interrupted)
            raise
        except Exception:
            logger.exception("Command execution failed for agent %s", agent.id)
            result = CommandError(
                status="failed",
                message="Command execution failed; its effects may already have occurred.",
            )
        if isinstance(result, CommandResult | CommandError):
            if receipt is not None:
                await self.threads.record_command_outcome(receipt.id, result)
            yield result
        else:
            assert receipt is not None
            yield await stack.enter_async_context(
                self.record_stream(stream, result, receipt.id, interrupted)
            )

record_stream async

record_stream(
    stream: AsyncGenerator[
        CommandOutcome | ReactStreamEvent[OutputT], None
    ],
    first_event: ReactStreamEvent[OutputT],
    receipt_id: UUID,
    interrupted: CommandError,
) -> AsyncGenerator[
    AsyncGenerator[ReactStreamEvent[OutputT], None]
]

Record completion only after consumption and runtime cleanup both finish.

Events remain owned by the normal stream consumer; a duplicate delivery receives completion status rather than replayed events. An abandoned or failed stream keeps the failure outcome prepared before dispatch.

Source code in octomate/managers/commands.py

@asynccontextmanager
async def record_stream[OutputT](
    self,
    stream: AsyncGenerator[CommandOutcome | ReactStreamEvent[OutputT], None],
    first_event: ReactStreamEvent[OutputT],
    receipt_id: uuid.UUID,
    interrupted: CommandError,
) -> AsyncGenerator[AsyncGenerator[ReactStreamEvent[OutputT], None]]:
    """Record completion only after consumption and runtime cleanup both finish.

    Events remain owned by the normal stream consumer; a duplicate delivery
    receives completion status rather than replayed events. An abandoned or
    failed stream keeps the failure outcome prepared before dispatch.
    """
    completed = asyncio.Event()
    outcome: CommandOutcome = interrupted

    async def forward() -> AsyncGenerator[ReactStreamEvent[OutputT], None]:
        yield first_event
        async for event in stream:
            if isinstance(event, CommandResult | CommandError):
                raise RuntimeError(
                    "An agent command stream yielded a direct outcome"
                )
            yield event
        completed.set()

    try:
        async with aclosing(stream), aclosing(forward()) as forwarded:
            yield forwarded
        if completed.is_set():
            outcome = CommandResult()
    finally:
        await self.threads.record_command_outcome(receipt_id, outcome)

invalidate

invalidate(
    *,
    agent_id: str | None = None,
    conversation_id: UUID | None = None,
) -> None

Forget matching catalogs; pending results cannot repopulate them.

No filters clears all. Invalidated waiters receive loading and must resolve context again before requesting another catalog.

Source code in octomate/managers/commands.py

def invalidate(
    self,
    *,
    agent_id: str | None = None,
    conversation_id: uuid.UUID | None = None,
) -> None:
    """Forget matching catalogs; pending results cannot repopulate them.

    No filters clears all. Invalidated waiters receive loading and must
    resolve context again before requesting another catalog.
    """
    for key in list(self.catalogs):
        if agent_id is not None and key.agent_id != agent_id:
            continue
        if conversation_id is not None and key.conversation_id != conversation_id:
            continue
        del self.catalogs[key]

close async

close() -> None

Cancel and drain discovery before the host shuts down its agents.

Source code in octomate/managers/commands.py

async def close(self) -> None:
    """Cancel and drain discovery before the host shuts down its agents."""
    probes = list(self.probes.values())
    for task in probes:
        task.cancel()
    await asyncio.gather(*probes, return_exceptions=True)
    self.catalogs.clear()

octomate.commands

Authenticated command discovery and execution for browser clients.

command_router module-attribute

command_router = APIRouter(
    prefix="/api/commands",
    tags=["commands"],
    dependencies=[Depends(browser_request)],
)

command_context async

command_context(
    agent_id: Annotated[
        str,
        Body(
            min_length=1, description="The selected agent."
        ),
    ],
    address: Annotated[
        ChannelAddress,
        Body(
            description="The originating channel surface."
        ),
    ],
    user: Annotated[User, Depends(current_user)],
    app: Annotated[Octomate, Depends(application)],
    conversations: Annotated[
        ConversationManager, Depends(conversation_manager)
    ],
    threads: Annotated[
        ThreadManager, Depends(thread_manager)
    ],
    workspaces: Annotated[
        WorkspaceManager, Depends(workspace_manager)
    ],
    conversation_id: Annotated[
        UUID | None,
        Body(
            description="An existing conversation; omit for a new composer."
        ),
    ] = None,
    model: Annotated[
        str | None,
        Body(
            description="A model selection for a new composer; existing conversations use their stored model."
        ),
    ] = None,
    permission_mode: Annotated[
        str | None,
        Body(
            description="An approval posture for a new composer; existing conversations use their stored permissions."
        ),
    ] = None,
) -> CommandContext

Resolve inspection context; reject foreign-agent conversations with HTTP 422 before discovery, workspace lookup, or graph execution.

Source code in octomate/commands.py

async def command_context(
    agent_id: Annotated[str, Body(min_length=1, description="The selected agent.")],
    address: Annotated[
        ChannelAddress,
        Body(description="The originating channel surface."),
    ],
    user: Annotated[User, Depends(current_user)],
    app: Annotated[Octomate, Depends(application)],
    conversations: Annotated[ConversationManager, Depends(conversation_manager)],
    threads: Annotated[ThreadManager, Depends(thread_manager)],
    workspaces: Annotated[WorkspaceManager, Depends(workspace_manager)],
    conversation_id: Annotated[
        uuid.UUID | None,
        Body(description="An existing conversation; omit for a new composer."),
    ] = None,
    model: Annotated[
        str | None,
        Body(
            description="A model selection for a new composer; existing conversations use their stored model."
        ),
    ] = None,
    permission_mode: Annotated[
        str | None,
        Body(
            description="An approval posture for a new composer; existing conversations use their stored permissions."
        ),
    ] = None,
) -> CommandContext:
    """Resolve inspection context; reject foreign-agent conversations with HTTP 422
    before discovery, workspace lookup, or graph execution.
    """
    channel = app.channels.get(address.channel_tentacle_id)
    if channel is None:
        raise HTTPException(status_code=404, detail="Command channel is unavailable")
    agent = app.agents.get(agent_id)
    if agent is None or agent_id not in channel.agent_ids:
        raise HTTPException(
            status_code=404, detail="Command agent is unavailable on this channel"
        )

    conversation = None
    cwd = None
    if conversation_id is not None:
        try:
            conversation = await conversations.get(conversation_id, with_history=False)
        except ValueError as error:
            raise HTTPException(
                status_code=404, detail="Command conversation is unavailable"
            ) from error
        thread = await threads.get(conversation.thread_id, with_messages=False)
        if thread is None:
            raise HTTPException(
                status_code=404, detail="Command conversation is unavailable"
            )
        if thread.kind == "native_thread":
            raise HTTPException(409, "Synced native sessions are read-only.")
        if conversation.agent_tentacle_id != agent.id:
            raise HTTPException(
                status_code=422,
                detail="Command conversation belongs to another agent",
            )
        surface = await threads.surface(thread)
        model = surface.active_model
        permission_mode = conversation.permission_mode or agent.default_permission_mode
        cwd = workspaces.open(thread.id, await thread.project).path

    try:
        model = agent.resolve_model(model)
        if permission_mode is not None:
            agent.check_permission_mode(permission_mode)
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error)) from error
    return CommandContext(
        agent_id=agent_id,
        user_id=user.id,
        address=address,
        cwd=cwd,
        conversation=conversation,
        model=model,
        permission_mode=permission_mode or agent.default_permission_mode,
    )

discover_commands async

discover_commands(
    context: Annotated[
        CommandContext, Depends(command_context)
    ],
    app: Annotated[Octomate, Depends(application)],
    refresh: Annotated[
        bool,
        Body(
            description="Rediscover rather than reuse a completed catalog."
        ),
    ] = False,
    prefix: Annotated[
        str,
        Body(
            description="Case-insensitive command-name prefix; omitted means the full catalog."
        ),
    ] = "",
) -> CommandCatalog

Discover or complete commands without creating a conversation or model turn.

Source code in octomate/commands.py

@command_router.post(
    "/catalog",
    response_model_exclude={
        "context": {
            "conversation": {"runs", "messages", "external_id", "allowed_tools"}
        }
    },
)
async def discover_commands(
    context: Annotated[CommandContext, Depends(command_context)],
    app: Annotated[Octomate, Depends(application)],
    refresh: Annotated[
        bool, Body(description="Rediscover rather than reuse a completed catalog.")
    ] = False,
    prefix: Annotated[
        str,
        Body(
            description="Case-insensitive command-name prefix; omitted means the full catalog."
        ),
    ] = "",
) -> CommandCatalog:
    """Discover or complete commands without creating a conversation or model turn."""
    agent = app.agents[context.agent_id]
    return await agent.discover_commands(context, refresh=refresh, prefix=prefix)

execute_command async

execute_command(
    context: Annotated[
        CommandContext, Depends(command_context)
    ],
    app: Annotated[Octomate, Depends(application)],
    command_id: Annotated[
        str,
        Body(
            min_length=1,
            description="The selected descriptor's opaque ID.",
        ),
    ],
    delivery_id: Annotated[
        str,
        Body(
            min_length=1,
            description="Delivery ID. Existing conversations deduplicate matching retries; executions without a conversation have no persisted receipt.",
        ),
    ],
    arguments: Annotated[
        str,
        Body(
            description="Raw command arguments, preserved unchanged."
        ),
    ] = "",
    attachments: Annotated[
        list[JsonValue] | None,
        Body(
            max_length=0,
            description="Must be empty until browser uploads are supported.",
        ),
    ] = None,
) -> CommandOutcomeEvent

Execute explicit command intent in the resolved context.

A descriptor may permit a direct result before a conversation exists. Such execution creates no thread, conversation, run or receipt; it is not replayable.

Every execution answers with the terminal command outcome, after cleanup and persistence; clients must not render it as another channel message. Every channel executes through the Reflex command node, which publishes feedback, run output and approvals on the nerve: the destination channel draws them, and the thread's live socket carries them to a watching console, up to the reflex falling asleep. A failed execution answers no outcome. Authentication and request-validation errors remain non-2xx JSON responses.

Source code in octomate/commands.py

@command_router.post("/execute")
async def execute_command(
    context: Annotated[CommandContext, Depends(command_context)],
    app: Annotated[Octomate, Depends(application)],
    command_id: Annotated[
        str, Body(min_length=1, description="The selected descriptor's opaque ID.")
    ],
    delivery_id: Annotated[
        str,
        Body(
            min_length=1,
            description="Delivery ID. Existing conversations deduplicate matching retries; "
            "executions without a conversation have no persisted receipt.",
        ),
    ],
    arguments: Annotated[
        str, Body(description="Raw command arguments, preserved unchanged.")
    ] = "",
    attachments: Annotated[
        list[JsonValue] | None,
        Body(
            max_length=0,
            description="Must be empty until browser uploads are supported.",
        ),
    ] = None,
) -> CommandOutcomeEvent:
    """Execute explicit command intent in the resolved context.

    A descriptor may permit a direct result before a conversation exists. Such
    execution creates no thread, conversation, run or receipt; it is not replayable.

    Every execution answers with the terminal command outcome, after cleanup and
    persistence; clients must not render it as another channel message. Every
    channel executes through the Reflex command node, which publishes feedback,
    run output and approvals on the nerve: the destination channel draws them,
    and the thread's live socket carries them to a watching console, up to the
    reflex falling asleep. A failed execution answers no outcome.
    Authentication and request-validation errors remain non-2xx JSON responses.
    """
    invocation = CommandInvocation(command_id=command_id, arguments=arguments)
    try:
        reflex = await app.kick(CommandSignal(context, invocation, delivery_id))
    except StillAwake as error:
        raise HTTPException(status_code=409, detail=str(error)) from error
    result = None if reflex is None else await reflex
    outcome = (
        result if isinstance(result, CommandResult | CommandError) else CommandResult()
    )
    return CommandOutcomeEvent(outcome=outcome)