Trunkline

octomate.tentacles.trunkline.routes

The Trunkline console HTTP surface, mounted under /api/trunkline.

The console is an entry and a reader, and it reads the domain in the shape the domain has: a thread, the conversations under it, the runs under those, and the thread's chat messages. Every read answers with the transmuter itself rather than a console-shaped copy of it, so the payload is the row.

What a read leaves out it leaves out deliberately. A thread's messages are their own request, because a listing that carried every thread's ledger would be the ledger; and a run's model messages leave only a page at a time, since one run can hold thousands. Both pages are Arcanus keyset pages, newest first.

Both halves of leaving something out are load-bearing. response_model_exclude keeps a relation out of the payload; it does nothing about the query, and every relation named here is lazy="selectin". The managers suppress the load itself (ThreadManager.get(with_messages=...), ConversationManager.for_thread), so these reads do not fetch a ledger to drop it.

Every read under a thread reads the thread first, and 404s on an id the relay does not know — a sub-resource of a thread that does not exist must say so rather than answer the empty list its query would return. Only the ledger read asks for the messages; the rest want the row.

Directives create or continue threads on the trunkline channel itself. Directives, batch responses and thread operations are answered as soon as their kick is on its way, and what the run does reaches the console over /live: a websocket that carries a thread's wire events for as long as it is open, named by the thread's row id or, for a Trunkline thread with no row yet, by the key its directives carry. The kick's reflex_asleep arrives there last.

Thread operations act on a thread of any channel the user can access. GET threads/{id}/operations answers a ThreadOperations, which the console reads when its operation controls open. GET threads/{id}/channels/{channel}/addresses lists one level of a connected channel's destinations as ChannelAddress rows whose metadata carries a name: a row with metadata.inside is a place to open, by passing that value back as ?inside=, and metadata.barred says why a thread cannot land in a row. A channel that cannot be browsed, has no connected agent or is not linked to the user answers 409. A listed address is a suggestion, validated again when Teleport submits it. POST threads/{id}/teleport takes a TeleportBody and POST threads/{id}/summon a SummonBody. Both refuse a thread with a turn running or approvals or questions pending; the turn then shows on the live socket as a gateway event when the conversation moves, a run_started event opening each run, and a run_error when the turn fails.

Where the work happened — a thread's project and a run's directory — is read here and nowhere written: a thread's project is frozen when its row is written, and both are learned from the session that ran, so no endpoint takes either.

A conversation's approval posture is the exception: it is the console's to change mid-thread, and the conversation is what remembers it. It is switched through PATCH, or — while the thread is still being composed and has no row to switch — carried on the directive that creates it.

RouteInfo pydantic-model

Bases: BaseModel

One route a directive may name, as GET /routes lists it.

Fields:

• id (str)
• agent (str)
• model (AgentRouteModelName | None)

Source code in octomate/tentacles/trunkline/routes.py

class RouteInfo(BaseModel):
    """One route a directive may name, as `GET /routes` lists it."""

    id: str = Field(description="The route id a directive's `model` field names.")
    agent: str
    model: AgentRouteModelName | None

id pydantic-field

id: str

The route id a directive's model field names.

agent pydantic-field

agent: str

model pydantic-field

model: AgentRouteModelName | None

ChannelInfo pydantic-model

Bases: BaseModel

One connected channel tentacle, as GET /channels lists it.

Fields:

• id (str)
• kind (str)

Source code in octomate/tentacles/trunkline/routes.py

class ChannelInfo(BaseModel):
    """One connected channel tentacle, as `GET /channels` lists it."""

    id: str = Field(description="The connected channel tentacle's id.")
    kind: str = Field(description="The tentacle class name — SlackTentacle, …")

id pydantic-field

id: str

The connected channel tentacle's id.

kind pydantic-field

kind: str

The tentacle class name — SlackTentacle, …

DirectiveBody pydantic-model

Bases: BaseModel

The body of a directive sent to a thread.

The text, and the route, project, and posture a thread's first directive may pick.

Fields:

• text (str)
• message_id (str | None)
• model (str | None)
• project (str | None)
• permission_mode (AgentPermissionMode | None)
• effort (str | None)

Source code in octomate/tentacles/trunkline/routes.py

