Threads and conversations

octomate.managers.thread

The thread chat ledger: threads, their messages, bindings to model messages, handoffs, and history search.

TITLE_MAX module-attribute

TITLE_MAX = 72

BindRefusal

Bases: ValueError

A bind refused by policy — a thread that is not work, or one already bound — carrying the sentence a model may be told and correct from. A ValueError still, so nothing that never told the two apart changes.

Source code in octomate/managers/thread.py

class BindRefusal(ValueError):
    """A bind refused by policy — a thread that is not work, or one already bound —
    carrying the sentence a model may be told and correct from. A `ValueError`
    still, so nothing that never told the two apart changes."""

ThreadManager

Bases: Manager, Locks[ThreadKey]

Owns durable thread chat ledger persistence.

Source code in octomate/managers/thread.py

class ThreadManager(Manager, Locks[ThreadKey]):
    """Owns durable thread chat ledger persistence."""

    users: UserManager
    nerve: Nerve

    def __init__(self, *, users: UserManager, nerve: Nerve) -> None:
        # Every ledger row references its sender's registry profile — the host
        # constructs this manager around its one identity registry.
        self.users = users
        self.nerve = nerve

    async def ensure(
        self,
        address_or_key: ChannelAddress | ThreadKey,
        *,
        project: Project | None = None,
        title: str | None = None,
    ) -> Thread:
        """The thread this key names, created if it is new.

        A key names a surface, so it resolves the thread nobody opened. A chat
        room's sub-threads share its address and are reached by id, through it.

        A channel-supplied `title` refreshes the surface's stored name, including
        existing groups and DMs previously named from an opening line.

        `project` attributes a thread being created and is ignored for one that
        already exists: a thread's project is frozen once the row is written, so
        declaring a project later attributes new threads rather than rewriting old
        ones. It is not part of the key either — the key is what the thread's history
        is filed under, and attribution must never move it.

        Only a thread and a native thread can take one; naming a project for a DM or a
        group chat raises rather than dropping it silently.
        """
        if isinstance(address_or_key, ChannelAddress):
            key = ThreadKey.from_address(address_or_key)
        else:
            key = address_or_key
        if project is not None and key.kind not in ATTRIBUTABLE_KINDS:
            raise ValueError(
                f"{key} is a {key.kind} and cannot be attributed to project "
                f"{project.name!r}: only a thread or a native_thread is work."
            )
        # The lock spans the read, the insert and the commit: a loser that woke
        # after the winner's read but before its commit would still find nothing.
        async with self.lock(key), async_session() as session:
            thread = await session.one_or_none(
                Thread,
                options=[selectinload(Thread["conversations"]).noload("*")],
                expressions=keyed(key),
            )
            if thread is None:
                thread = Thread(
                    channel_tentacle_id=key.channel_tentacle_id,
                    chat_type=key.chat_type,
                    chat_id=key.chat_id,
                    channel_thread_id=key.channel_thread_id,
                    kind=key.kind,
                    project_id=project.id if project is not None else None,
                )
                session.add(thread)
            title_changed = title is not None and thread.title != title
            if title_changed:
                thread.title = title
                thread.updated_at = datetime.now(UTC)
            await session.flush()
            await thread.handoffs
            await thread.project
            await session.commit()
        if title_changed:
            await self.publish(
                thread,
                ThreadTitleChangedEvent(
                    thread_id=thread.id,
                    title=thread.title,
                    updated_at=thread.updated_at,
                ),
            )
        return thread

    async def open_sub_thread(self, parent: Thread) -> Thread:
        """The thread one kick in `parent` does its work in.

        A dm and a group chat have no end, so a conversation held against one would
        accumulate for as long as the chat room does. Answering happens in a thread
        of its own instead, and everything a thread already governs governs it — one
        conversation per agent, runs continuing one by one. Only the chat everyone
        sees stays on the chat room, which is what `surface` resolves back to.

        Opened, never looked up: a sub-thread has no address of its own to be found
        by, and each kick wants its own. A redelivered event therefore opens a
        second — that is a duplicate to stop where it is made, in the ledger, not
        here.
        """
        if parent.parent_thread_id is not None:
            raise ValueError(
                f"thread {parent.id} is already a sub-thread of "
                f"{parent.parent_thread_id}: a chat room nests one level"
            )
        key = parent.key
        thread = Thread(
            channel_tentacle_id=key.channel_tentacle_id,
            chat_type=key.chat_type,
            chat_id=key.chat_id,
            # The chat room's own, verbatim: the platform never made this thread, so
            # there is no id of its own to carry, and anything invented here would be
            # sent back to the platform as a real one. Sharing the address is what
            # makes a sub-thread's key resolve to the chat room it was opened in.
            channel_thread_id=key.channel_thread_id,
            kind=key.kind,
            parent_thread_id=parent.id,
        )
        async with async_session() as session:
            session.add(thread)
            await session.flush()
            await thread.handoffs
            await thread.project
            await session.commit()
        return thread

    async def enter(
        self,
        address_or_key: ChannelAddress | ThreadKey,
        *,
        current: Thread | None = None,
    ) -> Thread:
        """The thread a turn arriving at this surface does its work in.

        A thread and a native thread are a piece of work already, so the turn runs in
        the one the key names. A dm and a group chat are not — they have no end, and a
        conversation held against one would accumulate for as long as the chat room
        does — so the turn runs in a sub-thread opened inside the chat room instead.

        `current` is the thread the turn is already in, and it is kept when this
        surface is the chat room that thread was opened in: an agent taking the
        conversation over in place is still answering the same kick, and a second
        sub-thread would split one answer across two model contexts.
        """
        surface = await self.ensure(address_or_key)
        if current is not None and current.parent_thread_id == surface.id:
            return current
        if surface.kind in ATTRIBUTABLE_KINDS:
            return surface
        return await self.open_sub_thread(surface)

    async def surface(self, thread: Thread) -> Thread:
        """The surface this thread is on: itself, or the chat room it was opened in.

        A sub-thread has none of its own — no address, no chat ledger, no prompt
        cursor — so this is the one place the split between where a run happens and
        where it is seen gets made. Nesting is one level, so it never walks.
        """
        if thread.parent_thread_id is None:
            return thread
        parent = await self.get(thread.parent_thread_id, with_messages=False)
        if parent is None:
            raise ValueError(
                f"thread {thread.id} names parent {thread.parent_thread_id}, "
                "which does not exist"
            )
        return parent

    @staticmethod
    def fork_title(source: Conversation, thread: Thread) -> str | None:
        """What a fork of `source`, from `thread`, is first called: what the source
        goes by, the name its runtime gave the session or else the thread's title.
        None when it goes by nothing, and the fork takes a name the way a new thread
        does."""
        return thread_title(source.name or thread.title)

    async def rename(self, thread: Thread, title: str | None) -> Thread:
        """Give the thread the name the runtime running it grabbed for itself.

        Unlike the opening line `store_message` falls back to, this is a name for
        the work rather than for how it was asked for, so it overwrites — and goes
        on overwriting, because the runtime revises it as the session goes on. A
        name with nothing in it is not one, and leaves the thread as it was.
        A group's main chat keeps its platform name, and a DM keeps the bot's
        name; only their work sub-threads take generated titles.
        """
        if thread.kind in {"group", "dm"} and thread.parent_thread_id is None:
            return thread
        named = thread_title(title)
        if named is None or thread.title == named:
            return thread
        async with async_session() as session:
            stored = await session.get(Thread, thread.id)
            if stored is None:
                raise ValueError(f"unknown thread {thread.id}")
            stored.title = named
            stored.updated_at = datetime.now(UTC)
            await session.commit()
        thread.title = named
        await self.publish(
            thread,
            ThreadTitleChangedEvent(
                thread_id=thread.id,
                title=named,
                updated_at=stored.updated_at,
            ),
        )
        return thread

    async def bind(self, thread_id: UUID7, project: Project) -> Thread:
        """Attribute a thread that is in no project to `project`, once.

        The one exception to a project being written when the row is created: a
        chat thread exists before anyone has said what it is about, so the
        attribution has to be able to arrive later. It arrives once. A second bind
        is refused rather than switching — the first project's workspace holds work
        nobody has reviewed yet, and a thread whose project moved is a record of
        what it did somewhere it no longer says it was. A different project is a
        different thread.

        A DM and a group chat are refused for the reason `ATTRIBUTABLE_KINDS`
        gives: they outlive every project in them. Both refusals are `BindRefusal`;
        a thread that does not exist is a plain `ValueError`, a wiring bug.

        No workspace is made here. A run's working directory is fixed when its
        process spawns, so this takes effect on the thread's next turn and the
        caller has to say so.
        """
        async with async_session() as session:
            thread = await session.get(Thread, thread_id)
            if thread is None:
                raise ValueError(f"no thread {thread_id} to bind")
            if thread.kind not in ATTRIBUTABLE_KINDS:
                raise BindRefusal(
                    f"{thread.key} is a {thread.kind} and cannot be attributed to "
                    f"project {project.name!r}: only a thread or a native_thread "
                    f"is work."
                )
            current = await thread.project
            if current is not None:
                raise BindRefusal(
                    f"{thread.key} is already about {current.name!r} and a thread "
                    f"binds once; work on {project.name!r} in its own thread."
                )
            thread.project_id = project.id
            await session.commit()

        # Read the thread back rather than returning the row that changed. `project`
        # is eagerly loaded, so the copy that set the id is still holding the answer
        # it was loaded with.
        bound = await self.get(thread_id)
        if bound is None:
            raise ValueError(f"thread {thread_id} vanished while binding")
        await self.publish(
            bound,
            ThreadProjectBoundEvent(
                thread_id=bound.id,
                project_id=project.id,
                updated_at=bound.updated_at,
            ),
        )
        return bound

    async def get(
        self,
        thread_id: UUID7 | ThreadKey,
        *,
        with_messages: bool = True,
        user_id: UUID7 | None = None,
    ) -> Thread | None:
        """The thread by primary key, or by the key it is filed under, or None —
        its handoffs with the row, its ledger only when asked for. A key never
        creates one, unlike `ensure`.

        `with_messages=True` is the whole ledger, deliberately: it is what a reader
        rebuilding a thread wants, and it is unbounded in a room, so a caller after
        one row wants `find_message` and a caller after the recent ones wants
        `chat_messages_before`.

        Either way the model ledger stays behind: it hangs off a message, and
        `related_model_messages` is how a caller asks for it — dragging it here
        would put a query per message behind every thread read.

        `user_id` restricts access to participation in the thread's own ledger;
        messages inherited from another thread grant no access.
        """
        options = (
            [selectinload(Thread["messages"]).noload(ThreadMessage["model_messages"])]
            if with_messages
            else []
        )
        options.append(selectinload(Thread["conversations"]).noload("*"))
        async with async_session() as session:
            expressions = (
                keyed(thread_id)
                if isinstance(thread_id, ThreadKey)
                else [Thread["id"] == thread_id]
            )
            if user_id is not None:
                expressions.append(
                    Thread["messages"].any(
                        (ThreadMessage["thread_id"] == Thread["id"])
                        & ThreadMessage["sender"].has(UserProfile["user_id"] == user_id)
                    )
                )
            thread = await session.one_or_none(
                Thread, expressions=expressions, options=options
            )
            if thread is None:
                return None
        return thread

    async def list_threads(
        self,
        channel_tentacle_id: str | None = None,
        *,
        limit: int = 100,
        user_id: UUID7 | None = None,
    ) -> list[Thread]:
        """Threads most recently touched first — one channel's, or every
        channel's when `channel_tentacle_id` is None. Sub-threads are not listed;
        the threads a chat room's kicks work in are reached through the chat room.

        Handoffs come with the rows (lazy="selectin", one batched pass); the
        ledgers do not. A listing that loaded them would read every message of
        every thread it names, and no caller has ever wanted that — the one
        reader of a thread's messages asks for that thread.
        """
        # Sub-threads are left out: a listing names the surfaces a person can open,
        # and the threads a chat room's kicks work in are reached through it.
        expressions = [Thread["parent_thread_id"].is_(None)]
        if user_id is not None:
            expressions.append(
                Thread["messages"].any(
                    (ThreadMessage["thread_id"] == Thread["id"])
                    & ThreadMessage["sender"].has(UserProfile["user_id"] == user_id)
                )
            )
        if channel_tentacle_id is not None:
            expressions.append(Thread["channel_tentacle_id"] == channel_tentacle_id)
        async with async_session() as session:
            rows = await session.list(
                Thread,
                limit=limit,
                options=[selectinload(Thread["conversations"]).noload("*")],
                order_bys=[Thread["updated_at"].desc(), Thread["id"].desc()],
                expressions=expressions,
            )
        return list(rows)

    async def publish(self, thread: Thread, *events: ThreadEvent) -> None:
        """Publish committed thread events to each eligible user's topic.

        Resolve registered participants once for the supplied events, without a cache,
        matching listing and access checks. Inherited history grants no access.
        Sub-threads are not listed. Token events never call this method.
        """
        if thread.parent_thread_id is not None:
            return
        async with async_session() as session:
            user_ids = await session.scalars(
                select(UserProfile["user_id"])
                .where(
                    UserProfile["user_id"].is_not(None),
                    UserProfile["id"].in_(
                        select(ThreadMessage["sender_id"]).where(
                            ThreadMessage["thread_id"] == thread.id
                        )
                    ),
                )
                .distinct()
            )
        impulses = [Impulse(event) for event in events]
        for user_id in user_ids:
            for impulse in impulses:
                self.nerve.deliver(UserTopic(user_id), impulse)

    async def store_message(self, message: ThreadMessage, thread: Thread) -> None:
        """Persist a ledger row.

        The row is touched, not only appended to. `updated_at` fires on an update
        of the thread and writing to its ledger is not one, so a session being
        tailed right now would keep the timestamp it was created with and sink, in
        a listing ordered by it, under threads nothing has ever said a word in.

        A thread nobody has named takes its name from this row when a person wrote
        it. The listing carries no messages, so the first thing said is the only
        name a row has until a runtime grabs one of its own (`rename`).
        """
        async with async_session() as session:
            session.add(message)
            row = await session.get(Thread, thread.id)
            if row is None:
                raise ValueError(f"unknown thread {thread.id}")
            previous_title = row.title
            row.updated_at = datetime.now(UTC)
            if (
                row.title is None
                and (
                    row.kind not in {"group", "dm"} or row.parent_thread_id is not None
                )
                and message.direction == "inbound"
                and message.actor_kind == "human"
            ):
                row.title = thread_title(message.message_text)
            await session.commit()
        events: list[ThreadEvent] = [
            ThreadMessageRecordedEvent(
                thread_id=thread.id,
                message_id=message.id,
                updated_at=row.updated_at,
            )
        ]
        if row.title != previous_title:
            events.append(
                ThreadTitleChangedEvent(
                    thread_id=thread.id, title=row.title, updated_at=row.updated_at
                )
            )
        await self.publish(thread, *events)

    async def record_artifact(
        self,
        thread_id: UUID7,
        artifact: Artifact,
        *,
        sender: UserProfile,
        agent_tentacle_id: str,
    ) -> ThreadArtifact:
        """Commit an artifact and its ledger reference together before publication."""
        thread = await self.get(thread_id, with_messages=False)
        if thread is None:
            raise ValueError(f"unknown thread {thread_id}")
        sender = await self.users.ensure_profile(thread.channel_tentacle_id, sender)
        message = ThreadArtifact(
            thread_id=thread.id,
            direction="outbound",
            actor_kind="agent",
            sender_id=sender.id,
            agent_tentacle_id=agent_tentacle_id,
            artifact_id=artifact.id,
            artifact=Relation(artifact),
        )
        await self.store_message(message, thread)
        return message

    async def record_command_outcome(
        self, receipt_id: uuid.UUID, outcome: CommandOutcome
    ) -> None:
        """Persist an outcome without mutating a caller's receipt snapshot. Under the
        thread's lock, only the outcome and the thread's `updated_at` are written."""
        async with async_session() as session:
            receipt = await session.get(
                ThreadCommand, receipt_id, options=[noload("*")]
            )
            if receipt is None:
                raise ValueError(f"command receipt {receipt_id} does not exist")
            thread = await session.get(Thread, receipt.thread_id, options=[noload("*")])
            if thread is None:
                raise ValueError(f"unknown thread {receipt.thread_id}")
            async with self.lock(thread.key):
                receipt.outcome = outcome
                thread.updated_at = datetime.now(UTC)
                await session.commit()
        await self.publish(
            thread,
            ThreadCommandOutcomeRecordedEvent(
                thread_id=thread.id,
                receipt_id=receipt.id,
                outcome=outcome,
                updated_at=thread.updated_at,
            ),
        )

    async def record_inbound(
        self,
        event: MessageEvent,
        *,
        actor_kind: ChannelActorKind = "human",
        agent_tentacle_id: str | None = None,
        happened_at: datetime | None = None,
    ) -> ThreadMessage:
        """Write the event to the thread's ledger, or hand back the row it is
        already on.

        A platform re-sends a message when it misses an ack — Slack does — and the
        second send is the same message, not a second one. Writing it twice would put
        a duplicate in everyone's scroll-back, so one delivery is one row and asking
        again is answered with the row that exists.

        Only a message the platform named can be recognised — an event with no
        `message_id` is always recorded, because there is nothing to match it on.
        Two deliveries racing this read both reach the insert, and the ledger's
        unique key refuses the loser loudly rather than letting it through.

        Recording is idempotent; answering is not. A caller that wakes a turn on
        what it records asks `find_message` first, because a row that already exists
        is a message somebody is already answering.

        Swaps `event.sender` for its registry row as a side effect, so the
        event's sender line resolves the owning identity via `event.sender.user`.

        Pass `happened_at` only to overrule the event's own clock: a transcript replay
        knows when the turn really happened, which a hook-built event cannot carry.
        Left out, the event's clock is used, and failing that the moment it arrived —
        the row is never undated, because an undated row has no place in the order.
        """
        thread = await self.ensure(
            ThreadKey(
                channel_tentacle_id=event.tentacle_id,
                chat_type=event.chat_type,
                chat_id=event.chat_id,
                channel_thread_id=event.channel_thread_id,
            )
        )
        platform_message_id = event.message_id or None
        if platform_message_id is not None:
            recorded = await self.find_message(
                thread.id, platform_message_id, "inbound"
            )
            if recorded is not None:
                return recorded
        if happened_at is None and event.timestamp > 0:
            happened_at = datetime.fromtimestamp(event.timestamp, UTC)
        sender = await self.users.ensure_profile(
            thread.channel_tentacle_id, event.sender
        )
        # Swap in the registry row (owner eagerly loaded) so the event's sender
        # line resolves `user:{username}` through `event.sender.user`.
        event.sender = sender
        message = ThreadMessage(
            thread_id=thread.id,
            platform_message_id=platform_message_id,
            reply_id=event.reply_id,
            happened_at=happened_at or datetime.now(UTC),
            direction="inbound",
            actor_kind=actor_kind,
            user_id=event.user_id,
            agent_tentacle_id=agent_tentacle_id,
            sender_id=sender.id,
            segments=event.segments,
            message_text=message_text_from_segments(event.segments),
            raw=event.raw,
        )
        await self.store_message(message, thread)
        return message

    async def record_outbound(
        self,
        thread_or_address: Thread | ChannelAddress | ThreadKey,
        *,
        agent_tentacle_id: str,
        segments: list[MessageSegment],
        platform_message_id: str | None = None,
        reply_id: str = "",
        happened_at: datetime | None = None,
        sender: UserProfile,
        actor_kind: ChannelActorKind = "agent",
        message_text: str | None = None,
        raw: str = "",
    ) -> ThreadMessage:
        """Write the agent's reply to the thread's ledger.
        Pass `happened_at` only when something knows better than this moment — a
        transcript replay does; an agent answering now does not.
        """
        if isinstance(thread_or_address, Thread):
            thread = thread_or_address
        else:
            thread = await self.ensure(thread_or_address)
        sender = await self.users.ensure_profile(thread.channel_tentacle_id, sender)
        message = ThreadMessage(
            thread_id=thread.id,
            platform_message_id=platform_message_id,
            reply_id=reply_id,
            happened_at=happened_at or datetime.now(UTC),
            direction="outbound",
            actor_kind=actor_kind,
            user_id="",
            agent_tentacle_id=agent_tentacle_id,
            sender_id=sender.id,
            segments=segments,
            message_text=message_text or message_text_from_segments(segments),
            raw=raw,
        )
        await self.store_message(message, thread)
        return message

    async def mark_presented(
        self, message_id: UUID7, platform_message_id: str | None
    ) -> None:
        """Name the platform message a ledger row was rendered as, once the
        channel that rendered it knows; a rendering with no id leaves it be."""
        if platform_message_id is None:
            return
        async with async_session() as session:
            stored = await session.one_or_none(
                ThreadMessage, expressions=[ThreadMessage["id"] == message_id]
            )
            if stored is None:
                raise ValueError(f"thread message {message_id} does not exist")
            stored.platform_message_id = platform_message_id
            await session.commit()

    async def redate_message(
        self, message: ThreadMessage, happened_at: datetime
    ) -> ThreadMessage:
        """Overrule a ledger row's clock after the fact: a transcript replay knows
        when a message really happened, while a row the live hooks wrote was stamped
        at receipt — a beat later than the transcript line, which is enough to sort a
        run's work above the prompt that caused it. Under the thread's lock, only the
        row's `happened_at` is written."""
        if message.happened_at == happened_at:
            return message
        async with async_session() as session:
            thread = await session.get(Thread, message.thread_id, options=[noload("*")])
            if thread is None:
                raise ValueError(f"unknown thread {message.thread_id}")
            async with self.lock(thread.key):
                stored = await session.get(
                    ThreadMessage, message.id, options=[noload("*")]
                )
                if stored is None:
                    raise ValueError(f"thread message {message.id} does not exist")
                stored.happened_at = happened_at
                await session.commit()
        message.happened_at = happened_at
        return message

    async def pending_prompt_messages(
        self,
        thread: Thread,
        trigger_message_id: UUID7,
        active_agent_id: str,
    ) -> list[ThreadMessage]:
        # A sub-thread's key is the chat room's, so this resolves the chat either
        # way — which is where both the ledger and the cursor are.
        fresh = await self.ensure(thread.key)

        async with async_session() as session:
            expressions = [
                in_ledger(fresh.id),
                ThreadMessage["id"] <= trigger_message_id,
                ThreadMessage["kind"] == "message",
                or_(
                    ThreadMessage["actor_kind"] != "agent",
                    ThreadMessage["agent_tentacle_id"] != active_agent_id,
                    ThreadMessage["agent_tentacle_id"].is_(None),
                ),
            ]
            if fresh.source_cursor_message_id is not None:
                expressions.append(ThreadMessage["id"] > fresh.source_cursor_message_id)
            rows = await session.list(
                ThreadMessage,
                limit=None,
                order_bys=[ThreadMessage["id"]],
                expressions=expressions,
            )
        return list(rows)

    async def advance_prompt_cursor(
        self,
        thread: Thread,
        message_id: UUID7,
    ) -> Thread | None:
        async with async_session() as session:
            stored = await session.get(Thread, thread.id)
            if stored is None:
                return None
            stored.source_cursor_message_id = message_id
            await session.commit()
        thread.source_cursor_message_id = message_id
        return thread

    async def record_fork(
        self,
        source: Conversation,
        target: Thread,
        *,
        sender: UserProfile,
        title: str | None,
    ) -> Thread:
        """Publish a user-attributed system notice after a successful fork. It names
        the source by `title`, the source's own name that the fork starts with, or by
        the source's id when it goes by nothing."""
        profile = await self.users.ensure_profile(target.channel_tentacle_id, sender)
        async with async_session() as session:
            thread = await session.get(
                Thread,
                target.id,
                options=[selectinload(Thread["conversations"]).noload("*")],
            )
            if thread is None:
                raise ValueError(f"unknown thread {target.id}")
            address = thread.key.address(profile.channel_user_id)
            text = (
                f'Forked from "{title}".\n\n'
                if title is not None
                else f"Forked from conversation {source.id}.\n\n"
            )
            if thread.project_id is None:
                text += (
                    "This fork has no server project. The source working directory "
                    "and its files were not transferred.\n\n"
                )
            text += f"Current channel address:\n{address}."
            notice = ThreadMessage(
                thread_id=thread.id,
                direction="inbound",
                actor_kind="system",
                user_id=profile.channel_user_id,
                sender_id=profile.id,
                segments=[TextSegment(data={"text": text})],
                message_text=text,
            )
            session.add(notice)
            previous_title = thread.title
            thread.title = title
            thread.updated_at = notice.happened_at
            await thread.handoffs
            await session.commit()
        events: list[ThreadEvent] = [
            ThreadMessageRecordedEvent(
                thread_id=thread.id,
                message_id=notice.id,
                updated_at=thread.updated_at,
            )
        ]
        if title != previous_title:
            events.append(
                ThreadTitleChangedEvent(
                    thread_id=thread.id, title=title, updated_at=thread.updated_at
                )
            )
        await self.publish(thread, *events)
        return thread

    async def record_handoff(
        self,
        thread_or_address: Thread | ChannelAddress | ThreadKey,
        *,
        to_agent_tentacle_id: str,
        source_agent_tentacle_id: str | None = None,
        to_model: AgentRouteModelName | None = None,
        reason: str = "",
        hint: str = "",
        brief: str = "",
        source_conversation_id: UUID7 | None = None,
        target_conversation_id: UUID7 | None = None,
        source_run_id: str | None = None,
        source_model_message_id: UUID7 | None = None,
    ) -> Handoff:
        if isinstance(thread_or_address, Thread):
            thread = thread_or_address
        else:
            thread = await self.ensure(thread_or_address)
        handoff = Handoff(
            thread_id=thread.id,
            source_agent_tentacle_id=source_agent_tentacle_id,
            to_agent_tentacle_id=to_agent_tentacle_id,
            to_model=to_model,
            reason=reason,
            hint=hint,
            brief=brief,
            source_conversation_id=source_conversation_id,
            target_conversation_id=target_conversation_id,
            source_run_id=source_run_id,
            source_model_message_id=source_model_message_id,
        )
        async with async_session() as session:
            session.add(handoff)
            await session.commit()
        thread.handoffs.append(handoff)
        await self.publish(
            thread,
            ThreadHandoffRecordedEvent(thread_id=thread.id, handoff=handoff),
        )
        return handoff

    async def chat_message(self, profile: UserProfile, handle: str) -> ThreadMessage:
        """The chat message `handle` names within this person's history — the threads
        they have spoken in, on this account or any the registry links to it. A row
        id, or the `#msg:<id>` handle shown beside it: the platform's own id, which
        is what a brief written from the other side of a handoff has to cite."""
        if handle.startswith("#msg:"):
            named = ThreadMessage["platform_message_id"] == handle.removeprefix("#msg:")
        else:
            named = ThreadMessage["id"] == uuid.UUID(handle)
        senders = [
            profile.id,
            *(linked.id for linked in await self.users.linked_profiles(profile)),
        ]
        async with async_session() as session:
            rows = await session.list(
                ThreadMessage,
                limit=2,
                expressions=[named, in_history_of(senders)],
            )
        if not rows:
            raise ValueError(f"no message {handle} in this person's history")
        if len(rows) > 1:
            raise ValueError(f"{handle} names more than one message; use a row id")
        return rows[0]

    async def find_message(
        self,
        thread_id: UUID7,
        platform_message_id: str,
        direction: ThreadMessageDirection,
    ) -> ThreadMessage | None:
        """The row delivered to the thread for one platform message — a turn's
        prompt, or the reply that answered it. Oldest wins, as scanning the ordered
        ledger did. A row the thread shows from a ledger a fork froze into it was
        delivered elsewhere, so it is never this one.

        Both halves of the key are indexed. The hooks and tailers ask this per turn,
        on a ledger with no ceiling, so the row has to be found by the index rather
        than by reading the room's whole tenure to match one id in Python.
        """
        async with async_session() as session:
            rows = await session.list(
                ThreadMessage,
                limit=1,
                order_bys=[ThreadMessage["happened_at"], ThreadMessage["id"]],
                # The model ledger hangs off the row and no caller here reads it;
                # `related_model_messages` is how someone asks.
                options=[noload(ThreadMessage["model_messages"])],
                expressions=[
                    ThreadMessage["thread_id"] == thread_id,
                    ThreadMessage["platform_message_id"] == platform_message_id,
                    ThreadMessage["direction"] == direction,
                ],
            )
        return rows[0] if rows else None

    async def record_native_prompt(
        self,
        address: ChannelAddress,
        *,
        message_id: str,
        text: str,
        sender: UserProfile,
        timestamp: datetime,
    ) -> ThreadMessage:
        """Record a transcript input once, before the run commits.

        The native message identity survives reconnects. The caller holds its
        session lock across this lookup and write; the row grants the stream's
        owner access to a new thread while its model history is still folding.
        """
        thread = await self.ensure(ThreadKey.from_address(address))
        inbound = await self.find_message(thread.id, message_id, "inbound")
        if inbound is not None:
            if inbound.happened_at != timestamp:
                await self.redate_message(inbound, timestamp)
            return inbound
        return await self.record_inbound(
            MessageEvent(
                tentacle_id=address.channel_tentacle_id,
                message_id=message_id,
                chat_id=address.chat_id,
                chat_type=address.chat_type,
                user_id=sender.channel_user_id,
                sender=sender,
                segments=[TextSegment(data={"text": text})],
            ),
            happened_at=timestamp,
        )

    async def bind_native_run(
        self,
        address: ChannelAddress,
        run: ExternalAgentRun,
        *,
        prompts: dict[UUID7, UUID7],
        answer: str,
        sender: UserProfile,
    ) -> UUID7 | None:
        """Bind transcript prompts and the final reply after the run commits.

        Each prompt maps its ledger id to its committed model message id. The caller holds the session lock and publishes `run_ended`
        only after this returns, so a reader can immediately reload the ledger.
        """
        for prompt_id, message_id in prompts.items():
            await self.bind_messages(
                [prompt_id],
                message_id,
                kind="request_source",
                run_id=run.id,
            )
        if not answer:
            return None
        thread = await self.ensure(ThreadKey.from_address(address))
        outbound = await self.find_message(thread.id, run.id, "outbound")
        answered = next(
            (
                message
                for message in reversed(run.messages)
                if isinstance(message, ModelResponse)
            ),
            None,
        )
        if outbound is None:
            outbound = await self.record_outbound(
                thread,
                agent_tentacle_id=address.channel_tentacle_id,
                segments=[MarkdownSegment(data={"text": answer})],
                sender=sender,
                platform_message_id=run.id,
                happened_at=answered.timestamp if answered is not None else None,
            )
        elif (
            answered is not None
            and answered.timestamp is not None
            and outbound.happened_at != answered.timestamp
        ):
            await self.redate_message(outbound, answered.timestamp)
        await self.bind_assistant_replies([outbound.id], run_id=run.id)
        return outbound.id

    async def bind_messages(
        self,
        thread_message_ids: list[UUID7],
        model_message_id: UUID7,
        *,
        kind: MessageBindingKind,
        run_id: str,
        tool_call_id: str | None = None,
    ) -> list[MessageBinding]:
        bindings: list[MessageBinding] = []
        async with async_session() as session:
            for position, thread_message_id in enumerate(thread_message_ids):
                binding = MessageBinding(
                    thread_message_id=thread_message_id,
                    model_message_id=model_message_id,
                    kind=kind,
                    run_id=run_id,
                    tool_call_id=tool_call_id,
                    position=position,
                )
                session.add(binding)
                bindings.append(binding)
            await session.commit()
        return bindings

    async def bind_assistant_replies(
        self,
        thread_message_ids: list[UUID7],
        *,
        run_id: str,
    ) -> list[MessageBinding]:
        if not thread_message_ids:
            return []
        async with async_session() as session:
            responses = await session.list(
                ModelResponse,
                limit=None,
                order_bys=[ModelResponse["id"]],
                expressions=[
                    ModelResponse["run_id"] == run_id,
                    ModelResponse["role"] == "assistant",
                    ModelResponse["message_text"].is_not(None),
                    ModelResponse["message_text"] != "",
                ],
            )
        assistant_replies = list(responses)
        if not assistant_replies:
            return []
        if len(assistant_replies) == len(thread_message_ids):
            bindings: list[MessageBinding] = []
            async with async_session() as session:
                for thread_message_id, response in zip(
                    thread_message_ids, assistant_replies, strict=True
                ):
                    binding = MessageBinding(
                        thread_message_id=thread_message_id,
                        model_message_id=response.id,
                        kind="assistant_reply",
                        run_id=run_id,
                        position=0,
                    )
                    session.add(binding)
                    bindings.append(binding)
                await session.commit()
            return bindings
        return await self.bind_messages(
            thread_message_ids,
            assistant_replies[-1].id,
            kind="assistant_reply",
            run_id=run_id,
        )

    async def search_chat_messages(
        self,
        profile: UserProfile,
        query: str,
        *,
        actor_kind: ChannelActorKind | None = None,
        limit: int = 10,
    ) -> list[ThreadMessage]:
        """The messages matching every English term in `query` across this person's
        history — the threads they have spoken in, on this account or any the registry
        links to it. Best lexical match first, newest first when scores tie. A
        visitor's history is one account's."""
        match_query = history_match_query(query)
        senders = [
            profile.id,
            *(linked.id for linked in await self.users.linked_profiles(profile)),
        ]
        expressions = [
            ThreadMessageFTS["message_text"].match(match_query),
            in_history_of(senders),
        ]
        if actor_kind is not None:
            expressions.append(ThreadMessage["actor_kind"] == actor_kind)
        async with async_session() as session:
            statement = (
                select(ThreadMessage)
                .join(
                    ThreadMessageFTS,
                    ThreadMessage["id"] == ThreadMessageFTS["message_id"],
                )
                .where(*expressions)
                .options(noload(ThreadMessage["model_messages"]))
                .order_by(
                    ThreadMessageFTS["rank"],
                    ThreadMessage["happened_at"].desc(),
                    ThreadMessage["id"].desc(),
                )
                .limit(limit)
            )
            rows = (await session.execute(statement)).scalars().all()
        return list(rows)

    async def chat_messages_before(
        self,
        thread_id: UUID7,
        anchor_id: UUID7,
        *,
        limit: int = 5,
    ) -> list[ThreadMessage]:
        """The rows standing before `anchor_id` in the thread, oldest last of them
        first. Neighbours in the conversation, which is what the ledger's order means —
        so the anchor is compared on the same key the rows are sorted by."""
        async with async_session() as session:
            anchor = await session.one_or_none(
                ThreadMessage, expressions=[ThreadMessage["id"] == anchor_id]
            )
            if anchor is None:
                raise ValueError(f"thread message {anchor_id} does not exist")
            rows = await session.list(
                ThreadMessage,
                limit=limit,
                order_bys=[
                    ThreadMessage["happened_at"].desc(),
                    ThreadMessage["id"].desc(),
                ],
                expressions=[
                    in_ledger(thread_id),
                    or_(
                        ThreadMessage["happened_at"] < anchor.happened_at,
                        and_(
                            ThreadMessage["happened_at"] == anchor.happened_at,
                            ThreadMessage["id"] < anchor.id,
                        ),
                    ),
                ],
            )
        return list(reversed(rows))

    async def chat_messages_after(
        self,
        thread_id: UUID7,
        anchor_id: UUID7,
        *,
        limit: int = 5,
    ) -> list[ThreadMessage]:
        """The rows standing after `anchor_id` in the thread, oldest first — the mirror
        of `chat_messages_before`, and anchored on the same key."""
        async with async_session() as session:
            anchor = await session.one_or_none(
                ThreadMessage, expressions=[ThreadMessage["id"] == anchor_id]
            )
            if anchor is None:
                raise ValueError(f"thread message {anchor_id} does not exist")
            rows = await session.list(
                ThreadMessage,
                limit=limit,
                order_bys=[ThreadMessage["happened_at"], ThreadMessage["id"]],
                expressions=[
                    in_ledger(thread_id),
                    or_(
                        ThreadMessage["happened_at"] > anchor.happened_at,
                        and_(
                            ThreadMessage["happened_at"] == anchor.happened_at,
                            ThreadMessage["id"] > anchor.id,
                        ),
                    ),
                ],
            )
        return list(rows)

    async def messages_page(
        self,
        thread_id: UUID7,
        cursor: str | None = None,
        *,
        unbound: bool = False,
        limit: int,
    ) -> Page[ThreadMessage]:
        """The rows the thread shows — its own ledger's, and a forked parent's up to
        the cut — a page at a time, newest first, continued from an Arcanus
        `cursor`. Ids are uuid7 and a ledger is appended to, so id order is the
        order rows were written in.

        `unbound` keeps to the rows no model message of the thread's own
        conversations was built from or answered with: a fork or teleport notice,
        or a chat room's talk, whose work runs in sub-threads. Everything else is
        read through those model messages, which carry the rows bound to them.
        """
        scope: list[Expression[bool] | ColumnElement[bool]]
        scope = [in_ledger(thread_id)]
        if unbound:
            scope.append(
                ~ThreadMessage["model_messages"].any(
                    ModelMessage["run_id"].in_(own_runs(thread_id))
                )
            )
        bookmark = None if cursor is None else Cursor[ThreadMessage](cursor).bookmark
        after = (
            []
            if bookmark is None or bookmark.expression is None
            else [bookmark.expression]
        )
        newest_first = (ThreadMessage["id"].desc(),)
        async with async_session() as session:
            rows = await session.list(
                ThreadMessage,
                limit=limit + 1,
                order_bys=newest_first,
                options=[noload(ThreadMessage["model_messages"])],
                expressions=[*scope, *after],
            )
            total = await session.count(ThreadMessage, expressions=scope)
        items = tuple(rows[:limit])
        last = ThreadMessage["id"] < items[-1].id if items else None
        return Page(
            items=items,
            total=total,
            next_cursor=str(
                Cursor[ThreadMessage].from_expressions(
                    bookmark=last, order_bys=newest_first, limit=limit
                )
            ),
            has_more=len(rows) > limit,
        )

    async def related_model_messages(
        self,
        thread_message_id: UUID7,
    ) -> list[ModelRequest | ModelResponse]:
        async with async_session() as session:
            message = await session.get(
                ThreadMessage,
                thread_message_id,
                options=[selectinload(ThreadMessage["model_messages"])],
            )
            if message is None:
                return []
            await message.model_messages
        return list(message.model_messages)