class DirectiveBody(BaseModel):
    """The body of a directive sent to a thread.

    The text, and the route, project, and posture a thread's first directive may
    pick.
    """

    text: str
    message_id: str | None = None
    model: str | None = Field(
        default=None,
        description="A route id from GET /routes ('agent:model'). Honored only "
        "on a thread's first directive; a differing pick on an owned thread is "
        "refused with 409 (re-routing needs a manual handoff).",
    )
    project: str | None = Field(
        default=None,
        description="A project name from GET /projects, filing the thread under "
        "it. Honored only on a thread's first directive, since a thread's project "
        "is frozen when the row is written; null is a real answer, and leaves the "
        "thread a chat that is in no project at all.",
    )
    permission_mode: AgentPermissionMode | None = Field(
        default=None,
        description="A posture from GET /permissions, in the vocabulary of the "
        "agent this thread is routed to, stored on that agent's conversation before "
        "the run. This is how a thread that does not exist yet gets one; an owned "
        "thread's is switched through PATCH /conversations/{id}/permission-mode. "
        "Null asks for no change, and never clears a posture already stored.",
    )
    effort: str | None = Field(
        default=None,
        description="An effort level advertised by the selected model, stored before "
        "the first run. Null leaves the conversation's selection unchanged.",
    )

text pydantic-field

text: str

message_id pydantic-field

message_id: str | None = None

model pydantic-field

model: str | None = None

A route id from GET /routes ('agent:model'). Honored only on a thread's first directive; a differing pick on an owned thread is refused with 409 (re-routing needs a manual handoff).

project pydantic-field

project: str | None = None

A project name from GET /projects, filing the thread under it. Honored only on a thread's first directive, since a thread's project is frozen when the row is written; null is a real answer, and leaves the thread a chat that is in no project at all.

permission_mode pydantic-field

permission_mode: AgentPermissionMode | None = None

A posture from GET /permissions, in the vocabulary of the agent this thread is routed to, stored on that agent's conversation before the run. This is how a thread that does not exist yet gets one; an owned thread's is switched through PATCH /conversations/{id}/permission-mode. Null asks for no change, and never clears a posture already stored.

effort pydantic-field

effort: str | None = None

An effort level advertised by the selected model, stored before the first run. Null leaves the conversation's selection unchanged.

TeleportBody

Bases: TypedDict

A teleport asked from the console. new_thread is true unless sent; the console sends none, so Teleport opens a thread at the destination picked. A prompt is the user's next message in the Trunkline thread the move lands in, and any other destination refuses it.

Source code in octomate/tentacles/trunkline/routes.py

class TeleportBody(TypedDict):
    """A teleport asked from the console. `new_thread` is true unless sent; the
    console sends none, so Teleport opens a thread at the destination picked. A
    `prompt` is the user's next message in the Trunkline thread the move lands
    in, and any other destination refuses it."""

    destination: ChannelAddress
    new_thread: NotRequired[bool]
    hint: Annotated[str, Field(min_length=1, max_length=1_000)]
    # Sent to the Trunkline thread the move lands in, as its user's next message.
    prompt: NotRequired[Annotated[str, Field(min_length=1, max_length=8_000)]]

destination instance-attribute

destination: ChannelAddress

new_thread instance-attribute

new_thread: NotRequired[bool]

hint instance-attribute

hint: Annotated[str, Field(min_length=1, max_length=1000)]

prompt instance-attribute

prompt: NotRequired[
    Annotated[str, Field(min_length=1, max_length=8000)]
]

SummonBody

Bases: TypedDict

A summon asked from the console: the agent and model that take the conversation over where it is, the brief they start from, and the hint recorded with the handoff.

Source code in octomate/tentacles/trunkline/routes.py

class SummonBody(TypedDict):
    """A summon asked from the console: the agent and model that take the
    conversation over where it is, the brief they start from, and the hint
    recorded with the handoff."""

    agent_id: str
    model: str
    brief: Annotated[str, Field(min_length=1, max_length=8_000)]
    hint: Annotated[str, Field(min_length=1, max_length=1_000)]
    effort: NotRequired[str | None]

agent_id instance-attribute

agent_id: str

model instance-attribute

model: str

brief instance-attribute

brief: Annotated[str, Field(min_length=1, max_length=8000)]

hint instance-attribute

hint: Annotated[str, Field(min_length=1, max_length=1000)]

effort instance-attribute

effort: NotRequired[str | None]

AgentPostures pydantic-model

Bases: BaseModel

An agent's approval modes and default, as GET /permissions lists them.