users instance-attribute

users: UserManager = users

nerve instance-attribute

nerve: Nerve = nerve

ensure async

ensure(
    address_or_key: ChannelAddress | ThreadKey,
    *,
    project: Project | None = None,
    title: str | None = None,
) -> Thread

The thread this key names, created if it is new.

A key names a surface, so it resolves the thread nobody opened. A chat room's sub-threads share its address and are reached by id, through it.

A channel-supplied title refreshes the surface's stored name, including existing groups and DMs previously named from an opening line.

project attributes a thread being created and is ignored for one that already exists: a thread's project is frozen once the row is written, so declaring a project later attributes new threads rather than rewriting old ones. It is not part of the key either — the key is what the thread's history is filed under, and attribution must never move it.

Only a thread and a native thread can take one; naming a project for a DM or a group chat raises rather than dropping it silently.

Source code in octomate/managers/thread.py

async def ensure(
    self,
    address_or_key: ChannelAddress | ThreadKey,
    *,
    project: Project | None = None,
    title: str | None = None,
) -> Thread:
    """The thread this key names, created if it is new.

    A key names a surface, so it resolves the thread nobody opened. A chat
    room's sub-threads share its address and are reached by id, through it.

    A channel-supplied `title` refreshes the surface's stored name, including
    existing groups and DMs previously named from an opening line.

    `project` attributes a thread being created and is ignored for one that
    already exists: a thread's project is frozen once the row is written, so
    declaring a project later attributes new threads rather than rewriting old
    ones. It is not part of the key either — the key is what the thread's history
    is filed under, and attribution must never move it.

    Only a thread and a native thread can take one; naming a project for a DM or a
    group chat raises rather than dropping it silently.
    """
    if isinstance(address_or_key, ChannelAddress):
        key = ThreadKey.from_address(address_or_key)
    else:
        key = address_or_key
    if project is not None and key.kind not in ATTRIBUTABLE_KINDS:
        raise ValueError(
            f"{key} is a {key.kind} and cannot be attributed to project "
            f"{project.name!r}: only a thread or a native_thread is work."
        )
    # The lock spans the read, the insert and the commit: a loser that woke
    # after the winner's read but before its commit would still find nothing.
    async with self.lock(key), async_session() as session:
        thread = await session.one_or_none(
            Thread,
            options=[selectinload(Thread["conversations"]).noload("*")],
            expressions=keyed(key),
        )
        if thread is None:
            thread = Thread(
                channel_tentacle_id=key.channel_tentacle_id,
                chat_type=key.chat_type,
                chat_id=key.chat_id,
                channel_thread_id=key.channel_thread_id,
                kind=key.kind,
                project_id=project.id if project is not None else None,
            )
            session.add(thread)
        title_changed = title is not None and thread.title != title
        if title_changed:
            thread.title = title
            thread.updated_at = datetime.now(UTC)
        await session.flush()
        await thread.handoffs
        await thread.project
        await session.commit()
    if title_changed:
        await self.publish(
            thread,
            ThreadTitleChangedEvent(
                thread_id=thread.id,
                title=thread.title,
                updated_at=thread.updated_at,
            ),
        )
    return thread

open_sub_thread async

open_sub_thread(parent: Thread) -> Thread

The thread one kick in parent does its work in.

A dm and a group chat have no end, so a conversation held against one would accumulate for as long as the chat room does. Answering happens in a thread of its own instead, and everything a thread already governs governs it — one conversation per agent, runs continuing one by one. Only the chat everyone sees stays on the chat room, which is what surface resolves back to.

Opened, never looked up: a sub-thread has no address of its own to be found by, and each kick wants its own. A redelivered event therefore opens a second — that is a duplicate to stop where it is made, in the ledger, not here.

Source code in octomate/managers/thread.py

async def open_sub_thread(self, parent: Thread) -> Thread:
    """The thread one kick in `parent` does its work in.

    A dm and a group chat have no end, so a conversation held against one would
    accumulate for as long as the chat room does. Answering happens in a thread
    of its own instead, and everything a thread already governs governs it — one
    conversation per agent, runs continuing one by one. Only the chat everyone
    sees stays on the chat room, which is what `surface` resolves back to.

    Opened, never looked up: a sub-thread has no address of its own to be found
    by, and each kick wants its own. A redelivered event therefore opens a
    second — that is a duplicate to stop where it is made, in the ledger, not
    here.
    """
    if parent.parent_thread_id is not None:
        raise ValueError(
            f"thread {parent.id} is already a sub-thread of "
            f"{parent.parent_thread_id}: a chat room nests one level"
        )
    key = parent.key
    thread = Thread(
        channel_tentacle_id=key.channel_tentacle_id,
        chat_type=key.chat_type,
        chat_id=key.chat_id,
        # The chat room's own, verbatim: the platform never made this thread, so
        # there is no id of its own to carry, and anything invented here would be
        # sent back to the platform as a real one. Sharing the address is what
        # makes a sub-thread's key resolve to the chat room it was opened in.
        channel_thread_id=key.channel_thread_id,
        kind=key.kind,
        parent_thread_id=parent.id,
    )
    async with async_session() as session:
        session.add(thread)
        await session.flush()
        await thread.handoffs
        await thread.project
        await session.commit()
    return thread

enter async

enter(
    address_or_key: ChannelAddress | ThreadKey,
    *,
    current: Thread | None = None,
) -> Thread

The thread a turn arriving at this surface does its work in.

A thread and a native thread are a piece of work already, so the turn runs in the one the key names. A dm and a group chat are not — they have no end, and a conversation held against one would accumulate for as long as the chat room does — so the turn runs in a sub-thread opened inside the chat room instead.

current is the thread the turn is already in, and it is kept when this surface is the chat room that thread was opened in: an agent taking the conversation over in place is still answering the same kick, and a second sub-thread would split one answer across two model contexts.

Source code in octomate/managers/thread.py

async def enter(
    self,
    address_or_key: ChannelAddress | ThreadKey,
    *,
    current: Thread | None = None,
) -> Thread:
    """The thread a turn arriving at this surface does its work in.

    A thread and a native thread are a piece of work already, so the turn runs in
    the one the key names. A dm and a group chat are not — they have no end, and a
    conversation held against one would accumulate for as long as the chat room
    does — so the turn runs in a sub-thread opened inside the chat room instead.

    `current` is the thread the turn is already in, and it is kept when this
    surface is the chat room that thread was opened in: an agent taking the
    conversation over in place is still answering the same kick, and a second
    sub-thread would split one answer across two model contexts.
    """
    surface = await self.ensure(address_or_key)
    if current is not None and current.parent_thread_id == surface.id:
        return current
    if surface.kind in ATTRIBUTABLE_KINDS:
        return surface
    return await self.open_sub_thread(surface)

surface async

surface(thread: Thread) -> Thread

The surface this thread is on: itself, or the chat room it was opened in.

A sub-thread has none of its own — no address, no chat ledger, no prompt cursor — so this is the one place the split between where a run happens and where it is seen gets made. Nesting is one level, so it never walks.

Source code in octomate/managers/thread.py

async def surface(self, thread: Thread) -> Thread:
    """The surface this thread is on: itself, or the chat room it was opened in.

    A sub-thread has none of its own — no address, no chat ledger, no prompt
    cursor — so this is the one place the split between where a run happens and
    where it is seen gets made. Nesting is one level, so it never walks.
    """
    if thread.parent_thread_id is None:
        return thread
    parent = await self.get(thread.parent_thread_id, with_messages=False)
    if parent is None:
        raise ValueError(
            f"thread {thread.id} names parent {thread.parent_thread_id}, "
            "which does not exist"
        )
    return parent

fork_title staticmethod

fork_title(
    source: Conversation, thread: Thread
) -> str | None

What a fork of source, from thread, is first called: what the source goes by, the name its runtime gave the session or else the thread's title. None when it goes by nothing, and the fork takes a name the way a new thread does.

Source code in octomate/managers/thread.py

@staticmethod
def fork_title(source: Conversation, thread: Thread) -> str | None:
    """What a fork of `source`, from `thread`, is first called: what the source
    goes by, the name its runtime gave the session or else the thread's title.
    None when it goes by nothing, and the fork takes a name the way a new thread
    does."""
    return thread_title(source.name or thread.title)

rename async

rename(thread: Thread, title: str | None) -> Thread

Give the thread the name the runtime running it grabbed for itself.

Unlike the opening line store_message falls back to, this is a name for the work rather than for how it was asked for, so it overwrites — and goes on overwriting, because the runtime revises it as the session goes on. A name with nothing in it is not one, and leaves the thread as it was. A group's main chat keeps its platform name, and a DM keeps the bot's name; only their work sub-threads take generated titles.

Source code in octomate/managers/thread.py

async def rename(self, thread: Thread, title: str | None) -> Thread:
    """Give the thread the name the runtime running it grabbed for itself.

    Unlike the opening line `store_message` falls back to, this is a name for
    the work rather than for how it was asked for, so it overwrites — and goes
    on overwriting, because the runtime revises it as the session goes on. A
    name with nothing in it is not one, and leaves the thread as it was.
    A group's main chat keeps its platform name, and a DM keeps the bot's
    name; only their work sub-threads take generated titles.
    """
    if thread.kind in {"group", "dm"} and thread.parent_thread_id is None:
        return thread
    named = thread_title(title)
    if named is None or thread.title == named:
        return thread
    async with async_session() as session:
        stored = await session.get(Thread, thread.id)
        if stored is None:
            raise ValueError(f"unknown thread {thread.id}")
        stored.title = named
        stored.updated_at = datetime.now(UTC)
        await session.commit()
    thread.title = named
    await self.publish(
        thread,
        ThreadTitleChangedEvent(
            thread_id=thread.id,
            title=named,
            updated_at=stored.updated_at,
        ),
    )
    return thread

bind async

bind(thread_id: UUID7, project: Project) -> Thread

Attribute a thread that is in no project to project, once.

The one exception to a project being written when the row is created: a chat thread exists before anyone has said what it is about, so the attribution has to be able to arrive later. It arrives once. A second bind is refused rather than switching — the first project's workspace holds work nobody has reviewed yet, and a thread whose project moved is a record of what it did somewhere it no longer says it was. A different project is a different thread.

A DM and a group chat are refused for the reason ATTRIBUTABLE_KINDS gives: they outlive every project in them. Both refusals are BindRefusal; a thread that does not exist is a plain ValueError, a wiring bug.

No workspace is made here. A run's working directory is fixed when its process spawns, so this takes effect on the thread's next turn and the caller has to say so.

Source code in octomate/managers/thread.py

async def bind(self, thread_id: UUID7, project: Project) -> Thread:
    """Attribute a thread that is in no project to `project`, once.

    The one exception to a project being written when the row is created: a
    chat thread exists before anyone has said what it is about, so the
    attribution has to be able to arrive later. It arrives once. A second bind
    is refused rather than switching — the first project's workspace holds work
    nobody has reviewed yet, and a thread whose project moved is a record of
    what it did somewhere it no longer says it was. A different project is a
    different thread.

    A DM and a group chat are refused for the reason `ATTRIBUTABLE_KINDS`
    gives: they outlive every project in them. Both refusals are `BindRefusal`;
    a thread that does not exist is a plain `ValueError`, a wiring bug.

    No workspace is made here. A run's working directory is fixed when its
    process spawns, so this takes effect on the thread's next turn and the
    caller has to say so.
    """
    async with async_session() as session:
        thread = await session.get(Thread, thread_id)
        if thread is None:
            raise ValueError(f"no thread {thread_id} to bind")
        if thread.kind not in ATTRIBUTABLE_KINDS:
            raise BindRefusal(
                f"{thread.key} is a {thread.kind} and cannot be attributed to "
                f"project {project.name!r}: only a thread or a native_thread "
                f"is work."
            )
        current = await thread.project
        if current is not None:
            raise BindRefusal(
                f"{thread.key} is already about {current.name!r} and a thread "
                f"binds once; work on {project.name!r} in its own thread."
            )
        thread.project_id = project.id
        await session.commit()

    # Read the thread back rather than returning the row that changed. `project`
    # is eagerly loaded, so the copy that set the id is still holding the answer
    # it was loaded with.
    bound = await self.get(thread_id)
    if bound is None:
        raise ValueError(f"thread {thread_id} vanished while binding")
    await self.publish(
        bound,
        ThreadProjectBoundEvent(
            thread_id=bound.id,
            project_id=project.id,
            updated_at=bound.updated_at,
        ),
    )
    return bound

get async

get(
    thread_id: UUID7 | ThreadKey,
    *,
    with_messages: bool = True,
    user_id: UUID7 | None = None,
) -> Thread | None

The thread by primary key, or by the key it is filed under, or None — its handoffs with the row, its ledger only when asked for. A key never creates one, unlike ensure.

with_messages=True is the whole ledger, deliberately: it is what a reader rebuilding a thread wants, and it is unbounded in a room, so a caller after one row wants find_message and a caller after the recent ones wants chat_messages_before.

Either way the model ledger stays behind: it hangs off a message, and related_model_messages is how a caller asks for it — dragging it here would put a query per message behind every thread read.

user_id restricts access to participation in the thread's own ledger; messages inherited from another thread grant no access.

Source code in octomate/managers/thread.py

async def get(
    self,
    thread_id: UUID7 | ThreadKey,
    *,
    with_messages: bool = True,
    user_id: UUID7 | None = None,
) -> Thread | None:
    """The thread by primary key, or by the key it is filed under, or None —
    its handoffs with the row, its ledger only when asked for. A key never
    creates one, unlike `ensure`.

    `with_messages=True` is the whole ledger, deliberately: it is what a reader
    rebuilding a thread wants, and it is unbounded in a room, so a caller after
    one row wants `find_message` and a caller after the recent ones wants
    `chat_messages_before`.

    Either way the model ledger stays behind: it hangs off a message, and
    `related_model_messages` is how a caller asks for it — dragging it here
    would put a query per message behind every thread read.

    `user_id` restricts access to participation in the thread's own ledger;
    messages inherited from another thread grant no access.
    """
    options = (
        [selectinload(Thread["messages"]).noload(ThreadMessage["model_messages"])]
        if with_messages
        else []
    )
    options.append(selectinload(Thread["conversations"]).noload("*"))
    async with async_session() as session:
        expressions = (
            keyed(thread_id)
            if isinstance(thread_id, ThreadKey)
            else [Thread["id"] == thread_id]
        )
        if user_id is not None:
            expressions.append(
                Thread["messages"].any(
                    (ThreadMessage["thread_id"] == Thread["id"])
                    & ThreadMessage["sender"].has(UserProfile["user_id"] == user_id)
                )
            )
        thread = await session.one_or_none(
            Thread, expressions=expressions, options=options
        )
        if thread is None:
            return None
    return thread

list_threads async

list_threads(
    channel_tentacle_id: str | None = None,
    *,
    limit: int = 100,
    user_id: UUID7 | None = None,
) -> list[Thread]

Threads most recently touched first — one channel's, or every channel's when channel_tentacle_id is None. Sub-threads are not listed; the threads a chat room's kicks work in are reached through the chat room.

Handoffs come with the rows (lazy="selectin", one batched pass); the ledgers do not. A listing that loaded them would read every message of every thread it names, and no caller has ever wanted that — the one reader of a thread's messages asks for that thread.

Source code in octomate/managers/thread.py

async def list_threads(
    self,
    channel_tentacle_id: str | None = None,
    *,
    limit: int = 100,
    user_id: UUID7 | None = None,
) -> list[Thread]:
    """Threads most recently touched first — one channel's, or every
    channel's when `channel_tentacle_id` is None. Sub-threads are not listed;
    the threads a chat room's kicks work in are reached through the chat room.

    Handoffs come with the rows (lazy="selectin", one batched pass); the
    ledgers do not. A listing that loaded them would read every message of
    every thread it names, and no caller has ever wanted that — the one
    reader of a thread's messages asks for that thread.
    """
    # Sub-threads are left out: a listing names the surfaces a person can open,
    # and the threads a chat room's kicks work in are reached through it.
    expressions = [Thread["parent_thread_id"].is_(None)]
    if user_id is not None:
        expressions.append(
            Thread["messages"].any(
                (ThreadMessage["thread_id"] == Thread["id"])
                & ThreadMessage["sender"].has(UserProfile["user_id"] == user_id)
            )
        )
    if channel_tentacle_id is not None:
        expressions.append(Thread["channel_tentacle_id"] == channel_tentacle_id)
    async with async_session() as session:
        rows = await session.list(
            Thread,
            limit=limit,
            options=[selectinload(Thread["conversations"]).noload("*")],
            order_bys=[Thread["updated_at"].desc(), Thread["id"].desc()],
            expressions=expressions,
        )
    return list(rows)

publish async

publish(thread: Thread, *events: ThreadEvent) -> None

Publish committed thread events to each eligible user's topic.

Resolve registered participants once for the supplied events, without a cache, matching listing and access checks. Inherited history grants no access. Sub-threads are not listed. Token events never call this method.

Source code in octomate/managers/thread.py

async def publish(self, thread: Thread, *events: ThreadEvent) -> None:
    """Publish committed thread events to each eligible user's topic.

    Resolve registered participants once for the supplied events, without a cache,
    matching listing and access checks. Inherited history grants no access.
    Sub-threads are not listed. Token events never call this method.
    """
    if thread.parent_thread_id is not None:
        return
    async with async_session() as session:
        user_ids = await session.scalars(
            select(UserProfile["user_id"])
            .where(
                UserProfile["user_id"].is_not(None),
                UserProfile["id"].in_(
                    select(ThreadMessage["sender_id"]).where(
                        ThreadMessage["thread_id"] == thread.id
                    )
                ),
            )
            .distinct()
        )
    impulses = [Impulse(event) for event in events]
    for user_id in user_ids:
        for impulse in impulses:
            self.nerve.deliver(UserTopic(user_id), impulse)

store_message async

store_message(
    message: ThreadMessage, thread: Thread
) -> None

Persist a ledger row.

The row is touched, not only appended to. updated_at fires on an update of the thread and writing to its ledger is not one, so a session being tailed right now would keep the timestamp it was created with and sink, in a listing ordered by it, under threads nothing has ever said a word in.

A thread nobody has named takes its name from this row when a person wrote it. The listing carries no messages, so the first thing said is the only name a row has until a runtime grabs one of its own (rename).

Source code in octomate/managers/thread.py

async def store_message(self, message: ThreadMessage, thread: Thread) -> None:
    """Persist a ledger row.

    The row is touched, not only appended to. `updated_at` fires on an update
    of the thread and writing to its ledger is not one, so a session being
    tailed right now would keep the timestamp it was created with and sink, in
    a listing ordered by it, under threads nothing has ever said a word in.

    A thread nobody has named takes its name from this row when a person wrote
    it. The listing carries no messages, so the first thing said is the only
    name a row has until a runtime grabs one of its own (`rename`).
    """
    async with async_session() as session:
        session.add(message)
        row = await session.get(Thread, thread.id)
        if row is None:
            raise ValueError(f"unknown thread {thread.id}")
        previous_title = row.title
        row.updated_at = datetime.now(UTC)
        if (
            row.title is None
            and (
                row.kind not in {"group", "dm"} or row.parent_thread_id is not None
            )
            and message.direction == "inbound"
            and message.actor_kind == "human"
        ):
            row.title = thread_title(message.message_text)
        await session.commit()
    events: list[ThreadEvent] = [
        ThreadMessageRecordedEvent(
            thread_id=thread.id,
            message_id=message.id,
            updated_at=row.updated_at,
        )
    ]
    if row.title != previous_title:
        events.append(
            ThreadTitleChangedEvent(
                thread_id=thread.id, title=row.title, updated_at=row.updated_at
            )
        )
    await self.publish(thread, *events)

record_artifact async

record_artifact(
    thread_id: UUID7,
    artifact: Artifact,
    *,
    sender: UserProfile,
    agent_tentacle_id: str,
) -> ThreadArtifact

Commit an artifact and its ledger reference together before publication.

Source code in octomate/managers/thread.py

async def record_artifact(
    self,
    thread_id: UUID7,
    artifact: Artifact,
    *,
    sender: UserProfile,
    agent_tentacle_id: str,
) -> ThreadArtifact:
    """Commit an artifact and its ledger reference together before publication."""
    thread = await self.get(thread_id, with_messages=False)
    if thread is None:
        raise ValueError(f"unknown thread {thread_id}")
    sender = await self.users.ensure_profile(thread.channel_tentacle_id, sender)
    message = ThreadArtifact(
        thread_id=thread.id,
        direction="outbound",
        actor_kind="agent",
        sender_id=sender.id,
        agent_tentacle_id=agent_tentacle_id,
        artifact_id=artifact.id,
        artifact=Relation(artifact),
    )
    await self.store_message(message, thread)
    return message

record_command_outcome async

record_command_outcome(
    receipt_id: UUID, outcome: CommandOutcome
) -> None

Persist an outcome without mutating a caller's receipt snapshot. Under the thread's lock, only the outcome and the thread's updated_at are written.

Source code in octomate/managers/thread.py

async def record_command_outcome(
    self, receipt_id: uuid.UUID, outcome: CommandOutcome
) -> None:
    """Persist an outcome without mutating a caller's receipt snapshot. Under the
    thread's lock, only the outcome and the thread's `updated_at` are written."""
    async with async_session() as session:
        receipt = await session.get(
            ThreadCommand, receipt_id, options=[noload("*")]
        )
        if receipt is None:
            raise ValueError(f"command receipt {receipt_id} does not exist")
        thread = await session.get(Thread, receipt.thread_id, options=[noload("*")])
        if thread is None:
            raise ValueError(f"unknown thread {receipt.thread_id}")
        async with self.lock(thread.key):
            receipt.outcome = outcome
            thread.updated_at = datetime.now(UTC)
            await session.commit()
    await self.publish(
        thread,
        ThreadCommandOutcomeRecordedEvent(
            thread_id=thread.id,
            receipt_id=receipt.id,
            outcome=outcome,
            updated_at=thread.updated_at,
        ),
    )

record_inbound async

record_inbound(
    event: MessageEvent,
    *,
    actor_kind: ChannelActorKind = "human",
    agent_tentacle_id: str | None = None,
    happened_at: datetime | None = None,
) -> ThreadMessage

Write the event to the thread's ledger, or hand back the row it is already on.

A platform re-sends a message when it misses an ack — Slack does — and the second send is the same message, not a second one. Writing it twice would put a duplicate in everyone's scroll-back, so one delivery is one row and asking again is answered with the row that exists.

Only a message the platform named can be recognised — an event with no message_id is always recorded, because there is nothing to match it on. Two deliveries racing this read both reach the insert, and the ledger's unique key refuses the loser loudly rather than letting it through.

Recording is idempotent; answering is not. A caller that wakes a turn on what it records asks find_message first, because a row that already exists is a message somebody is already answering.

Swaps event.sender for its registry row as a side effect, so the event's sender line resolves the owning identity via event.sender.user.

Pass happened_at only to overrule the event's own clock: a transcript replay knows when the turn really happened, which a hook-built event cannot carry. Left out, the event's clock is used, and failing that the moment it arrived — the row is never undated, because an undated row has no place in the order.

Source code in octomate/managers/thread.py

async def record_inbound(
    self,
    event: MessageEvent,
    *,
    actor_kind: ChannelActorKind = "human",
    agent_tentacle_id: str | None = None,
    happened_at: datetime | None = None,
) -> ThreadMessage:
    """Write the event to the thread's ledger, or hand back the row it is
    already on.

    A platform re-sends a message when it misses an ack — Slack does — and the
    second send is the same message, not a second one. Writing it twice would put
    a duplicate in everyone's scroll-back, so one delivery is one row and asking
    again is answered with the row that exists.

    Only a message the platform named can be recognised — an event with no
    `message_id` is always recorded, because there is nothing to match it on.
    Two deliveries racing this read both reach the insert, and the ledger's
    unique key refuses the loser loudly rather than letting it through.

    Recording is idempotent; answering is not. A caller that wakes a turn on
    what it records asks `find_message` first, because a row that already exists
    is a message somebody is already answering.

    Swaps `event.sender` for its registry row as a side effect, so the
    event's sender line resolves the owning identity via `event.sender.user`.

    Pass `happened_at` only to overrule the event's own clock: a transcript replay
    knows when the turn really happened, which a hook-built event cannot carry.
    Left out, the event's clock is used, and failing that the moment it arrived —
    the row is never undated, because an undated row has no place in the order.
    """
    thread = await self.ensure(
        ThreadKey(
            channel_tentacle_id=event.tentacle_id,
            chat_type=event.chat_type,
            chat_id=event.chat_id,
            channel_thread_id=event.channel_thread_id,
        )
    )
    platform_message_id = event.message_id or None
    if platform_message_id is not None:
        recorded = await self.find_message(
            thread.id, platform_message_id, "inbound"
        )
        if recorded is not None:
            return recorded
    if happened_at is None and event.timestamp > 0:
        happened_at = datetime.fromtimestamp(event.timestamp, UTC)
    sender = await self.users.ensure_profile(
        thread.channel_tentacle_id, event.sender
    )
    # Swap in the registry row (owner eagerly loaded) so the event's sender
    # line resolves `user:{username}` through `event.sender.user`.
    event.sender = sender
    message = ThreadMessage(
        thread_id=thread.id,
        platform_message_id=platform_message_id,
        reply_id=event.reply_id,
        happened_at=happened_at or datetime.now(UTC),
        direction="inbound",
        actor_kind=actor_kind,
        user_id=event.user_id,
        agent_tentacle_id=agent_tentacle_id,
        sender_id=sender.id,
        segments=event.segments,
        message_text=message_text_from_segments(event.segments),
        raw=event.raw,
    )
    await self.store_message(message, thread)
    return message

record_outbound async

record_outbound(
    thread_or_address: Thread | ChannelAddress | ThreadKey,
    *,
    agent_tentacle_id: str,
    segments: list[MessageSegment],
    platform_message_id: str | None = None,
    reply_id: str = "",
    happened_at: datetime | None = None,
    sender: UserProfile,
    actor_kind: ChannelActorKind = "agent",
    message_text: str | None = None,
    raw: str = "",
) -> ThreadMessage

Write the agent's reply to the thread's ledger. Pass happened_at only when something knows better than this moment — a transcript replay does; an agent answering now does not.

Source code in octomate/managers/thread.py

async def record_outbound(
    self,
    thread_or_address: Thread | ChannelAddress | ThreadKey,
    *,
    agent_tentacle_id: str,
    segments: list[MessageSegment],
    platform_message_id: str | None = None,
    reply_id: str = "",
    happened_at: datetime | None = None,
    sender: UserProfile,
    actor_kind: ChannelActorKind = "agent",
    message_text: str | None = None,
    raw: str = "",
) -> ThreadMessage:
    """Write the agent's reply to the thread's ledger.
    Pass `happened_at` only when something knows better than this moment — a
    transcript replay does; an agent answering now does not.
    """
    if isinstance(thread_or_address, Thread):
        thread = thread_or_address
    else:
        thread = await self.ensure(thread_or_address)
    sender = await self.users.ensure_profile(thread.channel_tentacle_id, sender)
    message = ThreadMessage(
        thread_id=thread.id,
        platform_message_id=platform_message_id,
        reply_id=reply_id,
        happened_at=happened_at or datetime.now(UTC),
        direction="outbound",
        actor_kind=actor_kind,
        user_id="",
        agent_tentacle_id=agent_tentacle_id,
        sender_id=sender.id,
        segments=segments,
        message_text=message_text or message_text_from_segments(segments),
        raw=raw,
    )
    await self.store_message(message, thread)
    return message

mark_presented async

mark_presented(
    message_id: UUID7, platform_message_id: str | None
) -> None

Name the platform message a ledger row was rendered as, once the channel that rendered it knows; a rendering with no id leaves it be.

Source code in octomate/managers/thread.py

async def mark_presented(
    self, message_id: UUID7, platform_message_id: str | None
) -> None:
    """Name the platform message a ledger row was rendered as, once the
    channel that rendered it knows; a rendering with no id leaves it be."""
    if platform_message_id is None:
        return
    async with async_session() as session:
        stored = await session.one_or_none(
            ThreadMessage, expressions=[ThreadMessage["id"] == message_id]
        )
        if stored is None:
            raise ValueError(f"thread message {message_id} does not exist")
        stored.platform_message_id = platform_message_id
        await session.commit()

redate_message async

redate_message(
    message: ThreadMessage, happened_at: datetime
) -> ThreadMessage

Overrule a ledger row's clock after the fact: a transcript replay knows when a message really happened, while a row the live hooks wrote was stamped at receipt — a beat later than the transcript line, which is enough to sort a run's work above the prompt that caused it. Under the thread's lock, only the row's happened_at is written.

Source code in octomate/managers/thread.py

async def redate_message(
    self, message: ThreadMessage, happened_at: datetime
) -> ThreadMessage:
    """Overrule a ledger row's clock after the fact: a transcript replay knows
    when a message really happened, while a row the live hooks wrote was stamped
    at receipt — a beat later than the transcript line, which is enough to sort a
    run's work above the prompt that caused it. Under the thread's lock, only the
    row's `happened_at` is written."""
    if message.happened_at == happened_at:
        return message
    async with async_session() as session:
        thread = await session.get(Thread, message.thread_id, options=[noload("*")])
        if thread is None:
            raise ValueError(f"unknown thread {message.thread_id}")
        async with self.lock(thread.key):
            stored = await session.get(
                ThreadMessage, message.id, options=[noload("*")]
            )
            if stored is None:
                raise ValueError(f"thread message {message.id} does not exist")
            stored.happened_at = happened_at
            await session.commit()
    message.happened_at = happened_at
    return message

pending_prompt_messages async

pending_prompt_messages(
    thread: Thread,
    trigger_message_id: UUID7,
    active_agent_id: str,
) -> list[ThreadMessage]

Source code in octomate/managers/thread.py

async def pending_prompt_messages(
    self,
    thread: Thread,
    trigger_message_id: UUID7,
    active_agent_id: str,
) -> list[ThreadMessage]:
    # A sub-thread's key is the chat room's, so this resolves the chat either
    # way — which is where both the ledger and the cursor are.
    fresh = await self.ensure(thread.key)

    async with async_session() as session:
        expressions = [
            in_ledger(fresh.id),
            ThreadMessage["id"] <= trigger_message_id,
            ThreadMessage["kind"] == "message",
            or_(
                ThreadMessage["actor_kind"] != "agent",
                ThreadMessage["agent_tentacle_id"] != active_agent_id,
                ThreadMessage["agent_tentacle_id"].is_(None),
            ),
        ]
        if fresh.source_cursor_message_id is not None:
            expressions.append(ThreadMessage["id"] > fresh.source_cursor_message_id)
        rows = await session.list(
            ThreadMessage,
            limit=None,
            order_bys=[ThreadMessage["id"]],
            expressions=expressions,
        )
    return list(rows)

advance_prompt_cursor async

advance_prompt_cursor(
    thread: Thread, message_id: UUID7
) -> Thread | None

Source code in octomate/managers/thread.py

async def advance_prompt_cursor(
    self,
    thread: Thread,
    message_id: UUID7,
) -> Thread | None:
    async with async_session() as session:
        stored = await session.get(Thread, thread.id)
        if stored is None:
            return None
        stored.source_cursor_message_id = message_id
        await session.commit()
    thread.source_cursor_message_id = message_id
    return thread

record_fork async

record_fork(
    source: Conversation,
    target: Thread,
    *,
    sender: UserProfile,
    title: str | None,
) -> Thread

Publish a user-attributed system notice after a successful fork. It names the source by title, the source's own name that the fork starts with, or by the source's id when it goes by nothing.

Source code in octomate/managers/thread.py

async def record_fork(
    self,
    source: Conversation,
    target: Thread,
    *,
    sender: UserProfile,
    title: str | None,
) -> Thread:
    """Publish a user-attributed system notice after a successful fork. It names
    the source by `title`, the source's own name that the fork starts with, or by
    the source's id when it goes by nothing."""
    profile = await self.users.ensure_profile(target.channel_tentacle_id, sender)
    async with async_session() as session:
        thread = await session.get(
            Thread,
            target.id,
            options=[selectinload(Thread["conversations"]).noload("*")],
        )
        if thread is None:
            raise ValueError(f"unknown thread {target.id}")
        address = thread.key.address(profile.channel_user_id)
        text = (
            f'Forked from "{title}".\n\n'
            if title is not None
            else f"Forked from conversation {source.id}.\n\n"
        )
        if thread.project_id is None:
            text += (
                "This fork has no server project. The source working directory "
                "and its files were not transferred.\n\n"
            )
        text += f"Current channel address:\n{address}."
        notice = ThreadMessage(
            thread_id=thread.id,
            direction="inbound",
            actor_kind="system",
            user_id=profile.channel_user_id,
            sender_id=profile.id,
            segments=[TextSegment(data={"text": text})],
            message_text=text,
        )
        session.add(notice)
        previous_title = thread.title
        thread.title = title
        thread.updated_at = notice.happened_at
        await thread.handoffs
        await session.commit()
    events: list[ThreadEvent] = [
        ThreadMessageRecordedEvent(
            thread_id=thread.id,
            message_id=notice.id,
            updated_at=thread.updated_at,
        )
    ]
    if title != previous_title:
        events.append(
            ThreadTitleChangedEvent(
                thread_id=thread.id, title=title, updated_at=thread.updated_at
            )
        )
    await self.publish(thread, *events)
    return thread

record_handoff async

record_handoff(
    thread_or_address: Thread | ChannelAddress | ThreadKey,
    *,
    to_agent_tentacle_id: str,
    source_agent_tentacle_id: str | None = None,
    to_model: AgentRouteModelName | None = None,
    reason: str = "",
    hint: str = "",
    brief: str = "",
    source_conversation_id: UUID7 | None = None,
    target_conversation_id: UUID7 | None = None,
    source_run_id: str | None = None,
    source_model_message_id: UUID7 | None = None,
) -> Handoff

Source code in octomate/managers/thread.py

async def record_handoff(
    self,
    thread_or_address: Thread | ChannelAddress | ThreadKey,
    *,
    to_agent_tentacle_id: str,
    source_agent_tentacle_id: str | None = None,
    to_model: AgentRouteModelName | None = None,
    reason: str = "",
    hint: str = "",
    brief: str = "",
    source_conversation_id: UUID7 | None = None,
    target_conversation_id: UUID7 | None = None,
    source_run_id: str | None = None,
    source_model_message_id: UUID7 | None = None,
) -> Handoff:
    if isinstance(thread_or_address, Thread):
        thread = thread_or_address
    else:
        thread = await self.ensure(thread_or_address)
    handoff = Handoff(
        thread_id=thread.id,
        source_agent_tentacle_id=source_agent_tentacle_id,
        to_agent_tentacle_id=to_agent_tentacle_id,
        to_model=to_model,
        reason=reason,
        hint=hint,
        brief=brief,
        source_conversation_id=source_conversation_id,
        target_conversation_id=target_conversation_id,
        source_run_id=source_run_id,
        source_model_message_id=source_model_message_id,
    )
    async with async_session() as session:
        session.add(handoff)
        await session.commit()
    thread.handoffs.append(handoff)
    await self.publish(
        thread,
        ThreadHandoffRecordedEvent(thread_id=thread.id, handoff=handoff),
    )
    return handoff

chat_message async

chat_message(
    profile: UserProfile, handle: str
) -> ThreadMessage

The chat message handle names within this person's history — the threads they have spoken in, on this account or any the registry links to it. A row id, or the #msg:<id> handle shown beside it: the platform's own id, which is what a brief written from the other side of a handoff has to cite.

Source code in octomate/managers/thread.py

async def chat_message(self, profile: UserProfile, handle: str) -> ThreadMessage:
    """The chat message `handle` names within this person's history — the threads
    they have spoken in, on this account or any the registry links to it. A row
    id, or the `#msg:<id>` handle shown beside it: the platform's own id, which
    is what a brief written from the other side of a handoff has to cite."""
    if handle.startswith("#msg:"):
        named = ThreadMessage["platform_message_id"] == handle.removeprefix("#msg:")
    else:
        named = ThreadMessage["id"] == uuid.UUID(handle)
    senders = [
        profile.id,
        *(linked.id for linked in await self.users.linked_profiles(profile)),
    ]
    async with async_session() as session:
        rows = await session.list(
            ThreadMessage,
            limit=2,
            expressions=[named, in_history_of(senders)],
        )
    if not rows:
        raise ValueError(f"no message {handle} in this person's history")
    if len(rows) > 1:
        raise ValueError(f"{handle} names more than one message; use a row id")
    return rows[0]

find_message async

find_message(
    thread_id: UUID7,
    platform_message_id: str,
    direction: ThreadMessageDirection,
) -> ThreadMessage | None

The row delivered to the thread for one platform message — a turn's prompt, or the reply that answered it. Oldest wins, as scanning the ordered ledger did. A row the thread shows from a ledger a fork froze into it was delivered elsewhere, so it is never this one.

Both halves of the key are indexed. The hooks and tailers ask this per turn, on a ledger with no ceiling, so the row has to be found by the index rather than by reading the room's whole tenure to match one id in Python.

Source code in octomate/managers/thread.py

async def find_message(
    self,
    thread_id: UUID7,
    platform_message_id: str,
    direction: ThreadMessageDirection,
) -> ThreadMessage | None:
    """The row delivered to the thread for one platform message — a turn's
    prompt, or the reply that answered it. Oldest wins, as scanning the ordered
    ledger did. A row the thread shows from a ledger a fork froze into it was
    delivered elsewhere, so it is never this one.

    Both halves of the key are indexed. The hooks and tailers ask this per turn,
    on a ledger with no ceiling, so the row has to be found by the index rather
    than by reading the room's whole tenure to match one id in Python.
    """
    async with async_session() as session:
        rows = await session.list(
            ThreadMessage,
            limit=1,
            order_bys=[ThreadMessage["happened_at"], ThreadMessage["id"]],
            # The model ledger hangs off the row and no caller here reads it;
            # `related_model_messages` is how someone asks.
            options=[noload(ThreadMessage["model_messages"])],
            expressions=[
                ThreadMessage["thread_id"] == thread_id,
                ThreadMessage["platform_message_id"] == platform_message_id,
                ThreadMessage["direction"] == direction,
            ],
        )
    return rows[0] if rows else None

record_native_prompt async

record_native_prompt(
    address: ChannelAddress,
    *,
    message_id: str,
    text: str,
    sender: UserProfile,
    timestamp: datetime,
) -> ThreadMessage

Record a transcript input once, before the run commits.

The native message identity survives reconnects. The caller holds its session lock across this lookup and write; the row grants the stream's owner access to a new thread while its model history is still folding.

Source code in octomate/managers/thread.py

async def record_native_prompt(
    self,
    address: ChannelAddress,
    *,
    message_id: str,
    text: str,
    sender: UserProfile,
    timestamp: datetime,
) -> ThreadMessage:
    """Record a transcript input once, before the run commits.

    The native message identity survives reconnects. The caller holds its
    session lock across this lookup and write; the row grants the stream's
    owner access to a new thread while its model history is still folding.
    """
    thread = await self.ensure(ThreadKey.from_address(address))
    inbound = await self.find_message(thread.id, message_id, "inbound")
    if inbound is not None:
        if inbound.happened_at != timestamp:
            await self.redate_message(inbound, timestamp)
        return inbound
    return await self.record_inbound(
        MessageEvent(
            tentacle_id=address.channel_tentacle_id,
            message_id=message_id,
            chat_id=address.chat_id,
            chat_type=address.chat_type,
            user_id=sender.channel_user_id,
            sender=sender,
            segments=[TextSegment(data={"text": text})],
        ),
        happened_at=timestamp,
    )

bind_native_run async

bind_native_run(
    address: ChannelAddress,
    run: ExternalAgentRun,
    *,
    prompts: dict[UUID7, UUID7],
    answer: str,
    sender: UserProfile,
) -> UUID7 | None

Bind transcript prompts and the final reply after the run commits.

Each prompt maps its ledger id to its committed model message id. The caller holds the session lock and publishes run_ended only after this returns, so a reader can immediately reload the ledger.

Source code in octomate/managers/thread.py

async def bind_native_run(
    self,
    address: ChannelAddress,
    run: ExternalAgentRun,
    *,
    prompts: dict[UUID7, UUID7],
    answer: str,
    sender: UserProfile,
) -> UUID7 | None:
    """Bind transcript prompts and the final reply after the run commits.

    Each prompt maps its ledger id to its committed model message id. The caller holds the session lock and publishes `run_ended`
    only after this returns, so a reader can immediately reload the ledger.
    """
    for prompt_id, message_id in prompts.items():
        await self.bind_messages(
            [prompt_id],
            message_id,
            kind="request_source",
            run_id=run.id,
        )
    if not answer:
        return None
    thread = await self.ensure(ThreadKey.from_address(address))
    outbound = await self.find_message(thread.id, run.id, "outbound")
    answered = next(
        (
            message
            for message in reversed(run.messages)
            if isinstance(message, ModelResponse)
        ),
        None,
    )
    if outbound is None:
        outbound = await self.record_outbound(
            thread,
            agent_tentacle_id=address.channel_tentacle_id,
            segments=[MarkdownSegment(data={"text": answer})],
            sender=sender,
            platform_message_id=run.id,
            happened_at=answered.timestamp if answered is not None else None,
        )
    elif (
        answered is not None
        and answered.timestamp is not None
        and outbound.happened_at != answered.timestamp
    ):
        await self.redate_message(outbound, answered.timestamp)
    await self.bind_assistant_replies([outbound.id], run_id=run.id)
    return outbound.id

bind_messages async

bind_messages(
    thread_message_ids: list[UUID7],
    model_message_id: UUID7,
    *,
    kind: MessageBindingKind,
    run_id: str,
    tool_call_id: str | None = None,
) -> list[MessageBinding]

Source code in octomate/managers/thread.py

async def bind_messages(
    self,
    thread_message_ids: list[UUID7],
    model_message_id: UUID7,
    *,
    kind: MessageBindingKind,
    run_id: str,
    tool_call_id: str | None = None,
) -> list[MessageBinding]:
    bindings: list[MessageBinding] = []
    async with async_session() as session:
        for position, thread_message_id in enumerate(thread_message_ids):
            binding = MessageBinding(
                thread_message_id=thread_message_id,
                model_message_id=model_message_id,
                kind=kind,
                run_id=run_id,
                tool_call_id=tool_call_id,
                position=position,
            )
            session.add(binding)
            bindings.append(binding)
        await session.commit()
    return bindings

bind_assistant_replies async

bind_assistant_replies(
    thread_message_ids: list[UUID7], *, run_id: str
) -> list[MessageBinding]

Source code in octomate/managers/thread.py

async def bind_assistant_replies(
    self,
    thread_message_ids: list[UUID7],
    *,
    run_id: str,
) -> list[MessageBinding]:
    if not thread_message_ids:
        return []
    async with async_session() as session:
        responses = await session.list(
            ModelResponse,
            limit=None,
            order_bys=[ModelResponse["id"]],
            expressions=[
                ModelResponse["run_id"] == run_id,
                ModelResponse["role"] == "assistant",
                ModelResponse["message_text"].is_not(None),
                ModelResponse["message_text"] != "",
            ],
        )
    assistant_replies = list(responses)
    if not assistant_replies:
        return []
    if len(assistant_replies) == len(thread_message_ids):
        bindings: list[MessageBinding] = []
        async with async_session() as session:
            for thread_message_id, response in zip(
                thread_message_ids, assistant_replies, strict=True
            ):
                binding = MessageBinding(
                    thread_message_id=thread_message_id,
                    model_message_id=response.id,
                    kind="assistant_reply",
                    run_id=run_id,
                    position=0,
                )
                session.add(binding)
                bindings.append(binding)
            await session.commit()
        return bindings
    return await self.bind_messages(
        thread_message_ids,
        assistant_replies[-1].id,
        kind="assistant_reply",
        run_id=run_id,
    )

search_chat_messages async

search_chat_messages(
    profile: UserProfile,
    query: str,
    *,
    actor_kind: ChannelActorKind | None = None,
    limit: int = 10,
) -> list[ThreadMessage]

The messages matching every English term in query across this person's history — the threads they have spoken in, on this account or any the registry links to it. Best lexical match first, newest first when scores tie. A visitor's history is one account's.

Source code in octomate/managers/thread.py

async def search_chat_messages(
    self,
    profile: UserProfile,
    query: str,
    *,
    actor_kind: ChannelActorKind | None = None,
    limit: int = 10,
) -> list[ThreadMessage]:
    """The messages matching every English term in `query` across this person's
    history — the threads they have spoken in, on this account or any the registry
    links to it. Best lexical match first, newest first when scores tie. A
    visitor's history is one account's."""
    match_query = history_match_query(query)
    senders = [
        profile.id,
        *(linked.id for linked in await self.users.linked_profiles(profile)),
    ]
    expressions = [
        ThreadMessageFTS["message_text"].match(match_query),
        in_history_of(senders),
    ]
    if actor_kind is not None:
        expressions.append(ThreadMessage["actor_kind"] == actor_kind)
    async with async_session() as session:
        statement = (
            select(ThreadMessage)
            .join(
                ThreadMessageFTS,
                ThreadMessage["id"] == ThreadMessageFTS["message_id"],
            )
            .where(*expressions)
            .options(noload(ThreadMessage["model_messages"]))
            .order_by(
                ThreadMessageFTS["rank"],
                ThreadMessage["happened_at"].desc(),
                ThreadMessage["id"].desc(),
            )
            .limit(limit)
        )
        rows = (await session.execute(statement)).scalars().all()
    return list(rows)

chat_messages_before async

chat_messages_before(
    thread_id: UUID7, anchor_id: UUID7, *, limit: int = 5
) -> list[ThreadMessage]

The rows standing before anchor_id in the thread, oldest last of them first. Neighbours in the conversation, which is what the ledger's order means — so the anchor is compared on the same key the rows are sorted by.

Source code in octomate/managers/thread.py

async def chat_messages_before(
    self,
    thread_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ThreadMessage]:
    """The rows standing before `anchor_id` in the thread, oldest last of them
    first. Neighbours in the conversation, which is what the ledger's order means —
    so the anchor is compared on the same key the rows are sorted by."""
    async with async_session() as session:
        anchor = await session.one_or_none(
            ThreadMessage, expressions=[ThreadMessage["id"] == anchor_id]
        )
        if anchor is None:
            raise ValueError(f"thread message {anchor_id} does not exist")
        rows = await session.list(
            ThreadMessage,
            limit=limit,
            order_bys=[
                ThreadMessage["happened_at"].desc(),
                ThreadMessage["id"].desc(),
            ],
            expressions=[
                in_ledger(thread_id),
                or_(
                    ThreadMessage["happened_at"] < anchor.happened_at,
                    and_(
                        ThreadMessage["happened_at"] == anchor.happened_at,
                        ThreadMessage["id"] < anchor.id,
                    ),
                ),
            ],
        )
    return list(reversed(rows))

chat_messages_after async

chat_messages_after(
    thread_id: UUID7, anchor_id: UUID7, *, limit: int = 5
) -> list[ThreadMessage]

The rows standing after anchor_id in the thread, oldest first — the mirror of chat_messages_before, and anchored on the same key.

Source code in octomate/managers/thread.py

async def chat_messages_after(
    self,
    thread_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ThreadMessage]:
    """The rows standing after `anchor_id` in the thread, oldest first — the mirror
    of `chat_messages_before`, and anchored on the same key."""
    async with async_session() as session:
        anchor = await session.one_or_none(
            ThreadMessage, expressions=[ThreadMessage["id"] == anchor_id]
        )
        if anchor is None:
            raise ValueError(f"thread message {anchor_id} does not exist")
        rows = await session.list(
            ThreadMessage,
            limit=limit,
            order_bys=[ThreadMessage["happened_at"], ThreadMessage["id"]],
            expressions=[
                in_ledger(thread_id),
                or_(
                    ThreadMessage["happened_at"] > anchor.happened_at,
                    and_(
                        ThreadMessage["happened_at"] == anchor.happened_at,
                        ThreadMessage["id"] > anchor.id,
                    ),
                ),
            ],
        )
    return list(rows)

messages_page async

messages_page(
    thread_id: UUID7,
    cursor: str | None = None,
    *,
    unbound: bool = False,
    limit: int,
) -> Page[ThreadMessage]

The rows the thread shows — its own ledger's, and a forked parent's up to the cut — a page at a time, newest first, continued from an Arcanus cursor. Ids are uuid7 and a ledger is appended to, so id order is the order rows were written in.

unbound keeps to the rows no model message of the thread's own conversations was built from or answered with: a fork or teleport notice, or a chat room's talk, whose work runs in sub-threads. Everything else is read through those model messages, which carry the rows bound to them.

Source code in octomate/managers/thread.py

async def messages_page(
    self,
    thread_id: UUID7,
    cursor: str | None = None,
    *,
    unbound: bool = False,
    limit: int,
) -> Page[ThreadMessage]:
    """The rows the thread shows — its own ledger's, and a forked parent's up to
    the cut — a page at a time, newest first, continued from an Arcanus
    `cursor`. Ids are uuid7 and a ledger is appended to, so id order is the
    order rows were written in.

    `unbound` keeps to the rows no model message of the thread's own
    conversations was built from or answered with: a fork or teleport notice,
    or a chat room's talk, whose work runs in sub-threads. Everything else is
    read through those model messages, which carry the rows bound to them.
    """
    scope: list[Expression[bool] | ColumnElement[bool]]
    scope = [in_ledger(thread_id)]
    if unbound:
        scope.append(
            ~ThreadMessage["model_messages"].any(
                ModelMessage["run_id"].in_(own_runs(thread_id))
            )
        )
    bookmark = None if cursor is None else Cursor[ThreadMessage](cursor).bookmark
    after = (
        []
        if bookmark is None or bookmark.expression is None
        else [bookmark.expression]
    )
    newest_first = (ThreadMessage["id"].desc(),)
    async with async_session() as session:
        rows = await session.list(
            ThreadMessage,
            limit=limit + 1,
            order_bys=newest_first,
            options=[noload(ThreadMessage["model_messages"])],
            expressions=[*scope, *after],
        )
        total = await session.count(ThreadMessage, expressions=scope)
    items = tuple(rows[:limit])
    last = ThreadMessage["id"] < items[-1].id if items else None
    return Page(
        items=items,
        total=total,
        next_cursor=str(
            Cursor[ThreadMessage].from_expressions(
                bookmark=last, order_bys=newest_first, limit=limit
            )
        ),
        has_more=len(rows) > limit,
    )

related_model_messages async

related_model_messages(
    thread_message_id: UUID7,
) -> list[ModelRequest | ModelResponse]

Source code in octomate/managers/thread.py

async def related_model_messages(
    self,
    thread_message_id: UUID7,
) -> list[ModelRequest | ModelResponse]:
    async with async_session() as session:
        message = await session.get(
            ThreadMessage,
            thread_message_id,
            options=[selectinload(ThreadMessage["model_messages"])],
        )
        if message is None:
            return []
        await message.model_messages
    return list(message.model_messages)

history_match_query

history_match_query(query: str) -> str

An English plain-text query expressed safely in FTS5's query language.

Each whitespace-delimited term is its own quoted phrase, so every term must match while operators and punctuation supplied by a caller stay literal.

Source code in octomate/managers/thread.py

def history_match_query(query: str) -> str:
    """An English plain-text query expressed safely in FTS5's query language.

    Each whitespace-delimited term is its own quoted phrase, so every term must
    match while operators and punctuation supplied by a caller stay literal.
    """
    terms = query.split()
    if not terms:
        raise ValueError("history search query must contain at least one word")
    escaped = [term.replace('"', '""') for term in terms]
    return " AND ".join(f'"{term}"' for term in escaped)

message_text_from_segments

message_text_from_segments(
    segments: list[MessageSegment],
) -> str | None

Source code in octomate/managers/thread.py

def message_text_from_segments(segments: list[MessageSegment]) -> str | None:
    parts = [
        segment.data["text"]
        for segment in segments
        if isinstance(segment, (TextSegment, MarkdownSegment))
    ]
    if not parts:
        return None
    return "\n".join(parts)

keyed

keyed(key: ThreadKey) -> list[Expression[bool]]

The row a key names. A sub-thread shares its chat room's address, so a key names the thread nobody opened — the line the unique index draws.

Source code in octomate/managers/thread.py

def keyed(key: ThreadKey) -> list[Expression[bool]]:
    """The row a key names. A sub-thread shares its chat room's address, so a key
    names the thread nobody opened — the line the unique index draws."""
    return [
        Thread["channel_tentacle_id"] == key.channel_tentacle_id,
        Thread["chat_type"] == key.chat_type,
        Thread["chat_id"] == key.chat_id,
        Thread["channel_thread_id"] == key.channel_thread_id,
        Thread["parent_thread_id"].is_(None),
    ]

shown_by

shown_by(threads: Expression[bool]) -> ColumnElement[bool]

The messages shown by the threads threads picks out of thread_ledgers: those delivered to one of their ledgers, up to that ledger's cut.

Source code in octomate/managers/thread.py

def shown_by(threads: Expression[bool]) -> ColumnElement[bool]:
    """The messages shown by the threads `threads` picks out of `thread_ledgers`:
    those delivered to one of their ledgers, up to that ledger's cut."""
    return (
        select(ThreadLedger["thread_id"])
        .where(
            threads,
            ThreadLedger["ledger_id"] == ThreadMessage["thread_id"],
            or_(
                ThreadLedger["cut_message_id"].is_(None),
                ThreadMessage["id"] <= ThreadLedger["cut_message_id"],
            ),
        )
        .exists()
    )

in_ledger

in_ledger(thread_id: UUID7) -> ColumnElement[bool]

The messages a thread shows.

Source code in octomate/managers/thread.py

def in_ledger(thread_id: UUID7) -> ColumnElement[bool]:
    """The messages a thread shows."""
    return shown_by(ThreadLedger["thread_id"] == thread_id)

in_history_of

in_history_of(senders: list[UUID7]) -> ColumnElement[bool]

The messages shown by threads that received a message from senders.

Inherited messages are readable history, not participation in the destination.

Source code in octomate/managers/thread.py

def in_history_of(senders: list[UUID7]) -> ColumnElement[bool]:
    """The messages shown by threads that received a message from `senders`.

    Inherited messages are readable history, not participation in the destination.
    """
    # Aliased, so the rows said here never correlate with the rows being shown.
    said = aliased(ThreadMessage)
    spoken_in = select(said["thread_id"]).where(
        said["sender_id"].in_(senders),
    )
    return shown_by(ThreadLedger["thread_id"].in_(spoken_in))

thread_title

thread_title(text: str | None) -> str | None

A name for a thread, out of one line of what was said in it.

Whitespace is folded because a directive arrives with its own newlines and a listing has one line to give it. A message with nothing in it names nothing, and the row goes on falling back to its surface.

Source code in octomate/managers/thread.py

def thread_title(text: str | None) -> str | None:
    """A name for a thread, out of one line of what was said in it.

    Whitespace is folded because a directive arrives with its own newlines and a
    listing has one line to give it. A message with nothing in it names nothing,
    and the row goes on falling back to its surface.
    """
    line = " ".join((text or "").split())
    if not line:
        return None
    return line if len(line) <= TITLE_MAX else f"{line[:TITLE_MAX].rstrip()}…"

octomate.managers.conversation

Agent conversations and their model message history.

RunT module-attribute

RunT = TypeVar('RunT', bound=AgentRun)