Fields:

• modes (tuple[PermissionMode, ...])
• default (AgentPermissionMode)

Source code in octomate/tentacles/trunkline/routes.py

class AgentPostures(BaseModel):
    """An agent's approval modes and default, as `GET /permissions` lists them."""

    modes: tuple[PermissionMode, ...] = Field(
        description="This agent's whole vocabulary, in the order a picker steps "
        "through it."
    )
    default: AgentPermissionMode = Field(
        description="What a conversation of this agent's runs under when it declares "
        "nothing of its own — the configured default, which is what a NULL "
        "`permission_mode` means rather than 'no posture'."
    )

modes pydantic-field

modes: tuple[PermissionMode, ...]

This agent's whole vocabulary, in the order a picker steps through it.

default pydantic-field

default: AgentPermissionMode

What a conversation of this agent's runs under when it declares nothing of its own — the configured default, which is what a NULL permission_mode means rather than 'no posture'.

accessible_thread async

accessible_thread(
    thread_id: UUID7,
    threads: Annotated[
        ThreadManager, Depends(thread_manager)
    ],
    user: Annotated[User, Depends(current_user)],
) -> Thread

Source code in octomate/tentacles/trunkline/routes.py

async def accessible_thread(
    thread_id: UUID7,
    threads: Annotated[ThreadManager, Depends(thread_manager)],
    user: Annotated[User, Depends(current_user)],
) -> Thread:
    thread = await threads.get(thread_id, with_messages=False, user_id=user.id)
    if thread is None:
        raise HTTPException(status_code=404, detail=f"no thread {thread_id}")
    return thread

thread_gateway async

thread_gateway(
    thread: Annotated[Thread, Depends(accessible_thread)],
    user: Annotated[User, Depends(current_user)],
    app: Annotated[Octomate, Depends(application)],
    gateway: Annotated[
        GatewayManager, Depends(gateway_manager)
    ],
    actions: Annotated[
        DeferredActionManager,
        Depends(deferred_action_manager),
    ],
    users: Annotated[UserManager, Depends(user_manager)],
    threads: Annotated[
        ThreadManager, Depends(thread_manager)
    ],
    workspaces: Annotated[
        WorkspaceManager, Depends(workspace_manager)
    ],
) -> OctomateSession

Source code in octomate/tentacles/trunkline/routes.py

async def thread_gateway(
    thread: Annotated[Thread, Depends(accessible_thread)],
    user: Annotated[User, Depends(current_user)],
    app: Annotated[Octomate, Depends(application)],
    gateway: Annotated[GatewayManager, Depends(gateway_manager)],
    actions: Annotated[DeferredActionManager, Depends(deferred_action_manager)],
    users: Annotated[UserManager, Depends(user_manager)],
    threads: Annotated[ThreadManager, Depends(thread_manager)],
    workspaces: Annotated[WorkspaceManager, Depends(workspace_manager)],
) -> OctomateSession:
    agent_id = thread.active_agent_tentacle_id
    if agent_id is None:
        raise HTTPException(409, "This thread has no active agent.")
    if any(gateway.get(c.id) is not None for c in thread.conversations):
        raise HTTPException(409, "Wait for the active turn to finish.")
    if await actions.pending_for_thread(thread.id):
        raise HTTPException(409, "Resolve the pending questions or approvals first.")
    native = thread.kind == "native_thread"
    channel = app.channels.get(thread.channel_tentacle_id)
    if not native and (channel is None or agent_id not in app.agents):
        raise HTTPException(409, "The source channel or agent is not connected.")
    profile = UserProfile(user_id=user.id, channel_user_id=str(user.id), name=user.name)
    if isinstance(channel, TrunklineTentacle):
        # The row a directive's sender resolves to, so what the operation's turn
        # writes under this profile — an authorization, a handoff — names a stored one.
        profile = await users.ensure_profile(channel.id, profile)
    elif channel is not None:
        linked = await users.linked_profiles(profile)
        source_profile = next(
            (one for one in linked if one.channel_tentacle_id == channel.id), None
        )
        if source_profile is None:
            raise HTTPException(403, "Link your account on the source channel first.")
        profile = source_profile
    else:
        profile.channel_tentacle_id = thread.channel_tentacle_id
    address = thread.key.address(profile.channel_user_id)
    return OctomateSession(
        channel_routes=gateway.available_routes(app.channels, app.agents),
        current_agent_id=agent_id,
        channels=app.channels,
        agents=app.agents,
        users=users,
        user_profile=profile,
        thread_id=thread.id,
        native=native,
        # A thread's row does not say who can read it, so its channel reads that
        # off the address. A native session has no channel and nobody else in it.
        conversation_address=replace(
            address, shared=channel is not None and channel.is_shared(address)
        ),
        threads=threads,
        workspaces=workspaces,
    )