request_usage_adapter module-attribute

request_usage_adapter: TypeAdapter[RequestUsage] = (
    TypeAdapter(RequestUsage)
)

ConversationManager

Bases: Manager, Locks[tuple[UUID7, str, str]]

Resolves and persists agent Conversation entities, and owns their model message history.

A Conversation is one agent's model context within a Thread — not a human chat log; the user-facing chat ledger is ThreadManager's ThreadMessage rows. A thread owns one conversation per agent, so the identity is (thread_id, agent_tentacle_id, subagent_id): every sender in a group thread keys to the same owning agent's conversation, regardless of who woke it.

Source code in octomate/managers/conversation.py

class ConversationManager(Manager, Locks[tuple[UUID7, str, str]]):
    """Resolves and persists agent `Conversation` entities, and owns their model
    message history.

    A `Conversation` is one agent's model context within a `Thread` — not a human
    chat log; the user-facing chat ledger is `ThreadManager`'s `ThreadMessage`
    rows. A thread owns one conversation per agent, so the identity is
    `(thread_id, agent_tentacle_id, subagent_id)`: every sender in a group thread
    keys to the same owning agent's conversation, regardless of who woke it.
    """

    async def latest_model(self, conversation_id: UUID7) -> str | None:
        """Read the last reported model without loading the conversation's history."""
        async with async_session() as session:
            return await session.scalar(
                select(AgentRun["model_name"])
                .where(
                    AgentRun["id"].in_(
                        select(ConversationRun["run_id"]).where(
                            ConversationRun["conversation_id"] == conversation_id
                        )
                    ),
                    AgentRun["model_name"].is_not(None),
                )
                .order_by(AgentRun["started_at"].desc(), AgentRun["id"].desc())
                .limit(1)
            )

    async def ensure(
        self,
        thread_id: UUID7,
        *,
        agent_tentacle_id: str,
        subagent_id: str = "",
        parent_conversation_id: UUID7 | None = None,
        with_history: bool = True,
    ) -> Conversation:
        """Resolve the conversation owned by `agent_tentacle_id` in `thread_id`,
        creating it if it does not yet exist. Its `conversation.messages` is the
        live model history. The owning agent is part of the identity — two agents
        in the same thread keep separate conversations — so callers must always
        supply it. `subagent_id` narrows to one subagent's own context; empty is
        the agent's long-lived conversation in the thread. A subagent context
        names the conversation that spawned it, and only a subagent context may.
        `with_history=False` leaves runs and messages unloaded for callers that
        only need identity, permissions, or the native session's resume id."""
        if bool(subagent_id) != (parent_conversation_id is not None):
            raise ValueError(
                "a subagent conversation requires both subagent_id and "
                "parent_conversation_id; a bare conversation takes neither"
            )
        # Serialize first sightings so the loser reads the committed row.
        async with async_session() as session:
            async with self.lock((thread_id, agent_tentacle_id, subagent_id)):
                conversation = await session.one_or_none(
                    Conversation,
                    options=[
                        lazyload(Conversation["runs"]),
                        lazyload(Conversation["messages"]),
                    ],
                    expressions=[
                        Conversation["thread_id"] == thread_id,
                        Conversation["agent_tentacle_id"] == agent_tentacle_id,
                        Conversation["subagent_id"] == subagent_id,
                    ],
                )
                if conversation is None:
                    conversation = Conversation(
                        thread_id=thread_id,
                        agent_tentacle_id=agent_tentacle_id,
                        subagent_id=subagent_id,
                        parent_conversation_id=parent_conversation_id,
                    )
                    session.add(conversation)
                await session.flush()
                await session.commit()
            if with_history:
                await conversation.runs
                await conversation.messages
            return conversation

    async def get(
        self, conversation_id: UUID7, *, with_history: bool = True
    ) -> Conversation:
        """Resolve a conversation by id — one fresh read; raises on an unknown
        id. This is the by-id path for a run addressed at a pre-ensured
        conversation (a commissioned accomplice's child context): the caller
        ensured it and owns any thread/agent validation. `with_history=False`
        leaves runs and messages unloaded, as on `ensure`."""
        if not with_history:
            options = [
                lazyload(Conversation["runs"]),
                lazyload(Conversation["messages"]),
            ]
        else:
            options = []
        async with async_session() as session:
            conversation = await session.get(
                Conversation,
                conversation_id,
                options=options,
            )
            if conversation is None:
                raise ValueError(f"unknown conversation {conversation_id}")
            if with_history:
                await conversation.runs
                await conversation.messages
        return conversation

    async def link_parent_run(
        self,
        run_id: str,
        *,
        parent_run_id: str,
        parent_tool_call_id: str | None,
    ) -> None:
        """Stamp a recorded run's place in the run tree, after the fact. The
        runner stays ignorant of why it ran; the caller that spawned it links
        the turn to its own run and tool call once the child returns. A run
        that recorded nothing has no row to link — not an error."""
        async with async_session() as session:
            run = await session.one_or_none(
                AgentRun, expressions=[AgentRun["id"] == run_id]
            )
            if run is None:
                return
            run.parent_run_id = parent_run_id
            run.parent_tool_call_id = parent_tool_call_id
            await session.commit()

    async def subagents(self, parent_conversation_id: UUID7) -> list[Conversation]:
        """The subagent conversations spawned from `parent_conversation_id` — the
        live accomplices a `whisper` can reach. Rows only; callers resolve a
        chosen one through `ensure`."""
        async with async_session() as session:
            rows = await session.list(
                Conversation,
                limit=None,
                expressions=[
                    Conversation["parent_conversation_id"] == parent_conversation_id
                ],
            )
        return list(rows)

    async def for_thread(self, thread_id: UUID7) -> list[Conversation]:
        """The thread's agent conversations, subagents included, each with its runs.

        Both message relations here are lazy="selectin", so the plain query would read
        every model message the thread ever produced twice over — once under the
        conversation and once under the run that owns it. Both are dropped: a reader
        after the messages pages through them with `messages_page`.
        """
        async with async_session() as session:
            rows = await session.list(
                Conversation,
                limit=None,
                options=[
                    noload(Conversation["messages"]),
                    selectinload(Conversation["runs"]).noload(AgentRun["messages"]),
                ],
                expressions=[Conversation["thread_id"] == thread_id],
            )
        return list(rows)

    async def messages_page(
        self, thread_id: UUID7, cursor: str | None = None, *, limit: int
    ) -> Page[ModelRequest | ModelResponse]:
        """The model messages of the thread's own conversations, newest first, a
        page at a time and continued from an Arcanus `cursor`. Ids are uuid7 and a
        run's messages are appended in processing order, so id order preserves
        each transcript's sequence across committed turns.

        Each carries the chat rows bound to it — the ones a request was built from,
        the ones a response was delivered as — so a reader has who said what
        without the prompt's dressing. A subagent's messages stay out: they are its
        own conversation's, and the spawn call in its parent's run stands for them.
        """
        scope = ModelMessage["run_id"].in_(own_runs(thread_id))
        bookmark = None if cursor is None else Cursor[ModelMessage](cursor).bookmark
        after = (
            []
            if bookmark is None or bookmark.expression is None
            else [bookmark.expression]
        )
        newest_first = (ModelMessage["id"].desc(),)
        async with async_session() as session:
            rows = await session.list(
                ModelMessage,
                limit=limit + 1,
                order_bys=newest_first,
                options=[
                    selectinload(ModelMessage["thread_messages"]).noload(
                        ThreadMessage["model_messages"]
                    )
                ],
                expressions=[scope, *after],
            )
            total = await session.count(ModelMessage, expressions=[scope])
            # The polymorphic read materializes every row as one of the two.
            items = tuple(
                row
                for row in rows[:limit]
                if isinstance(row, ModelRequest | ModelResponse)
            )
            for item in items:
                await item.thread_messages
        last = ModelMessage["id"] < items[-1].id if items else None
        return Page(
            items=items,
            total=total,
            next_cursor=str(
                Cursor[ModelMessage].from_expressions(
                    bookmark=last, order_bys=newest_first, limit=limit
                )
            ),
            has_more=len(rows) > limit,
        )

    async def usage(self, thread_id: UUID7) -> RequestUsage:
        """The tokens the thread's own conversations spent, every response's usage
        summed. A subagent's are left to its own conversation, as its messages are.
        Only the usage column is read: summing a thread must not load its parts."""
        async with async_session() as session:
            usages = await session.scalars(
                select(ModelResponse["usage"]).where(
                    ModelResponse["run_id"].in_(own_runs(thread_id)),
                    ModelResponse["usage"].is_not(None),
                )
            )
            return sum(
                map(request_usage_adapter.validate_python, usages),
                RequestUsage(),
            )

    async def thread_id(self, conversation_id: UUID7) -> UUID7 | None:
        async with async_session() as session:
            conversation = await session.get(Conversation, conversation_id)
        if conversation is None:
            return None
        return conversation.thread_id

    async def record_agent_run(
        self,
        conversation: Conversation,
        run_id: str,
        messages: Sequence[PydanticModelMessage],
        *,
        name: str | None = None,
        model_name: str | None = None,
        permission_mode: AgentPermissionMode | None = None,
        cwd: Path | None = None,
        external_id: str | None = None,
        native_id: str | None = None,
        native_turn_id: str | None = None,
        parent_run_id: str | None = None,
        parent_tool_call_id: str | None = None,
    ) -> AgentRun | None:
        """Persist a fresh agent run.
        `cwd` is the directory the run happened in, None when its caller has none.
        `external_id`, when given, updates the conversation's resumable agent
        session handle in the same commit (external-runtime agents own their own
        session). The parent pair marks a subagent's turn: the run whose tool
        call spawned it, and which call it answers."""
        if not messages:
            return None
        run = AgentRun(
            id=run_id,
            conversation_id=conversation.id,
            native_id=native_id,
            native_session_id=external_id,
            native_turn_id=native_turn_id,
            name=name,
            model_name=model_name,
            permission_mode=permission_mode,
            cwd=cwd,
            parent_run_id=parent_run_id,
            parent_tool_call_id=parent_tool_call_id,
            started_at=messages[0].timestamp,
            # Shallow `vars(m)` dicts, so pydantic routes each through the blessed
            # `ModelRequest | ModelResponse` union at construction — the raw pydantic-ai
            # dataclass is rejected, not being an instance of our blessed subclass.
            messages=[vars(m) for m in messages],  # pyright: ignore[reportArgumentType]
        )
        return await self.persist_run(
            run, conversation_id=conversation.id, external_id=external_id
        )

    async def driven_run(
        self, native_id: str, session_id: str, turn_id: str
    ) -> AgentRun | None:
        """Find the driven owner of a native turn without loading its messages."""
        async with async_session() as session:
            return await session.one_or_none(
                AgentRun,
                expressions=[
                    AgentRun["kind"] == "octomate",
                    AgentRun["native_id"] == native_id,
                    AgentRun["native_session_id"] == session_id,
                    AgentRun["native_turn_id"] == turn_id,
                ],
                options=[noload(AgentRun["messages"])],
            )

    async def record_external_run(
        self,
        conversation: Conversation,
        run_id: str,
        messages: Sequence[PydanticModelMessage] = (),
        *,
        name: str | None = None,
        model_name: str | None = None,
        permission_mode: AgentPermissionMode | None = None,
        cwd: Path | None = None,
        native_session_id: str,
        source: str | None = None,
        start_offset: int | None = None,
        end_offset: int | None = None,
        last_line_uuid: str | None = None,
        parent_run_id: str | None = None,
        parent_tool_call_id: str | None = None,
    ) -> ExternalAgentRun | None:
        """Persist a turn of a native runtime's session as the
        `external` variant. `cwd` is the directory the turn ran in, as the hook or
        the transcript reported it. `native_session_id` doubles as the
        conversation's resumable handle (`external_id`).

        A turn already recorded by the driven runtime keeps its original run and
        messages. Its runtime identity lets native ingest skip the replay without
        copying it into the native ledger or saving transcript coordinates on it.

        The byte range is what marks a turn finished. A run carrying `end_offset` was
        assembled from the transcript and is final, so recording it again — a recovery
        overlapping a live tail, or a resume re-covering committed bytes — is a no-op
        returning `None` rather than a primary-key collision. That guard belongs here,
        at the durable sink: a caller's in-memory guard only knows the runs *it* wrote,
        never one another tailer already committed.

        Omitting messages reserves a run identity without changing an existing run.
        Native hooks use this so the transcript remains the only message writer.

        Completing a reserved run fills its messages in transcript order while
        preserving its identity. Older incomplete imports may contain provisional
        messages; the completed transcript replaces those in the same commit.
        """
        driven = await self.driven_run(
            conversation.agent_tentacle_id, native_session_id, run_id
        )
        if driven is not None:
            return None
        run = ExternalAgentRun(
            id=run_id,
            conversation_id=conversation.id,
            native_id=conversation.agent_tentacle_id,
            native_turn_id=run_id,
            name=name,
            model_name=model_name,
            permission_mode=permission_mode,
            cwd=cwd,
            parent_run_id=parent_run_id,
            parent_tool_call_id=parent_tool_call_id,
            started_at=messages[0].timestamp if messages else None,
            messages=[vars(m) for m in messages],  # pyright: ignore[reportArgumentType]
            native_session_id=native_session_id,
            source=source,
            start_offset=start_offset,
            end_offset=end_offset,
            last_line_uuid=last_line_uuid,
        )
        async with async_session() as session:
            stored = await session.one_or_none(
                AgentRun, expressions=[AgentRun["id"] == run_id]
            )
            if stored is not None:
                if not messages and end_offset is None:
                    return stored if isinstance(stored, ExternalAgentRun) else None
                if (
                    not isinstance(stored, ExternalAgentRun)
                    or stored.end_offset is not None
                ):
                    return None
                # `model_messages.run_id` is NOT NULL, so the sketch's messages are
                # deleted rather than left behind by the merged collection.
                sketch = list(stored.messages)
                run = await session.merge(run)
                for message in sketch:
                    await session.delete(message)
                owner = await session.get(
                    Conversation,
                    conversation.id,
                    options=[
                        noload(Conversation["runs"]),
                        noload(Conversation["messages"]),
                    ],
                )
                if owner is not None:
                    owner.external_id = native_session_id
                await session.commit()
                return run
        return await self.persist_run(
            run, conversation_id=conversation.id, external_id=native_session_id
        )

    async def persist_run(
        self,
        run: RunT,
        *,
        conversation_id: UUID7,
        external_id: str | None,
    ) -> RunT:
        """Persist a freshly built run, in the history of the conversation it ran
        in. `external_id`, when given, updates the resumable agent-session handle in
        the same commit.

        Only the new run and its messages are added, never the caller's whole
        conversation graph: re-merging it (prior runs and their message
        collections) can make SQLAlchemy believe a message was dropped from a
        stale run collection and null its NOT NULL `run_id`, raising an
        IntegrityError.
        """
        async with async_session() as session:
            session.add(run)
            reloaded = await session.one_or_none(
                Conversation,
                expressions=[Conversation["id"] == conversation_id],
                options=[
                    noload(Conversation["runs"]),
                    noload(Conversation["messages"]),
                ],
            )
            if reloaded is not None:
                reloaded.runs.append(run)
                reloaded.external_id = external_id
            await session.commit()
            await run.messages
        return run

    async def fork(
        self,
        source: Conversation,
        target: Conversation,
        *,
        carry_external_id: bool = False,
        external_id: str | None = None,
        transcript: Jsonl | None = None,
        end_offset: int | None = None,
        permission_mode: AgentPermissionMode | None = None,
    ) -> str | None:
        """Continue `source`'s history in an empty `target`, in one transaction.

        Nothing is copied. The target's history points at the source's runs through
        `conversation_runs`, and its thread shows the source thread's ledgers through
        `thread_ledgers`, each frozen where it stands now: whatever either side says
        afterwards is its own. Only the selected conversation receives the
        independently forked runtime handle.

        A native import passes `end_offset`, where the transcript's last completed
        turn ends (an uploaded snapshot's size by default), to leave out the turns
        that end past it and cut the source's ledger before their chat.
        `carry_external_id` moves the existing handle instead, for runtimes whose
        continuation requires it. Return the id of the last run the target's history
        includes, if any.
        """
        if source.thread_id == target.thread_id:
            raise ValueError("A fork requires an independent destination thread")
        if transcript is not None and (external_id is None or carry_external_id):
            raise ValueError("An imported transcript requires an independent session")
        if external_id is not None and (
            carry_external_id or external_id == source.external_id
        ):
            raise ValueError(
                "a fork requires a new external id without carrying the source"
            )
        if end_offset is None and transcript is not None:
            end_offset = transcript.size

        async with async_session() as session:
            stored_target = await session.get(
                Conversation,
                target.id,
                options=[
                    noload(Conversation["runs"]),
                    noload(Conversation["messages"]),
                ],
            )
            if stored_target is None:
                raise ValueError(f"unknown conversation {target.id}")
            target_thread = await session.get(Thread, target.thread_id)
            if target_thread is None:
                raise ValueError(f"unknown thread {target.thread_id}")
            if await session.count(
                Conversation,
                expressions=[
                    Conversation["thread_id"] == target.thread_id,
                    or_(Conversation["id"] != target.id, Conversation["runs"].any()),
                ],
            ) or await session.count(
                Thread,
                expressions=[
                    Thread["id"] == target.thread_id,
                    Thread["messages"].any(ThreadMessage["actor_kind"] != "system"),
                ],
            ):
                raise ValueError(
                    "fork target already holds history; refusing to splice histories"
                )

            source_thread = await session.get(Thread, source.thread_id)
            if source_thread is None:
                raise ValueError(f"unknown thread {source.thread_id}")
            origin = await session.get(
                Conversation,
                source.id,
                options=[
                    noload(Conversation["messages"]),
                    selectinload(Conversation["runs"]).noload(AgentRun["messages"]),
                ],
            )
            if origin is None:
                raise ValueError(f"unknown conversation {source.id}")
            dropped: set[str] = set()
            # Without an end offset the cut is the ledger's last message. With one,
            # it falls after the last chat of a completed turn: bound to one, or
            # neither bound to nor filed under an unfinished turn (a native ledger
            # names a turn's prompt by the turn's id), and said no later than the last
            # message the fork keeps.
            cut = select(func.max(ThreadMessage["id"])).where(
                ThreadMessage["thread_id"] == source.thread_id
            )
            if end_offset is not None:
                # A turn of the origin's own session that ends past the offset is
                # unfinished, and so is the subagent work it spawned. The offset is a
                # position in that session's transcript, so it judges no other run.
                unfinished = (
                    select(ExternalAgentRun["id"])
                    .where(
                        ExternalAgentRun["id"].in_(
                            select(ConversationRun["run_id"]).where(
                                ConversationRun["conversation_id"] == source.id
                            )
                        ),
                        ExternalAgentRun["native_session_id"] == origin.external_id,
                        or_(
                            ExternalAgentRun["end_offset"].is_(None),
                            ExternalAgentRun["end_offset"] > end_offset,
                        ),
                    )
                    .cte(recursive=True)
                )
                unfinished = unfinished.union(
                    select(AgentRun["id"]).where(
                        AgentRun["parent_run_id"] == unfinished.c.id
                    )
                )
                dropped = set(await session.scalars(select(unfinished.c.id)))
                # Every run in the history of a conversation in the source thread,
                # but the unfinished ones.
                kept = select(ConversationRun["run_id"]).where(
                    ConversationRun["conversation_id"].in_(
                        select(Conversation["id"]).where(
                            Conversation["thread_id"] == source.thread_id
                        )
                    ),
                    ConversationRun["run_id"].not_in(dropped),
                )
                bound = select(MessageBinding["thread_message_id"])
                cut = cut.where(
                    or_(
                        ThreadMessage["id"].in_(
                            bound.where(MessageBinding["run_id"].in_(kept))
                        ),
                        and_(
                            ThreadMessage["id"].not_in(
                                bound.where(MessageBinding["run_id"].in_(dropped))
                            ),
                            or_(
                                ThreadMessage["platform_message_id"].is_(None),
                                ThreadMessage["platform_message_id"].not_in(dropped),
                            ),
                            ThreadMessage["happened_at"]
                            <= select(func.max(ModelMessage["timestamp"]))
                            .where(ModelMessage["run_id"].in_(kept))
                            .scalar_subquery(),
                        ),
                    )
                )
            own_cut = await session.scalar(cut)
            ledgers = await session.list(
                ThreadLedger,
                limit=None,
                order_bys=[ThreadLedger["ledger_id"]],
                expressions=[ThreadLedger["thread_id"] == source.thread_id],
            )
            # Every ledger the source shows reaches the fork frozen: its own at the
            # cut, and the ones a fork froze into it where they already stop.
            cuts = {ledger.ledger_id: ledger.cut_message_id for ledger in ledgers}
            cuts[source.thread_id] = own_cut
            session.add_all(
                ThreadLedger(
                    thread_id=target.thread_id,
                    ledger_id=ledger_id,
                    cut_message_id=cut,
                )
                for ledger_id, cut in cuts.items()
                if cut is not None
            )

            await session.execute(
                insert(ConversationRun).from_select(
                    ["conversation_id", "run_id"],
                    select(literal(target.id), ConversationRun["run_id"]).where(
                        ConversationRun["conversation_id"] == source.id,
                        ConversationRun["run_id"].not_in(dropped),
                    ),
                )
            )
            # The cursor stops where the source's did, within what the fork shows.
            shown = max(
                (cut for cut in cuts.values() if cut is not None),
                default=None,
            )
            cursor = source_thread.source_cursor_message_id
            target_thread.source_cursor_message_id = (
                shown if cursor is None or shown is None else min(cursor, shown)
            )
            stored_target.name = origin.name
            stored_target.permission_mode = permission_mode or origin.permission_mode
            stored_target.effort = origin.effort
            stored_target.allowed_tools = list(origin.allowed_tools)
            if transcript is not None:
                session.add(transcript)
                await session.flush()
                stored_target.transcript_file_id = transcript.id
            if carry_external_id:
                stored_target.external_id = origin.external_id
                origin.external_id = None
            elif external_id is not None:
                stored_target.external_id = external_id
            await session.commit()
        if carry_external_id:
            source.external_id = None
        elif external_id is not None:
            target.external_id = external_id
        forked = [run.id for run in origin.runs if run.id not in dropped]
        return forked[-1] if forked else None

    async def store_transcript(
        self,
        uploaded: UploadFile,
        start: int,
        *,
        conversation: Conversation,
        files: FileManager,
        owner_id: UUID7,
    ) -> FileVariant:
        """Persist new bytes of a native conversation's owner-scoped transcript.

        Create the first file and conversation reference in one transaction;
        subsequent uploads append to that file. Never mutate the caller's conversation.
        """
        async with self.lock(conversation.key):
            current = await self.get(conversation.id, with_history=False)
            previous_id = current.transcript_file_id
            if previous_id is not None:
                return await files.append(
                    previous_id, uploaded.file, offset=start, owner_id=owner_id
                )
            if start != 0:
                raise ValueError("The first transcript upload must start at zero")
            async with (
                files.upload(uploaded, owner_id=owner_id) as transcript,
                async_session() as session,
            ):
                stored = await session.get(Conversation, conversation.id)
                if stored is None:
                    raise ValueError(f"unknown conversation {conversation.id}")
                session.add(transcript)
                await session.flush()
                stored.transcript_file_id = transcript.id
                await session.commit()
            return transcript

    async def set_external_id(
        self, conversation: Conversation, external_id: str
    ) -> None:
        """Bind a conversation to its replacement runtime session."""
        async with async_session() as session:
            stored = await session.get(Conversation, conversation.id)
            if stored is None:
                raise ValueError(f"unknown conversation {conversation.id}")
            stored.external_id = external_id
            await session.commit()
        conversation.external_id = external_id

    async def set_permission_mode(
        self,
        conversation: Conversation,
        mode: AgentPermissionMode | None,
    ) -> Conversation:
        """Store the approval posture this conversation's agent works under.

        Callers selecting a mode validate against the running agent's catalog.
        Native observers also store modes that are no longer selectable. None clears
        the selection and lets the agent's configured default decide.

        This method only persists the selection. User changes go through the
        owning agent tentacle, which also updates its live runtime when supported.
        """
        async with async_session() as session:
            stored = await session.get(Conversation, conversation.id)
            if stored is None:
                raise ValueError(f"unknown conversation {conversation.id}")
            stored.permission_mode = mode
            await session.commit()
        conversation.permission_mode = mode
        return conversation

    async def set_effort(
        self, conversation: Conversation, effort: str | None
    ) -> Conversation:
        """Store the reasoning effort this conversation's runs ask for.

        Callers validate the level against the route the conversation runs on;
        None clears it and lets the runtime's own default decide.
        """
        async with async_session() as session:
            stored = await session.get(Conversation, conversation.id)
            if stored is None:
                raise ValueError(f"unknown conversation {conversation.id}")
            stored.effort = effort
            await session.commit()
        conversation.effort = effort
        return conversation

    async def set_name(self, conversation: Conversation, name: str) -> Conversation:
        """Store the name the runtime running this session grabbed for itself.

        Only a runtime that names its own sessions has one to store, and it revises
        it as the session goes on, so this overwrites. A name with nothing in it is
        not one, and leaves the conversation as it was.
        """
        named = name.strip()
        if not named or conversation.name == named:
            return conversation
        async with async_session() as session:
            stored = await session.get(Conversation, conversation.id)
            if stored is None:
                raise ValueError(f"unknown conversation {conversation.id}")
            stored.name = named
            await session.commit()
        conversation.name = named
        return conversation

    async def grant_session_tool(
        self,
        conversation: Conversation,
        tool_name: str,
    ) -> None:
        """Persist an `allow for session` grant on the conversation, so the tool
        auto-approves on later turns."""
        fresh = await self.ensure(
            conversation.thread_id,
            agent_tentacle_id=conversation.agent_tentacle_id,
        )
        if tool_name in fresh.allowed_tools:
            return
        async with async_session() as session:
            stored = await session.get(Conversation, fresh.id)
            if stored is None:
                return
            stored.allowed_tools = [*stored.allowed_tools, tool_name]
            await session.commit()

    async def search_messages(
        self,
        conversation_id: UUID7,
        query: str,
        *,
        role: Literal["user", "assistant"] | None = None,
        limit: int = 10,
    ) -> list[ModelMessage]:
        """Conversation messages whose `message_text` contains `query`
        (case-insensitive), ordered chronologically. One polymorphic select spans
        both kinds, so an unfiltered search returns user and assistant hits in a
        single pass; `role` narrows it when set. Tool-call/thinking messages carry
        no `message_text` and never match — reach them via
        `messages_before`/`messages_after`."""
        expressions = [
            ModelMessage["conversation_id"] == str(conversation_id),
            ModelMessage["message_text"].ilike(f"%{query}%"),
        ]
        if role is not None:
            expressions.append(ModelMessage["role"] == role)
        async with async_session() as session:
            rows = await session.list(
                ModelMessage,
                limit=limit,
                order_bys=[ModelMessage["id"]],
                expressions=expressions,
            )
        return list(rows)

    async def messages_before(
        self,
        conversation_id: UUID7,
        anchor_id: UUID7,
        *,
        limit: int = 5,
    ) -> list[ModelMessage]:
        """The `limit` messages immediately preceding `anchor_id` in the
        conversation (all kinds), oldest-first."""
        async with async_session() as session:
            rows = await session.list(
                ModelMessage,
                limit=limit,
                order_bys=[ModelMessage["id"].desc()],
                expressions=[
                    ModelMessage["conversation_id"] == str(conversation_id),
                    ModelMessage["id"] < anchor_id,
                ],
            )
        return list(reversed(rows))

    async def messages_after(
        self,
        conversation_id: UUID7,
        anchor_id: UUID7,
        *,
        limit: int = 5,
    ) -> list[ModelMessage]:
        """The `limit` messages immediately following `anchor_id` in the
        conversation (all kinds), oldest-first."""
        async with async_session() as session:
            rows = await session.list(
                ModelMessage,
                limit=limit,
                order_bys=[ModelMessage["id"]],
                expressions=[
                    ModelMessage["conversation_id"] == str(conversation_id),
                    ModelMessage["id"] > anchor_id,
                ],
            )
        return list(rows)

    async def related_chat_messages(
        self,
        model_message_id: UUID7,
    ) -> list[ThreadMessage]:
        async with async_session() as session:
            message = await session.one_or_none(
                ModelMessage,
                options=[selectinload(ModelMessage["thread_messages"])],
                expressions=[ModelMessage["id"] == model_message_id],
            )
            if message is None:
                return []
            return [
                ThreadMessage.model_validate(thread_message)
                for thread_message in message.thread_messages
            ]

latest_model async

latest_model(conversation_id: UUID7) -> str | None

Read the last reported model without loading the conversation's history.

Source code in octomate/managers/conversation.py

async def latest_model(self, conversation_id: UUID7) -> str | None:
    """Read the last reported model without loading the conversation's history."""
    async with async_session() as session:
        return await session.scalar(
            select(AgentRun["model_name"])
            .where(
                AgentRun["id"].in_(
                    select(ConversationRun["run_id"]).where(
                        ConversationRun["conversation_id"] == conversation_id
                    )
                ),
                AgentRun["model_name"].is_not(None),
            )
            .order_by(AgentRun["started_at"].desc(), AgentRun["id"].desc())
            .limit(1)
        )

ensure async

ensure(
    thread_id: UUID7,
    *,
    agent_tentacle_id: str,
    subagent_id: str = "",
    parent_conversation_id: UUID7 | None = None,
    with_history: bool = True,
) -> Conversation

Resolve the conversation owned by agent_tentacle_id in thread_id, creating it if it does not yet exist. Its conversation.messages is the live model history. The owning agent is part of the identity — two agents in the same thread keep separate conversations — so callers must always supply it. subagent_id narrows to one subagent's own context; empty is the agent's long-lived conversation in the thread. A subagent context names the conversation that spawned it, and only a subagent context may. with_history=False leaves runs and messages unloaded for callers that only need identity, permissions, or the native session's resume id.

Source code in octomate/managers/conversation.py

async def ensure(
    self,
    thread_id: UUID7,
    *,
    agent_tentacle_id: str,
    subagent_id: str = "",
    parent_conversation_id: UUID7 | None = None,
    with_history: bool = True,
) -> Conversation:
    """Resolve the conversation owned by `agent_tentacle_id` in `thread_id`,
    creating it if it does not yet exist. Its `conversation.messages` is the
    live model history. The owning agent is part of the identity — two agents
    in the same thread keep separate conversations — so callers must always
    supply it. `subagent_id` narrows to one subagent's own context; empty is
    the agent's long-lived conversation in the thread. A subagent context
    names the conversation that spawned it, and only a subagent context may.
    `with_history=False` leaves runs and messages unloaded for callers that
    only need identity, permissions, or the native session's resume id."""
    if bool(subagent_id) != (parent_conversation_id is not None):
        raise ValueError(
            "a subagent conversation requires both subagent_id and "
            "parent_conversation_id; a bare conversation takes neither"
        )
    # Serialize first sightings so the loser reads the committed row.
    async with async_session() as session:
        async with self.lock((thread_id, agent_tentacle_id, subagent_id)):
            conversation = await session.one_or_none(
                Conversation,
                options=[
                    lazyload(Conversation["runs"]),
                    lazyload(Conversation["messages"]),
                ],
                expressions=[
                    Conversation["thread_id"] == thread_id,
                    Conversation["agent_tentacle_id"] == agent_tentacle_id,
                    Conversation["subagent_id"] == subagent_id,
                ],
            )
            if conversation is None:
                conversation = Conversation(
                    thread_id=thread_id,
                    agent_tentacle_id=agent_tentacle_id,
                    subagent_id=subagent_id,
                    parent_conversation_id=parent_conversation_id,
                )
                session.add(conversation)
            await session.flush()
            await session.commit()
        if with_history:
            await conversation.runs
            await conversation.messages
        return conversation

get async

get(
    conversation_id: UUID7, *, with_history: bool = True
) -> Conversation

Resolve a conversation by id — one fresh read; raises on an unknown id. This is the by-id path for a run addressed at a pre-ensured conversation (a commissioned accomplice's child context): the caller ensured it and owns any thread/agent validation. with_history=False leaves runs and messages unloaded, as on ensure.

Source code in octomate/managers/conversation.py

async def get(
    self, conversation_id: UUID7, *, with_history: bool = True
) -> Conversation:
    """Resolve a conversation by id — one fresh read; raises on an unknown
    id. This is the by-id path for a run addressed at a pre-ensured
    conversation (a commissioned accomplice's child context): the caller
    ensured it and owns any thread/agent validation. `with_history=False`
    leaves runs and messages unloaded, as on `ensure`."""
    if not with_history:
        options = [
            lazyload(Conversation["runs"]),
            lazyload(Conversation["messages"]),
        ]
    else:
        options = []
    async with async_session() as session:
        conversation = await session.get(
            Conversation,
            conversation_id,
            options=options,
        )
        if conversation is None:
            raise ValueError(f"unknown conversation {conversation_id}")
        if with_history:
            await conversation.runs
            await conversation.messages
    return conversation

link_parent_run async

link_parent_run(
    run_id: str,
    *,
    parent_run_id: str,
    parent_tool_call_id: str | None,
) -> None

Stamp a recorded run's place in the run tree, after the fact. The runner stays ignorant of why it ran; the caller that spawned it links the turn to its own run and tool call once the child returns. A run that recorded nothing has no row to link — not an error.

Source code in octomate/managers/conversation.py

async def link_parent_run(
    self,
    run_id: str,
    *,
    parent_run_id: str,
    parent_tool_call_id: str | None,
) -> None:
    """Stamp a recorded run's place in the run tree, after the fact. The
    runner stays ignorant of why it ran; the caller that spawned it links
    the turn to its own run and tool call once the child returns. A run
    that recorded nothing has no row to link — not an error."""
    async with async_session() as session:
        run = await session.one_or_none(
            AgentRun, expressions=[AgentRun["id"] == run_id]
        )
        if run is None:
            return
        run.parent_run_id = parent_run_id
        run.parent_tool_call_id = parent_tool_call_id
        await session.commit()

subagents async

subagents(
    parent_conversation_id: UUID7,
) -> list[Conversation]

The subagent conversations spawned from parent_conversation_id — the live accomplices a whisper can reach. Rows only; callers resolve a chosen one through ensure.

Source code in octomate/managers/conversation.py

async def subagents(self, parent_conversation_id: UUID7) -> list[Conversation]:
    """The subagent conversations spawned from `parent_conversation_id` — the
    live accomplices a `whisper` can reach. Rows only; callers resolve a
    chosen one through `ensure`."""
    async with async_session() as session:
        rows = await session.list(
            Conversation,
            limit=None,
            expressions=[
                Conversation["parent_conversation_id"] == parent_conversation_id
            ],
        )
    return list(rows)

for_thread async

for_thread(thread_id: UUID7) -> list[Conversation]

The thread's agent conversations, subagents included, each with its runs.

Both message relations here are lazy="selectin", so the plain query would read every model message the thread ever produced twice over — once under the conversation and once under the run that owns it. Both are dropped: a reader after the messages pages through them with messages_page.

Source code in octomate/managers/conversation.py

async def for_thread(self, thread_id: UUID7) -> list[Conversation]:
    """The thread's agent conversations, subagents included, each with its runs.

    Both message relations here are lazy="selectin", so the plain query would read
    every model message the thread ever produced twice over — once under the
    conversation and once under the run that owns it. Both are dropped: a reader
    after the messages pages through them with `messages_page`.
    """
    async with async_session() as session:
        rows = await session.list(
            Conversation,
            limit=None,
            options=[
                noload(Conversation["messages"]),
                selectinload(Conversation["runs"]).noload(AgentRun["messages"]),
            ],
            expressions=[Conversation["thread_id"] == thread_id],
        )
    return list(rows)

messages_page async

messages_page(
    thread_id: UUID7,
    cursor: str | None = None,
    *,
    limit: int,
) -> Page[ModelRequest | ModelResponse]

The model messages of the thread's own conversations, newest first, a page at a time and continued from an Arcanus cursor. Ids are uuid7 and a run's messages are appended in processing order, so id order preserves each transcript's sequence across committed turns.

Each carries the chat rows bound to it — the ones a request was built from, the ones a response was delivered as — so a reader has who said what without the prompt's dressing. A subagent's messages stay out: they are its own conversation's, and the spawn call in its parent's run stands for them.

Source code in octomate/managers/conversation.py

async def messages_page(
    self, thread_id: UUID7, cursor: str | None = None, *, limit: int
) -> Page[ModelRequest | ModelResponse]:
    """The model messages of the thread's own conversations, newest first, a
    page at a time and continued from an Arcanus `cursor`. Ids are uuid7 and a
    run's messages are appended in processing order, so id order preserves
    each transcript's sequence across committed turns.

    Each carries the chat rows bound to it — the ones a request was built from,
    the ones a response was delivered as — so a reader has who said what
    without the prompt's dressing. A subagent's messages stay out: they are its
    own conversation's, and the spawn call in its parent's run stands for them.
    """
    scope = ModelMessage["run_id"].in_(own_runs(thread_id))
    bookmark = None if cursor is None else Cursor[ModelMessage](cursor).bookmark
    after = (
        []
        if bookmark is None or bookmark.expression is None
        else [bookmark.expression]
    )
    newest_first = (ModelMessage["id"].desc(),)
    async with async_session() as session:
        rows = await session.list(
            ModelMessage,
            limit=limit + 1,
            order_bys=newest_first,
            options=[
                selectinload(ModelMessage["thread_messages"]).noload(
                    ThreadMessage["model_messages"]
                )
            ],
            expressions=[scope, *after],
        )
        total = await session.count(ModelMessage, expressions=[scope])
        # The polymorphic read materializes every row as one of the two.
        items = tuple(
            row
            for row in rows[:limit]
            if isinstance(row, ModelRequest | ModelResponse)
        )
        for item in items:
            await item.thread_messages
    last = ModelMessage["id"] < items[-1].id if items else None
    return Page(
        items=items,
        total=total,
        next_cursor=str(
            Cursor[ModelMessage].from_expressions(
                bookmark=last, order_bys=newest_first, limit=limit
            )
        ),
        has_more=len(rows) > limit,
    )

usage async

usage(thread_id: UUID7) -> RequestUsage

The tokens the thread's own conversations spent, every response's usage summed. A subagent's are left to its own conversation, as its messages are. Only the usage column is read: summing a thread must not load its parts.

Source code in octomate/managers/conversation.py

async def usage(self, thread_id: UUID7) -> RequestUsage:
    """The tokens the thread's own conversations spent, every response's usage
    summed. A subagent's are left to its own conversation, as its messages are.
    Only the usage column is read: summing a thread must not load its parts."""
    async with async_session() as session:
        usages = await session.scalars(
            select(ModelResponse["usage"]).where(
                ModelResponse["run_id"].in_(own_runs(thread_id)),
                ModelResponse["usage"].is_not(None),
            )
        )
        return sum(
            map(request_usage_adapter.validate_python, usages),
            RequestUsage(),
        )

thread_id async

thread_id(conversation_id: UUID7) -> UUID7 | None

Source code in octomate/managers/conversation.py

async def thread_id(self, conversation_id: UUID7) -> UUID7 | None:
    async with async_session() as session:
        conversation = await session.get(Conversation, conversation_id)
    if conversation is None:
        return None
    return conversation.thread_id

record_agent_run async

record_agent_run(
    conversation: Conversation,
    run_id: str,
    messages: Sequence[ModelMessage],
    *,
    name: str | None = None,
    model_name: str | None = None,
    permission_mode: AgentPermissionMode | None = None,
    cwd: Path | None = None,
    external_id: str | None = None,
    native_id: str | None = None,
    native_turn_id: str | None = None,
    parent_run_id: str | None = None,
    parent_tool_call_id: str | None = None,
) -> AgentRun | None

Persist a fresh agent run. cwd is the directory the run happened in, None when its caller has none. external_id, when given, updates the conversation's resumable agent session handle in the same commit (external-runtime agents own their own session). The parent pair marks a subagent's turn: the run whose tool call spawned it, and which call it answers.

Source code in octomate/managers/conversation.py

async def record_agent_run(
    self,
    conversation: Conversation,
    run_id: str,
    messages: Sequence[PydanticModelMessage],
    *,
    name: str | None = None,
    model_name: str | None = None,
    permission_mode: AgentPermissionMode | None = None,
    cwd: Path | None = None,
    external_id: str | None = None,
    native_id: str | None = None,
    native_turn_id: str | None = None,
    parent_run_id: str | None = None,
    parent_tool_call_id: str | None = None,
) -> AgentRun | None:
    """Persist a fresh agent run.
    `cwd` is the directory the run happened in, None when its caller has none.
    `external_id`, when given, updates the conversation's resumable agent
    session handle in the same commit (external-runtime agents own their own
    session). The parent pair marks a subagent's turn: the run whose tool
    call spawned it, and which call it answers."""
    if not messages:
        return None
    run = AgentRun(
        id=run_id,
        conversation_id=conversation.id,
        native_id=native_id,
        native_session_id=external_id,
        native_turn_id=native_turn_id,
        name=name,
        model_name=model_name,
        permission_mode=permission_mode,
        cwd=cwd,
        parent_run_id=parent_run_id,
        parent_tool_call_id=parent_tool_call_id,
        started_at=messages[0].timestamp,
        # Shallow `vars(m)` dicts, so pydantic routes each through the blessed
        # `ModelRequest | ModelResponse` union at construction — the raw pydantic-ai
        # dataclass is rejected, not being an instance of our blessed subclass.
        messages=[vars(m) for m in messages],  # pyright: ignore[reportArgumentType]
    )
    return await self.persist_run(
        run, conversation_id=conversation.id, external_id=external_id
    )

driven_run async

driven_run(
    native_id: str, session_id: str, turn_id: str
) -> AgentRun | None

Find the driven owner of a native turn without loading its messages.

Source code in octomate/managers/conversation.py

async def driven_run(
    self, native_id: str, session_id: str, turn_id: str
) -> AgentRun | None:
    """Find the driven owner of a native turn without loading its messages."""
    async with async_session() as session:
        return await session.one_or_none(
            AgentRun,
            expressions=[
                AgentRun["kind"] == "octomate",
                AgentRun["native_id"] == native_id,
                AgentRun["native_session_id"] == session_id,
                AgentRun["native_turn_id"] == turn_id,
            ],
            options=[noload(AgentRun["messages"])],
        )

record_external_run async

record_external_run(
    conversation: Conversation,
    run_id: str,
    messages: Sequence[ModelMessage] = (),
    *,
    name: str | None = None,
    model_name: str | None = None,
    permission_mode: AgentPermissionMode | None = None,
    cwd: Path | None = None,
    native_session_id: str,
    source: str | None = None,
    start_offset: int | None = None,
    end_offset: int | None = None,
    last_line_uuid: str | None = None,
    parent_run_id: str | None = None,
    parent_tool_call_id: str | None = None,
) -> ExternalAgentRun | None

Persist a turn of a native runtime's session as the external variant. cwd is the directory the turn ran in, as the hook or the transcript reported it. native_session_id doubles as the conversation's resumable handle (external_id).

A turn already recorded by the driven runtime keeps its original run and messages. Its runtime identity lets native ingest skip the replay without copying it into the native ledger or saving transcript coordinates on it.

The byte range is what marks a turn finished. A run carrying end_offset was assembled from the transcript and is final, so recording it again — a recovery overlapping a live tail, or a resume re-covering committed bytes — is a no-op returning None rather than a primary-key collision. That guard belongs here, at the durable sink: a caller's in-memory guard only knows the runs it wrote, never one another tailer already committed.

Omitting messages reserves a run identity without changing an existing run. Native hooks use this so the transcript remains the only message writer.

Completing a reserved run fills its messages in transcript order while preserving its identity. Older incomplete imports may contain provisional messages; the completed transcript replaces those in the same commit.

Source code in octomate/managers/conversation.py

async def record_external_run(
    self,
    conversation: Conversation,
    run_id: str,
    messages: Sequence[PydanticModelMessage] = (),
    *,
    name: str | None = None,
    model_name: str | None = None,
    permission_mode: AgentPermissionMode | None = None,
    cwd: Path | None = None,
    native_session_id: str,
    source: str | None = None,
    start_offset: int | None = None,
    end_offset: int | None = None,
    last_line_uuid: str | None = None,
    parent_run_id: str | None = None,
    parent_tool_call_id: str | None = None,
) -> ExternalAgentRun | None:
    """Persist a turn of a native runtime's session as the
    `external` variant. `cwd` is the directory the turn ran in, as the hook or
    the transcript reported it. `native_session_id` doubles as the
    conversation's resumable handle (`external_id`).

    A turn already recorded by the driven runtime keeps its original run and
    messages. Its runtime identity lets native ingest skip the replay without
    copying it into the native ledger or saving transcript coordinates on it.

    The byte range is what marks a turn finished. A run carrying `end_offset` was
    assembled from the transcript and is final, so recording it again — a recovery
    overlapping a live tail, or a resume re-covering committed bytes — is a no-op
    returning `None` rather than a primary-key collision. That guard belongs here,
    at the durable sink: a caller's in-memory guard only knows the runs *it* wrote,
    never one another tailer already committed.

    Omitting messages reserves a run identity without changing an existing run.
    Native hooks use this so the transcript remains the only message writer.

    Completing a reserved run fills its messages in transcript order while
    preserving its identity. Older incomplete imports may contain provisional
    messages; the completed transcript replaces those in the same commit.
    """
    driven = await self.driven_run(
        conversation.agent_tentacle_id, native_session_id, run_id
    )
    if driven is not None:
        return None
    run = ExternalAgentRun(
        id=run_id,
        conversation_id=conversation.id,
        native_id=conversation.agent_tentacle_id,
        native_turn_id=run_id,
        name=name,
        model_name=model_name,
        permission_mode=permission_mode,
        cwd=cwd,
        parent_run_id=parent_run_id,
        parent_tool_call_id=parent_tool_call_id,
        started_at=messages[0].timestamp if messages else None,
        messages=[vars(m) for m in messages],  # pyright: ignore[reportArgumentType]
        native_session_id=native_session_id,
        source=source,
        start_offset=start_offset,
        end_offset=end_offset,
        last_line_uuid=last_line_uuid,
    )
    async with async_session() as session:
        stored = await session.one_or_none(
            AgentRun, expressions=[AgentRun["id"] == run_id]
        )
        if stored is not None:
            if not messages and end_offset is None:
                return stored if isinstance(stored, ExternalAgentRun) else None
            if (
                not isinstance(stored, ExternalAgentRun)
                or stored.end_offset is not None
            ):
                return None
            # `model_messages.run_id` is NOT NULL, so the sketch's messages are
            # deleted rather than left behind by the merged collection.
            sketch = list(stored.messages)
            run = await session.merge(run)
            for message in sketch:
                await session.delete(message)
            owner = await session.get(
                Conversation,
                conversation.id,
                options=[
                    noload(Conversation["runs"]),
                    noload(Conversation["messages"]),
                ],
            )
            if owner is not None:
                owner.external_id = native_session_id
            await session.commit()
            return run
    return await self.persist_run(
        run, conversation_id=conversation.id, external_id=native_session_id
    )

persist_run async

persist_run(
    run: RunT,
    *,
    conversation_id: UUID7,
    external_id: str | None,
) -> RunT

Persist a freshly built run, in the history of the conversation it ran in. external_id, when given, updates the resumable agent-session handle in the same commit.

Only the new run and its messages are added, never the caller's whole conversation graph: re-merging it (prior runs and their message collections) can make SQLAlchemy believe a message was dropped from a stale run collection and null its NOT NULL run_id, raising an IntegrityError.

Source code in octomate/managers/conversation.py

async def persist_run(
    self,
    run: RunT,
    *,
    conversation_id: UUID7,
    external_id: str | None,
) -> RunT:
    """Persist a freshly built run, in the history of the conversation it ran
    in. `external_id`, when given, updates the resumable agent-session handle in
    the same commit.

    Only the new run and its messages are added, never the caller's whole
    conversation graph: re-merging it (prior runs and their message
    collections) can make SQLAlchemy believe a message was dropped from a
    stale run collection and null its NOT NULL `run_id`, raising an
    IntegrityError.
    """
    async with async_session() as session:
        session.add(run)
        reloaded = await session.one_or_none(
            Conversation,
            expressions=[Conversation["id"] == conversation_id],
            options=[
                noload(Conversation["runs"]),
                noload(Conversation["messages"]),
            ],
        )
        if reloaded is not None:
            reloaded.runs.append(run)
            reloaded.external_id = external_id
        await session.commit()
        await run.messages
    return run

fork async

fork(
    source: Conversation,
    target: Conversation,
    *,
    carry_external_id: bool = False,
    external_id: str | None = None,
    transcript: Jsonl | None = None,
    end_offset: int | None = None,
    permission_mode: AgentPermissionMode | None = None,
) -> str | None

Continue source's history in an empty target, in one transaction.

Nothing is copied. The target's history points at the source's runs through conversation_runs, and its thread shows the source thread's ledgers through thread_ledgers, each frozen where it stands now: whatever either side says afterwards is its own. Only the selected conversation receives the independently forked runtime handle.

A native import passes end_offset, where the transcript's last completed turn ends (an uploaded snapshot's size by default), to leave out the turns that end past it and cut the source's ledger before their chat. carry_external_id moves the existing handle instead, for runtimes whose continuation requires it. Return the id of the last run the target's history includes, if any.

Source code in octomate/managers/conversation.py

async def fork(
    self,
    source: Conversation,
    target: Conversation,
    *,
    carry_external_id: bool = False,
    external_id: str | None = None,
    transcript: Jsonl | None = None,
    end_offset: int | None = None,
    permission_mode: AgentPermissionMode | None = None,
) -> str | None:
    """Continue `source`'s history in an empty `target`, in one transaction.

    Nothing is copied. The target's history points at the source's runs through
    `conversation_runs`, and its thread shows the source thread's ledgers through
    `thread_ledgers`, each frozen where it stands now: whatever either side says
    afterwards is its own. Only the selected conversation receives the
    independently forked runtime handle.

    A native import passes `end_offset`, where the transcript's last completed
    turn ends (an uploaded snapshot's size by default), to leave out the turns
    that end past it and cut the source's ledger before their chat.
    `carry_external_id` moves the existing handle instead, for runtimes whose
    continuation requires it. Return the id of the last run the target's history
    includes, if any.
    """
    if source.thread_id == target.thread_id:
        raise ValueError("A fork requires an independent destination thread")
    if transcript is not None and (external_id is None or carry_external_id):
        raise ValueError("An imported transcript requires an independent session")
    if external_id is not None and (
        carry_external_id or external_id == source.external_id
    ):
        raise ValueError(
            "a fork requires a new external id without carrying the source"
        )
    if end_offset is None and transcript is not None:
        end_offset = transcript.size

    async with async_session() as session:
        stored_target = await session.get(
            Conversation,
            target.id,
            options=[
                noload(Conversation["runs"]),
                noload(Conversation["messages"]),
            ],
        )
        if stored_target is None:
            raise ValueError(f"unknown conversation {target.id}")
        target_thread = await session.get(Thread, target.thread_id)
        if target_thread is None:
            raise ValueError(f"unknown thread {target.thread_id}")
        if await session.count(
            Conversation,
            expressions=[
                Conversation["thread_id"] == target.thread_id,
                or_(Conversation["id"] != target.id, Conversation["runs"].any()),
            ],
        ) or await session.count(
            Thread,
            expressions=[
                Thread["id"] == target.thread_id,
                Thread["messages"].any(ThreadMessage["actor_kind"] != "system"),
            ],
        ):
            raise ValueError(
                "fork target already holds history; refusing to splice histories"
            )

        source_thread = await session.get(Thread, source.thread_id)
        if source_thread is None:
            raise ValueError(f"unknown thread {source.thread_id}")
        origin = await session.get(
            Conversation,
            source.id,
            options=[
                noload(Conversation["messages"]),
                selectinload(Conversation["runs"]).noload(AgentRun["messages"]),
            ],
        )
        if origin is None:
            raise ValueError(f"unknown conversation {source.id}")
        dropped: set[str] = set()
        # Without an end offset the cut is the ledger's last message. With one,
        # it falls after the last chat of a completed turn: bound to one, or
        # neither bound to nor filed under an unfinished turn (a native ledger
        # names a turn's prompt by the turn's id), and said no later than the last
        # message the fork keeps.
        cut = select(func.max(ThreadMessage["id"])).where(
            ThreadMessage["thread_id"] == source.thread_id
        )
        if end_offset is not None:
            # A turn of the origin's own session that ends past the offset is
            # unfinished, and so is the subagent work it spawned. The offset is a
            # position in that session's transcript, so it judges no other run.
            unfinished = (
                select(ExternalAgentRun["id"])
                .where(
                    ExternalAgentRun["id"].in_(
                        select(ConversationRun["run_id"]).where(
                            ConversationRun["conversation_id"] == source.id
                        )
                    ),
                    ExternalAgentRun["native_session_id"] == origin.external_id,
                    or_(
                        ExternalAgentRun["end_offset"].is_(None),
                        ExternalAgentRun["end_offset"] > end_offset,
                    ),
                )
                .cte(recursive=True)
            )
            unfinished = unfinished.union(
                select(AgentRun["id"]).where(
                    AgentRun["parent_run_id"] == unfinished.c.id
                )
            )
            dropped = set(await session.scalars(select(unfinished.c.id)))
            # Every run in the history of a conversation in the source thread,
            # but the unfinished ones.
            kept = select(ConversationRun["run_id"]).where(
                ConversationRun["conversation_id"].in_(
                    select(Conversation["id"]).where(
                        Conversation["thread_id"] == source.thread_id
                    )
                ),
                ConversationRun["run_id"].not_in(dropped),
            )
            bound = select(MessageBinding["thread_message_id"])
            cut = cut.where(
                or_(
                    ThreadMessage["id"].in_(
                        bound.where(MessageBinding["run_id"].in_(kept))
                    ),
                    and_(
                        ThreadMessage["id"].not_in(
                            bound.where(MessageBinding["run_id"].in_(dropped))
                        ),
                        or_(
                            ThreadMessage["platform_message_id"].is_(None),
                            ThreadMessage["platform_message_id"].not_in(dropped),
                        ),
                        ThreadMessage["happened_at"]
                        <= select(func.max(ModelMessage["timestamp"]))
                        .where(ModelMessage["run_id"].in_(kept))
                        .scalar_subquery(),
                    ),
                )
            )
        own_cut = await session.scalar(cut)
        ledgers = await session.list(
            ThreadLedger,
            limit=None,
            order_bys=[ThreadLedger["ledger_id"]],
            expressions=[ThreadLedger["thread_id"] == source.thread_id],
        )
        # Every ledger the source shows reaches the fork frozen: its own at the
        # cut, and the ones a fork froze into it where they already stop.
        cuts = {ledger.ledger_id: ledger.cut_message_id for ledger in ledgers}
        cuts[source.thread_id] = own_cut
        session.add_all(
            ThreadLedger(
                thread_id=target.thread_id,
                ledger_id=ledger_id,
                cut_message_id=cut,
            )
            for ledger_id, cut in cuts.items()
            if cut is not None
        )

        await session.execute(
            insert(ConversationRun).from_select(
                ["conversation_id", "run_id"],
                select(literal(target.id), ConversationRun["run_id"]).where(
                    ConversationRun["conversation_id"] == source.id,
                    ConversationRun["run_id"].not_in(dropped),
                ),
            )
        )
        # The cursor stops where the source's did, within what the fork shows.
        shown = max(
            (cut for cut in cuts.values() if cut is not None),
            default=None,
        )
        cursor = source_thread.source_cursor_message_id
        target_thread.source_cursor_message_id = (
            shown if cursor is None or shown is None else min(cursor, shown)
        )
        stored_target.name = origin.name
        stored_target.permission_mode = permission_mode or origin.permission_mode
        stored_target.effort = origin.effort
        stored_target.allowed_tools = list(origin.allowed_tools)
        if transcript is not None:
            session.add(transcript)
            await session.flush()
            stored_target.transcript_file_id = transcript.id
        if carry_external_id:
            stored_target.external_id = origin.external_id
            origin.external_id = None
        elif external_id is not None:
            stored_target.external_id = external_id
        await session.commit()
    if carry_external_id:
        source.external_id = None
    elif external_id is not None:
        target.external_id = external_id
    forked = [run.id for run in origin.runs if run.id not in dropped]
    return forked[-1] if forked else None

store_transcript async

store_transcript(
    uploaded: UploadFile,
    start: int,
    *,
    conversation: Conversation,
    files: FileManager,
    owner_id: UUID7,
) -> FileVariant

Persist new bytes of a native conversation's owner-scoped transcript.

Create the first file and conversation reference in one transaction; subsequent uploads append to that file. Never mutate the caller's conversation.

Source code in octomate/managers/conversation.py

async def store_transcript(
    self,
    uploaded: UploadFile,
    start: int,
    *,
    conversation: Conversation,
    files: FileManager,
    owner_id: UUID7,
) -> FileVariant:
    """Persist new bytes of a native conversation's owner-scoped transcript.

    Create the first file and conversation reference in one transaction;
    subsequent uploads append to that file. Never mutate the caller's conversation.
    """
    async with self.lock(conversation.key):
        current = await self.get(conversation.id, with_history=False)
        previous_id = current.transcript_file_id
        if previous_id is not None:
            return await files.append(
                previous_id, uploaded.file, offset=start, owner_id=owner_id
            )
        if start != 0:
            raise ValueError("The first transcript upload must start at zero")
        async with (
            files.upload(uploaded, owner_id=owner_id) as transcript,
            async_session() as session,
        ):
            stored = await session.get(Conversation, conversation.id)
            if stored is None:
                raise ValueError(f"unknown conversation {conversation.id}")
            session.add(transcript)
            await session.flush()
            stored.transcript_file_id = transcript.id
            await session.commit()
        return transcript

set_external_id async

set_external_id(
    conversation: Conversation, external_id: str
) -> None

Bind a conversation to its replacement runtime session.

Source code in octomate/managers/conversation.py

async def set_external_id(
    self, conversation: Conversation, external_id: str
) -> None:
    """Bind a conversation to its replacement runtime session."""
    async with async_session() as session:
        stored = await session.get(Conversation, conversation.id)
        if stored is None:
            raise ValueError(f"unknown conversation {conversation.id}")
        stored.external_id = external_id
        await session.commit()
    conversation.external_id = external_id

set_permission_mode async

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

Store the approval posture this conversation's agent works under.

Callers selecting a mode validate against the running agent's catalog. Native observers also store modes that are no longer selectable. None clears the selection and lets the agent's configured default decide.

This method only persists the selection. User changes go through the owning agent tentacle, which also updates its live runtime when supported.

Source code in octomate/managers/conversation.py

async def set_permission_mode(
    self,
    conversation: Conversation,
    mode: AgentPermissionMode | None,
) -> Conversation:
    """Store the approval posture this conversation's agent works under.

    Callers selecting a mode validate against the running agent's catalog.
    Native observers also store modes that are no longer selectable. None clears
    the selection and lets the agent's configured default decide.

    This method only persists the selection. User changes go through the
    owning agent tentacle, which also updates its live runtime when supported.
    """
    async with async_session() as session:
        stored = await session.get(Conversation, conversation.id)
        if stored is None:
            raise ValueError(f"unknown conversation {conversation.id}")
        stored.permission_mode = mode
        await session.commit()
    conversation.permission_mode = mode
    return conversation

set_effort async

set_effort(
    conversation: Conversation, effort: str | None
) -> Conversation

Store the reasoning effort this conversation's runs ask for.

Callers validate the level against the route the conversation runs on; None clears it and lets the runtime's own default decide.

Source code in octomate/managers/conversation.py

async def set_effort(
    self, conversation: Conversation, effort: str | None
) -> Conversation:
    """Store the reasoning effort this conversation's runs ask for.

    Callers validate the level against the route the conversation runs on;
    None clears it and lets the runtime's own default decide.
    """
    async with async_session() as session:
        stored = await session.get(Conversation, conversation.id)
        if stored is None:
            raise ValueError(f"unknown conversation {conversation.id}")
        stored.effort = effort
        await session.commit()
    conversation.effort = effort
    return conversation

set_name async

set_name(
    conversation: Conversation, name: str
) -> Conversation

Store the name the runtime running this session grabbed for itself.

Only a runtime that names its own sessions has one to store, and it revises it as the session goes on, so this overwrites. A name with nothing in it is not one, and leaves the conversation as it was.

Source code in octomate/managers/conversation.py

async def set_name(self, conversation: Conversation, name: str) -> Conversation:
    """Store the name the runtime running this session grabbed for itself.

    Only a runtime that names its own sessions has one to store, and it revises
    it as the session goes on, so this overwrites. A name with nothing in it is
    not one, and leaves the conversation as it was.
    """
    named = name.strip()
    if not named or conversation.name == named:
        return conversation
    async with async_session() as session:
        stored = await session.get(Conversation, conversation.id)
        if stored is None:
            raise ValueError(f"unknown conversation {conversation.id}")
        stored.name = named
        await session.commit()
    conversation.name = named
    return conversation

grant_session_tool async

grant_session_tool(
    conversation: Conversation, tool_name: str
) -> None

Persist an allow for session grant on the conversation, so the tool auto-approves on later turns.

Source code in octomate/managers/conversation.py

async def grant_session_tool(
    self,
    conversation: Conversation,
    tool_name: str,
) -> None:
    """Persist an `allow for session` grant on the conversation, so the tool
    auto-approves on later turns."""
    fresh = await self.ensure(
        conversation.thread_id,
        agent_tentacle_id=conversation.agent_tentacle_id,
    )
    if tool_name in fresh.allowed_tools:
        return
    async with async_session() as session:
        stored = await session.get(Conversation, fresh.id)
        if stored is None:
            return
        stored.allowed_tools = [*stored.allowed_tools, tool_name]
        await session.commit()

search_messages async

search_messages(
    conversation_id: UUID7,
    query: str,
    *,
    role: Literal["user", "assistant"] | None = None,
    limit: int = 10,
) -> list[ModelMessage]

Conversation messages whose message_text contains query (case-insensitive), ordered chronologically. One polymorphic select spans both kinds, so an unfiltered search returns user and assistant hits in a single pass; role narrows it when set. Tool-call/thinking messages carry no message_text and never match — reach them via messages_before/messages_after.

Source code in octomate/managers/conversation.py