build_trunkline_router

build_trunkline_router(
    octomate: Octomate, *, channel_id: str = "trunkline"
) -> APIRouter

Build the console HTTP router bound to a registered TrunklineTentacle.

Source code in octomate/tentacles/trunkline/routes.py

def build_trunkline_router(
    octomate: Octomate,
    *,
    channel_id: str = "trunkline",
) -> APIRouter:
    """Build the console HTTP router bound to a registered TrunklineTentacle."""
    registered = octomate.channels.get(channel_id)
    if not isinstance(registered, TrunklineTentacle):
        raise ValueError(
            f"Trunkline router requires a registered TrunklineTentacle at "
            f"{channel_id!r}"
        )
    channel: TrunklineTentacle = registered

    router = APIRouter(
        prefix="/api/trunkline",
        tags=["trunkline"],
        dependencies=[Depends(browser_request), Depends(current_user)],
    )

    @router.get("/threads/{thread_id}/operations")
    async def thread_operations(
        session: Annotated[OctomateSession, Depends(thread_gateway)],
    ) -> ThreadOperations:
        return await session.operations

    @router.get(
        "/threads/{thread_id}/channels/{channel_id}/addresses",
        summary="List one level of a connected channel's destinations",
    )
    async def list_addresses(
        channel_id: str,
        session: Annotated[OctomateSession, Depends(thread_gateway)],
        inside: Annotated[str | None, Query(max_length=200)] = None,
    ) -> list[ChannelAddress]:
        try:
            return await session.list_addresses(channel_id, inside)
        except GatewayRefusal as exc:
            raise HTTPException(409, str(exc)) from exc

    @router.post("/threads/{thread_id}/teleport", status_code=202, response_model=None)
    async def teleport_thread(
        body: TeleportBody,
        session: Annotated[OctomateSession, Depends(thread_gateway)],
    ) -> None:
        prompt = body.get("prompt")
        # Only a thread here takes a message from this console.
        if prompt is not None and body["destination"].channel_tentacle_id != channel.id:
            raise HTTPException(
                409, "A prompt can follow a teleport only into a Trunkline thread."
            )
        try:
            await session.teleport(
                destination=body["destination"],
                new_thread=body.get("new_thread", True),
                hint=body["hint"],
                prompt=prompt,
            )
            await octomate.kick(session.thread_operation())
        except (GatewayRefusal, StillAwake) as exc:
            raise HTTPException(409, str(exc)) from exc

    @router.post("/threads/{thread_id}/summon", status_code=202, response_model=None)
    async def summon_thread(
        body: SummonBody,
        session: Annotated[OctomateSession, Depends(thread_gateway)],
    ) -> None:
        try:
            await session.summon(
                agent_id=body["agent_id"],
                model=body["model"],
                hint=body["hint"],
                reason="Summon requested from Trunkline",
                brief=body["brief"],
                effort=body.get("effort"),
            )
            await octomate.kick(session.thread_operation())
        except (GatewayRefusal, StillAwake) as exc:
            raise HTTPException(409, str(exc)) from exc

    @router.get("/health", include_in_schema=False)
    async def health() -> JSONResponse:
        return JSONResponse({"ok": True})

    @router.get("/routes")
    async def routes() -> list[RouteInfo]:
        return [
            RouteInfo(
                id=f"{agent_config.agent}{ROUTE_SEP}{agent_config.model or ''}",
                agent=agent_config.agent,
                model=agent_config.model,
            )
            for agent_config in channel.routable_agents()
        ]

    @router.get("/projects")
    async def list_projects(
        projects: Annotated[ProjectManager, Depends(project_manager)],
    ) -> list[Project]:
        """The projects a new thread can be filed under — the enabled ones, since
        a project whose root disk has lost is nowhere to work."""
        return [project for project in projects.list() if project.enabled]

    @router.get("/permissions")
    async def permission_modes() -> dict[str, AgentPostures]:
        """Each registered agent's approval vocabulary, in the order a picker steps
        through it — a provider's own scale, never a shared one, so the list a
        conversation may be given is keyed by the agent that reads it.

        Only the agents this instance runs: a posture is stored on a conversation,
        and an agent nobody can be routed to has no conversation to store one on. The
        tailed runtimes are absent for the same reason, which is what keeps the console
        reporting their posture rather than offering to switch it."""
        postures: dict[str, AgentPostures] = {}
        for agent_id, agent in octomate.agents.items():
            configured = agent.default_permission_mode
            # An agent that failed to start has no catalog, only its configuration.
            if configured is None or not agent.permission_modes:
                continue
            agent.check_permission_mode(configured)
            postures[agent_id] = AgentPostures(
                modes=agent.permission_modes, default=configured
            )
        return postures

    @router.get("/channels")
    async def list_channels() -> list[ChannelInfo]:
        """The channels this instance actually connected — the sidebar's rail
        mirrors this, so a channel disabled in config never appears."""
        return [
            ChannelInfo(id=channel_id, kind=type(tentacle).__name__)
            for channel_id, tentacle in octomate.channels.items()
        ]

    @router.get("/agents")
    async def list_agents() -> list[AgentInfo]:
        """The registered agents with their catalogs — every model each one can be
        routed to, what that route claims to be for, and the effort levels it takes.

        `GET /routes` answers the composer's question, which is narrower: the ids a
        directive may name, with the entry agent's default alone. This answers the
        Agents page's question, which is what the instance is made of, so it keeps
        the models that question drops and adds what no route id carries.

        The session counts are the live ones, read off the tentacle as the request
        passes: what this instance is driving now, and what it is only reading.
        """
        return [agent.info for agent in octomate.agents.values()]

    @router.get("/profile")
    async def profile(user: Annotated[User, Depends(current_user)]) -> ProfileInfo:
        """The signed-in user, linked channel profiles, and OAuth MCP authorizations."""
        return await octomate.profile(user)

    @router.get(
        "/threads",
        summary="This user's threads, most recently touched first",
        response_model_exclude={"__all__": {"messages", "parent"}},
    )
    async def list_threads(
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        user: Annotated[User, Depends(current_user)],
    ) -> list[Thread]:
        return await threads.list_threads(user_id=user.id)

    @router.get("/threads/{thread_id}", response_model_exclude={"messages", "parent"})
    async def read_thread(
        thread: Annotated[Thread, Depends(accessible_thread)],
    ) -> Thread:
        """One thread and its handoffs, by row id — any channel's, not only the
        console's own."""
        return thread

    @router.get(
        "/threads/{thread_id}/messages",
        summary="The thread's chat ledger, newest first, a page at a time",
        response_model=Page[ThreadMessage | ThreadCommand | ThreadArtifact],
        response_model_exclude={"items": {"__all__": {"model_messages"}}},
    )
    async def thread_messages(
        thread: Annotated[Thread, Depends(accessible_thread)],
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        cursor: str | None = None,
        unbound: bool = False,
        limit: Annotated[int, Query(ge=1, le=500)] = 100,
    ) -> Page[ThreadMessage]:
        """Every row the thread shows, including command receipts and artifact references.
        `cursor` is a page's `next_cursor`, read further back. `unbound` keeps to the rows no
        model message of the thread carries — notices, commands, a chat room's
        talk — since the rest arrive bound to the model messages they prompted or
        delivered."""
        try:
            return await threads.messages_page(
                thread.id, cursor, unbound=unbound, limit=limit
            )
        except ValidationError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

    @router.get(
        "/threads/{thread_id}/conversations/messages",
        summary="The model messages of the thread's own conversations, newest first",
        response_model=Page[ModelRequest | ModelResponse],
        response_model_exclude={
            "items": {"__all__": {"thread_messages": {"__all__": {"model_messages"}}}}
        },
    )
    async def thread_model_messages(
        thread: Annotated[Thread, Depends(accessible_thread)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
        cursor: str | None = None,
        limit: Annotated[int, Query(ge=1, le=500)] = 100,
    ) -> Page[ModelRequest | ModelResponse]:
        """Every run of the thread's own agents, as one stream across its
        conversations, a page at a time: where the thinking and the tool calls are.
        Each message carries the chat rows bound to it — who asked, in their own
        words, and what was delivered. `cursor` is a page's `next_cursor`."""
        try:
            return await conversations.messages_page(thread.id, cursor, limit=limit)
        except ValidationError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

    @router.get("/threads/{thread_id}/usage")
    async def thread_usage(
        thread: Annotated[Thread, Depends(accessible_thread)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
    ) -> RequestUsage:
        """Every token the thread's own conversations spent, summed over their
        responses — the total a reader paging the ledger cannot add up itself."""
        return await conversations.usage(thread.id)

    @router.get(
        "/threads/{thread_id}/conversations",
        summary="The thread's agent conversations, each with its runs",
        response_model_exclude={
            "__all__": {"messages": True, "runs": {"__all__": {"messages"}}}
        },
    )
    async def thread_conversations(
        thread: Annotated[Thread, Depends(accessible_thread)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
    ) -> list[Conversation]:
        """Subagent conversations included — they name their parent, so a reader
        can fold them under the run whose tool call spawned them.

        The runs come without their model messages: a thread's run can hold
        thousands, which `/conversations/messages` serves a page at a time."""
        return await conversations.for_thread(thread.id)

    @router.get("/threads/{thread_id}/project")
    async def thread_project(
        thread: Annotated[Thread, Depends(accessible_thread)],
    ) -> Project | None:
        """The project this thread's work is in; null for a thread no project
        claims. Frozen: it is set when the thread is created, from the directory
        the session ran in, and no endpoint changes it."""
        return await thread.project

    @router.get(
        "/threads/{thread_id}/batches",
        summary="Unanswered action batches, oldest first",
        response_model_exclude={"__all__": {"requests"}},
    )
    async def thread_batches(
        thread: Annotated[Thread, Depends(accessible_thread)],
        deferred_actions: Annotated[
            DeferredActionManager, Depends(deferred_action_manager)
        ],
    ) -> list[DeferredActionBatch]:
        """The waiting questions and approvals, so a reload re-renders the
        feelers a run is blocked on. `requests` stays behind: it is the agent's
        own tool-call payload, and the actions carry what a reader asks."""
        return await deferred_actions.pending_for_thread(thread.id)

    @router.post(
        "/threads/{thread_key}/messages",
        status_code=202,
        response_model=None,
        summary="Send a directive; its run shows on the thread's live socket",
        description="`thread_key` is the trunkline platform thread id (not the "
        "row id): a fresh key creates the thread, an existing one continues it.",
    )
    async def send_directive(
        thread_key: str,
        body: DirectiveBody,
        user: Annotated[User, Depends(current_user)],
    ) -> None:
        directive = TrunklineDirective(
            thread_id=thread_key,
            user=user,
            text=body.text,
            message_id=body.message_id,
            model=body.model,
            project=body.project,
            permission_mode=body.permission_mode,
            effort=body.effort,
        )
        try:
            await channel.handle_directive(directive)
        except (RouteLockedError, StillAwake) as error:
            raise HTTPException(status_code=409, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

    @router.patch(
        "/conversations/{conversation_id}/permission-mode",
        summary="Switch the approval posture this conversation's agent works under",
        response_model_exclude={"messages", "runs"},
    )
    async def set_permission_mode(
        conversation_id: UUID7,
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
        user: Annotated[User, Depends(current_user)],
        permission_mode: Annotated[
            AgentPermissionMode | None,
            Body(
                embed=True,
                description="A posture from GET /permissions, in this conversation's "
                "own agent's vocabulary. Null clears it: nothing is declared and the agent's "
                "configured default decides again.",
            ),
        ] = None,
    ) -> Conversation:
        """Change the selection through the owning agent, including live updates
        where its runtime supports them. Pending approval cards stay separate.
        """
        try:
            conversation = await conversations.get(conversation_id)
        except ValueError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        thread = await accessible_thread(conversation.thread_id, threads, user)
        if thread.kind == "native_thread":
            raise HTTPException(409, "Synced native sessions are read-only.")
        try:
            agent = octomate.agents.get(conversation.agent_tentacle_id)
            if agent is None:
                raise ValueError(
                    "This conversation has no driven agent to set permissions on"
                )
            return await agent.set_permission_mode(conversation, permission_mode)
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

    @router.patch(
        "/conversations/{conversation_id}/effort",
        summary="Set the reasoning effort this conversation's runs ask for",
        response_model_exclude={"messages", "runs"},
    )
    async def set_effort(
        conversation_id: UUID7,
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
        user: Annotated[User, Depends(current_user)],
        effort: Annotated[
            str | None,
            Body(
                embed=True,
                description="A level the thread's route claims, as GET /agents lists them. "
                "Null clears it: nothing is declared and the runtime's own default decides.",
            ),
        ] = None,
    ) -> Conversation:
        """Validate the level against the model the thread runs on, then store it.
        The next run reads it; a run already going keeps the level it started at.
        """
        try:
            conversation = await conversations.get(conversation_id, with_history=False)
        except ValueError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        thread = await accessible_thread(conversation.thread_id, threads, user)
        agent = octomate.agents.get(conversation.agent_tentacle_id)
        if thread.kind == "native_thread":
            raise HTTPException(409, "Synced native sessions are read-only.")
        if agent is None:
            raise HTTPException(
                status_code=422,
                detail="This conversation has no driven agent to set effort on",
            )
        if effort is not None:
            surface = await threads.surface(thread)
            try:
                await agent.resolve_effort(
                    conversation, model=surface.active_model, effort=effort
                )
            except ValueError as error:
                raise HTTPException(status_code=422, detail=str(error)) from error
        return await conversations.set_effort(conversation, effort)

    @router.post(
        "/batches/{batch_id}/resolve",
        status_code=202,
        response_model=None,
        summary="Answer a deferred-action batch; the resumed run shows on the "
        "thread's live socket",
    )
    async def resolve_batch(
        batch_id: UUID7,
        user: Annotated[User, Depends(current_user)],
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        conversations: Annotated[ConversationManager, Depends(conversation_manager)],
        deferred_actions: Annotated[
            DeferredActionManager, Depends(deferred_action_manager)
        ],
        answers: Annotated[
            dict[UUID7, QuestionAnswer],
            Body(default_factory=dict, description="Answers keyed by action id."),
        ],
        approvals: Annotated[
            dict[UUID7, bool],
            Body(default_factory=dict, description="Approvals keyed by action id."),
        ],
        allow_session: Annotated[
            bool,
            Body(
                description="Keep approved tools allowed for the rest of the session."
            ),
        ] = False,
    ) -> None:
        try:
            batch = await deferred_actions.get_batch(batch_id)
        except ValueError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        conversation = await conversations.get(batch.conversation_id)
        thread = await accessible_thread(conversation.thread_id, threads, user)
        if thread.kind == "native_thread":
            raise HTTPException(409, "Synced native sessions are read-only.")
        if batch.status != "pending":
            # A resolved batch must not resume twice (double-click, retry).
            raise HTTPException(status_code=409, detail=f"batch already {batch.status}")
        if (
            batch.response_mode == "resume"
            and ThreadKey.from_address(batch.source_address) in octomate.awakes
        ):
            # The kick would wait out the turn running there, holding this
            # request open for as long.
            raise HTTPException(status_code=409, detail=str(StillAwake()))
        await octomate.kick(
            DeferredActionBatchResponse(
                batch_id=batch_id,
                responder_id=str(user.id),
                answers=answers,
                approvals=approvals,
                allow_session=allow_session,
            )
        )

    return router

build_trunkline_live_router

build_trunkline_live_router(
    channel: TrunklineTentacle,
) -> APIRouter

Build the console's live socket router. It is a router of its own because the console router runs the same-origin write guard on every request, and a handshake has no method for that guard to read.

Source code in octomate/tentacles/trunkline/routes.py

def build_trunkline_live_router(channel: TrunklineTentacle) -> APIRouter:
    """Build the console's live socket router. It is a router of its own because
    the console router runs the same-origin write guard on every request, and a
    handshake has no method for that guard to read."""
    router = APIRouter(prefix="/api/trunkline", tags=["trunkline"])

    @router.websocket("/live/user")
    async def watch_user(
        websocket: WebSocket,
        user: Annotated[User, Depends(current_user)],
    ) -> None:
        """Subscribe to the authenticated user's topic, without replay.

        The subscription precedes acceptance. Read the thread list after the
        handshake, then apply the events received while that read was pending.
        """
        await channel.siphon.watch(websocket, UserTopic(user.id))

    @router.websocket("/live")
    async def watch_thread(
        websocket: WebSocket,
        user: Annotated[User, Depends(current_user)],
        threads: Annotated[ThreadManager, Depends(thread_manager)],
        thread_id: Annotated[
            UUID7 | None,
            Query(description="A thread the user can read, by row id — any channel's."),
        ] = None,
        thread_key: Annotated[
            str | None,
            Query(
                description="A Trunkline thread by the key its directives carry, "
                "so a thread being composed can be watched before its first "
                "directive; empty for the composer itself, which a command "
                "sent before any thread exists answers on."
            ),
        ] = None,
    ) -> None:
        """The thread's events as they happen, for as long as the socket is open:
        what a directive, a batch answer or a thread operation sets going, and
        the kick's `reflex_asleep` last. The thread is named by `thread_id` or by
        `thread_key`, one of the two; a socket naming neither, or both, is
        closed with 1008."""
        if thread_id is not None and thread_key is None:
            thread = await threads.get(thread_id, with_messages=False, user_id=user.id)
            if thread is None:
                await websocket.close(code=1008, reason=f"no thread {thread_id}")
                return
            key = thread.key
        elif thread_key is not None and thread_id is None:
            # The key a directive from this user files the thread under.
            key = ThreadKey(channel.id, "thread", str(user.id), thread_key or None)
        else:
            await websocket.close(
                code=1008, reason="Name the thread by its id or by its key, not both"
            )
            return
        await channel.siphon.watch(websocket, ThreadTopic(key))

    return router

octomate.schemas.operations

Thread operations exposed to authenticated clients.

OperationAvailability pydantic-model

Bases: BaseModel

Whether one operation can run on a thread, and with what.

Fields:

• destinations (list[ChannelAddress])
• here (ChannelAddress | None)
• routes (dict[str, list[AgentRoute]])
• reason (str | None)

Source code in octomate/schemas/operations.py

class OperationAvailability(BaseModel):
    """Whether one operation can run on a thread, and with what."""

    destinations: list[ChannelAddress] = Field(
        default_factory=list,
        description="Suggested places to go. An empty list sets no reason by "
        "itself: the destination browser can still find one.",
    )
    here: ChannelAddress | None = Field(
        default=None,
        description="For Summon, this conversation's own address, where it is "
        "handed over.",
    )
    routes: dict[str, list[AgentRoute]] = Field(
        default_factory=dict,
        description="What the operation can run, keyed by channel id: for Summon, "
        "every other agent on this conversation's own channel; for Teleport, the "
        "conversation's own on each connected channel, where a channel with none "
        "cannot be the destination.",
    )
    reason: str | None = Field(
        default=None,
        description="Why the operation is unavailable; set only then.",
    )

destinations pydantic-field

destinations: list[ChannelAddress]

Suggested places to go. An empty list sets no reason by itself: the destination browser can still find one.

here pydantic-field

here: ChannelAddress | None = None

For Summon, this conversation's own address, where it is handed over.

routes pydantic-field

routes: dict[str, list[AgentRoute]]

What the operation can run, keyed by channel id: for Summon, every other agent on this conversation's own channel; for Teleport, the conversation's own on each connected channel, where a channel with none cannot be the destination.

reason pydantic-field

reason: str | None = None

Why the operation is unavailable; set only then.

ThreadOperations pydantic-model

Bases: BaseModel

What the console may do with a thread: each operation's availability, the thread's own surface, and the channels nothing can land in.

Fields:

• source (ChannelAddress | None)
• teleport (OperationAvailability)
• summon (OperationAvailability)
• barred (dict[str, str])

Source code in octomate/schemas/operations.py

class ThreadOperations(BaseModel):
    """What the console may do with a thread: each operation's availability, the
    thread's own surface, and the channels nothing can land in."""

    source: ChannelAddress | None = Field(
        description="The surface this conversation is on. Its `shared`, against a "
        "destination's, is what a client warns by before private history lands in "
        "a shared place."
    )
    teleport: OperationAvailability
    summon: OperationAvailability
    barred: dict[str, str] = Field(
        default_factory=dict,
        description="Connected channels where no conversation can land, keyed by "
        "channel id, with why: shown to either operation, never picked.",
    )

source pydantic-field

source: ChannelAddress | None

The surface this conversation is on. Its shared, against a destination's, is what a client warns by before private history lands in a shared place.

teleport pydantic-field

teleport: OperationAvailability

summon pydantic-field

summon: OperationAvailability

barred pydantic-field

barred: dict[str, str]

Connected channels where no conversation can land, keyed by channel id, with why: shown to either operation, never picked.