async def search_messages(
    self,
    conversation_id: UUID7,
    query: str,
    *,
    role: Literal["user", "assistant"] | None = None,
    limit: int = 10,
) -> list[ModelMessage]:
    """Conversation messages whose `message_text` contains `query`
    (case-insensitive), ordered chronologically. One polymorphic select spans
    both kinds, so an unfiltered search returns user and assistant hits in a
    single pass; `role` narrows it when set. Tool-call/thinking messages carry
    no `message_text` and never match — reach them via
    `messages_before`/`messages_after`."""
    expressions = [
        ModelMessage["conversation_id"] == str(conversation_id),
        ModelMessage["message_text"].ilike(f"%{query}%"),
    ]
    if role is not None:
        expressions.append(ModelMessage["role"] == role)
    async with async_session() as session:
        rows = await session.list(
            ModelMessage,
            limit=limit,
            order_bys=[ModelMessage["id"]],
            expressions=expressions,
        )
    return list(rows)

messages_before async

messages_before(
    conversation_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ModelMessage]

The limit messages immediately preceding anchor_id in the conversation (all kinds), oldest-first.

Source code in octomate/managers/conversation.py

async def messages_before(
    self,
    conversation_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ModelMessage]:
    """The `limit` messages immediately preceding `anchor_id` in the
    conversation (all kinds), oldest-first."""
    async with async_session() as session:
        rows = await session.list(
            ModelMessage,
            limit=limit,
            order_bys=[ModelMessage["id"].desc()],
            expressions=[
                ModelMessage["conversation_id"] == str(conversation_id),
                ModelMessage["id"] < anchor_id,
            ],
        )
    return list(reversed(rows))

messages_after async

messages_after(
    conversation_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ModelMessage]

The limit messages immediately following anchor_id in the conversation (all kinds), oldest-first.

Source code in octomate/managers/conversation.py

async def messages_after(
    self,
    conversation_id: UUID7,
    anchor_id: UUID7,
    *,
    limit: int = 5,
) -> list[ModelMessage]:
    """The `limit` messages immediately following `anchor_id` in the
    conversation (all kinds), oldest-first."""
    async with async_session() as session:
        rows = await session.list(
            ModelMessage,
            limit=limit,
            order_bys=[ModelMessage["id"]],
            expressions=[
                ModelMessage["conversation_id"] == str(conversation_id),
                ModelMessage["id"] > anchor_id,
            ],
        )
    return list(rows)

related_chat_messages async

related_chat_messages(
    model_message_id: UUID7,
) -> list[ThreadMessage]

Source code in octomate/managers/conversation.py

async def related_chat_messages(
    self,
    model_message_id: UUID7,
) -> list[ThreadMessage]:
    async with async_session() as session:
        message = await session.one_or_none(
            ModelMessage,
            options=[selectinload(ModelMessage["thread_messages"])],
            expressions=[ModelMessage["id"] == model_message_id],
        )
        if message is None:
            return []
        return [
            ThreadMessage.model_validate(thread_message)
            for thread_message in message.thread_messages
        ]

own_runs

own_runs(thread_id: UUID7) -> Select[tuple[str]]

The runs in the history of the thread's own conversations — its agents', not the subagents' those agents spawned. A fork's history holds the runs it was forked with, so those are its own too.

Source code in octomate/managers/conversation.py

def own_runs(thread_id: UUID7) -> Select[tuple[str]]:
    """The runs in the history of the thread's own conversations — its agents', not
    the subagents' those agents spawned. A fork's history holds the runs it was
    forked with, so those are its own too."""
    return (
        select(ConversationRun["run_id"])
        .join(Conversation, Conversation["id"] == ConversationRun["conversation_id"])
        .where(
            Conversation["thread_id"] == thread_id, Conversation["subagent_id"] == ""
        )
    )

octomate.managers.deferred

Persistence of deferred-action batches — the questions and approvals a run parks for a human — and their resolution from replies.

DeferredActionManager

Bases: Manager

Persists deferred-action batches and resolves them from user replies.

Source code in octomate/managers/deferred.py

class DeferredActionManager(Manager):
    """Persists deferred-action batches and resolves them from user replies."""

    async def create_batch(
        self,
        *,
        conversation: Conversation,
        agent_tentacle_id: str,
        run_name: str | None,
        source_address: ChannelAddress,
        target_address: ChannelAddress,
        target_mode: ResponseTargetMode,
        decision: SummonDecision | None,
        requests: DeferredToolRequests,
        response_mode: DeferredResponseMode,
        batch_id: UUID7 | None = None,
    ) -> DeferredActionBatch:
        with deferred_logfire.span("deferred.create_batch", run_name=run_name) as span:
            actions = DeferredActionCollection.validate_python(requests)
            questions = [
                action for action in actions if isinstance(action, DeferredQuestion)
            ]
            approvals = [
                action for action in actions if isinstance(action, DeferredApproval)
            ]
            batch = DeferredActionBatch(
                id=batch_id or uuid7(),
                response_mode=response_mode,
                conversation_id=conversation.id,
                agent_tentacle_id=agent_tentacle_id,
                run_name=run_name,
                source_address=source_address,
                target_address=target_address,
                target_mode=target_mode,
                decision=decision,
                requests=requests,
                questions=RelationCollection(questions),
                approvals=RelationCollection(approvals),
            )

            span.set_attribute("batch_id", str(batch.id))
            span.set_attribute("questions", len(questions))
            span.set_attribute("approvals", len(approvals))

            async with async_session() as session:
                session.add(batch)
                await session.commit()
                await batch.questions
                await batch.approvals

            async with async_session() as session:
                return await session.one(DeferredActionBatch, id=batch.id)

    async def pending_for_thread(
        self,
        thread_id: UUID7,
    ) -> list[DeferredActionBatch]:
        """The thread's unanswered batches, oldest first, so a channel can
        re-present waiting questions and approvals when the thread is reloaded."""
        async with async_session() as session:
            conversations = await session.list(
                Conversation,
                # Only the ids are wanted. Both of these are lazy="selectin", so
                # without the suppression this walks every run and every model
                # message in the thread to collect a handful of uuids.
                options=[
                    noload(Conversation["messages"]),
                    noload(Conversation["runs"]),
                ],
                expressions=[Conversation["thread_id"] == thread_id],
            )
            conversation_ids = [conversation.id for conversation in conversations]
            if not conversation_ids:
                return []
            # questions/approvals are lazy="selectin": the list query loads them
            # in one batched pass, so no per-row loading is needed here.
            batches = await session.list(
                DeferredActionBatch,
                order_bys=[DeferredActionBatch["id"]],
                expressions=[
                    DeferredActionBatch["conversation_id"].in_(conversation_ids),
                    DeferredActionBatch["status"] == "pending",
                ],
            )
        return list(batches)

    async def get_batch(
        self,
        batch_id: UUID7,
    ) -> DeferredActionBatch:
        async with async_session() as session:
            batch = await session.get(DeferredActionBatch, batch_id)
            if batch is None:
                raise ValueError(f"unknown deferred action batch {batch_id}")
            await batch.questions
            await batch.approvals
            return batch

    async def resolve_batch(
        self,
        awake: DeferredActionBatchResponse,
    ) -> DeferredActionBatch:
        with deferred_logfire.span(
            "deferred.resolve_batch",
            batch_id=str(awake.batch_id),
            answers=len(awake.answers),
            approvals=len(awake.approvals),
        ) as span:
            async with async_session() as session:
                batch = await session.get(DeferredActionBatch, awake.batch_id)
                if batch is None:
                    raise ValueError(f"unknown deferred action batch {awake.batch_id}")
                await batch.questions
                await batch.approvals
                if batch.status in {"completed", "resuming"}:
                    span.set_attribute("batch.status", batch.status)
                    span.set_attribute("batch.completed", batch.completed)
                    return batch
                questions_by_id = {action.id: action for action in batch.questions}
                approvals_by_id = {action.id: action for action in batch.approvals}
                now = datetime.now(UTC)

                for action_id, answer in awake.answers.items():
                    action = questions_by_id.get(action_id)
                    if action is None:
                        raise ValueError(
                            f"unknown deferred action {action_id} in batch "
                            f"{awake.batch_id}"
                        )
                    if action.status != "pending":
                        continue
                    action.status = "answered"
                    action.result = answer
                    action.responder_id = awake.responder_id or None
                    action.resolved_at = now
                    action.updated_at = now

                for action_id, approved in awake.approvals.items():
                    action = approvals_by_id.get(action_id)
                    if action is None:
                        raise ValueError(
                            f"unknown deferred action {action_id} in batch "
                            f"{awake.batch_id}"
                        )
                    if action.status != "pending":
                        continue
                    action.status = "approved" if approved else "denied"
                    action.result = approved
                    action.metadata = {
                        **action.metadata,
                        "allow_session": awake.allow_session,
                    }
                    action.responder_id = awake.responder_id or None
                    action.resolved_at = now
                    action.updated_at = now

                if batch.completed:
                    batch.status = "resolved"
                batch.updated_at = now
                await session.commit()
                span.set_attribute("batch.status", batch.status)
                span.set_attribute("batch.completed", batch.completed)
                return batch

    async def mark_action_presented(
        self,
        action_id: UUID7,
        platform_message_id: str | None,
    ) -> None:
        if not platform_message_id:
            return
        async with async_session() as session:
            action = await session.get(DeferredAction, action_id)
            if action is None:
                return
            action.platform_message_id = platform_message_id
            action.updated_at = datetime.now(UTC)
            await session.commit()

    async def mark_batch(
        self,
        batch_id: UUID7,
        status: DeferredBatchStatus,
        *,
        completed: bool = False,
    ) -> None:
        async with async_session() as session:
            batch = await session.get(DeferredActionBatch, batch_id)
            if batch is None:
                return
            now = datetime.now(UTC)
            batch.status = status
            batch.updated_at = now
            if completed:
                batch.completed_at = now
            await session.commit()

    async def expire_live(self) -> None:
        """Expire every live request still pending. Its waiter lived in the process
        that asked, so after a restart a reply has nowhere to go; expired, the reply
        is refused rather than accepted and lost."""
        async with async_session() as session:
            await session.execute(
                update(DeferredActionBatch)
                .where(
                    DeferredActionBatch["response_mode"] == "live",
                    DeferredActionBatch["status"] == "pending",
                )
                .values(status="expired", updated_at=datetime.now(UTC))
            )
            await session.commit()

create_batch async

create_batch(
    *,
    conversation: Conversation,
    agent_tentacle_id: str,
    run_name: str | None,
    source_address: ChannelAddress,
    target_address: ChannelAddress,
    target_mode: ResponseTargetMode,
    decision: SummonDecision | None,
    requests: DeferredToolRequests,
    response_mode: DeferredResponseMode,
    batch_id: UUID7 | None = None,
) -> DeferredActionBatch

Source code in octomate/managers/deferred.py

async def create_batch(
    self,
    *,
    conversation: Conversation,
    agent_tentacle_id: str,
    run_name: str | None,
    source_address: ChannelAddress,
    target_address: ChannelAddress,
    target_mode: ResponseTargetMode,
    decision: SummonDecision | None,
    requests: DeferredToolRequests,
    response_mode: DeferredResponseMode,
    batch_id: UUID7 | None = None,
) -> DeferredActionBatch:
    with deferred_logfire.span("deferred.create_batch", run_name=run_name) as span:
        actions = DeferredActionCollection.validate_python(requests)
        questions = [
            action for action in actions if isinstance(action, DeferredQuestion)
        ]
        approvals = [
            action for action in actions if isinstance(action, DeferredApproval)
        ]
        batch = DeferredActionBatch(
            id=batch_id or uuid7(),
            response_mode=response_mode,
            conversation_id=conversation.id,
            agent_tentacle_id=agent_tentacle_id,
            run_name=run_name,
            source_address=source_address,
            target_address=target_address,
            target_mode=target_mode,
            decision=decision,
            requests=requests,
            questions=RelationCollection(questions),
            approvals=RelationCollection(approvals),
        )

        span.set_attribute("batch_id", str(batch.id))
        span.set_attribute("questions", len(questions))
        span.set_attribute("approvals", len(approvals))

        async with async_session() as session:
            session.add(batch)
            await session.commit()
            await batch.questions
            await batch.approvals

        async with async_session() as session:
            return await session.one(DeferredActionBatch, id=batch.id)

pending_for_thread async

pending_for_thread(
    thread_id: UUID7,
) -> list[DeferredActionBatch]

The thread's unanswered batches, oldest first, so a channel can re-present waiting questions and approvals when the thread is reloaded.

Source code in octomate/managers/deferred.py

async def pending_for_thread(
    self,
    thread_id: UUID7,
) -> list[DeferredActionBatch]:
    """The thread's unanswered batches, oldest first, so a channel can
    re-present waiting questions and approvals when the thread is reloaded."""
    async with async_session() as session:
        conversations = await session.list(
            Conversation,
            # Only the ids are wanted. Both of these are lazy="selectin", so
            # without the suppression this walks every run and every model
            # message in the thread to collect a handful of uuids.
            options=[
                noload(Conversation["messages"]),
                noload(Conversation["runs"]),
            ],
            expressions=[Conversation["thread_id"] == thread_id],
        )
        conversation_ids = [conversation.id for conversation in conversations]
        if not conversation_ids:
            return []
        # questions/approvals are lazy="selectin": the list query loads them
        # in one batched pass, so no per-row loading is needed here.
        batches = await session.list(
            DeferredActionBatch,
            order_bys=[DeferredActionBatch["id"]],
            expressions=[
                DeferredActionBatch["conversation_id"].in_(conversation_ids),
                DeferredActionBatch["status"] == "pending",
            ],
        )
    return list(batches)

get_batch async

get_batch(batch_id: UUID7) -> DeferredActionBatch

Source code in octomate/managers/deferred.py

async def get_batch(
    self,
    batch_id: UUID7,
) -> DeferredActionBatch:
    async with async_session() as session:
        batch = await session.get(DeferredActionBatch, batch_id)
        if batch is None:
            raise ValueError(f"unknown deferred action batch {batch_id}")
        await batch.questions
        await batch.approvals
        return batch

resolve_batch async

resolve_batch(
    awake: DeferredActionBatchResponse,
) -> DeferredActionBatch

Source code in octomate/managers/deferred.py

async def resolve_batch(
    self,
    awake: DeferredActionBatchResponse,
) -> DeferredActionBatch:
    with deferred_logfire.span(
        "deferred.resolve_batch",
        batch_id=str(awake.batch_id),
        answers=len(awake.answers),
        approvals=len(awake.approvals),
    ) as span:
        async with async_session() as session:
            batch = await session.get(DeferredActionBatch, awake.batch_id)
            if batch is None:
                raise ValueError(f"unknown deferred action batch {awake.batch_id}")
            await batch.questions
            await batch.approvals
            if batch.status in {"completed", "resuming"}:
                span.set_attribute("batch.status", batch.status)
                span.set_attribute("batch.completed", batch.completed)
                return batch
            questions_by_id = {action.id: action for action in batch.questions}
            approvals_by_id = {action.id: action for action in batch.approvals}
            now = datetime.now(UTC)

            for action_id, answer in awake.answers.items():
                action = questions_by_id.get(action_id)
                if action is None:
                    raise ValueError(
                        f"unknown deferred action {action_id} in batch "
                        f"{awake.batch_id}"
                    )
                if action.status != "pending":
                    continue
                action.status = "answered"
                action.result = answer
                action.responder_id = awake.responder_id or None
                action.resolved_at = now
                action.updated_at = now

            for action_id, approved in awake.approvals.items():
                action = approvals_by_id.get(action_id)
                if action is None:
                    raise ValueError(
                        f"unknown deferred action {action_id} in batch "
                        f"{awake.batch_id}"
                    )
                if action.status != "pending":
                    continue
                action.status = "approved" if approved else "denied"
                action.result = approved
                action.metadata = {
                    **action.metadata,
                    "allow_session": awake.allow_session,
                }
                action.responder_id = awake.responder_id or None
                action.resolved_at = now
                action.updated_at = now

            if batch.completed:
                batch.status = "resolved"
            batch.updated_at = now
            await session.commit()
            span.set_attribute("batch.status", batch.status)
            span.set_attribute("batch.completed", batch.completed)
            return batch

mark_action_presented async

mark_action_presented(
    action_id: UUID7, platform_message_id: str | None
) -> None

Source code in octomate/managers/deferred.py

async def mark_action_presented(
    self,
    action_id: UUID7,
    platform_message_id: str | None,
) -> None:
    if not platform_message_id:
        return
    async with async_session() as session:
        action = await session.get(DeferredAction, action_id)
        if action is None:
            return
        action.platform_message_id = platform_message_id
        action.updated_at = datetime.now(UTC)
        await session.commit()

mark_batch async

mark_batch(
    batch_id: UUID7,
    status: DeferredBatchStatus,
    *,
    completed: bool = False,
) -> None

Source code in octomate/managers/deferred.py

async def mark_batch(
    self,
    batch_id: UUID7,
    status: DeferredBatchStatus,
    *,
    completed: bool = False,
) -> None:
    async with async_session() as session:
        batch = await session.get(DeferredActionBatch, batch_id)
        if batch is None:
            return
        now = datetime.now(UTC)
        batch.status = status
        batch.updated_at = now
        if completed:
            batch.completed_at = now
        await session.commit()

expire_live async

expire_live() -> None

Expire every live request still pending. Its waiter lived in the process that asked, so after a restart a reply has nowhere to go; expired, the reply is refused rather than accepted and lost.

Source code in octomate/managers/deferred.py

async def expire_live(self) -> None:
    """Expire every live request still pending. Its waiter lived in the process
    that asked, so after a restart a reply has nowhere to go; expired, the reply
    is refused rather than accepted and lost."""
    async with async_session() as session:
        await session.execute(
            update(DeferredActionBatch)
            .where(
                DeferredActionBatch["response_mode"] == "live",
                DeferredActionBatch["status"] == "pending",
            )
            .values(status="expired", updated_at=datetime.now(UTC))
        )
        await session.commit()

octomate.managers.gateway

Gateway policy: where a conversation can go, decided the same way for every agent.

OctomateSession is one turn's gateway — the route and destination resolution, the validation of what a spell names, and the typed decision it records for the reflex graph to act on after the turn. It speaks no pydantic-ai: the Inkling capability translates its refusals into ModelRetry and its teleport into a deferral, and any other runtime's adapter translates them into its own tool errors, so every runtime meets the same policy through its own tool mechanism. GatewayManager is the in-process registry an external runtime's tool call resolves its driving turn's session from.

PrivateBlocker module-attribute

PrivateBlocker = Literal[
    "no_surface", "already_private", "no_user"
]

PRIVATE_REFUSALS module-attribute

PRIVATE_REFUSALS: dict[PrivateBlocker, str] = {
    "no_surface": "This channel has no direct messages.",
    "already_private": "This conversation is already that user's direct messages, so there is nowhere to move it to.",
    "no_user": "This run has no single user whose direct messages could be opened.",
}

GatewayRefusal

Bases: Exception

A spell that cannot proceed as named, with the sentence that teaches the caller what to name instead. Neutral on purpose: the Inkling capability re-raises it as ModelRetry, and an MCP surface returns it as a tool error the model corrects from — one refusal, worded once, for every runtime.

Source code in octomate/managers/gateway.py

class GatewayRefusal(Exception):
    """A spell that cannot proceed as named, with the sentence that teaches the
    caller what to name instead. Neutral on purpose: the Inkling capability
    re-raises it as `ModelRetry`, and an MCP surface returns it as a tool error
    the model corrects from — one refusal, worded once, for every runtime."""

OctomateSession dataclass

One turn's gateway: the policy the spells share, and the decision they record.

Built per turn by whoever drives one — the react node for a driven agent — and read again when the turn ends: a deciding spell stores its typed decision here, and the reflex graph performs the move. The spells themselves are projected onto each runtime's own tool mechanism by a thin adapter that owns nothing but the translation.

Source code in octomate/managers/gateway.py

@dataclass
class OctomateSession:
    """One turn's gateway: the policy the spells share, and the decision they record.

    Built per turn by whoever drives one — the react node for a driven agent — and
    read again when the turn ends: a deciding spell stores its typed decision here,
    and the reflex graph performs the move. The spells themselves are projected onto
    each runtime's own tool mechanism by a thin adapter that owns nothing but the
    translation.
    """

    # What each channel can route to, keyed by channel id — not one list, because a
    # spell that crosses lands on a channel with its own idea of who runs there.
    # This run's own channel answers `routes`; the rest answer a crossing.
    channel_routes: dict[str, list[AgentRoute]]
    current_agent_id: str
    # Every connected channel, so `surfaces` can be read for any of them and not
    # just this run's own — what a cross-channel move needs.
    channels: dict[str, ChannelTentacle] = field(default_factory=dict)
    # The identity registry and the run's own profile: together they say where else
    # this person is reachable. Both None on a gateway built only to route locally.
    users: UserManager | None = None
    user_profile: UserProfile | None = None
    # Which agents are live, narrowing destinations to channels somebody
    # actually serves; also what the accomplice spells run with.
    agents: dict[str, AgentTentacle] | None = None
    thread_id: UUID7 | None = None
    # Where this run lives; also what `allow_here` and `private_blocked_by` read.
    conversation_address: ChannelAddress | None = None
    # The conversation whose turn this session belongs to — the key an external
    # runtime's tool call presents to `GatewayManager`. None on a gateway built
    # outside a driven turn, which is then never registered.
    conversation_id: UUID7 | None = None
    # An anonymous native session — a terminal run reaching the served gateway with
    # a runtime attribution and nothing else. Policy, never schema: there is no
    # here or sub-thread to land on, every destination is a crossing, nothing here
    # can be summoned, and a teleport or scheme is kicked as its own turn instead of
    # being read after this one.
    native: bool = False
    # What the project spells work with — a `teleport` that binds, `dismiss`, the
    # `projects` facet: the ledger a binding is written to and read from, and the
    # registry and forks. Both None on a gateway built only to route.
    threads: ThreadManager | None = None
    workspaces: WorkspaceManager | None = None
    # The kick's end of the nerve when a kick drives this turn, for a served spell
    # to publish on the turn's own stream; None for a call no kick drives.
    impulses: ObjectSendStream[Impulse] | None = None
    # The host's nerve, which a spell no kick drives publishes on. None on a
    # gateway built only to route, and on one a kick drives through `impulses`.
    nerve: Nerve | None = None
    decision: GatewayDecision | None = field(default=None, init=False)
    # Whether `dismiss` was cast: the thread's workspace goes when the turn ends,
    # the graph's doing once the run is out of it.
    dismissing: bool = field(default=False, init=False)
    # Every route on this run's own channel but the current agent's own — the info
    # shared with the agent to decide where to go, and what a spell landing here
    # validates a chosen route against. A crossing validates against its own.
    other_routes: list[AgentRoute] = field(init=False, repr=False)

    # Discovery is lazy and reused for this gateway session.
    destination_cache: list[ChannelAddress] | None = field(
        default=None, init=False, repr=False
    )

    def __post_init__(self) -> None:
        address = self.conversation_address
        here = (
            self.channel_routes.get(address.channel_tentacle_id, [])
            if address is not None
            else []
        )
        self.other_routes = [
            route for route in here if route.agent_id != self.current_agent_id
        ]

    async def publish(
        self, address: ChannelAddress, event: WireEvent, *, private: bool = False
    ) -> None:
        """Put what a served spell did on the nerve, heard on the thread at
        `address`: on the kick's stream when one drives this turn, so the thread
        the kick was asked on hears it too unless it is `private`, else on that
        thread alone."""
        impulse = Impulse(event, address, private=private)
        if self.impulses is not None:
            await self.impulses.send(impulse)
            return
        if self.nerve is None:
            raise RuntimeError("a spell no kick drives publishes on the host's nerve")
        self.nerve.deliver(ChannelTopic(address.channel_tentacle_id), impulse)
        self.nerve.deliver(ThreadTopic(ThreadKey.from_address(address)), impulse)

    @property
    def allow_here(self) -> bool:
        """Whether `summon here` may hand this conversation over in place.

        False on a group's main channel, where pinning an owner would route every
        gated-in message, from any user, to one agent — and false for a native
        session, which has no conversation Octomate could hand over."""
        if self.native:
            return False
        address = self.conversation_address
        if address is None:
            return True
        return address.chat_type != "group"

    @property
    def private_blocked_by(self) -> PrivateBlocker | None:
        """Why `scheme` has nowhere to land from this run, or None when it does.

        A gateway that knows no channels can reach no direct messages."""
        address = self.conversation_address
        if address is None:
            return "no_user"
        channel = self.channels.get(address.channel_tentacle_id)
        if channel is None or not channel.surfaces.direct_message:
            return "no_surface"
        # Read the surface, not the type: a thread in a Slack DM and a Lark p2p topic
        # are threads that only one person can read, and moving them to "their direct
        # messages" would land beside where they already are, under another owner.
        if not address.shared:
            return "already_private"
        if not address.user_id:
            return "no_user"
        return None

    @property
    def lands_privately(self) -> bool:
        """Whether a landing should open so only its user can read it: this
        conversation is theirs alone, and moving it should not change that."""
        source = self.conversation_address
        return source is not None and not source.shared

    async def destinations(self) -> list[ChannelAddress]:
        """Cache optional address suggestions; execution validates its address independently."""
        if self.destination_cache is not None:
            return self.destination_cache
        linked = (
            await self.users.linked_profiles(self.user_profile)
            if self.users is not None and self.user_profile is not None
            else []
        )
        profiles = {profile.channel_tentacle_id: profile for profile in linked}
        if self.user_profile is not None:
            profiles[self.user_profile.channel_tentacle_id] = self.user_profile
        source = self.conversation_address
        if source is not None and source.channel_tentacle_id not in profiles:
            profiles[source.channel_tentacle_id] = UserProfile(
                channel_tentacle_id=source.channel_tentacle_id,
                channel_user_id=source.user_id,
            )
        addresses: list[ChannelAddress] = []
        if source is not None and self.private_blocked_by is None:
            addresses.append(
                ChannelAddress(source.channel_tentacle_id, "dm", "", source.user_id)
            )
        for channel in self.channels.values():
            profile = profiles.get(channel.id, self.user_profile)
            if profile is None:
                continue
            user_id = channel.thread_user_id(profile)
            if (
                user_id is not None
                and channel.surfaces.direct_message
                and any(
                    self.agents is None or agent_id in self.agents
                    for agent_id in channel.agent_ids
                )
                and (
                    source is None
                    or source.channel_tentacle_id != channel.id
                    or self.private_blocked_by is None
                )
            ):
                addresses.append(ChannelAddress(channel.id, "dm", "", user_id))
            if self.channel_routes.get(channel.id):
                addresses.extend(await channel.suggest_addresses(profile, source))
        suggested: list[ChannelAddress] = []
        for address in dict.fromkeys(addresses):
            channel = self.channels.get(address.channel_tentacle_id)
            if channel is None:
                suggested.append(address)
            elif reason := channel.landing_unavailable:
                # Kept, so an agent asked to go there can say why it cannot.
                suggested.append(replace(address, metadata={"barred": reason}))
            elif address.chat_type != "dm" or channel.accepts_sub_thread(address):
                suggested.append(address)
        self.destination_cache = suggested
        return self.destination_cache

    async def list_addresses(
        self, channel_id: str, inside: str | None = None
    ) -> list[ChannelAddress]:
        """List one level of a connected channel when it is opened, as the
        requester's linked identity.

        A listed address is a suggestion: execution validates the one it is given."""
        channel = self.channels.get(channel_id)
        if channel is None:
            raise GatewayRefusal(f"No connected channel {channel_id!r}.")
        if reason := channel.landing_unavailable:
            raise GatewayRefusal(reason)
        if not self.channel_routes.get(channel.id):
            raise GatewayRefusal("No connected agent serves the destination channel.")
        profile = await self.channel_profile(channel)
        try:
            return await channel.list_addresses(
                profile, inside, private=self.lands_privately
            )
        except ValueError as error:
            raise GatewayRefusal(str(error)) from error

    @property
    async def operations(self) -> ThreadOperations:
        """Offer addresses and channel routes without introducing another target type."""
        suggestions = [
            address
            for address in await self.destinations()
            if address.chat_type == "thread"
            or self.channels[address.channel_tentacle_id].accepts_sub_thread(address)
        ]
        source = self.conversation_address
        # Summon only hands this conversation over where it is; moving it is Teleport's.
        handover = self.summon_unavailable
        summon = (
            OperationAvailability(
                here=source, routes={source.channel_tentacle_id: self.other_routes}
            )
            if handover is None and source is not None
            else OperationAvailability(reason=handover)
        )
        reason = self.teleport_unavailable
        # What each channel offers that can carry this conversation on; a channel
        # with none takes no teleport from another one.
        carriers = {
            channel: [
                route for route in offered if self.is_compatible_agent(route.agent_id)
            ]
            for channel, offered in self.channel_routes.items()
        }
        teleport = [
            one
            for one in suggestions
            if not reason
            and (source is None or not source.shared or one == source)
            and (one == source or carriers.get(one.channel_tentacle_id))
        ]
        # An empty offer closes nothing by itself: browsing can still find a place.
        if not reason and not teleport:
            if source is not None and source.shared:
                reason = (
                    "Shared history may only move into a sub-thread of the current "
                    "chat, and none can start here."
                )
            elif not any(carriers.values()):
                reason = (
                    "No connected agent can import this native history."
                    if self.native
                    else "No connected channel runs this conversation's agent."
                )
        return ThreadOperations(
            source=source,
            teleport=OperationAvailability(
                destinations=teleport, routes=carriers, reason=reason
            ),
            summon=summon,
            barred={
                channel.id: reason
                for channel in self.channels.values()
                if (reason := channel.landing_unavailable)
            },
        )

    def is_compatible_agent(self, agent_id: str) -> bool:
        """A driven source keeps its agent; native history needs a matching importer."""
        if not self.native:
            return agent_id == self.current_agent_id
        agent = self.agents.get(agent_id) if self.agents is not None else None
        return (
            agent is not None
            and agent.native_id == self.current_agent_id
            and type(agent).fork_transcript is not AgentTentacle.fork_transcript
        )

    @property
    def summon_unavailable(self) -> str | None:
        """Why this conversation cannot be handed to another agent where it is."""
        if self.native:
            return "A native session cannot be handed over; teleport it into a thread first."
        if not self.allow_here:
            return "A group's main channel cannot be handed to one agent; teleport into a thread first."
        if not self.other_routes:
            return "No other agent runs on this channel."
        return None

    @property
    def teleport_unavailable(self) -> str | None:
        """Only a stored, owned history can be imported from a native runtime."""
        if self.native and self.thread_id is None:
            return "Native teleport requires an uploaded thread; use Trunkline to select it."
        if not self.native and self.agents is not None:
            agent = self.agents.get(self.current_agent_id)
            if agent is None:
                return "The source agent is not connected."
            if not agent.supports_session_fork:
                return "The source agent does not support independent session forking."
        return None

    async def channel_profile(self, channel: ChannelTentacle) -> UserProfile:
        """The requester's identity usable by this channel, never supplied by a tool."""
        profile = self.user_profile
        source = self.conversation_address
        if (
            profile is None
            and source is not None
            and source.channel_tentacle_id == channel.id
        ):
            profile = UserProfile(
                channel_tentacle_id=channel.id, channel_user_id=source.user_id
            )
        if profile is not None and channel.thread_user_id(profile) is not None:
            return profile
        if profile is not None and self.users is not None:
            linked = await self.users.linked_profiles(profile)
            found = next(
                (one for one in linked if one.channel_tentacle_id == channel.id), None
            )
            if found is not None:
                return found
        raise GatewayRefusal("Link your profile on the destination channel first.")

    async def prepare_address(self, destination: ChannelAddress) -> ChannelAddress:
        """Verify identity, then let the owning channel validate its address."""
        channel = self.channels.get(destination.channel_tentacle_id)
        if channel is None:
            raise GatewayRefusal(
                f"No connected channel {destination.channel_tentacle_id!r}."
            )
        if reason := channel.landing_unavailable:
            raise GatewayRefusal(reason)
        profile = await self.channel_profile(channel)
        if destination.user_id != channel.thread_user_id(profile):
            raise GatewayRefusal("The address does not belong to the requesting user.")
        try:
            return await channel.prepare_address(
                destination, self.conversation_address, private=self.lands_privately
            )
        except ValueError as error:
            raise GatewayRefusal(str(error)) from error

    async def direct_destination(self, target: DirectTarget) -> ChannelAddress:
        """Resolve the requesting user's DM on the selected channel."""
        source = self.conversation_address
        local = target.channel is None or (
            source is not None and target.channel == source.channel_tentacle_id
        )
        if local and (blocker := self.private_blocked_by) is not None:
            raise GatewayRefusal(PRIVATE_REFUSALS[blocker])
        channel_id = target.channel or (source.channel_tentacle_id if source else "")
        channel = self.channels.get(channel_id)
        if channel is None:
            raise GatewayRefusal(f"No connected channel {channel_id!r}.")
        if not channel.surfaces.direct_message:
            raise GatewayRefusal("This channel has no direct messages.")
        if not local and not any(
            self.agents is None or agent_id in self.agents
            for agent_id in channel.agent_ids
        ):
            raise GatewayRefusal("No connected agent serves the destination channel.")
        profile = await self.channel_profile(channel)
        return ChannelAddress(
            channel_tentacle_id=channel.id,
            chat_type="dm",
            chat_id="",
            user_id=profile.channel_user_id,
        )

    def claimed_route(
        self,
        agent_id: str,
        model: str,
        effort: str | None,
        *,
        spell: str,
        offered: list[AgentRoute] | None = None,
    ) -> AgentRoute:
        """The offered route for (agent_id, model), with the requested effort
        validated against its claim — the shared gatekeeping of `summon` and
        `commission`. Both arrive as free strings, so this is where an unrouteable
        pair is caught; callers build from the returned route, never from the args.

        `offered` is the list to check against, defaulting to this channel's. A
        summon that crosses passes the far channel's, since that is who will be
        asked to run it."""
        offered = self.other_routes if offered is None else offered
        route = next(
            (
                route
                for route in offered
                if route.agent_id == agent_id and str(route.model) == model
            ),
            None,
        )
        if route is None:
            available = "\n".join(str(route) for route in offered) or "- (none)"
            raise GatewayRefusal(
                f"Invalid {spell} route (agent_id={agent_id!r}, "
                f"model={model!r}). Copy an agent_id and model exactly "
                f"from one of these routes:\n{available}"
            )
        if effort is not None and effort not in route.claim.efforts:
            raise GatewayRefusal(
                f"Route (agent_id={agent_id!r}, model={model!r}) does "
                f"not accept effort {effort!r}; it claims "
                f"{'/'.join(route.claim.efforts)}. Pick one of those, or omit "
                f"effort."
            )
        return route

    @overload
    async def inspect(
        self,
        reveal: Literal["routes"],
        channel: str | None = None,
        inside: str | None = None,
    ) -> list[AgentRoute]: ...

    @overload
    async def inspect(
        self,
        reveal: Literal["destinations"],
        channel: str | None = None,
        inside: str | None = None,
    ) -> list[ChannelAddress]: ...

    @overload
    async def inspect(
        self,
        reveal: Literal["projects"],
        channel: str | None = None,
        inside: str | None = None,
    ) -> list[ProjectSummary]: ...

    async def inspect(
        self,
        reveal: InspectFacet,
        channel: str | None = None,
        inside: str | None = None,
    ) -> list[AgentRoute] | list[ChannelAddress] | list[ProjectSummary]:
        """Reveal current or selected-channel routes, addresses, or projects.

        Only the requested facet is computed. Destinations with no channel are the
        suggested addresses; with one they are a level of that channel, the top
        or what `inside` holds, as `list_addresses` lists it. Neither limits which
        addresses execution can validate.
        """
        if reveal == "destinations" and channel is not None:
            return await self.list_addresses(channel, inside)
        if inside is not None:
            raise GatewayRefusal(
                "`inside` opens a place listed by `destinations` for a channel."
            )
        if channel is not None:
            if reveal != "routes":
                raise GatewayRefusal(
                    "Only routes and destinations can be inspected for a "
                    "specific channel."
                )
            target = self.channels.get(channel)
            if target is None:
                raise GatewayRefusal(f"No connected channel {channel!r}.")
            await self.channel_profile(target)
            return [
                route
                for route in self.channel_routes.get(channel, [])
                if route.agent_id != self.current_agent_id
            ]
        match reveal:
            case "routes":
                return self.other_routes
            case "destinations":
                return await self.destinations()
            case "projects":
                return await self.projects()

    async def projects(self) -> list[ProjectSummary]:
        """The projects this deployment can work on — a registered user's to see,
        being theirs to bind; a visitor is refused rather than shown nothing."""
        if (
            self.users is None
            or self.user_profile is None
            or await self.users.owner(self.user_profile) is None
        ):
            raise GatewayRefusal(
                "This session speaks for no registered user, and the projects are a "
                "registered user's to see."
            )
        if self.workspaces is None:
            raise RuntimeError("the project facet needs the workspace manager")
        return [
            ProjectSummary(name=project.name, description=project.description)
            for project in self.workspaces.projects.list()
            if project.enabled
        ]

    async def summon(
        self,
        *,
        agent_id: str,
        model: str,
        hint: str,
        reason: str,
        brief: str,
        effort: str | None = None,
    ) -> str:
        """Validate and record a handoff of this conversation, where it is, for the
        graph to perform. Continuing somewhere else is `teleport`'s."""
        if agent_id == self.current_agent_id:
            raise GatewayRefusal(
                f"Cannot summon yourself {self.current_agent_id!r}. "
                f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` to choose a valid route.'
            )
        if refusal := self.summon_unavailable:
            raise GatewayRefusal(refusal)
        route = self.claimed_route(
            agent_id, model, effort, spell="summon", offered=self.other_routes
        )
        self.decision = SummonDecision(
            agent_id=route.agent_id,
            model=route.model,
            effort=effort,
            hint=hint,
            reason=reason,
            brief=brief,
        )
        return f"Summoning {route.agent_id} ({route.model}) to take over here."

    async def teleport(
        self,
        *,
        hint: str,
        destination: ChannelAddress | None = None,
        new_thread: bool = True,
        project: str | None = None,
        ref: str | None = None,
        resume: bool = False,
        prompt: str | None = None,
    ) -> TeleportDecision:
        """Validate and record a teleport decision — the same agent continuing
        somewhere else, its history with it. How the move happens is the graph's:
        the run ends on the decision and the Teleport node performs it.

        A thread about a project carries it: the graph lands the move in a thread
        about the same project, with the work as it stands, so a teleport never
        switches one and naming a project there is refused. A `project` binds a
        thread about none: the thread landed in — a new sub-thread or crossing,
        or this thread when `new_thread` is false, the one case a teleport may
        stay put. The project and the ref are validated here, where a refusal
        reaches the model; the binding itself is the graph's, on the thread that
        turns out to be landed in. Binding is a trust act (a project's own
        `AGENTS.md` reaches the agent as instructions), so it takes a registered
        user; only a thread binds, so staying put is refused for a DM or a group,
        before a mirror is synced for nothing.

        Native history requires a stored thread and a runtime that can import its
        uploaded transcript. Anonymous native tool sessions still cannot teleport."""
        agent_id = self.current_agent_id
        if self.native and (reason := self.teleport_unavailable):
            raise GatewayRefusal(reason)
        destination = destination or self.conversation_address
        if destination is None:
            raise GatewayRefusal("Teleport requires a destination address.")
        if self.native and not new_thread:
            raise GatewayRefusal("Native teleport requires a new destination.")
        if self.native and project is not None:
            raise GatewayRefusal(
                "A native teleport keeps the project its session is about."
            )
        if project is not None and (
            self.users is None
            or self.user_profile is None
            or await self.users.owner(self.user_profile) is None
        ):
            raise GatewayRefusal(
                "This session speaks for no registered user, and binding a thread to "
                "a project is a registered user's act."
            )
        if project is not None:
            if self.thread_id is None or self.threads is None:
                raise RuntimeError(
                    "a teleport naming a project needs the thread this turn is in, "
                    "and the ledger it is written to"
                )
            thread = await self.threads.get(self.thread_id)
            if thread is None:
                raise RuntimeError(f"thread {self.thread_id} vanished")
            current = await thread.project
            if current is not None:
                raise GatewayRefusal(
                    f"This thread is about {current.name!r}, and a teleport carries "
                    "it: the thread you land in is about the same project, with the "
                    "work as it stands. A teleport never switches projects, so omit "
                    "`project`."
                )
            if not new_thread and thread.kind not in ATTRIBUTABLE_KINDS:
                raise GatewayRefusal(
                    f"This conversation is a {thread.kind}, and a DM or a group chat "
                    "outlives every project in it — only a thread binds. Teleport "
                    f"with `new_thread=true` to open one about {project!r}."
                )
        if not new_thread:
            if destination != self.conversation_address:
                raise GatewayRefusal("Only the current conversation can be reused.")
            if project is None:
                raise GatewayRefusal(
                    "A teleport that stays put is you carrying on. Name a project to "
                    "bind this thread to, or somewhere to go."
                )
        else:
            source = self.conversation_address
            if source is not None and source.shared and destination != source:
                raise GatewayRefusal(
                    "Shared history may only move into a sub-thread of the current chat."
                )
            destination = await self.prepare_address(destination)
        if new_thread and (self.native or destination != self.conversation_address):
            route = next(
                (
                    route
                    for route in self.channel_routes.get(
                        destination.channel_tentacle_id, []
                    )
                    if self.is_compatible_agent(route.agent_id)
                ),
                None,
            )
            if route is None:
                channel = self.channels[destination.channel_tentacle_id]
                raise GatewayRefusal(
                    f"{channel.name} has no agent that can import this native history."
                    if self.native
                    else f"{channel.name} does not run {self.current_agent_id}, and "
                    f"a teleport keeps its agent. Stay here, or "
                    f"`{GatewayTool.SUMMON}` an agent it does run."
                )
            agent_id = route.agent_id
        if reason := self.teleport_unavailable:
            raise GatewayRefusal(reason)
        if project is not None:
            if self.workspaces is None:
                raise RuntimeError("a teleport into a project needs the workspaces")
            registered = self.workspaces.projects.get(project)
            if registered is None or not registered.enabled:
                available = sorted(
                    other.name
                    for other in self.workspaces.projects.list()
                    if other.enabled
                )
                raise GatewayRefusal(
                    f"No project called {project!r} is registered here. "
                    f"Available: {', '.join(available) or 'none'}."
                )
            # Ahead of the move, because a thread binds once: a ref that turns out
            # not to resolve would otherwise leave the thread committed to the
            # project with no way to ask again for the branch that was meant.
            mirror = await self.workspaces.mirrors.sync(registered)
            if ref is not None and not await run_git("ls-remote", str(mirror), ref):
                raise GatewayRefusal(
                    f"{registered.name!r} has no {ref!r} to start from. Name a "
                    "branch, tag or commit its mirror has, or omit it for the "
                    "default branch."
                )
        decision = TeleportDecision(
            agent_id=agent_id,
            hint=hint,
            destination=destination,
            new_thread=new_thread,
            project=project,
            ref=ref,
            resume=resume,
            prompt=prompt,
        )
        self.decision = decision
        return decision

    async def dismiss(self) -> str:
        """Record that this thread's workspace may go: its agent has said the work
        in it is done. Validated here, where a refusal reaches the model; the
        release is the graph's, once the turn ends and its work is saved, so no
        tree is pulled out from under a run still in it.

        No registered user is needed, unlike a binding: what a release costs is a
        fork on the next turn, never work. A thread about no project has nothing
        to release — its tree goes with the turn — and a native session runs in a
        directory of its own that Octomate never forked."""
        if self.native:
            raise GatewayRefusal(
                "This session lives in your terminal, in a directory of its own — "
                "Octomate forked nothing for it, and has nothing to release."
            )
        if self.thread_id is None or self.threads is None:
            raise RuntimeError(
                "a dismiss needs the thread this turn is in, and the ledger it is "
                "read from"
            )
        thread = await self.threads.get(self.thread_id)
        if thread is None:
            raise RuntimeError(f"thread {self.thread_id} vanished")
        if await thread.project is None:
            raise GatewayRefusal(
                "This thread is about no project, and its tree is thrown away when "
                "this turn ends — there is nothing to release."
            )
        self.dismissing = True
        return (
            "Releasing this thread's workspace when this turn ends, once its work is "
            "saved to the project's mirror. Nothing is lost — a later message here "
            "forks it afresh from there — but finish up now: your working directory "
            "goes with it."
        )

    async def scheme(
        self,
        *,
        hint: str,
        brief: str,
        destination: SchemeTarget = DIRECT_TARGET,
    ) -> str:
        """Validate and record a scheme decision, returning the sentence the
        scheming agent is told. The move itself is the graph's, after the turn."""
        address = await self.direct_destination(destination)
        self.decision = SchemeDecision(hint=hint, brief=brief, destination=address)
        return f"Taking this to your direct messages on {address.channel_tentacle_id}."

    async def resolve_send(self, destination: SendTarget) -> ChannelAddress | None:
        """The address a send delivers to, or None for this conversation itself.

        Being in their direct messages already stops a `scheme` — nowhere to move
        the conversation to — but never a send: that *is* where it was asked to go,
        so it lands here rather than being refused."""
        if isinstance(destination, HereTarget):
            return None
        source = self.conversation_address
        local = destination.channel is None or (
            source is not None and destination.channel == source.channel_tentacle_id
        )
        if local and self.private_blocked_by == "already_private":
            return None
        return await self.direct_destination(destination)

    async def attach_native_thread(self, session_id: str) -> None:
        """Bind a native call to the thread its own session is filed under, as
        Trunkline's thread operations bind theirs, so a spell it casts acts on that
        history. Only a thread the caller has written to is found: naming another
        session, theirs or anyone's, finds nothing."""
        profile = self.user_profile
        if not self.native or self.threads is None or profile is None:
            raise RuntimeError("only a native session with a ledger binds a thread")
        thread = await self.threads.get(
            ThreadKey(self.current_agent_id, "thread", session_id),
            with_messages=False,
            user_id=profile.user_id,
        )
        if thread is None:
            raise GatewayRefusal(
                "Octomate holds no uploaded history for this session yet. Finish a "
                "turn with Octomate's hooks installed, then teleport again."
            )
        self.thread_id = thread.id
        self.conversation_address = thread.key.address(profile.channel_user_id)

    def thread_operation(self) -> DrivenGatewaySignal:
        """Package a validated user action for the graph's existing-thread entry."""
        if (
            self.thread_id is None
            or self.conversation_address is None
            or self.user_profile is None
            or not isinstance(self.decision, SummonDecision | TeleportDecision)
        ):
            raise ValueError(
                "A thread operation requires a thread, user and validated decision."
            )
        return DrivenGatewaySignal(
            thread_id=self.thread_id,
            source=self.conversation_address,
            agent_id=self.current_agent_id,
            user_profile=self.user_profile,
            decision=self.decision,
        )

    def native_handoff(self) -> NativeGatewaySignal:
        """This native session's recorded decision, packaged to be kicked as its
        own turn. Only a native spell leaves one, so anything else asking is a
        wiring bug, not a refusal a model could correct from."""
        decision = self.decision
        if not self.native or not isinstance(
            decision, SchemeDecision | TeleportDecision
        ):
            raise RuntimeError("only a native teleport or scheme kicks a handoff")
        return NativeGatewaySignal(
            decision=decision,
            agent_id=self.current_agent_id,
            user_profile=self.user_profile,
            # A teleport carries the thread it attached; the rest come from nowhere.
            source=self.conversation_address
            if isinstance(decision, TeleportDecision)
            else ChannelAddress(
                channel_tentacle_id=self.current_agent_id,
                chat_type="dm",
                chat_id="",
                user_id=NATIVE_CHANNEL_USER_ID,
            ),
        )

channel_routes instance-attribute

channel_routes: dict[str, list[AgentRoute]]

current_agent_id instance-attribute

current_agent_id: str

channels class-attribute instance-attribute

channels: dict[str, ChannelTentacle] = field(
    default_factory=dict
)

users class-attribute instance-attribute

users: UserManager | None = None

user_profile class-attribute instance-attribute

user_profile: UserProfile | None = None

agents class-attribute instance-attribute

agents: dict[str, AgentTentacle] | None = None

thread_id class-attribute instance-attribute

thread_id: UUID7 | None = None

conversation_address class-attribute instance-attribute

conversation_address: ChannelAddress | None = None

conversation_id class-attribute instance-attribute

conversation_id: UUID7 | None = None

native class-attribute instance-attribute

native: bool = False

threads class-attribute instance-attribute

threads: ThreadManager | None = None

workspaces class-attribute instance-attribute

workspaces: WorkspaceManager | None = None

impulses class-attribute instance-attribute

impulses: ObjectSendStream[Impulse] | None = None

nerve class-attribute instance-attribute

nerve: Nerve | None = None

decision class-attribute instance-attribute

decision: GatewayDecision | None = field(
    default=None, init=False
)

dismissing class-attribute instance-attribute

dismissing: bool = field(default=False, init=False)

other_routes class-attribute instance-attribute

other_routes: list[AgentRoute] = field(
    init=False, repr=False
)

destination_cache class-attribute instance-attribute

destination_cache: list[ChannelAddress] | None = field(
    default=None, init=False, repr=False
)

allow_here property

allow_here: bool

Whether summon here may hand this conversation over in place.

False on a group's main channel, where pinning an owner would route every gated-in message, from any user, to one agent — and false for a native session, which has no conversation Octomate could hand over.

private_blocked_by property

private_blocked_by: PrivateBlocker | None

Why scheme has nowhere to land from this run, or None when it does.

A gateway that knows no channels can reach no direct messages.

lands_privately property

lands_privately: bool

Whether a landing should open so only its user can read it: this conversation is theirs alone, and moving it should not change that.

operations async property

operations: ThreadOperations

Offer addresses and channel routes without introducing another target type.

summon_unavailable property

summon_unavailable: str | None

Why this conversation cannot be handed to another agent where it is.

teleport_unavailable property

teleport_unavailable: str | None

Only a stored, owned history can be imported from a native runtime.

publish async

publish(
    address: ChannelAddress,
    event: WireEvent,
    *,
    private: bool = False,
) -> None

Put what a served spell did on the nerve, heard on the thread at address: on the kick's stream when one drives this turn, so the thread the kick was asked on hears it too unless it is private, else on that thread alone.

Source code in octomate/managers/gateway.py

async def publish(
    self, address: ChannelAddress, event: WireEvent, *, private: bool = False
) -> None:
    """Put what a served spell did on the nerve, heard on the thread at
    `address`: on the kick's stream when one drives this turn, so the thread
    the kick was asked on hears it too unless it is `private`, else on that
    thread alone."""
    impulse = Impulse(event, address, private=private)
    if self.impulses is not None:
        await self.impulses.send(impulse)
        return
    if self.nerve is None:
        raise RuntimeError("a spell no kick drives publishes on the host's nerve")
    self.nerve.deliver(ChannelTopic(address.channel_tentacle_id), impulse)
    self.nerve.deliver(ThreadTopic(ThreadKey.from_address(address)), impulse)

destinations async

destinations() -> list[ChannelAddress]

Cache optional address suggestions; execution validates its address independently.

Source code in octomate/managers/gateway.py

async def destinations(self) -> list[ChannelAddress]:
    """Cache optional address suggestions; execution validates its address independently."""
    if self.destination_cache is not None:
        return self.destination_cache
    linked = (
        await self.users.linked_profiles(self.user_profile)
        if self.users is not None and self.user_profile is not None
        else []
    )
    profiles = {profile.channel_tentacle_id: profile for profile in linked}
    if self.user_profile is not None:
        profiles[self.user_profile.channel_tentacle_id] = self.user_profile
    source = self.conversation_address
    if source is not None and source.channel_tentacle_id not in profiles:
        profiles[source.channel_tentacle_id] = UserProfile(
            channel_tentacle_id=source.channel_tentacle_id,
            channel_user_id=source.user_id,
        )
    addresses: list[ChannelAddress] = []
    if source is not None and self.private_blocked_by is None:
        addresses.append(
            ChannelAddress(source.channel_tentacle_id, "dm", "", source.user_id)
        )
    for channel in self.channels.values():
        profile = profiles.get(channel.id, self.user_profile)
        if profile is None:
            continue
        user_id = channel.thread_user_id(profile)
        if (
            user_id is not None
            and channel.surfaces.direct_message
            and any(
                self.agents is None or agent_id in self.agents
                for agent_id in channel.agent_ids
            )
            and (
                source is None
                or source.channel_tentacle_id != channel.id
                or self.private_blocked_by is None
            )
        ):
            addresses.append(ChannelAddress(channel.id, "dm", "", user_id))
        if self.channel_routes.get(channel.id):
            addresses.extend(await channel.suggest_addresses(profile, source))
    suggested: list[ChannelAddress] = []
    for address in dict.fromkeys(addresses):
        channel = self.channels.get(address.channel_tentacle_id)
        if channel is None:
            suggested.append(address)
        elif reason := channel.landing_unavailable:
            # Kept, so an agent asked to go there can say why it cannot.
            suggested.append(replace(address, metadata={"barred": reason}))
        elif address.chat_type != "dm" or channel.accepts_sub_thread(address):
            suggested.append(address)
    self.destination_cache = suggested
    return self.destination_cache

list_addresses async

list_addresses(
    channel_id: str, inside: str | None = None
) -> list[ChannelAddress]

List one level of a connected channel when it is opened, as the requester's linked identity.

A listed address is a suggestion: execution validates the one it is given.

Source code in octomate/managers/gateway.py

async def list_addresses(
    self, channel_id: str, inside: str | None = None
) -> list[ChannelAddress]:
    """List one level of a connected channel when it is opened, as the
    requester's linked identity.

    A listed address is a suggestion: execution validates the one it is given."""
    channel = self.channels.get(channel_id)
    if channel is None:
        raise GatewayRefusal(f"No connected channel {channel_id!r}.")
    if reason := channel.landing_unavailable:
        raise GatewayRefusal(reason)
    if not self.channel_routes.get(channel.id):
        raise GatewayRefusal("No connected agent serves the destination channel.")
    profile = await self.channel_profile(channel)
    try:
        return await channel.list_addresses(
            profile, inside, private=self.lands_privately
        )
    except ValueError as error:
        raise GatewayRefusal(str(error)) from error

is_compatible_agent

is_compatible_agent(agent_id: str) -> bool

A driven source keeps its agent; native history needs a matching importer.

Source code in octomate/managers/gateway.py

def is_compatible_agent(self, agent_id: str) -> bool:
    """A driven source keeps its agent; native history needs a matching importer."""
    if not self.native:
        return agent_id == self.current_agent_id
    agent = self.agents.get(agent_id) if self.agents is not None else None
    return (
        agent is not None
        and agent.native_id == self.current_agent_id
        and type(agent).fork_transcript is not AgentTentacle.fork_transcript
    )

channel_profile async

channel_profile(channel: ChannelTentacle) -> UserProfile

The requester's identity usable by this channel, never supplied by a tool.

Source code in octomate/managers/gateway.py

async def channel_profile(self, channel: ChannelTentacle) -> UserProfile:
    """The requester's identity usable by this channel, never supplied by a tool."""
    profile = self.user_profile
    source = self.conversation_address
    if (
        profile is None
        and source is not None
        and source.channel_tentacle_id == channel.id
    ):
        profile = UserProfile(
            channel_tentacle_id=channel.id, channel_user_id=source.user_id
        )
    if profile is not None and channel.thread_user_id(profile) is not None:
        return profile
    if profile is not None and self.users is not None:
        linked = await self.users.linked_profiles(profile)
        found = next(
            (one for one in linked if one.channel_tentacle_id == channel.id), None
        )
        if found is not None:
            return found
    raise GatewayRefusal("Link your profile on the destination channel first.")

prepare_address async

prepare_address(
    destination: ChannelAddress,
) -> ChannelAddress

Verify identity, then let the owning channel validate its address.

Source code in octomate/managers/gateway.py

async def prepare_address(self, destination: ChannelAddress) -> ChannelAddress:
    """Verify identity, then let the owning channel validate its address."""
    channel = self.channels.get(destination.channel_tentacle_id)
    if channel is None:
        raise GatewayRefusal(
            f"No connected channel {destination.channel_tentacle_id!r}."
        )
    if reason := channel.landing_unavailable:
        raise GatewayRefusal(reason)
    profile = await self.channel_profile(channel)
    if destination.user_id != channel.thread_user_id(profile):
        raise GatewayRefusal("The address does not belong to the requesting user.")
    try:
        return await channel.prepare_address(
            destination, self.conversation_address, private=self.lands_privately
        )
    except ValueError as error:
        raise GatewayRefusal(str(error)) from error

direct_destination async

direct_destination(target: DirectTarget) -> ChannelAddress

Resolve the requesting user's DM on the selected channel.

Source code in octomate/managers/gateway.py

async def direct_destination(self, target: DirectTarget) -> ChannelAddress:
    """Resolve the requesting user's DM on the selected channel."""
    source = self.conversation_address
    local = target.channel is None or (
        source is not None and target.channel == source.channel_tentacle_id
    )
    if local and (blocker := self.private_blocked_by) is not None:
        raise GatewayRefusal(PRIVATE_REFUSALS[blocker])
    channel_id = target.channel or (source.channel_tentacle_id if source else "")
    channel = self.channels.get(channel_id)
    if channel is None:
        raise GatewayRefusal(f"No connected channel {channel_id!r}.")
    if not channel.surfaces.direct_message:
        raise GatewayRefusal("This channel has no direct messages.")
    if not local and not any(
        self.agents is None or agent_id in self.agents
        for agent_id in channel.agent_ids
    ):
        raise GatewayRefusal("No connected agent serves the destination channel.")
    profile = await self.channel_profile(channel)
    return ChannelAddress(
        channel_tentacle_id=channel.id,
        chat_type="dm",
        chat_id="",
        user_id=profile.channel_user_id,
    )

claimed_route

claimed_route(
    agent_id: str,
    model: str,
    effort: str | None,
    *,
    spell: str,
    offered: list[AgentRoute] | None = None,
) -> AgentRoute

The offered route for (agent_id, model), with the requested effort validated against its claim — the shared gatekeeping of summon and commission. Both arrive as free strings, so this is where an unrouteable pair is caught; callers build from the returned route, never from the args.

offered is the list to check against, defaulting to this channel's. A summon that crosses passes the far channel's, since that is who will be asked to run it.

Source code in octomate/managers/gateway.py

def claimed_route(
    self,
    agent_id: str,
    model: str,
    effort: str | None,
    *,
    spell: str,
    offered: list[AgentRoute] | None = None,
) -> AgentRoute:
    """The offered route for (agent_id, model), with the requested effort
    validated against its claim — the shared gatekeeping of `summon` and
    `commission`. Both arrive as free strings, so this is where an unrouteable
    pair is caught; callers build from the returned route, never from the args.

    `offered` is the list to check against, defaulting to this channel's. A
    summon that crosses passes the far channel's, since that is who will be
    asked to run it."""
    offered = self.other_routes if offered is None else offered
    route = next(
        (
            route
            for route in offered
            if route.agent_id == agent_id and str(route.model) == model
        ),
        None,
    )
    if route is None:
        available = "\n".join(str(route) for route in offered) or "- (none)"
        raise GatewayRefusal(
            f"Invalid {spell} route (agent_id={agent_id!r}, "
            f"model={model!r}). Copy an agent_id and model exactly "
            f"from one of these routes:\n{available}"
        )
    if effort is not None and effort not in route.claim.efforts:
        raise GatewayRefusal(
            f"Route (agent_id={agent_id!r}, model={model!r}) does "
            f"not accept effort {effort!r}; it claims "
            f"{'/'.join(route.claim.efforts)}. Pick one of those, or omit "
            f"effort."
        )
    return route

inspect async

inspect(
    reveal: Literal["routes"],
    channel: str | None = None,
    inside: str | None = None,
) -> list[AgentRoute]

inspect(
    reveal: Literal["destinations"],
    channel: str | None = None,
    inside: str | None = None,
) -> list[ChannelAddress]

inspect(
    reveal: Literal["projects"],
    channel: str | None = None,
    inside: str | None = None,
) -> list[ProjectSummary]

inspect(
    reveal: InspectFacet,
    channel: str | None = None,
    inside: str | None = None,
) -> (
    list[AgentRoute]
    | list[ChannelAddress]
    | list[ProjectSummary]
)

Reveal current or selected-channel routes, addresses, or projects.

Only the requested facet is computed. Destinations with no channel are the suggested addresses; with one they are a level of that channel, the top or what inside holds, as list_addresses lists it. Neither limits which addresses execution can validate.

Source code in octomate/managers/gateway.py

async def inspect(
    self,
    reveal: InspectFacet,
    channel: str | None = None,
    inside: str | None = None,
) -> list[AgentRoute] | list[ChannelAddress] | list[ProjectSummary]:
    """Reveal current or selected-channel routes, addresses, or projects.

    Only the requested facet is computed. Destinations with no channel are the
    suggested addresses; with one they are a level of that channel, the top
    or what `inside` holds, as `list_addresses` lists it. Neither limits which
    addresses execution can validate.
    """
    if reveal == "destinations" and channel is not None:
        return await self.list_addresses(channel, inside)
    if inside is not None:
        raise GatewayRefusal(
            "`inside` opens a place listed by `destinations` for a channel."
        )
    if channel is not None:
        if reveal != "routes":
            raise GatewayRefusal(
                "Only routes and destinations can be inspected for a "
                "specific channel."
            )
        target = self.channels.get(channel)
        if target is None:
            raise GatewayRefusal(f"No connected channel {channel!r}.")
        await self.channel_profile(target)
        return [
            route
            for route in self.channel_routes.get(channel, [])
            if route.agent_id != self.current_agent_id
        ]
    match reveal:
        case "routes":
            return self.other_routes
        case "destinations":
            return await self.destinations()
        case "projects":
            return await self.projects()

projects async

projects() -> list[ProjectSummary]

The projects this deployment can work on — a registered user's to see, being theirs to bind; a visitor is refused rather than shown nothing.

Source code in octomate/managers/gateway.py

async def projects(self) -> list[ProjectSummary]:
    """The projects this deployment can work on — a registered user's to see,
    being theirs to bind; a visitor is refused rather than shown nothing."""
    if (
        self.users is None
        or self.user_profile is None
        or await self.users.owner(self.user_profile) is None
    ):
        raise GatewayRefusal(
            "This session speaks for no registered user, and the projects are a "
            "registered user's to see."
        )
    if self.workspaces is None:
        raise RuntimeError("the project facet needs the workspace manager")
    return [
        ProjectSummary(name=project.name, description=project.description)
        for project in self.workspaces.projects.list()
        if project.enabled
    ]

summon async

summon(
    *,
    agent_id: str,
    model: str,
    hint: str,
    reason: str,
    brief: str,
    effort: str | None = None,
) -> str

Validate and record a handoff of this conversation, where it is, for the graph to perform. Continuing somewhere else is teleport's.

Source code in octomate/managers/gateway.py

async def summon(
    self,
    *,
    agent_id: str,
    model: str,
    hint: str,
    reason: str,
    brief: str,
    effort: str | None = None,
) -> str:
    """Validate and record a handoff of this conversation, where it is, for the
    graph to perform. Continuing somewhere else is `teleport`'s."""
    if agent_id == self.current_agent_id:
        raise GatewayRefusal(
            f"Cannot summon yourself {self.current_agent_id!r}. "
            f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` to choose a valid route.'
        )
    if refusal := self.summon_unavailable:
        raise GatewayRefusal(refusal)
    route = self.claimed_route(
        agent_id, model, effort, spell="summon", offered=self.other_routes
    )
    self.decision = SummonDecision(
        agent_id=route.agent_id,
        model=route.model,
        effort=effort,
        hint=hint,
        reason=reason,
        brief=brief,
    )
    return f"Summoning {route.agent_id} ({route.model}) to take over here."

teleport async

teleport(
    *,
    hint: str,
    destination: ChannelAddress | None = None,
    new_thread: bool = True,
    project: str | None = None,
    ref: str | None = None,
    resume: bool = False,
    prompt: str | None = None,
) -> TeleportDecision

Validate and record a teleport decision — the same agent continuing somewhere else, its history with it. How the move happens is the graph's: the run ends on the decision and the Teleport node performs it.

A thread about a project carries it: the graph lands the move in a thread about the same project, with the work as it stands, so a teleport never switches one and naming a project there is refused. A project binds a thread about none: the thread landed in — a new sub-thread or crossing, or this thread when new_thread is false, the one case a teleport may stay put. The project and the ref are validated here, where a refusal reaches the model; the binding itself is the graph's, on the thread that turns out to be landed in. Binding is a trust act (a project's own AGENTS.md reaches the agent as instructions), so it takes a registered user; only a thread binds, so staying put is refused for a DM or a group, before a mirror is synced for nothing.

Native history requires a stored thread and a runtime that can import its uploaded transcript. Anonymous native tool sessions still cannot teleport.

Source code in octomate/managers/gateway.py

async def teleport(
    self,
    *,
    hint: str,
    destination: ChannelAddress | None = None,
    new_thread: bool = True,
    project: str | None = None,
    ref: str | None = None,
    resume: bool = False,
    prompt: str | None = None,
) -> TeleportDecision:
    """Validate and record a teleport decision — the same agent continuing
    somewhere else, its history with it. How the move happens is the graph's:
    the run ends on the decision and the Teleport node performs it.

    A thread about a project carries it: the graph lands the move in a thread
    about the same project, with the work as it stands, so a teleport never
    switches one and naming a project there is refused. A `project` binds a
    thread about none: the thread landed in — a new sub-thread or crossing,
    or this thread when `new_thread` is false, the one case a teleport may
    stay put. The project and the ref are validated here, where a refusal
    reaches the model; the binding itself is the graph's, on the thread that
    turns out to be landed in. Binding is a trust act (a project's own
    `AGENTS.md` reaches the agent as instructions), so it takes a registered
    user; only a thread binds, so staying put is refused for a DM or a group,
    before a mirror is synced for nothing.

    Native history requires a stored thread and a runtime that can import its
    uploaded transcript. Anonymous native tool sessions still cannot teleport."""
    agent_id = self.current_agent_id
    if self.native and (reason := self.teleport_unavailable):
        raise GatewayRefusal(reason)
    destination = destination or self.conversation_address
    if destination is None:
        raise GatewayRefusal("Teleport requires a destination address.")
    if self.native and not new_thread:
        raise GatewayRefusal("Native teleport requires a new destination.")
    if self.native and project is not None:
        raise GatewayRefusal(
            "A native teleport keeps the project its session is about."
        )
    if project is not None and (
        self.users is None
        or self.user_profile is None
        or await self.users.owner(self.user_profile) is None
    ):
        raise GatewayRefusal(
            "This session speaks for no registered user, and binding a thread to "
            "a project is a registered user's act."
        )
    if project is not None:
        if self.thread_id is None or self.threads is None:
            raise RuntimeError(
                "a teleport naming a project needs the thread this turn is in, "
                "and the ledger it is written to"
            )
        thread = await self.threads.get(self.thread_id)
        if thread is None:
            raise RuntimeError(f"thread {self.thread_id} vanished")
        current = await thread.project
        if current is not None:
            raise GatewayRefusal(
                f"This thread is about {current.name!r}, and a teleport carries "
                "it: the thread you land in is about the same project, with the "
                "work as it stands. A teleport never switches projects, so omit "
                "`project`."
            )
        if not new_thread and thread.kind not in ATTRIBUTABLE_KINDS:
            raise GatewayRefusal(
                f"This conversation is a {thread.kind}, and a DM or a group chat "
                "outlives every project in it — only a thread binds. Teleport "
                f"with `new_thread=true` to open one about {project!r}."
            )
    if not new_thread:
        if destination != self.conversation_address:
            raise GatewayRefusal("Only the current conversation can be reused.")
        if project is None:
            raise GatewayRefusal(
                "A teleport that stays put is you carrying on. Name a project to "
                "bind this thread to, or somewhere to go."
            )
    else:
        source = self.conversation_address
        if source is not None and source.shared and destination != source:
            raise GatewayRefusal(
                "Shared history may only move into a sub-thread of the current chat."
            )
        destination = await self.prepare_address(destination)
    if new_thread and (self.native or destination != self.conversation_address):
        route = next(
            (
                route
                for route in self.channel_routes.get(
                    destination.channel_tentacle_id, []
                )
                if self.is_compatible_agent(route.agent_id)
            ),
            None,
        )
        if route is None:
            channel = self.channels[destination.channel_tentacle_id]
            raise GatewayRefusal(
                f"{channel.name} has no agent that can import this native history."
                if self.native
                else f"{channel.name} does not run {self.current_agent_id}, and "
                f"a teleport keeps its agent. Stay here, or "
                f"`{GatewayTool.SUMMON}` an agent it does run."
            )
        agent_id = route.agent_id
    if reason := self.teleport_unavailable:
        raise GatewayRefusal(reason)
    if project is not None:
        if self.workspaces is None:
            raise RuntimeError("a teleport into a project needs the workspaces")
        registered = self.workspaces.projects.get(project)
        if registered is None or not registered.enabled:
            available = sorted(
                other.name
                for other in self.workspaces.projects.list()
                if other.enabled
            )
            raise GatewayRefusal(
                f"No project called {project!r} is registered here. "
                f"Available: {', '.join(available) or 'none'}."
            )
        # Ahead of the move, because a thread binds once: a ref that turns out
        # not to resolve would otherwise leave the thread committed to the
        # project with no way to ask again for the branch that was meant.
        mirror = await self.workspaces.mirrors.sync(registered)
        if ref is not None and not await run_git("ls-remote", str(mirror), ref):
            raise GatewayRefusal(
                f"{registered.name!r} has no {ref!r} to start from. Name a "
                "branch, tag or commit its mirror has, or omit it for the "
                "default branch."
            )
    decision = TeleportDecision(
        agent_id=agent_id,
        hint=hint,
        destination=destination,
        new_thread=new_thread,
        project=project,
        ref=ref,
        resume=resume,
        prompt=prompt,
    )
    self.decision = decision
    return decision

dismiss async

dismiss() -> str

Record that this thread's workspace may go: its agent has said the work in it is done. Validated here, where a refusal reaches the model; the release is the graph's, once the turn ends and its work is saved, so no tree is pulled out from under a run still in it.

No registered user is needed, unlike a binding: what a release costs is a fork on the next turn, never work. A thread about no project has nothing to release — its tree goes with the turn — and a native session runs in a directory of its own that Octomate never forked.

Source code in octomate/managers/gateway.py

async def dismiss(self) -> str:
    """Record that this thread's workspace may go: its agent has said the work
    in it is done. Validated here, where a refusal reaches the model; the
    release is the graph's, once the turn ends and its work is saved, so no
    tree is pulled out from under a run still in it.

    No registered user is needed, unlike a binding: what a release costs is a
    fork on the next turn, never work. A thread about no project has nothing
    to release — its tree goes with the turn — and a native session runs in a
    directory of its own that Octomate never forked."""
    if self.native:
        raise GatewayRefusal(
            "This session lives in your terminal, in a directory of its own — "
            "Octomate forked nothing for it, and has nothing to release."
        )
    if self.thread_id is None or self.threads is None:
        raise RuntimeError(
            "a dismiss needs the thread this turn is in, and the ledger it is "
            "read from"
        )
    thread = await self.threads.get(self.thread_id)
    if thread is None:
        raise RuntimeError(f"thread {self.thread_id} vanished")
    if await thread.project is None:
        raise GatewayRefusal(
            "This thread is about no project, and its tree is thrown away when "
            "this turn ends — there is nothing to release."
        )
    self.dismissing = True
    return (
        "Releasing this thread's workspace when this turn ends, once its work is "
        "saved to the project's mirror. Nothing is lost — a later message here "
        "forks it afresh from there — but finish up now: your working directory "
        "goes with it."
    )

scheme async

scheme(
    *,
    hint: str,
    brief: str,
    destination: SchemeTarget = DIRECT_TARGET,
) -> str

Validate and record a scheme decision, returning the sentence the scheming agent is told. The move itself is the graph's, after the turn.

Source code in octomate/managers/gateway.py

async def scheme(
    self,
    *,
    hint: str,
    brief: str,
    destination: SchemeTarget = DIRECT_TARGET,
) -> str:
    """Validate and record a scheme decision, returning the sentence the
    scheming agent is told. The move itself is the graph's, after the turn."""
    address = await self.direct_destination(destination)
    self.decision = SchemeDecision(hint=hint, brief=brief, destination=address)
    return f"Taking this to your direct messages on {address.channel_tentacle_id}."

resolve_send async

resolve_send(
    destination: SendTarget,
) -> ChannelAddress | None

The address a send delivers to, or None for this conversation itself.

Being in their direct messages already stops a scheme — nowhere to move the conversation to — but never a send: that is where it was asked to go, so it lands here rather than being refused.

Source code in octomate/managers/gateway.py

async def resolve_send(self, destination: SendTarget) -> ChannelAddress | None:
    """The address a send delivers to, or None for this conversation itself.

    Being in their direct messages already stops a `scheme` — nowhere to move
    the conversation to — but never a send: that *is* where it was asked to go,
    so it lands here rather than being refused."""
    if isinstance(destination, HereTarget):
        return None
    source = self.conversation_address
    local = destination.channel is None or (
        source is not None and destination.channel == source.channel_tentacle_id
    )
    if local and self.private_blocked_by == "already_private":
        return None
    return await self.direct_destination(destination)

attach_native_thread async

attach_native_thread(session_id: str) -> None

Bind a native call to the thread its own session is filed under, as Trunkline's thread operations bind theirs, so a spell it casts acts on that history. Only a thread the caller has written to is found: naming another session, theirs or anyone's, finds nothing.

Source code in octomate/managers/gateway.py

async def attach_native_thread(self, session_id: str) -> None:
    """Bind a native call to the thread its own session is filed under, as
    Trunkline's thread operations bind theirs, so a spell it casts acts on that
    history. Only a thread the caller has written to is found: naming another
    session, theirs or anyone's, finds nothing."""
    profile = self.user_profile
    if not self.native or self.threads is None or profile is None:
        raise RuntimeError("only a native session with a ledger binds a thread")
    thread = await self.threads.get(
        ThreadKey(self.current_agent_id, "thread", session_id),
        with_messages=False,
        user_id=profile.user_id,
    )
    if thread is None:
        raise GatewayRefusal(
            "Octomate holds no uploaded history for this session yet. Finish a "
            "turn with Octomate's hooks installed, then teleport again."
        )
    self.thread_id = thread.id
    self.conversation_address = thread.key.address(profile.channel_user_id)

thread_operation

thread_operation() -> DrivenGatewaySignal

Package a validated user action for the graph's existing-thread entry.

Source code in octomate/managers/gateway.py

def thread_operation(self) -> DrivenGatewaySignal:
    """Package a validated user action for the graph's existing-thread entry."""
    if (
        self.thread_id is None
        or self.conversation_address is None
        or self.user_profile is None
        or not isinstance(self.decision, SummonDecision | TeleportDecision)
    ):
        raise ValueError(
            "A thread operation requires a thread, user and validated decision."
        )
    return DrivenGatewaySignal(
        thread_id=self.thread_id,
        source=self.conversation_address,
        agent_id=self.current_agent_id,
        user_profile=self.user_profile,
        decision=self.decision,
    )

native_handoff

native_handoff() -> NativeGatewaySignal

This native session's recorded decision, packaged to be kicked as its own turn. Only a native spell leaves one, so anything else asking is a wiring bug, not a refusal a model could correct from.

Source code in octomate/managers/gateway.py

def native_handoff(self) -> NativeGatewaySignal:
    """This native session's recorded decision, packaged to be kicked as its
    own turn. Only a native spell leaves one, so anything else asking is a
    wiring bug, not a refusal a model could correct from."""
    decision = self.decision
    if not self.native or not isinstance(
        decision, SchemeDecision | TeleportDecision
    ):
        raise RuntimeError("only a native teleport or scheme kicks a handoff")
    return NativeGatewaySignal(
        decision=decision,
        agent_id=self.current_agent_id,
        user_profile=self.user_profile,
        # A teleport carries the thread it attached; the rest come from nowhere.
        source=self.conversation_address
        if isinstance(decision, TeleportDecision)
        else ChannelAddress(
            channel_tentacle_id=self.current_agent_id,
            chat_type="dm",
            chat_id="",
            user_id=NATIVE_CHANNEL_USER_ID,
        ),
    )

GatewayManager

Bases: Manager

The live Octomate sessions, one per driven turn, keyed by conversation id.

In-process on purpose: a session is only meaningful while its turn is in flight, so a restart rightly forgets them all. An external runtime's tool call presents a conversation id, and this is where it finds the session of the turn driving it.

Source code in octomate/managers/gateway.py

class GatewayManager(Manager):
    """The live Octomate sessions, one per driven turn, keyed by conversation id.

    In-process on purpose: a session is only meaningful while its turn is in
    flight, so a restart rightly forgets them all. An external runtime's tool call
    presents a conversation id, and this is where it finds the session of the turn
    driving it."""

    def __init__(self) -> None:
        # None holds a turn whose agent does not expose gateway spells.
        self.sessions: dict[UUID7, OctomateSession | None] = {}

    def register(self, session: OctomateSession) -> None:
        """Hold the conversation for `session`, first arrival only.

        Nothing else serialises turns of one conversation, so two can overlap; an
        external caller naming the conversation must find exactly one session, and
        a second run racing the first would otherwise have its calls land on the
        first's. So the second is refused outright rather than queued or ignored.
        """
        if session.conversation_id is None:
            raise ValueError("a registered Octomate session needs its conversation id")
        holder = self.sessions.get(session.conversation_id)
        if session.conversation_id in self.sessions and holder is not session:
            raise RuntimeError(
                f"conversation {session.conversation_id} already has a turn at the "
                "gateway; a second run on it is refused until that turn ends"
            )
        self.sessions[session.conversation_id] = session

    def unregister(self, session: OctomateSession) -> None:
        # Only its own entry — a session that was never registered (no conversation
        # id, or refused) removes nothing.
        if (
            session.conversation_id is not None
            and self.sessions.get(session.conversation_id) is session
        ):
            del self.sessions[session.conversation_id]

    def get(self, conversation_id: UUID7) -> OctomateSession | None:
        return self.sessions.get(conversation_id)

    def available_routes(
        self,
        channels: dict[str, ChannelTentacle],
        agents: dict[str, AgentTentacle],
    ) -> dict[str, list[AgentRoute]]:
        """What each channel can route to: every connected agent it exposes, each
        answering with its own discovered or configured model routes.

        The one computation behind both `ReflexDeps.available_routes` and the
        served gateway's ephemeral native session, so a driven turn and a native
        call cannot disagree about what a channel offers."""
        return {
            channel_id: [
                route
                for agent_id in dict.fromkeys(
                    connection
                    for connection in channel.agent_ids
                    if connection in agents
                )
                for route in agents[agent_id].routes
            ]
            for channel_id, channel in channels.items()
        }

    @asynccontextmanager
    async def driving(
        self,
        session: OctomateSession | None,
        *,
        conversation_id: UUID7 | None = None,
    ) -> AsyncGenerator[None]:
        """The registration span of one driven turn: external tool calls reach
        `session` only while the run that mounted it is in flight, and a second
        turn of the same conversation is refused at the door. A caller without
        gateway spells supplies its conversation id to hold the same guard
        without exposing a gateway session. Calls with neither remain untracked.
        """
        if session is not None and session.conversation_id is not None:
            if (
                conversation_id is not None
                and conversation_id != session.conversation_id
            ):
                raise ValueError("gateway session belongs to another conversation")
            conversation_id = session.conversation_id
            self.register(session)
        elif conversation_id is not None:
            if conversation_id in self.sessions:
                raise RuntimeError(
                    f"conversation {conversation_id} already has a turn at the "
                    "gateway; a second run on it is refused until that turn ends"
                )
            self.sessions[conversation_id] = None
        try:
            yield
        finally:
            if session is not None and session.conversation_id is not None:
                self.unregister(session)
            elif (
                conversation_id is not None
                and self.sessions.get(conversation_id) is None
            ):
                self.sessions.pop(conversation_id, None)

sessions instance-attribute

sessions: dict[UUID7, OctomateSession | None] = {}

register

register(session: OctomateSession) -> None

Hold the conversation for session, first arrival only.

Nothing else serialises turns of one conversation, so two can overlap; an external caller naming the conversation must find exactly one session, and a second run racing the first would otherwise have its calls land on the first's. So the second is refused outright rather than queued or ignored.

Source code in octomate/managers/gateway.py

def register(self, session: OctomateSession) -> None:
    """Hold the conversation for `session`, first arrival only.

    Nothing else serialises turns of one conversation, so two can overlap; an
    external caller naming the conversation must find exactly one session, and
    a second run racing the first would otherwise have its calls land on the
    first's. So the second is refused outright rather than queued or ignored.
    """
    if session.conversation_id is None:
        raise ValueError("a registered Octomate session needs its conversation id")
    holder = self.sessions.get(session.conversation_id)
    if session.conversation_id in self.sessions and holder is not session:
        raise RuntimeError(
            f"conversation {session.conversation_id} already has a turn at the "
            "gateway; a second run on it is refused until that turn ends"
        )
    self.sessions[session.conversation_id] = session

unregister

unregister(session: OctomateSession) -> None

Source code in octomate/managers/gateway.py

def unregister(self, session: OctomateSession) -> None:
    # Only its own entry — a session that was never registered (no conversation
    # id, or refused) removes nothing.
    if (
        session.conversation_id is not None
        and self.sessions.get(session.conversation_id) is session
    ):
        del self.sessions[session.conversation_id]

get

get(conversation_id: UUID7) -> OctomateSession | None

Source code in octomate/managers/gateway.py

def get(self, conversation_id: UUID7) -> OctomateSession | None:
    return self.sessions.get(conversation_id)

available_routes

available_routes(
    channels: dict[str, ChannelTentacle],
    agents: dict[str, AgentTentacle],
) -> dict[str, list[AgentRoute]]

What each channel can route to: every connected agent it exposes, each answering with its own discovered or configured model routes.

The one computation behind both ReflexDeps.available_routes and the served gateway's ephemeral native session, so a driven turn and a native call cannot disagree about what a channel offers.

Source code in octomate/managers/gateway.py

def available_routes(
    self,
    channels: dict[str, ChannelTentacle],
    agents: dict[str, AgentTentacle],
) -> dict[str, list[AgentRoute]]:
    """What each channel can route to: every connected agent it exposes, each
    answering with its own discovered or configured model routes.

    The one computation behind both `ReflexDeps.available_routes` and the
    served gateway's ephemeral native session, so a driven turn and a native
    call cannot disagree about what a channel offers."""
    return {
        channel_id: [
            route
            for agent_id in dict.fromkeys(
                connection
                for connection in channel.agent_ids
                if connection in agents
            )
            for route in agents[agent_id].routes
        ]
        for channel_id, channel in channels.items()
    }

driving async

driving(
    session: OctomateSession | None,
    *,
    conversation_id: UUID7 | None = None,
) -> AsyncGenerator[None]

The registration span of one driven turn: external tool calls reach session only while the run that mounted it is in flight, and a second turn of the same conversation is refused at the door. A caller without gateway spells supplies its conversation id to hold the same guard without exposing a gateway session. Calls with neither remain untracked.

Source code in octomate/managers/gateway.py

@asynccontextmanager
async def driving(
    self,
    session: OctomateSession | None,
    *,
    conversation_id: UUID7 | None = None,
) -> AsyncGenerator[None]:
    """The registration span of one driven turn: external tool calls reach
    `session` only while the run that mounted it is in flight, and a second
    turn of the same conversation is refused at the door. A caller without
    gateway spells supplies its conversation id to hold the same guard
    without exposing a gateway session. Calls with neither remain untracked.
    """
    if session is not None and session.conversation_id is not None:
        if (
            conversation_id is not None
            and conversation_id != session.conversation_id
        ):
            raise ValueError("gateway session belongs to another conversation")
        conversation_id = session.conversation_id
        self.register(session)
    elif conversation_id is not None:
        if conversation_id in self.sessions:
            raise RuntimeError(
                f"conversation {conversation_id} already has a turn at the "
                "gateway; a second run on it is refused until that turn ends"
            )
        self.sessions[conversation_id] = None
    try:
        yield
    finally:
        if session is not None and session.conversation_id is not None:
            self.unregister(session)
        elif (
            conversation_id is not None
            and self.sessions.get(conversation_id) is None
        ):
            self.sessions.pop(conversation_id, None)

octomate.managers.todos

Conversation-scoped todo persistence.

TodoManager

Bases: Manager

Conversation-scoped todo persistence, keyed on the short hex ref.

Persistence only — event derivation and cycle/blocking validation live in the todo capability, which owns the agent-facing behaviour.

Source code in octomate/managers/todos.py

class TodoManager(Manager):
    """Conversation-scoped todo persistence, keyed on the short hex `ref`.

    Persistence only — event derivation and cycle/blocking validation live in the
    todo capability, which owns the agent-facing behaviour.
    """

    async def list_todos(self, conversation_id: UUID7) -> list[Todo]:
        async with async_session() as session:
            todos = await session.list(
                Todo,
                limit=None,
                order_bys=[Todo["position"], Todo["created_at"]],
                expressions=[Todo["conversation_id"] == conversation_id],
            )
        return list(todos)

    async def get_todo(self, conversation_id: UUID7, ref: str) -> Todo | None:
        async with async_session() as session:
            return await session.one_or_none(
                Todo,
                expressions=[
                    Todo["conversation_id"] == conversation_id,
                    Todo["ref"] == ref,
                ],
            )

    async def add_todo(
        self,
        conversation_id: UUID7,
        *,
        content: str,
        active_form: str,
        parent_ref: str | None = None,
    ) -> Todo:
        existing = await self.list_todos(conversation_id)
        todo = Todo(
            conversation_id=conversation_id,
            ref=fresh_ref({item.ref for item in existing}),
            content=content,
            active_form=active_form,
            parent_ref=parent_ref,
            position=max((item.position for item in existing), default=-1) + 1,
        )
        async with async_session() as session:
            session.add(todo)
            await session.commit()
        return todo

    async def update_todo(
        self,
        conversation_id: UUID7,
        ref: str,
        *,
        content: str | None = None,
        status: TodoStatus | None = None,
        active_form: str | None = None,
        parent_ref: str | None = None,
        depends_on: list[str] | None = None,
    ) -> Todo | None:
        async with async_session() as session:
            todo = await session.one_or_none(
                Todo,
                expressions=[
                    Todo["conversation_id"] == conversation_id,
                    Todo["ref"] == ref,
                ],
            )
            if todo is None:
                return None
            if content is not None:
                todo.content = content
            if status is not None:
                todo.status = status
            if active_form is not None:
                todo.active_form = active_form
            if parent_ref is not None:
                todo.parent_ref = parent_ref
            if depends_on is not None:
                todo.depends_on = depends_on
            todo.updated_at = datetime.now(UTC)
            await session.commit()
        return todo

    async def remove_todo(self, conversation_id: UUID7, ref: str) -> Todo | None:
        async with async_session() as session:
            todo = await session.one_or_none(
                Todo,
                expressions=[
                    Todo["conversation_id"] == conversation_id,
                    Todo["ref"] == ref,
                ],
            )
            if todo is None:
                return None
            await session.delete(todo)
            await session.commit()
        return todo

    async def write_todos(
        self,
        conversation_id: UUID7,
        items: list[TodoWrite],
    ) -> tuple[list[Todo], list[Todo]]:
        """Replace the conversation's todos with `items`; return (deleted, created)."""
        async with async_session() as session:
            deleted = list(
                await session.list(
                    Todo,
                    limit=None,
                    order_bys=[Todo["position"], Todo["created_at"]],
                    expressions=[Todo["conversation_id"] == conversation_id],
                )
            )
            for todo in deleted:
                await session.delete(todo)
            # Flush the deletes before inserting so a reused `ref` cannot trip the
            # (conversation_id, ref) unique constraint mid-flush.
            await session.commit()

            taken: set[str] = set()
            created: list[Todo] = []
            for position, item in enumerate(items):
                ref = item.get("ref") or fresh_ref(taken)
                taken.add(ref)
                todo = Todo(
                    conversation_id=conversation_id,
                    ref=ref,
                    content=item["content"],
                    status=item.get("status", "pending"),
                    active_form=item.get("active_form", ""),
                    parent_ref=item.get("parent_ref"),
                    depends_on=item.get("depends_on", []),
                    position=position,
                )
                session.add(todo)
                created.append(todo)
            await session.commit()
        return deleted, created

list_todos async

list_todos(conversation_id: UUID7) -> list[Todo]

Source code in octomate/managers/todos.py

async def list_todos(self, conversation_id: UUID7) -> list[Todo]:
    async with async_session() as session:
        todos = await session.list(
            Todo,
            limit=None,
            order_bys=[Todo["position"], Todo["created_at"]],
            expressions=[Todo["conversation_id"] == conversation_id],
        )
    return list(todos)

get_todo async

get_todo(conversation_id: UUID7, ref: str) -> Todo | None

Source code in octomate/managers/todos.py

async def get_todo(self, conversation_id: UUID7, ref: str) -> Todo | None:
    async with async_session() as session:
        return await session.one_or_none(
            Todo,
            expressions=[
                Todo["conversation_id"] == conversation_id,
                Todo["ref"] == ref,
            ],
        )

add_todo async

add_todo(
    conversation_id: UUID7,
    *,
    content: str,
    active_form: str,
    parent_ref: str | None = None,
) -> Todo

Source code in octomate/managers/todos.py

async def add_todo(
    self,
    conversation_id: UUID7,
    *,
    content: str,
    active_form: str,
    parent_ref: str | None = None,
) -> Todo:
    existing = await self.list_todos(conversation_id)
    todo = Todo(
        conversation_id=conversation_id,
        ref=fresh_ref({item.ref for item in existing}),
        content=content,
        active_form=active_form,
        parent_ref=parent_ref,
        position=max((item.position for item in existing), default=-1) + 1,
    )
    async with async_session() as session:
        session.add(todo)
        await session.commit()
    return todo

update_todo async

update_todo(
    conversation_id: UUID7,
    ref: str,
    *,
    content: str | None = None,
    status: TodoStatus | None = None,
    active_form: str | None = None,
    parent_ref: str | None = None,
    depends_on: list[str] | None = None,
) -> Todo | None

Source code in octomate/managers/todos.py

async def update_todo(
    self,
    conversation_id: UUID7,
    ref: str,
    *,
    content: str | None = None,
    status: TodoStatus | None = None,
    active_form: str | None = None,
    parent_ref: str | None = None,
    depends_on: list[str] | None = None,
) -> Todo | None:
    async with async_session() as session:
        todo = await session.one_or_none(
            Todo,
            expressions=[
                Todo["conversation_id"] == conversation_id,
                Todo["ref"] == ref,
            ],
        )
        if todo is None:
            return None
        if content is not None:
            todo.content = content
        if status is not None:
            todo.status = status
        if active_form is not None:
            todo.active_form = active_form
        if parent_ref is not None:
            todo.parent_ref = parent_ref
        if depends_on is not None:
            todo.depends_on = depends_on
        todo.updated_at = datetime.now(UTC)
        await session.commit()
    return todo

remove_todo async

remove_todo(
    conversation_id: UUID7, ref: str
) -> Todo | None

Source code in octomate/managers/todos.py

async def remove_todo(self, conversation_id: UUID7, ref: str) -> Todo | None:
    async with async_session() as session:
        todo = await session.one_or_none(
            Todo,
            expressions=[
                Todo["conversation_id"] == conversation_id,
                Todo["ref"] == ref,
            ],
        )
        if todo is None:
            return None
        await session.delete(todo)
        await session.commit()
    return todo

write_todos async

write_todos(
    conversation_id: UUID7, items: list[TodoWrite]
) -> tuple[list[Todo], list[Todo]]

Replace the conversation's todos with items; return (deleted, created).

Source code in octomate/managers/todos.py

async def write_todos(
    self,
    conversation_id: UUID7,
    items: list[TodoWrite],
) -> tuple[list[Todo], list[Todo]]:
    """Replace the conversation's todos with `items`; return (deleted, created)."""
    async with async_session() as session:
        deleted = list(
            await session.list(
                Todo,
                limit=None,
                order_bys=[Todo["position"], Todo["created_at"]],
                expressions=[Todo["conversation_id"] == conversation_id],
            )
        )
        for todo in deleted:
            await session.delete(todo)
        # Flush the deletes before inserting so a reused `ref` cannot trip the
        # (conversation_id, ref) unique constraint mid-flush.
        await session.commit()

        taken: set[str] = set()
        created: list[Todo] = []
        for position, item in enumerate(items):
            ref = item.get("ref") or fresh_ref(taken)
            taken.add(ref)
            todo = Todo(
                conversation_id=conversation_id,
                ref=ref,
                content=item["content"],
                status=item.get("status", "pending"),
                active_form=item.get("active_form", ""),
                parent_ref=item.get("parent_ref"),
                depends_on=item.get("depends_on", []),
                position=position,
            )
            session.add(todo)
            created.append(todo)
        await session.commit()
    return deleted, created

fresh_ref

fresh_ref(taken: set[str]) -> str

A short hex todo handle not already in taken.

Source code in octomate/managers/todos.py

def fresh_ref(taken: set[str]) -> str:
    """A short hex todo handle not already in `taken`."""
    ref = uuid.uuid4().hex[:8]
    while ref in taken:
        ref = uuid.uuid4().hex[:8]
    return ref