Feelers

octomate.tentacles.feelers.base

Base channel feeler aggregate surface.

Feelers dataclass

Per-channel human-interaction surface.

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

@dataclass
class Feelers:
    """Per-channel human-interaction surface."""

    markdown: MarkdownFeeler
    timeline: TimelineFeeler
    segments: SegmentsFeeler
    approvals: ApprovalFeeler
    ask_questions: QuestionFeeler
    oauth: OAuthFeeler
    # Saved artifacts presented independently of the run's reply.
    artifacts: ArtifactFeeler

    async def present(
        self, address: ChannelAddress, event: GatewayEvent | RunErrorEvent
    ) -> IMMessageID | None:
        """What the graph reports outside a run, as a message: a move's
        announcement, or a failure's trace id. A move that leaves no line says
        nothing here."""
        if isinstance(event, RunErrorEvent):
            return await self.markdown.present(
                address,
                "Something went wrong while handling your message. "
                f"Reference id for tracing the issue: `{event.trace_id}`.",
            )
        if event.announcement is None:
            return None
        return await self.markdown.present(address, event.announcement)

    async def present_actions(
        self,
        address: ChannelAddress,
        event: ActionBatchEvent,
        *,
        action_manager: DeferredActionManager,
    ) -> None:
        """A batch put up for a run that is not streamed, as the channel's cards."""
        with channel_logfire.span(
            "present_actions",
            target_address=str(address),
            approvals=len(event.approvals),
            questions=len(event.questions),
        ):
            approval_message_ids = await self.approvals.present(
                address, event.approvals
            )
            for action in event.approvals:
                await action_manager.mark_action_presented(
                    action.id,
                    approval_message_ids.get(action.id),
                )

            message_ids = await self.ask_questions.present(address, event.questions)
            for action in event.questions:
                await action_manager.mark_action_presented(
                    action.id,
                    message_ids.get(action.id),
                )

markdown instance-attribute

markdown: MarkdownFeeler

timeline instance-attribute

timeline: TimelineFeeler

segments instance-attribute

segments: SegmentsFeeler

approvals instance-attribute

approvals: ApprovalFeeler

ask_questions instance-attribute

ask_questions: QuestionFeeler

oauth instance-attribute

oauth: OAuthFeeler

artifacts instance-attribute

artifacts: ArtifactFeeler

present async

present(
    address: ChannelAddress,
    event: GatewayEvent | RunErrorEvent,
) -> IMMessageID | None

What the graph reports outside a run, as a message: a move's announcement, or a failure's trace id. A move that leaves no line says nothing here.

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

async def present(
    self, address: ChannelAddress, event: GatewayEvent | RunErrorEvent
) -> IMMessageID | None:
    """What the graph reports outside a run, as a message: a move's
    announcement, or a failure's trace id. A move that leaves no line says
    nothing here."""
    if isinstance(event, RunErrorEvent):
        return await self.markdown.present(
            address,
            "Something went wrong while handling your message. "
            f"Reference id for tracing the issue: `{event.trace_id}`.",
        )
    if event.announcement is None:
        return None
    return await self.markdown.present(address, event.announcement)

present_actions async

present_actions(
    address: ChannelAddress,
    event: ActionBatchEvent,
    *,
    action_manager: DeferredActionManager,
) -> None

A batch put up for a run that is not streamed, as the channel's cards.

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

async def present_actions(
    self,
    address: ChannelAddress,
    event: ActionBatchEvent,
    *,
    action_manager: DeferredActionManager,
) -> None:
    """A batch put up for a run that is not streamed, as the channel's cards."""
    with channel_logfire.span(
        "present_actions",
        target_address=str(address),
        approvals=len(event.approvals),
        questions=len(event.questions),
    ):
        approval_message_ids = await self.approvals.present(
            address, event.approvals
        )
        for action in event.approvals:
            await action_manager.mark_action_presented(
                action.id,
                approval_message_ids.get(action.id),
            )

        message_ids = await self.ask_questions.present(address, event.questions)
        for action in event.questions:
            await action_manager.mark_action_presented(
                action.id,
                message_ids.get(action.id),
            )

octomate.tentacles.feelers.output

Output rendering shared by every channel: markdown chunking, stream batching, the timeline state a run is rendered onto, and the default feelers for channels with no live transport.

logger module-attribute

logger = logging.getLogger(__name__)

StreamBlockType module-attribute

StreamBlockType = Literal[
    "answer",
    "thinking",
    "tool_call",
    "tool_result",
    "subagent",
]

StreamBlockStatus module-attribute

StreamBlockStatus = Literal['streaming', 'done', 'error']

SKIPPED_PLAN_TOOL_NAMES module-attribute

SKIPPED_PLAN_TOOL_NAMES = frozenset(
    {
        ASK_QUESTIONS_TOOL_NAME,
        DEFAULT_OUTPUT_TOOL_NAME,
        GatewayTool.SEND,
        GatewayTool.COMMISSION,
        GatewayTool.WHISPER,
        GatewayTool.TELEPORT,
    }
)

MAX_TASK_DETAIL_CHARS module-attribute

MAX_TASK_DETAIL_CHARS = 2000

IMMessageID

IMMessageID = str

MarkdownChunker dataclass

Splits markdown into chunks under a platform's message limit, preferring paragraph, line, sentence and word boundaries in that order.

Source code in octomate/tentacles/feelers/output.py

@dataclass(frozen=True)
class MarkdownChunker:
    """Splits markdown into chunks under a platform's message limit, preferring
    paragraph, line, sentence and word boundaries in that order."""

    DEFAULT_LIMIT: ClassVar[int] = 12_000

    limit: int = DEFAULT_LIMIT
    natural_min_size: int | None = None

    @property
    def effective_natural_min_size(self) -> int:
        if self.natural_min_size is not None:
            return self.natural_min_size
        return self.limit // 2

    def chunk(self, text: str) -> list[str]:
        if not text:
            return [""]

        chunks: list[str] = []
        remaining = text
        while len(remaining) > self.limit:
            split_at = self.split_index(remaining)
            chunks.append(remaining[:split_at])
            remaining = remaining[split_at:]
        if remaining:
            chunks.append(remaining)
        return chunks

    def split_index(self, text: str) -> int:
        for boundary in (
            self.last_separator_boundary(text, "\n\n"),
            self.last_separator_boundary(text, "\n"),
            self.last_sentence_boundary(text),
            self.last_whitespace_boundary(text),
        ):
            if boundary >= self.effective_natural_min_size:
                return boundary

        return self.last_whitespace_boundary(text) or self.limit

    def last_separator_boundary(self, text: str, separator: str) -> int:
        boundary = 0
        start = 0
        while True:
            index = text.find(separator, start, self.limit)
            if index < 0:
                return boundary
            candidate = index + len(separator)
            if candidate <= self.limit:
                boundary = candidate
            start = index + len(separator)

    def last_sentence_boundary(self, text: str) -> int:
        boundary = 0
        for match in re.finditer(r'[.!?][)"\']?\s+', text[: self.limit]):
            boundary = match.end()
        return boundary

    def last_whitespace_boundary(self, text: str) -> int:
        boundary = 0
        for match in re.finditer(r"\s+", text[: self.limit]):
            boundary = match.end()
        return boundary

DEFAULT_LIMIT class-attribute

DEFAULT_LIMIT: int = 12000

limit class-attribute instance-attribute

limit: int = DEFAULT_LIMIT

natural_min_size class-attribute instance-attribute

natural_min_size: int | None = None

effective_natural_min_size property

effective_natural_min_size: int

chunk

chunk(text: str) -> list[str]

Source code in octomate/tentacles/feelers/output.py

def chunk(self, text: str) -> list[str]:
    if not text:
        return [""]

    chunks: list[str] = []
    remaining = text
    while len(remaining) > self.limit:
        split_at = self.split_index(remaining)
        chunks.append(remaining[:split_at])
        remaining = remaining[split_at:]
    if remaining:
        chunks.append(remaining)
    return chunks

split_index

split_index(text: str) -> int

Source code in octomate/tentacles/feelers/output.py

def split_index(self, text: str) -> int:
    for boundary in (
        self.last_separator_boundary(text, "\n\n"),
        self.last_separator_boundary(text, "\n"),
        self.last_sentence_boundary(text),
        self.last_whitespace_boundary(text),
    ):
        if boundary >= self.effective_natural_min_size:
            return boundary

    return self.last_whitespace_boundary(text) or self.limit

last_separator_boundary

last_separator_boundary(text: str, separator: str) -> int

Source code in octomate/tentacles/feelers/output.py

def last_separator_boundary(self, text: str, separator: str) -> int:
    boundary = 0
    start = 0
    while True:
        index = text.find(separator, start, self.limit)
        if index < 0:
            return boundary
        candidate = index + len(separator)
        if candidate <= self.limit:
            boundary = candidate
        start = index + len(separator)

last_sentence_boundary

last_sentence_boundary(text: str) -> int

Source code in octomate/tentacles/feelers/output.py

def last_sentence_boundary(self, text: str) -> int:
    boundary = 0
    for match in re.finditer(r'[.!?][)"\']?\s+', text[: self.limit]):
        boundary = match.end()
    return boundary

last_whitespace_boundary

last_whitespace_boundary(text: str) -> int

Source code in octomate/tentacles/feelers/output.py

def last_whitespace_boundary(self, text: str) -> int:
    boundary = 0
    for match in re.finditer(r"\s+", text[: self.limit]):
        boundary = match.end()
    return boundary

StreamBlock dataclass

One block of a streamed reply — answer, thinking, tool call — by id.

Source code in octomate/tentacles/feelers/output.py

@dataclass(frozen=True)
class StreamBlock:
    """One block of a streamed reply — answer, thinking, tool call — by id."""

    id: str
    type: StreamBlockType = "answer"
    title: str = ""
    foldable: bool = False
    status: StreamBlockStatus = "streaming"

id instance-attribute

id: str

type class-attribute instance-attribute

type: StreamBlockType = 'answer'

title class-attribute instance-attribute

title: str = ''

foldable class-attribute instance-attribute

foldable: bool = False

status class-attribute instance-attribute

status: StreamBlockStatus = 'streaming'

BatchedTextUpdate dataclass

A flushed edit to one block: the new delta, the full text so far, and whether the block is final.

Source code in octomate/tentacles/feelers/output.py

@dataclass(frozen=True)
class BatchedTextUpdate:
    """A flushed edit to one block: the new delta, the full text so far, and
    whether the block is final."""

    block_id: str
    block_type: StreamBlockType
    title: str
    delta_text: str
    full_text: str
    sequence: int
    is_final: bool = False
    foldable: bool = False
    status: StreamBlockStatus = "streaming"

block_id instance-attribute

block_id: str

block_type instance-attribute

block_type: StreamBlockType

title instance-attribute

title: str

delta_text instance-attribute

delta_text: str

full_text instance-attribute

full_text: str

sequence instance-attribute

sequence: int

is_final class-attribute instance-attribute

is_final: bool = False

foldable class-attribute instance-attribute

foldable: bool = False

status class-attribute instance-attribute

status: StreamBlockStatus = 'streaming'

StreamEventDelta dataclass

A stream event reduced to text, tagged with the block it belongs to.

Source code in octomate/tentacles/feelers/output.py

@dataclass(frozen=True)
class StreamEventDelta:
    """A stream event reduced to text, tagged with the block it belongs to."""

    block: StreamBlock
    text: str

block instance-attribute

block: StreamBlock

text instance-attribute

text: str

TextStreamBuffer dataclass

Text accumulated for one block between flushes.

Source code in octomate/tentacles/feelers/output.py

@dataclass
class TextStreamBuffer:
    """Text accumulated for one block between flushes."""

    block: StreamBlock
    full_text: str = ""
    pending_delta: str = ""
    sequence: int = 0
    last_flush_at: float = 0.0

block instance-attribute

block: StreamBlock

full_text class-attribute instance-attribute

full_text: str = ''

pending_delta class-attribute instance-attribute

pending_delta: str = ''

sequence class-attribute instance-attribute

sequence: int = 0

last_flush_at class-attribute instance-attribute

last_flush_at: float = 0.0

TextStreamBatcher

Coalesces streamed text per block into paced updates, by interval and size.

Source code in octomate/tentacles/feelers/output.py

class TextStreamBatcher:
    """Coalesces streamed text per block into paced updates, by interval and
    size."""

    def __init__(
        self,
        *,
        flush_interval: float = 0.5,
        min_chars: int = 120,
        max_chars: int = 1000,
        fold_threshold: int = 1500,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        self.flush_interval = max(flush_interval, 0.0)
        self.min_chars = max(min_chars, 0)
        self.max_chars = max(max_chars, 1)
        self.fold_threshold = max(fold_threshold, 1)
        self.clock = clock
        self.buffers: dict[str, TextStreamBuffer] = {}
        self.active_block_id: str | None = None

    def push_text(
        self,
        text: str,
        *,
        block: StreamBlock | None = None,
    ) -> list[BatchedTextUpdate]:
        if not text:
            return []
        block = block or StreamBlock(id="answer")
        updates: list[BatchedTextUpdate] = []
        if self.active_block_id and self.active_block_id != block.id:
            previous = self.flush_block(self.active_block_id)
            if previous is not None:
                updates.append(previous)

        self.active_block_id = block.id
        buffer = self.buffers.get(block.id)
        if buffer is None:
            buffer = TextStreamBuffer(block=block, last_flush_at=self.clock())
            self.buffers[block.id] = buffer
        else:
            buffer.block = block

        buffer.full_text += text
        buffer.pending_delta += text
        if self.should_flush(buffer):
            update = self.flush_block(block.id)
            if update is not None:
                updates.append(update)
        return updates

    def flush_block(
        self,
        block_id: str,
        *,
        is_final: bool = False,
    ) -> BatchedTextUpdate | None:
        buffer = self.buffers.get(block_id)
        if buffer is None or not buffer.pending_delta:
            return None
        buffer.sequence += 1
        update = BatchedTextUpdate(
            block_id=buffer.block.id,
            block_type=buffer.block.type,
            title=buffer.block.title,
            delta_text=buffer.pending_delta,
            full_text=buffer.full_text,
            sequence=buffer.sequence,
            is_final=is_final,
            foldable=buffer.block.foldable
            or len(buffer.full_text) >= self.fold_threshold,
            status="done" if is_final else buffer.block.status,
        )
        buffer.pending_delta = ""
        buffer.last_flush_at = self.clock()
        return update

    def finish_all(self) -> list[BatchedTextUpdate]:
        updates: list[BatchedTextUpdate] = []
        if self.active_block_id:
            update = self.flush_block(self.active_block_id, is_final=True)
            if update is not None:
                updates.append(update)
        for block_id in list(self.buffers):
            if block_id == self.active_block_id:
                continue
            update = self.flush_block(block_id, is_final=True)
            if update is not None:
                updates.append(update)
        return updates

    def reset(self) -> None:
        """Drop accumulated buffers so the same batcher can back the next message
        without carrying over `full_text` (which drives the fold threshold). The
        configured pacing (flush_interval/min_chars/...) is preserved."""
        self.buffers.clear()
        self.active_block_id = None

    def full_text(self, block_id: str = "answer") -> str:
        buffer = self.buffers.get(block_id)
        return buffer.full_text if buffer is not None else ""

    def should_flush(self, buffer: TextStreamBuffer) -> bool:
        if buffer.sequence == 0:
            return True
        if self.flush_interval <= 0:
            return True
        if len(buffer.pending_delta) >= self.max_chars:
            return True
        if len(buffer.pending_delta) < self.min_chars:
            return False
        return self.clock() - buffer.last_flush_at >= self.flush_interval

flush_interval instance-attribute

flush_interval = max(flush_interval, 0.0)

min_chars instance-attribute

min_chars = max(min_chars, 0)

max_chars instance-attribute

max_chars = max(max_chars, 1)

fold_threshold instance-attribute

fold_threshold = max(fold_threshold, 1)

clock instance-attribute

clock = clock

buffers instance-attribute

buffers: dict[str, TextStreamBuffer] = {}

active_block_id instance-attribute

active_block_id: str | None = None

push_text

push_text(
    text: str, *, block: StreamBlock | None = None
) -> list[BatchedTextUpdate]

Source code in octomate/tentacles/feelers/output.py

def push_text(
    self,
    text: str,
    *,
    block: StreamBlock | None = None,
) -> list[BatchedTextUpdate]:
    if not text:
        return []
    block = block or StreamBlock(id="answer")
    updates: list[BatchedTextUpdate] = []
    if self.active_block_id and self.active_block_id != block.id:
        previous = self.flush_block(self.active_block_id)
        if previous is not None:
            updates.append(previous)

    self.active_block_id = block.id
    buffer = self.buffers.get(block.id)
    if buffer is None:
        buffer = TextStreamBuffer(block=block, last_flush_at=self.clock())
        self.buffers[block.id] = buffer
    else:
        buffer.block = block

    buffer.full_text += text
    buffer.pending_delta += text
    if self.should_flush(buffer):
        update = self.flush_block(block.id)
        if update is not None:
            updates.append(update)
    return updates

flush_block

flush_block(
    block_id: str, *, is_final: bool = False
) -> BatchedTextUpdate | None

Source code in octomate/tentacles/feelers/output.py

def flush_block(
    self,
    block_id: str,
    *,
    is_final: bool = False,
) -> BatchedTextUpdate | None:
    buffer = self.buffers.get(block_id)
    if buffer is None or not buffer.pending_delta:
        return None
    buffer.sequence += 1
    update = BatchedTextUpdate(
        block_id=buffer.block.id,
        block_type=buffer.block.type,
        title=buffer.block.title,
        delta_text=buffer.pending_delta,
        full_text=buffer.full_text,
        sequence=buffer.sequence,
        is_final=is_final,
        foldable=buffer.block.foldable
        or len(buffer.full_text) >= self.fold_threshold,
        status="done" if is_final else buffer.block.status,
    )
    buffer.pending_delta = ""
    buffer.last_flush_at = self.clock()
    return update

finish_all

finish_all() -> list[BatchedTextUpdate]

Source code in octomate/tentacles/feelers/output.py

def finish_all(self) -> list[BatchedTextUpdate]:
    updates: list[BatchedTextUpdate] = []
    if self.active_block_id:
        update = self.flush_block(self.active_block_id, is_final=True)
        if update is not None:
            updates.append(update)
    for block_id in list(self.buffers):
        if block_id == self.active_block_id:
            continue
        update = self.flush_block(block_id, is_final=True)
        if update is not None:
            updates.append(update)
    return updates

reset

reset() -> None

Drop accumulated buffers so the same batcher can back the next message without carrying over full_text (which drives the fold threshold). The configured pacing (flush_interval/min_chars/...) is preserved.

Source code in octomate/tentacles/feelers/output.py

def reset(self) -> None:
    """Drop accumulated buffers so the same batcher can back the next message
    without carrying over `full_text` (which drives the fold threshold). The
    configured pacing (flush_interval/min_chars/...) is preserved."""
    self.buffers.clear()
    self.active_block_id = None

full_text

full_text(block_id: str = 'answer') -> str

Source code in octomate/tentacles/feelers/output.py

def full_text(self, block_id: str = "answer") -> str:
    buffer = self.buffers.get(block_id)
    return buffer.full_text if buffer is not None else ""

should_flush

should_flush(buffer: TextStreamBuffer) -> bool

Source code in octomate/tentacles/feelers/output.py

def should_flush(self, buffer: TextStreamBuffer) -> bool:
    if buffer.sequence == 0:
        return True
    if self.flush_interval <= 0:
        return True
    if len(buffer.pending_delta) >= self.max_chars:
        return True
    if len(buffer.pending_delta) < self.min_chars:
        return False
    return self.clock() - buffer.last_flush_at >= self.flush_interval

StreamFlusher

Single-flight background flusher for streaming renders (thinking/answer, on every channel). signal() wakes a lone background task that runs the flush callback — which sends whatever has accumulated; deltas that arrive while a flush is in flight coalesce into the next pass, so the drive loop never blocks on a slow platform round-trip. drain() stops the task; the caller sends any final/settled state itself afterwards.

Source code in octomate/tentacles/feelers/output.py

class StreamFlusher:
    """Single-flight background flusher for streaming renders (thinking/answer,
    on every channel). `signal()` wakes a lone background task that runs the
    `flush` callback — which sends whatever has accumulated; deltas that arrive
    while a flush is in flight coalesce into the next pass, so the drive loop
    never blocks on a slow platform round-trip. `drain()` stops the task; the
    caller sends any final/settled state itself afterwards."""

    def __init__(self, flush: Callable[[], Awaitable[None]]) -> None:
        self.flush = flush
        self.task: asyncio.Task[None] | None = None
        self.event = asyncio.Event()
        self.closed = False

    def signal(self) -> None:
        if self.task is None or self.task.done():
            self.closed = False
            self.event.clear()
            self.task = asyncio.create_task(self.run())
        self.event.set()

    async def run(self) -> None:
        while True:
            await self.event.wait()
            self.event.clear()
            if self.closed:
                return
            await self.flush()

    async def drain(self) -> None:
        self.closed = True
        self.event.set()
        if self.task is not None:
            await self.task
            self.task = None

flush instance-attribute

flush = flush

task instance-attribute

task: Task[None] | None = None

event instance-attribute

event = asyncio.Event()

closed instance-attribute

closed = False

signal

signal() -> None

Source code in octomate/tentacles/feelers/output.py

def signal(self) -> None:
    if self.task is None or self.task.done():
        self.closed = False
        self.event.clear()
        self.task = asyncio.create_task(self.run())
    self.event.set()

run async

run() -> None

Source code in octomate/tentacles/feelers/output.py

async def run(self) -> None:
    while True:
        await self.event.wait()
        self.event.clear()
        if self.closed:
            return
        await self.flush()

drain async

drain() -> None

Source code in octomate/tentacles/feelers/output.py

async def drain(self) -> None:
    self.closed = True
    self.event.set()
    if self.task is not None:
        await self.task
        self.task = None

MarkdownFeeler

Bases: Protocol

Presents markdown to IM and returns platform message metadata only.

Return IMMessageID values for IM bookkeeping, such as a message id, timestamp, or card id. Feelers must not return agent/runtime results.

Source code in octomate/tentacles/feelers/output.py

class MarkdownFeeler(Protocol):
    """Presents markdown to IM and returns platform message metadata only.

    Return ``IMMessageID`` values for IM bookkeeping, such as a message id,
    timestamp, or card id. Feelers must not return agent/runtime results.
    """

    async def present(
        self,
        address: ChannelAddress,
        markdown: str,
    ) -> IMMessageID | None: ...

present async

present(
    address: ChannelAddress, markdown: str
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/output.py

async def present(
    self,
    address: ChannelAddress,
    markdown: str,
) -> IMMessageID | None: ...

SubagentTimelineState

One independently rendered commissioned child run.

Source code in octomate/tentacles/feelers/output.py

class SubagentTimelineState:
    """One independently rendered commissioned child run."""

    async def append_response(self, delta: str) -> None: ...

    async def settle(
        self,
        status: SubagentActivityStatus,
        detail: str | None = None,
    ) -> None: ...

append_response async

append_response(delta: str) -> None

Source code in octomate/tentacles/feelers/output.py

async def append_response(self, delta: str) -> None: ...

settle async

settle(
    status: SubagentActivityStatus,
    detail: str | None = None,
) -> None

Source code in octomate/tentacles/feelers/output.py

async def settle(
    self,
    status: SubagentActivityStatus,
    detail: str | None = None,
) -> None: ...

OpenSubagentTimeline dataclass

A subagent timeline held open: its context manager and the state its events land on.

Source code in octomate/tentacles/feelers/output.py

@dataclass
class OpenSubagentTimeline:
    """A subagent timeline held open: its context manager and the state its events
    land on."""

    context: AbstractAsyncContextManager[SubagentTimelineState]
    state: SubagentTimelineState

context instance-attribute

context: AbstractAsyncContextManager[SubagentTimelineState]

state instance-attribute

state: SubagentTimelineState

TimelineState

Streaming render hook for one run, and the event→render dispatch (drive) that walks the run stream. The feeler that opened the state calls bind with the per-run context (address + the deferred-action feelers), then hands the stream to drive, which maps one lifecycle method per real stream event: PartStart/PartDelta/PartEnd of the thinking and answer parts map to *_start / *_delta / *_end, and the tool call/result events to tool_start / tool_end. A per-run timeline subclasses this and overrides only the lifecycle points it actually renders; the base methods render nothing. message_id carries the platform id of the rendered reply once the run's context closes.

Mid-run notice contract: answer hooks that render reply text set noticed, and every hook that opens a NEW timeline entry (a thinking block, a tool call, a todo update — never a tool result, so in-flight work finishes where it began) calls begin_entry() first, which rotate()s once when a notice streamed in between. Overrides must keep doing both.

Source code in octomate/tentacles/feelers/output.py

class TimelineState:
    """Streaming render hook for one run, and the event→render dispatch (`drive`)
    that walks the run stream. The feeler that opened the state calls `bind` with
    the per-run context (address + the deferred-action feelers), then hands the stream
    to `drive`, which maps one lifecycle method per real stream event:
    `PartStart`/`PartDelta`/`PartEnd` of the thinking and answer parts map to
    `*_start` / `*_delta` / `*_end`, and the tool call/result events to
    `tool_start` / `tool_end`. A per-run timeline subclasses this and overrides
    only the lifecycle points it actually renders; the base methods render
    nothing. `message_id` carries the platform id of the rendered reply once
    the run's context closes.

    Mid-run notice contract: answer hooks that render reply text set `noticed`,
    and every hook that opens a NEW timeline entry (a thinking block, a tool
    call, a todo update — never a tool result, so in-flight work finishes where
    it began) calls `begin_entry()` first, which `rotate()`s once when a notice
    streamed in between. Overrides must keep doing both."""

    message_id: IMMessageID | None = None
    noticed: bool = False
    # The first `ReplySegment` seen sets this to the referenced message id.
    # Annotation-only here (no value) so dataclass subclasses that redeclare it
    # keep their own defaults; `capture_reply` assigns it, the concrete states
    # provide the field/default, and `reply_captured` keeps the first one from
    # being overwritten.
    reply_to: str | None
    reply_captured: bool = False
    # Per-run context the timeline feeler injects when it creates the state, so
    # `drive` can present deferred-action batches.
    address: ChannelAddress
    ask_questions: QuestionFeeler
    approvals: ApprovalFeeler
    oauth: OAuthFeeler
    deferred_actions: DeferredActionManager
    # Bound by the siphon before driving a run, alongside its channel's timeline.
    artifacts: ArtifactFeeler
    subagent_timelines: dict[str, OpenSubagentTimeline] | None = None

    @asynccontextmanager
    async def open_subagent(
        self,
        activity: SubagentStartedEvent,
    ) -> AsyncGenerator[SubagentTimelineState]:
        """Ignore child activity when this timeline has no streaming renderer."""
        yield SubagentTimelineState()

    async def start_subagent(self, event: SubagentStartedEvent) -> None:
        """Open an accomplice's own timeline; a renderer that fails is logged and
        never interrupts the parent's stream."""
        context = self.open_subagent(event)
        try:
            state = await context.__aenter__()
        except Exception:
            logger.warning(
                "Subagent timeline %s failed to open",
                event.invocation_id,
                exc_info=True,
            )
            return
        if self.subagent_timelines is None:
            self.subagent_timelines = {}
        self.subagent_timelines[event.invocation_id] = OpenSubagentTimeline(
            context=context,
            state=state,
        )

    async def finish_subagent(self, event: SubagentSettledEvent) -> None:
        if self.subagent_timelines is None:
            return
        timeline = self.subagent_timelines.pop(event.invocation_id, None)
        if timeline is None:
            return
        if event.response:
            try:
                await timeline.state.append_response(event.response)
            except Exception:
                logger.warning(
                    "Subagent timeline %s failed to render its response",
                    event.invocation_id,
                    exc_info=True,
                )
        try:
            await timeline.state.settle(event.status, event.detail)
        except Exception:
            logger.warning("Subagent timeline failed to settle", exc_info=True)
        try:
            await timeline.context.__aexit__(None, None, None)
        except Exception:
            logger.warning("Subagent timeline failed to close", exc_info=True)

    async def drive(self, stream: AsyncIterator[WireEvent]) -> None:
        """The single event→render dispatch over a run's wire events, as the nerve
        carries them; each typed event is drawn onto this state, which renders
        itself."""
        skipped_tools: set[str] = set()
        failed = False
        answered = False
        final_output: ChannelOutput = None
        async for event in stream:
            # Ahead of the render-failure skip: an accomplice's surface still settles.
            if isinstance(event, SubagentStartedEvent):
                await self.start_subagent(event)
            elif isinstance(event, SubagentSettledEvent):
                await self.finish_subagent(event)
            if failed:
                continue  # keep draining the stream even after a render failure
            try:
                match event:
                    case PartStartEvent(part=ThinkingPart(content=content)):
                        await self.thinking_start()
                        if content:
                            await self.thinking_delta(content)
                    case PartDeltaEvent(delta=ThinkingPartDelta(content_delta=delta)):
                        await self.thinking_delta(delta or "")
                    case PartEndEvent(part=ThinkingPart()):
                        await self.thinking_end()
                    case PartStartEvent(part=TextPart(content=content)):
                        answered = True
                        await self.answer_start()
                        if content:
                            await self.answer_delta(content)
                    case PartDeltaEvent(delta=TextPartDelta(content_delta=delta)):
                        answered = answered or bool(delta)
                        await self.answer_delta(delta or "")
                    case PartEndEvent(part=TextPart()):
                        await self.answer_end()
                    case FunctionToolCallEvent() | OutputToolCallEvent():
                        if should_skip_plan_tool(event.part.tool_name):
                            if event.part.tool_call_id:
                                skipped_tools.add(event.part.tool_call_id)
                        else:
                            await self.tool_start(event)
                    case FunctionToolResultEvent() | OutputToolResultEvent():
                        part = event.part
                        tool_call_id = getattr(part, "tool_call_id", None)
                        if should_skip_plan_tool(part.tool_name or "") or (
                            tool_call_id is not None and tool_call_id in skipped_tools
                        ):
                            skipped_tools.discard(tool_call_id or "")
                        else:
                            await self.tool_end(event)
                    case ResultSegmentEvent(segment=ReplySegment(data=reply_data)):
                        # The reply handle threads the message; it is not rendered.
                        self.capture_reply(reply_data["id"])
                    case ResultSegmentEvent():
                        answered = True
                        await self.answer_segment(event.segment)
                    case ResultTextDeltaEvent():
                        answered = True
                        await self.answer_delta(event.delta)
                    case (
                        TodoCreatedEvent()
                        | TodoUpdatedEvent()
                        | TodoStatusChangedEvent()
                        | TodoCompletedEvent()
                        | TodoDeletedEvent()
                    ):
                        await self.todo(event)
                    case MessageSentEvent():
                        answered = True
                        await self.message_sent(event)
                    case OAuthAuthorizationEvent():
                        answered = True
                        await self.oauth_authorization(event)
                    case ArtifactAvailableEvent():
                        await self.artifact_available(event)
                    case ActionBatchEvent():
                        answered = answered or bool(event.questions or event.approvals)
                        await self.present_actions(event)
                    case RunResultEvent():
                        final_output = event.output  # for the fallback below
                    case _:
                        pass  # PartEnd of other parts / passthrough
            except Exception:
                logger.warning(
                    "Channel %s: timeline render failed",
                    self.address.channel_tentacle_id,
                    exc_info=True,
                )
                failed = True
        # `failed` means a render already raised (the transport threw); this
        # "never streamed" backup is outside the try/except, so re-rendering after
        # a failure would propagate and abort the drain (or double-send a partial).
        # It only fires when nothing was drawn — not as an error-recovery retry.
        if (
            not failed
            and not answered
            and isinstance(final_output, Sequence)
            and all(isinstance(segment, Segment) for segment in final_output)
        ):
            reply_to, body = split_reply(cast("list[MessageSegment]", final_output))
            self.capture_reply(reply_to)
            for segment in body:
                await self.answer_segment(segment)
        elif (
            not failed
            and not answered
            and final_output is not None
            and not isinstance(final_output, DeferredToolRequests)
        ):
            # The reply never streamed; render the final output once so the turn
            # is not left blank. A deferral is the exception: the run parked rather
            # than finished, and there is nothing to show for it here — a `teleport`
            # resumes in the place it moved to, and a batch put up for review has
            # already drawn itself through `present_actions`. `str()` on one is its
            # own repr, tool call and metadata and all, which is what this used to
            # post into the chat.
            await self.answer_delta(str(final_output))

    async def present_actions(self, event: ActionBatchEvent) -> None:
        """Render a deferred-action batch as a unit, record each presented
        action's platform message id, then settle the surface."""
        if event.questions:
            message_ids = await self.ask_questions.present(
                self.address,
                event.questions,
            )
            for action in event.questions:
                await self.deferred_actions.mark_action_presented(
                    action.id, message_ids.get(action.id)
                )
        if event.approvals:
            message_ids = await self.approvals.present(self.address, event.approvals)
            for action in event.approvals:
                await self.deferred_actions.mark_action_presented(
                    action.id, message_ids.get(action.id)
                )
        # The cards are the load-bearing part and are already up; the settle is a
        # UI hint, and a render hiccup must not fail the presentation.
        try:
            await self.actions_presented()
        except Exception:
            logger.warning(
                "Channel %s: timeline failed to settle after presenting actions",
                self.address.channel_tentacle_id,
                exc_info=True,
            )

    async def actions_presented(self) -> None:
        """A deferred-action batch was just presented in this timeline's thread.

        A pending action does not mean the run has stopped: a live run carries on
        past it. Hooks may flush buffered output, but must keep active thinking and
        answer surfaces open for more events. Drain background flushers before
        flushing their buffers here to avoid overlapping writes. The base renders
        no status, so there is nothing to settle."""

    async def begin_entry(self) -> None:
        """A new timeline entry is opening: if answer content streamed since
        the last rotation, that content was a mid-run notice — rotate first."""
        if not self.noticed:
            return
        self.noticed = False
        await self.rotate()

    async def rotate(self) -> None:
        """A mid-run notice happened: finalize the current timeline surface
        (plan message, cards, buffered text) so the notice stays visible where
        it streamed, and let the new entries open a fresh surface below it.
        Work still in flight finishes — and closes — the previous surface."""

    async def thinking_start(self) -> None:
        await self.begin_entry()

    async def thinking_delta(self, text: str) -> None: ...
    async def thinking_end(self) -> None: ...
    async def answer_start(self) -> None: ...

    async def answer_delta(self, text: str) -> None:
        self.noticed = True

    async def answer_end(self) -> None: ...

    async def answer_segment(self, segment: MessageSegment) -> None:
        """One completed message segment from a segment-list output. Platforms
        with a media transport override this to render images/cards natively; the
        default renders the segment's text form."""
        await self.answer_delta(str(segment))

    async def tool_start(
        self, event: FunctionToolCallEvent | OutputToolCallEvent
    ) -> None:
        await self.begin_entry()

    async def tool_end(
        self, event: FunctionToolResultEvent | OutputToolResultEvent
    ) -> None: ...

    async def todo(self, event: TodoEvent) -> None:
        await self.begin_entry()

    def capture_reply(self, reply_to: str | None) -> None:
        """Apply the first reply target seen this run; ignore later replies."""
        if reply_to is not None and not self.reply_captured:
            self.reply_to = reply_to
            self.reply_captured = True

    async def message_sent(self, event: MessageSentEvent) -> None:
        """A `send` tool delivered these segments mid-run. Render them as
        reply content (images/cards land natively via `answer_segment`), then
        `begin_entry()` to rotate the just-rendered notice out as its own message —
        the run continues afterward, so the final reply lands separately."""
        reply_to, body = split_reply(event.segments)
        self.capture_reply(reply_to)
        for segment in body:
            await self.answer_segment(segment)
        await self.begin_entry()

    async def oauth_authorization(self, event: OAuthAuthorizationEvent) -> None:
        """An integration is waiting for this user to authorize it.

        Rotates the open entry first: the authorization goes out as its own message
        — a card where the platform has them — so whatever streamed before it stays
        above rather than merging into it.
        """
        await self.begin_entry()
        await self.oauth.present(self.address, event)

    async def artifact_available(self, event: ArtifactAvailableEvent) -> None:
        """Flush preceding output so the card separates it from the next entry.

        The artifact has its own ledger row and never supplies ``message_id`` for
        the final reply. It is not answer text and does not suppress that reply.
        """
        if not self.artifacts.enabled:
            return
        await self.rotate()
        self.noticed = False
        await self.artifacts.present(self.address, event)

message_id class-attribute instance-attribute

message_id: IMMessageID | None = None

noticed class-attribute instance-attribute

noticed: bool = False

reply_to instance-attribute

reply_to: str | None

reply_captured class-attribute instance-attribute

reply_captured: bool = False

address instance-attribute

address: ChannelAddress

ask_questions instance-attribute

ask_questions: QuestionFeeler

approvals instance-attribute

approvals: ApprovalFeeler

oauth instance-attribute

oauth: OAuthFeeler

deferred_actions instance-attribute

deferred_actions: DeferredActionManager

artifacts instance-attribute

artifacts: ArtifactFeeler

subagent_timelines class-attribute instance-attribute

subagent_timelines: (
    dict[str, OpenSubagentTimeline] | None
) = None

open_subagent async

open_subagent(
    activity: SubagentStartedEvent,
) -> AsyncGenerator[SubagentTimelineState]

Ignore child activity when this timeline has no streaming renderer.

Source code in octomate/tentacles/feelers/output.py

@asynccontextmanager
async def open_subagent(
    self,
    activity: SubagentStartedEvent,
) -> AsyncGenerator[SubagentTimelineState]:
    """Ignore child activity when this timeline has no streaming renderer."""
    yield SubagentTimelineState()

start_subagent async

start_subagent(event: SubagentStartedEvent) -> None

Open an accomplice's own timeline; a renderer that fails is logged and never interrupts the parent's stream.

Source code in octomate/tentacles/feelers/output.py

async def start_subagent(self, event: SubagentStartedEvent) -> None:
    """Open an accomplice's own timeline; a renderer that fails is logged and
    never interrupts the parent's stream."""
    context = self.open_subagent(event)
    try:
        state = await context.__aenter__()
    except Exception:
        logger.warning(
            "Subagent timeline %s failed to open",
            event.invocation_id,
            exc_info=True,
        )
        return
    if self.subagent_timelines is None:
        self.subagent_timelines = {}
    self.subagent_timelines[event.invocation_id] = OpenSubagentTimeline(
        context=context,
        state=state,
    )

finish_subagent async

finish_subagent(event: SubagentSettledEvent) -> None

Source code in octomate/tentacles/feelers/output.py

async def finish_subagent(self, event: SubagentSettledEvent) -> None:
    if self.subagent_timelines is None:
        return
    timeline = self.subagent_timelines.pop(event.invocation_id, None)
    if timeline is None:
        return
    if event.response:
        try:
            await timeline.state.append_response(event.response)
        except Exception:
            logger.warning(
                "Subagent timeline %s failed to render its response",
                event.invocation_id,
                exc_info=True,
            )
    try:
        await timeline.state.settle(event.status, event.detail)
    except Exception:
        logger.warning("Subagent timeline failed to settle", exc_info=True)
    try:
        await timeline.context.__aexit__(None, None, None)
    except Exception:
        logger.warning("Subagent timeline failed to close", exc_info=True)

drive async

drive(stream: AsyncIterator[WireEvent]) -> None

The single event→render dispatch over a run's wire events, as the nerve carries them; each typed event is drawn onto this state, which renders itself.

Source code in octomate/tentacles/feelers/output.py

async def drive(self, stream: AsyncIterator[WireEvent]) -> None:
    """The single event→render dispatch over a run's wire events, as the nerve
    carries them; each typed event is drawn onto this state, which renders
    itself."""
    skipped_tools: set[str] = set()
    failed = False
    answered = False
    final_output: ChannelOutput = None
    async for event in stream:
        # Ahead of the render-failure skip: an accomplice's surface still settles.
        if isinstance(event, SubagentStartedEvent):
            await self.start_subagent(event)
        elif isinstance(event, SubagentSettledEvent):
            await self.finish_subagent(event)
        if failed:
            continue  # keep draining the stream even after a render failure
        try:
            match event:
                case PartStartEvent(part=ThinkingPart(content=content)):
                    await self.thinking_start()
                    if content:
                        await self.thinking_delta(content)
                case PartDeltaEvent(delta=ThinkingPartDelta(content_delta=delta)):
                    await self.thinking_delta(delta or "")
                case PartEndEvent(part=ThinkingPart()):
                    await self.thinking_end()
                case PartStartEvent(part=TextPart(content=content)):
                    answered = True
                    await self.answer_start()
                    if content:
                        await self.answer_delta(content)
                case PartDeltaEvent(delta=TextPartDelta(content_delta=delta)):
                    answered = answered or bool(delta)
                    await self.answer_delta(delta or "")
                case PartEndEvent(part=TextPart()):
                    await self.answer_end()
                case FunctionToolCallEvent() | OutputToolCallEvent():
                    if should_skip_plan_tool(event.part.tool_name):
                        if event.part.tool_call_id:
                            skipped_tools.add(event.part.tool_call_id)
                    else:
                        await self.tool_start(event)
                case FunctionToolResultEvent() | OutputToolResultEvent():
                    part = event.part
                    tool_call_id = getattr(part, "tool_call_id", None)
                    if should_skip_plan_tool(part.tool_name or "") or (
                        tool_call_id is not None and tool_call_id in skipped_tools
                    ):
                        skipped_tools.discard(tool_call_id or "")
                    else:
                        await self.tool_end(event)
                case ResultSegmentEvent(segment=ReplySegment(data=reply_data)):
                    # The reply handle threads the message; it is not rendered.
                    self.capture_reply(reply_data["id"])
                case ResultSegmentEvent():
                    answered = True
                    await self.answer_segment(event.segment)
                case ResultTextDeltaEvent():
                    answered = True
                    await self.answer_delta(event.delta)
                case (
                    TodoCreatedEvent()
                    | TodoUpdatedEvent()
                    | TodoStatusChangedEvent()
                    | TodoCompletedEvent()
                    | TodoDeletedEvent()
                ):
                    await self.todo(event)
                case MessageSentEvent():
                    answered = True
                    await self.message_sent(event)
                case OAuthAuthorizationEvent():
                    answered = True
                    await self.oauth_authorization(event)
                case ArtifactAvailableEvent():
                    await self.artifact_available(event)
                case ActionBatchEvent():
                    answered = answered or bool(event.questions or event.approvals)
                    await self.present_actions(event)
                case RunResultEvent():
                    final_output = event.output  # for the fallback below
                case _:
                    pass  # PartEnd of other parts / passthrough
        except Exception:
            logger.warning(
                "Channel %s: timeline render failed",
                self.address.channel_tentacle_id,
                exc_info=True,
            )
            failed = True
    # `failed` means a render already raised (the transport threw); this
    # "never streamed" backup is outside the try/except, so re-rendering after
    # a failure would propagate and abort the drain (or double-send a partial).
    # It only fires when nothing was drawn — not as an error-recovery retry.
    if (
        not failed
        and not answered
        and isinstance(final_output, Sequence)
        and all(isinstance(segment, Segment) for segment in final_output)
    ):
        reply_to, body = split_reply(cast("list[MessageSegment]", final_output))
        self.capture_reply(reply_to)
        for segment in body:
            await self.answer_segment(segment)
    elif (
        not failed
        and not answered
        and final_output is not None
        and not isinstance(final_output, DeferredToolRequests)
    ):
        # The reply never streamed; render the final output once so the turn
        # is not left blank. A deferral is the exception: the run parked rather
        # than finished, and there is nothing to show for it here — a `teleport`
        # resumes in the place it moved to, and a batch put up for review has
        # already drawn itself through `present_actions`. `str()` on one is its
        # own repr, tool call and metadata and all, which is what this used to
        # post into the chat.
        await self.answer_delta(str(final_output))

present_actions async

present_actions(event: ActionBatchEvent) -> None

Render a deferred-action batch as a unit, record each presented action's platform message id, then settle the surface.

Source code in octomate/tentacles/feelers/output.py

async def present_actions(self, event: ActionBatchEvent) -> None:
    """Render a deferred-action batch as a unit, record each presented
    action's platform message id, then settle the surface."""
    if event.questions:
        message_ids = await self.ask_questions.present(
            self.address,
            event.questions,
        )
        for action in event.questions:
            await self.deferred_actions.mark_action_presented(
                action.id, message_ids.get(action.id)
            )
    if event.approvals:
        message_ids = await self.approvals.present(self.address, event.approvals)
        for action in event.approvals:
            await self.deferred_actions.mark_action_presented(
                action.id, message_ids.get(action.id)
            )
    # The cards are the load-bearing part and are already up; the settle is a
    # UI hint, and a render hiccup must not fail the presentation.
    try:
        await self.actions_presented()
    except Exception:
        logger.warning(
            "Channel %s: timeline failed to settle after presenting actions",
            self.address.channel_tentacle_id,
            exc_info=True,
        )

actions_presented async

actions_presented() -> None

A deferred-action batch was just presented in this timeline's thread.

A pending action does not mean the run has stopped: a live run carries on past it. Hooks may flush buffered output, but must keep active thinking and answer surfaces open for more events. Drain background flushers before flushing their buffers here to avoid overlapping writes. The base renders no status, so there is nothing to settle.

Source code in octomate/tentacles/feelers/output.py

async def actions_presented(self) -> None:
    """A deferred-action batch was just presented in this timeline's thread.

    A pending action does not mean the run has stopped: a live run carries on
    past it. Hooks may flush buffered output, but must keep active thinking and
    answer surfaces open for more events. Drain background flushers before
    flushing their buffers here to avoid overlapping writes. The base renders
    no status, so there is nothing to settle."""

begin_entry async

begin_entry() -> None

A new timeline entry is opening: if answer content streamed since the last rotation, that content was a mid-run notice — rotate first.

Source code in octomate/tentacles/feelers/output.py

async def begin_entry(self) -> None:
    """A new timeline entry is opening: if answer content streamed since
    the last rotation, that content was a mid-run notice — rotate first."""
    if not self.noticed:
        return
    self.noticed = False
    await self.rotate()

rotate async

rotate() -> None

A mid-run notice happened: finalize the current timeline surface (plan message, cards, buffered text) so the notice stays visible where it streamed, and let the new entries open a fresh surface below it. Work still in flight finishes — and closes — the previous surface.

Source code in octomate/tentacles/feelers/output.py

async def rotate(self) -> None:
    """A mid-run notice happened: finalize the current timeline surface
    (plan message, cards, buffered text) so the notice stays visible where
    it streamed, and let the new entries open a fresh surface below it.
    Work still in flight finishes — and closes — the previous surface."""

thinking_start async

thinking_start() -> None

Source code in octomate/tentacles/feelers/output.py

async def thinking_start(self) -> None:
    await self.begin_entry()

thinking_delta async

thinking_delta(text: str) -> None

Source code in octomate/tentacles/feelers/output.py

async def thinking_delta(self, text: str) -> None: ...

thinking_end async

thinking_end() -> None

Source code in octomate/tentacles/feelers/output.py

async def thinking_end(self) -> None: ...

answer_start async

answer_start() -> None

Source code in octomate/tentacles/feelers/output.py

async def answer_start(self) -> None: ...

answer_delta async

answer_delta(text: str) -> None

Source code in octomate/tentacles/feelers/output.py

async def answer_delta(self, text: str) -> None:
    self.noticed = True

answer_end async

answer_end() -> None

Source code in octomate/tentacles/feelers/output.py

async def answer_end(self) -> None: ...

answer_segment async

answer_segment(segment: MessageSegment) -> None

One completed message segment from a segment-list output. Platforms with a media transport override this to render images/cards natively; the default renders the segment's text form.

Source code in octomate/tentacles/feelers/output.py

async def answer_segment(self, segment: MessageSegment) -> None:
    """One completed message segment from a segment-list output. Platforms
    with a media transport override this to render images/cards natively; the
    default renders the segment's text form."""
    await self.answer_delta(str(segment))

tool_start async

tool_start(
    event: FunctionToolCallEvent | OutputToolCallEvent,
) -> None

Source code in octomate/tentacles/feelers/output.py

async def tool_start(
    self, event: FunctionToolCallEvent | OutputToolCallEvent
) -> None:
    await self.begin_entry()

tool_end async

tool_end(
    event: FunctionToolResultEvent | OutputToolResultEvent,
) -> None

Source code in octomate/tentacles/feelers/output.py

async def tool_end(
    self, event: FunctionToolResultEvent | OutputToolResultEvent
) -> None: ...

todo async

todo(event: TodoEvent) -> None

Source code in octomate/tentacles/feelers/output.py

async def todo(self, event: TodoEvent) -> None:
    await self.begin_entry()

capture_reply

capture_reply(reply_to: str | None) -> None

Apply the first reply target seen this run; ignore later replies.

Source code in octomate/tentacles/feelers/output.py

def capture_reply(self, reply_to: str | None) -> None:
    """Apply the first reply target seen this run; ignore later replies."""
    if reply_to is not None and not self.reply_captured:
        self.reply_to = reply_to
        self.reply_captured = True

message_sent async

message_sent(event: MessageSentEvent) -> None

A send tool delivered these segments mid-run. Render them as reply content (images/cards land natively via answer_segment), then begin_entry() to rotate the just-rendered notice out as its own message — the run continues afterward, so the final reply lands separately.

Source code in octomate/tentacles/feelers/output.py

async def message_sent(self, event: MessageSentEvent) -> None:
    """A `send` tool delivered these segments mid-run. Render them as
    reply content (images/cards land natively via `answer_segment`), then
    `begin_entry()` to rotate the just-rendered notice out as its own message —
    the run continues afterward, so the final reply lands separately."""
    reply_to, body = split_reply(event.segments)
    self.capture_reply(reply_to)
    for segment in body:
        await self.answer_segment(segment)
    await self.begin_entry()

oauth_authorization async

oauth_authorization(event: OAuthAuthorizationEvent) -> None

An integration is waiting for this user to authorize it.

Rotates the open entry first: the authorization goes out as its own message — a card where the platform has them — so whatever streamed before it stays above rather than merging into it.

Source code in octomate/tentacles/feelers/output.py

async def oauth_authorization(self, event: OAuthAuthorizationEvent) -> None:
    """An integration is waiting for this user to authorize it.

    Rotates the open entry first: the authorization goes out as its own message
    — a card where the platform has them — so whatever streamed before it stays
    above rather than merging into it.
    """
    await self.begin_entry()
    await self.oauth.present(self.address, event)

artifact_available async

artifact_available(event: ArtifactAvailableEvent) -> None

Flush preceding output so the card separates it from the next entry.

The artifact has its own ledger row and never supplies message_id for the final reply. It is not answer text and does not suppress that reply.

Source code in octomate/tentacles/feelers/output.py

async def artifact_available(self, event: ArtifactAvailableEvent) -> None:
    """Flush preceding output so the card separates it from the next entry.

    The artifact has its own ledger row and never supplies ``message_id`` for
    the final reply. It is not answer text and does not suppress that reply.
    """
    if not self.artifacts.enabled:
        return
    await self.rotate()
    self.noticed = False
    await self.artifacts.present(self.address, event)

TimelineFeeler

Bases: Protocol

Opens a per-run TimelineState for a channel. The per-channel Feelers.timeline is a single stateless instance: the channel's siphon does async with feelers.timeline.open(address) as state: — entering acquires the platform resource and yields the per-run hook (a card, a stream session, …), or the feeler itself when there is nothing to set up; state.drive(events) then renders each event onto it; exiting releases the resource and sets message_id.

Source code in octomate/tentacles/feelers/output.py

class TimelineFeeler(Protocol):
    """Opens a per-run `TimelineState` for a channel. The per-channel `Feelers.timeline`
    is a single stateless instance: the channel's siphon does
    `async with feelers.timeline.open(address) as state:` — entering acquires the
    platform resource and yields the per-run hook (a card, a stream session, …), or
    the feeler itself when there is nothing to set up; `state.drive(events)` then
    renders each event onto it; exiting releases the resource and sets `message_id`."""

    def open(
        self, address: ChannelAddress
    ) -> AbstractAsyncContextManager[TimelineState]: ...

open

open(
    address: ChannelAddress,
) -> AbstractAsyncContextManager[TimelineState]

Source code in octomate/tentacles/feelers/output.py

def open(
    self, address: ChannelAddress
) -> AbstractAsyncContextManager[TimelineState]: ...

DefaultMarkdownFeeler

Sends markdown through a channel's chromo and ink as plain messages.

Source code in octomate/tentacles/feelers/output.py

class DefaultMarkdownFeeler[RawT, MessageT]:
    """Sends markdown through a channel's chromo and ink as plain messages."""

    def __init__(self, *, ink: Ink[MessageT], chromo: Chromo[RawT, MessageT]) -> None:
        self.ink = ink
        self.chromo = chromo

    async def present(
        self,
        address: ChannelAddress,
        markdown: str,
    ) -> IMMessageID | None:
        return await present_markdown(
            ink=self.ink,
            chromo=self.chromo,
            address=address,
            markdown=markdown,
        )

ink instance-attribute

ink = ink

chromo instance-attribute

chromo = chromo

present async

present(
    address: ChannelAddress, markdown: str
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/output.py

async def present(
    self,
    address: ChannelAddress,
    markdown: str,
) -> IMMessageID | None:
    return await present_markdown(
        ink=self.ink,
        chromo=self.chromo,
        address=address,
        markdown=markdown,
    )

SegmentsFeeler

Bases: Protocol

Sends completed output segments to a conversation as one platform message.

Platforms with native media transport (e.g. NapCat images/files) implement this to deliver segments natively; the default joins their text forms.

Source code in octomate/tentacles/feelers/output.py

class SegmentsFeeler(Protocol):
    """Sends completed output segments to a conversation as one platform message.

    Platforms with native media transport (e.g. NapCat images/files) implement
    this to deliver segments natively; the default joins their text forms.
    """

    async def present(
        self,
        address: ChannelAddress,
        segments: list[MessageSegment],
    ) -> IMMessageID | None: ...

present async

present(
    address: ChannelAddress, segments: list[MessageSegment]
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/output.py

async def present(
    self,
    address: ChannelAddress,
    segments: list[MessageSegment],
) -> IMMessageID | None: ...

DefaultSegmentsFeeler

Sends output segments through a channel's chromo and ink, one message per payload.

Source code in octomate/tentacles/feelers/output.py

class DefaultSegmentsFeeler[RawT, MessageT]:
    """Sends output segments through a channel's chromo and ink, one message per
    payload."""

    def __init__(self, *, ink: Ink[MessageT], chromo: Chromo[RawT, MessageT]) -> None:
        self.ink = ink
        self.chromo = chromo

    async def present(
        self,
        address: ChannelAddress,
        segments: list[MessageSegment],
    ) -> IMMessageID | None:
        with channel_logfire.span(
            "default.segments.present",
            channel_id=address.channel_tentacle_id,
            conversation_address=str(address),
            segments=len(segments),
        ) as span:
            reply_to, body = split_reply(segments)
            messages = await self.chromo.outbound_segments(body)
            if not messages:
                return None
            chat_id = address.chat_id or address.user_id
            first_message_id: IMMessageID | None = None
            for message in messages:
                message_id = await self.ink.send_message(
                    chat_id,
                    address.chat_type,
                    [message],
                    reply_to=reply_to,
                    channel_thread_id=address.channel_thread_id or chat_id,
                )
                first_message_id = first_message_id or message_id
            span.set_attribute("message_id", str(first_message_id))
            return first_message_id

ink instance-attribute

ink = ink

chromo instance-attribute

chromo = chromo

present async

present(
    address: ChannelAddress, segments: list[MessageSegment]
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/output.py

async def present(
    self,
    address: ChannelAddress,
    segments: list[MessageSegment],
) -> IMMessageID | None:
    with channel_logfire.span(
        "default.segments.present",
        channel_id=address.channel_tentacle_id,
        conversation_address=str(address),
        segments=len(segments),
    ) as span:
        reply_to, body = split_reply(segments)
        messages = await self.chromo.outbound_segments(body)
        if not messages:
            return None
        chat_id = address.chat_id or address.user_id
        first_message_id: IMMessageID | None = None
        for message in messages:
            message_id = await self.ink.send_message(
                chat_id,
                address.chat_type,
                [message],
                reply_to=reply_to,
                channel_thread_id=address.channel_thread_id or chat_id,
            )
            first_message_id = first_message_id or message_id
        span.set_attribute("message_id", str(first_message_id))
        return first_message_id

DefaultTimelineState dataclass

Bases: TimelineState

The timeline for a channel with no live transport: collects the answer and todos, and sends them as one message when the run ends.

Source code in octomate/tentacles/feelers/output.py

@dataclass
class DefaultTimelineState[RawT, MessageT](TimelineState):
    """The timeline for a channel with no live transport: collects the answer and
    todos, and sends them as one message when the run ends."""

    segments: DefaultSegmentsFeeler[RawT, MessageT]
    address: ChannelAddress
    ask_questions: QuestionFeeler
    approvals: ApprovalFeeler
    oauth: OAuthFeeler
    deferred_actions: DeferredActionManager
    parts: list[MessageSegment] = field(default_factory=list)
    text_part: MarkdownSegment | None = None
    todos: dict[str, Todo] = field(default_factory=dict)
    message_id: IMMessageID | None = None
    reply_to: str | None = None

    async def answer_delta(self, text: str) -> None:
        self.noticed = True
        if self.text_part is None:
            self.text_part = MarkdownSegment(data={"text": text})
            self.parts.append(self.text_part)
        else:
            self.text_part.data["text"] += text

    async def answer_segment(self, segment: MessageSegment) -> None:
        self.noticed = True
        self.text_part = None
        self.parts.append(segment)

    async def todo(self, event: TodoEvent) -> None:
        await self.begin_entry()
        # No live transport: keep the latest snapshot; it rides as a checklist
        # under the final message.
        if isinstance(event, TodoDeletedEvent):
            self.todos.pop(event.todo.ref, None)
        else:
            self.todos[event.todo.ref] = event.todo

    async def rotate(self) -> None:
        # The accumulated text was a mid-run notice: send it as its own message
        # so the final reply arrives separately.
        await self.send_parts()

    async def send_parts(self, *, todo_block: bool = False) -> None:
        if todo_block and self.todos:
            block = "Tasks:\n" + "\n".join(render_todo_lines(self.todos.values()))
            await self.answer_delta(f"\n\n{block}" if self.parts else block)
        segments, self.parts = self.parts, []
        self.text_part = None
        if not segments:
            return
        logger.warning(
            "Channel %s: timeline has no streaming transport; "
            "sending the reply as a single message",
            self.address.channel_tentacle_id,
        )
        if self.reply_to is not None:
            segments.insert(0, ReplySegment(data={"id": self.reply_to}))
        self.message_id = await self.segments.present(self.address, segments)

segments instance-attribute

segments: DefaultSegmentsFeeler[RawT, MessageT]

address instance-attribute

address: ChannelAddress

ask_questions instance-attribute

ask_questions: QuestionFeeler

approvals instance-attribute

approvals: ApprovalFeeler

oauth instance-attribute

oauth: OAuthFeeler

deferred_actions instance-attribute

deferred_actions: DeferredActionManager

parts class-attribute instance-attribute

parts: list[MessageSegment] = field(default_factory=list)

text_part class-attribute instance-attribute

text_part: MarkdownSegment | None = None

todos class-attribute instance-attribute

todos: dict[str, Todo] = field(default_factory=dict)

message_id class-attribute instance-attribute

message_id: IMMessageID | None = None

reply_to class-attribute instance-attribute

reply_to: str | None = None

answer_delta async

answer_delta(text: str) -> None

Source code in octomate/tentacles/feelers/output.py

async def answer_delta(self, text: str) -> None:
    self.noticed = True
    if self.text_part is None:
        self.text_part = MarkdownSegment(data={"text": text})
        self.parts.append(self.text_part)
    else:
        self.text_part.data["text"] += text

answer_segment async

answer_segment(segment: MessageSegment) -> None

Source code in octomate/tentacles/feelers/output.py

async def answer_segment(self, segment: MessageSegment) -> None:
    self.noticed = True
    self.text_part = None
    self.parts.append(segment)

todo async

todo(event: TodoEvent) -> None

Source code in octomate/tentacles/feelers/output.py

async def todo(self, event: TodoEvent) -> None:
    await self.begin_entry()
    # No live transport: keep the latest snapshot; it rides as a checklist
    # under the final message.
    if isinstance(event, TodoDeletedEvent):
        self.todos.pop(event.todo.ref, None)
    else:
        self.todos[event.todo.ref] = event.todo

rotate async

rotate() -> None

Source code in octomate/tentacles/feelers/output.py

async def rotate(self) -> None:
    # The accumulated text was a mid-run notice: send it as its own message
    # so the final reply arrives separately.
    await self.send_parts()

send_parts async

send_parts(*, todo_block: bool = False) -> None

Source code in octomate/tentacles/feelers/output.py

async def send_parts(self, *, todo_block: bool = False) -> None:
    if todo_block and self.todos:
        block = "Tasks:\n" + "\n".join(render_todo_lines(self.todos.values()))
        await self.answer_delta(f"\n\n{block}" if self.parts else block)
    segments, self.parts = self.parts, []
    self.text_part = None
    if not segments:
        return
    logger.warning(
        "Channel %s: timeline has no streaming transport; "
        "sending the reply as a single message",
        self.address.channel_tentacle_id,
    )
    if self.reply_to is not None:
        segments.insert(0, ReplySegment(data={"id": self.reply_to}))
    self.message_id = await self.segments.present(self.address, segments)

DefaultTimelineFeeler

Bases: TimelineFeeler

Answer-only timeline for platforms with no streaming transport (NapCat).

Stateless; open(address) yields a fresh DefaultTimelineState. Thinking/tool are inherited no-ops; text deltas are accumulated and native segments are preserved, then sent on exit (a mid-run notice rotates out as its own message), with a warning that streaming was unavailable.

Source code in octomate/tentacles/feelers/output.py

class DefaultTimelineFeeler[RawT, MessageT](TimelineFeeler):
    """Answer-only timeline for platforms with no streaming transport (NapCat).

    Stateless; `open(address)` yields a fresh `DefaultTimelineState`. Thinking/tool
    are inherited no-ops; text deltas are accumulated and native segments are
    preserved, then sent on exit (a mid-run notice rotates out as its own
    message), with a warning that streaming was unavailable.
    """

    segments: DefaultSegmentsFeeler[RawT, MessageT]

    def __init__(
        self,
        *,
        ink: Ink[MessageT],
        chromo: Chromo[RawT, MessageT],
        ask_questions: QuestionFeeler,
        approvals: ApprovalFeeler,
        oauth: OAuthFeeler,
        deferred_actions: DeferredActionManager,
    ) -> None:
        self.segments = DefaultSegmentsFeeler(ink=ink, chromo=chromo)
        self.ask_questions = ask_questions
        self.approvals = approvals
        self.oauth = oauth
        self.deferred_actions = deferred_actions

    @asynccontextmanager
    async def open(
        self, address: ChannelAddress
    ) -> AsyncGenerator[DefaultTimelineState[RawT, MessageT]]:
        state = DefaultTimelineState(
            segments=self.segments,
            address=address,
            ask_questions=self.ask_questions,
            approvals=self.approvals,
            oauth=self.oauth,
            deferred_actions=self.deferred_actions,
        )
        try:
            yield state
        finally:
            await state.send_parts(todo_block=True)

segments instance-attribute

segments: DefaultSegmentsFeeler[RawT, MessageT] = (
    DefaultSegmentsFeeler(ink=ink, chromo=chromo)
)

ask_questions instance-attribute

ask_questions = ask_questions

approvals instance-attribute

approvals = approvals

oauth instance-attribute

oauth = oauth

deferred_actions instance-attribute

deferred_actions = deferred_actions

open async

open(
    address: ChannelAddress,
) -> AsyncGenerator[DefaultTimelineState[RawT, MessageT]]

Source code in octomate/tentacles/feelers/output.py

@asynccontextmanager
async def open(
    self, address: ChannelAddress
) -> AsyncGenerator[DefaultTimelineState[RawT, MessageT]]:
    state = DefaultTimelineState(
        segments=self.segments,
        address=address,
        ask_questions=self.ask_questions,
        approvals=self.approvals,
        oauth=self.oauth,
        deferred_actions=self.deferred_actions,
    )
    try:
        yield state
    finally:
        await state.send_parts(todo_block=True)

render_stream_event_delta

render_stream_event_delta(
    event: AgentStreamEvent | AgentRunResultEvent[OutputT],
) -> StreamEventDelta | None

Source code in octomate/tentacles/feelers/output.py

def render_stream_event_delta[
    OutputT: JsonValue | Sequence[MessageSegment] | DeferredToolRequests
](
    event: AgentStreamEvent | AgentRunResultEvent[OutputT],
) -> StreamEventDelta | None:
    if isinstance(event, PartStartEvent):
        if isinstance(event.part, TextPart):
            return StreamEventDelta(
                block=StreamBlock(id=f"answer-{event.index}", type="answer"),
                text=event.part.content,
            )
        if isinstance(event.part, ThinkingPart):
            return StreamEventDelta(
                block=StreamBlock(
                    id=f"thinking-{event.index}",
                    type="thinking",
                    title="Thinking",
                    foldable=True,
                ),
                text=event.part.content,
            )
    if isinstance(event, PartDeltaEvent):
        if isinstance(event.delta, TextPartDelta):
            return StreamEventDelta(
                block=StreamBlock(id=f"answer-{event.index}", type="answer"),
                text=event.delta.content_delta,
            )
        if isinstance(event.delta, ThinkingPartDelta):
            return StreamEventDelta(
                block=StreamBlock(
                    id=f"thinking-{event.index}",
                    type="thinking",
                    title="Thinking",
                    foldable=True,
                ),
                text=event.delta.content_delta or "",
            )
    if isinstance(event, FunctionToolCallEvent | OutputToolCallEvent):
        tool = event.part
        title = f"Tool call: {tool.tool_name}"
        if isinstance(event, OutputToolCallEvent):
            title = f"Output tool: {tool.tool_name}"
        return StreamEventDelta(
            block=StreamBlock(
                id=f"tool-call-{tool.tool_call_id}",
                type="tool_call",
                title=title,
                foldable=True,
                status="done",
            ),
            text=format_stream_value(cast(JsonValue, tool.args_as_dict())),
        )
    if isinstance(event, FunctionToolResultEvent | OutputToolResultEvent):
        part = event.part
        tool_name = part.tool_name or "output"
        title = f"Tool result: {tool_name}"
        if isinstance(event, OutputToolResultEvent):
            title = f"Output result: {tool_name}"
        return StreamEventDelta(
            block=StreamBlock(
                id=f"tool-result-{part.tool_call_id}",
                type="tool_result",
                title=title,
                foldable=True,
                status="done",
            ),
            text=format_stream_value(tool_result_text(part)),
        )
    return None

tool_result_text

tool_result_text(
    part: ToolReturnPart | RetryPromptPart,
) -> str

Source code in octomate/tentacles/feelers/output.py

def tool_result_text(part: ToolReturnPart | RetryPromptPart) -> str:
    if isinstance(part, ToolReturnPart):
        return part.model_response_str()
    return part.model_response()

format_stream_value

format_stream_value(
    value: JsonValue, *, max_chars: int = 2000
) -> str

Source code in octomate/tentacles/feelers/output.py

def format_stream_value(value: JsonValue, *, max_chars: int = 2000) -> str:
    if isinstance(value, str):
        text = value
    else:
        text = to_json(
            value,
            indent=2,
            ensure_ascii=False,
            fallback=str,
        ).decode()
    if len(text) <= max_chars:
        return text
    return f"{text[:max_chars].rstrip()}\n\n...[truncated]"

should_skip_plan_tool

should_skip_plan_tool(tool_name: str) -> bool

Source code in octomate/tentacles/feelers/output.py

def should_skip_plan_tool(tool_name: str) -> bool:
    return tool_name in SKIPPED_PLAN_TOOL_NAMES

format_field_name

format_field_name(value: object) -> str

Source code in octomate/tentacles/feelers/output.py

def format_field_name(value: object) -> str:
    text = str(value).replace("_", " ").strip()
    return text[:1].upper() + text[1:] if text else "Value"

humanize_tool_name

humanize_tool_name(tool_name: str) -> str

Source code in octomate/tentacles/feelers/output.py

def humanize_tool_name(tool_name: str) -> str:
    name = tool_name.rsplit("__", 1)[-1].replace("-", " ")
    return format_field_name(name)

truncate_task_detail

truncate_task_detail(
    text: str, *, max_chars: int = MAX_TASK_DETAIL_CHARS
) -> str

Source code in octomate/tentacles/feelers/output.py

def truncate_task_detail(text: str, *, max_chars: int = MAX_TASK_DETAIL_CHARS) -> str:
    if len(text) <= max_chars:
        return text
    return f"{text[: max_chars - 16].rstrip()}\n...[truncated]"

format_inline_value

format_inline_value(value: JsonValue) -> str

Source code in octomate/tentacles/feelers/output.py

def format_inline_value(value: JsonValue) -> str:
    if value is None:
        return "_None_"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, list):
        if not value:
            return "_None_"
        return ", ".join(format_inline_value(item) for item in value)
    if isinstance(value, dict):
        return "; ".join(
            f"{format_field_name(key)}: {format_inline_value(item)}"
            for key, item in value.items()
        )
    return truncate_task_detail(str(value), max_chars=500)

format_mapping_lines

format_mapping_lines(
    value: dict[str, Any],
    *,
    indent: int = 0,
    bold: str = "*",
) -> list[str]

Source code in octomate/tentacles/feelers/output.py

def format_mapping_lines(
    value: dict[str, Any], *, indent: int = 0, bold: str = "*"
) -> list[str]:
    lines: list[str] = []
    pad = "   " * indent
    for key, item in value.items():
        label = format_field_name(key)
        if isinstance(item, dict):
            lines.append(f"{pad}{bold}{label}:{bold}")
            lines.extend(format_mapping_lines(item, indent=indent + 1, bold=bold))
        elif (
            isinstance(item, list)
            and item
            and any(isinstance(entry, (dict, list)) for entry in item)
        ):
            lines.append(f"{pad}{bold}{label}:{bold}")
            lines.extend(format_list_lines(item, indent=indent + 1, bold=bold))
        else:
            lines.append(f"{pad}{bold}{label}:{bold} {format_inline_value(item)}")
    return lines

format_list_lines

format_list_lines(
    value: list[Any], *, indent: int = 0, bold: str = "*"
) -> list[str]

Source code in octomate/tentacles/feelers/output.py

def format_list_lines(
    value: list[Any], *, indent: int = 0, bold: str = "*"
) -> list[str]:
    lines: list[str] = []
    pad = "   " * indent
    for index, item in enumerate(value, start=1):
        if isinstance(item, dict):
            lines.append(f"{pad}{index}.")
            lines.extend(format_mapping_lines(item, indent=indent + 1, bold=bold))
        else:
            lines.append(f"{pad}{index}. {format_inline_value(item)}")
    return lines

format_fields

format_fields(
    value: dict[str, Any], *, bold: str = "*"
) -> str

Source code in octomate/tentacles/feelers/output.py

def format_fields(value: dict[str, Any], *, bold: str = "*") -> str:
    lines = format_mapping_lines(value, bold=bold)
    return truncate_task_detail("\n".join(lines) if lines else "_No details_")

render_todo_lines

render_todo_lines(todos: Iterable[Todo]) -> list[str]

User-facing checklist lines (- [x] … task-list markdown, plaintext-safe): an in-progress item shows its active form, subtasks indent under their parent.

Source code in octomate/tentacles/feelers/output.py

def render_todo_lines(todos: Iterable[Todo]) -> list[str]:
    """User-facing checklist lines (`- [x] …` task-list markdown, plaintext-safe):
    an in-progress item shows its active form, subtasks indent under their parent."""
    lines: list[str] = []
    for todo in sorted(todos, key=lambda todo: (todo.position, todo.ref)):
        label = (
            todo.active_form
            if todo.status == "in_progress" and todo.active_form
            else todo.content
        )
        indent = "  " if todo.parent_ref else ""
        lines.append(f"{indent}- {STATUS_MARKERS[todo.status]} {label}")
    return lines

markdown_from_output

markdown_from_output(
    output: JsonValue
    | Sequence[MessageSegment]
    | DeferredToolRequests,
) -> str | None

Source code in octomate/tentacles/feelers/output.py

def markdown_from_output(
    output: JsonValue | Sequence[MessageSegment] | DeferredToolRequests,
) -> str | None:
    if output is None or isinstance(output, DeferredToolRequests):
        return None
    if isinstance(output, str):
        return output
    if isinstance(output, Sequence):
        if all(isinstance(item, Segment) for item in output):
            # A leading reply threads the message; it is never rendered as text.
            return (
                "\n\n".join(
                    str(item) for item in output if not isinstance(item, ReplySegment)
                )
                or None
            )
        output = [item for item in output if not isinstance(item, Segment)]
    return f"```json\n{format_stream_value(output)}\n```"

split_reply

split_reply(
    segments: Sequence[MessageSegment],
) -> tuple[str | None, list[MessageSegment]]

Take the first ReplySegment anywhere in the list as the reply target (its data.id is passed to the platform as reply_to); the body is the remaining segments with every reply stripped out, so none renders as [reply: …] text. No reply segment leaves the list untouched.

Source code in octomate/tentacles/feelers/output.py

def split_reply(
    segments: Sequence[MessageSegment],
) -> tuple[str | None, list[MessageSegment]]:
    """Take the first `ReplySegment` anywhere in the list as the reply target (its
    `data.id` is passed to the platform as `reply_to`); the body is the remaining
    segments with every reply stripped out, so none renders as `[reply: …]` text.
    No reply segment leaves the list untouched."""
    reply_to: str | None = None
    body: list[MessageSegment] = []
    for seg in segments:
        if isinstance(seg, ReplySegment):
            reply_to = reply_to or seg.data["id"]
        else:
            body.append(seg)
    return reply_to, body

present_markdown async

present_markdown(
    *,
    ink: Ink[MessageT],
    chromo: Chromo[RawT, MessageT],
    address: ChannelAddress,
    markdown: str,
    reply_to: str | None = None,
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/output.py

async def present_markdown[MessageT, RawT](
    *,
    ink: Ink[MessageT],
    chromo: Chromo[RawT, MessageT],
    address: ChannelAddress,
    markdown: str,
    reply_to: str | None = None,
) -> IMMessageID | None:
    chat_id = address.chat_id or address.user_id
    chat_type = address.chat_type
    first_message_id: IMMessageID | None = None
    with channel_logfire.span(
        "default.markdown.present",
        channel_id=address.channel_tentacle_id,
        conversation_address=str(address),
        chat_type=chat_type,
        markdown_len=len(markdown),
    ) as span:
        for message in chromo.outbound_markdown(markdown):
            message_id = await ink.send_message(
                chat_id,
                chat_type,
                [message],
                reply_to=reply_to,
                channel_thread_id=address.channel_thread_id or chat_id,
            )
            first_message_id = first_message_id or message_id
        span.set_attribute("message_id", str(first_message_id))
        return first_message_id

octomate.tentacles.feelers.artifacts

Artifact overviews rendered on the channel where they were created.

ArtifactFeeler

Bases: ABC

Resolve an owned overview and record its delivery separately from the reply.

The original address is preserved, including its platform thread. Recipients see the saved title and provider; opening the viewer still requires ownership. Without a configured viewer origin, channel overviews are disabled. Subclasses implement only the platform presentation in send.

Source code in octomate/tentacles/feelers/artifacts.py

class ArtifactFeeler(ABC):
    """Resolve an owned overview and record its delivery separately from the reply.

    The original address is preserved, including its platform thread. Recipients
    see the saved title and provider; opening the viewer still requires ownership.
    Without a configured viewer origin, channel overviews are disabled.
    Subclasses implement only the platform presentation in ``send``.
    """

    def __init__(self, *, artifacts: ArtifactManager, users: UserManager) -> None:
        self.artifacts: ArtifactManager = artifacts
        self.users: UserManager = users

    @property
    def enabled(self) -> bool:
        """Whether this deployment can link to its artifact viewer."""
        return self.artifacts.config.viewer_origin is not None

    async def present(
        self, address: ChannelAddress, event: ArtifactAvailableEvent
    ) -> IMMessageID | None:
        # The artifact manager imports MCP, which loads channel classes at startup.
        from octomate.managers.artifacts import ArtifactUnavailable

        if not self.enabled:
            return None
        profile = await self.users.profile(address.channel_tentacle_id, address.user_id)
        if profile is None or profile.user_id is None:
            raise ArtifactUnavailable("Artifact owner unavailable")
        user = profile.user.value
        if user is None:
            raise ArtifactUnavailable("Artifact owner unavailable")
        overview = await self.artifacts.overview(event.artifact_id, user)
        message_id = await self.send(address, overview)
        await self.artifacts.threads.mark_presented(event.message_id, message_id)
        return message_id

    @abstractmethod
    async def send(
        self, address: ChannelAddress, overview: McpAppOverview
    ) -> IMMessageID | None:
        """Present already-authorized metadata without reading app content."""

artifacts instance-attribute

artifacts: ArtifactManager = artifacts

users instance-attribute

users: UserManager = users

enabled property

enabled: bool

Whether this deployment can link to its artifact viewer.

present async

present(
    address: ChannelAddress, event: ArtifactAvailableEvent
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/artifacts.py

async def present(
    self, address: ChannelAddress, event: ArtifactAvailableEvent
) -> IMMessageID | None:
    # The artifact manager imports MCP, which loads channel classes at startup.
    from octomate.managers.artifacts import ArtifactUnavailable

    if not self.enabled:
        return None
    profile = await self.users.profile(address.channel_tentacle_id, address.user_id)
    if profile is None or profile.user_id is None:
        raise ArtifactUnavailable("Artifact owner unavailable")
    user = profile.user.value
    if user is None:
        raise ArtifactUnavailable("Artifact owner unavailable")
    overview = await self.artifacts.overview(event.artifact_id, user)
    message_id = await self.send(address, overview)
    await self.artifacts.threads.mark_presented(event.message_id, message_id)
    return message_id

send abstractmethod async

send(
    address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None

Present already-authorized metadata without reading app content.

Source code in octomate/tentacles/feelers/artifacts.py

@abstractmethod
async def send(
    self, address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None:
    """Present already-authorized metadata without reading app content."""

PlainTextArtifactFeeler

Bases: ArtifactFeeler

A saved title, provider name, and literal viewer URL.

Source code in octomate/tentacles/feelers/artifacts.py

class PlainTextArtifactFeeler(ArtifactFeeler):
    """A saved title, provider name, and literal viewer URL."""

    def __init__(
        self,
        *,
        artifacts: ArtifactManager,
        users: UserManager,
        segments: SegmentsFeeler,
    ) -> None:
        super().__init__(artifacts=artifacts, users=users)
        self.segments: SegmentsFeeler = segments

    async def send(
        self, address: ChannelAddress, overview: McpAppOverview
    ) -> IMMessageID | None:
        return await self.segments.present(
            address,
            [
                TextSegment(
                    data={
                        "text": f"{overview.title}\nMCP app · {overview.provider_name}"
                        f"\n{overview.viewer_url}"
                    }
                )
            ],
        )

segments instance-attribute

segments: SegmentsFeeler = segments

send async

send(
    address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/artifacts.py

async def send(
    self, address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None:
    return await self.segments.present(
        address,
        [
            TextSegment(
                data={
                    "text": f"{overview.title}\nMCP app · {overview.provider_name}"
                    f"\n{overview.viewer_url}"
                }
            )
        ],
    )

octomate.tentacles.slack.feelers.artifacts

Slack artifact overviews, with a link to the authenticated viewer.

SlackArtifactFeeler

Bases: ArtifactFeeler

Show a native card with a sparkle icon, saved title/provider, and Open button.

Title and subtitle are bounded by Slack's 150-character card limits. No app resource is fetched; the viewer still checks ownership when the link opens.

Source code in octomate/tentacles/slack/feelers/artifacts.py

class SlackArtifactFeeler(ArtifactFeeler):
    """Show a native card with a sparkle icon, saved title/provider, and Open button.

    Title and subtitle are bounded by Slack's 150-character card limits. No app
    resource is fetched; the viewer still checks ownership when the link opens.
    """

    def __init__(
        self,
        *,
        artifacts: ArtifactManager,
        users: UserManager,
        ink: Ink[SlackOutboundMessage],
    ) -> None:
        super().__init__(artifacts=artifacts, users=users)
        self.ink: Ink[SlackOutboundMessage] = ink

    @slack_logfire.instrument("slack.artifact.send", extract_args=False)
    async def send(
        self, address: ChannelAddress, overview: McpAppOverview
    ) -> IMMessageID | None:
        title = overview.title[:150]
        provider = f"{overview.provider_name} · Interactive artifact"[:150]
        text = (
            f"{escape(title, quote=False)}\n{escape(provider, quote=False)}"
            f"\n{overview.viewer_url}"
        )
        return await self.ink.send_message(
            address.chat_id or address.user_id,
            address.chat_type,
            [
                SlackOutboundMessage(
                    text=text,
                    blocks=[
                        {
                            "type": "card",
                            "slack_icon": {"type": "icon", "name": "sparkle"},
                            "title": {"type": "plain_text", "text": title},
                            "subtitle": {"type": "plain_text", "text": provider},
                            "actions": [
                                {
                                    "type": "button",
                                    "text": {"type": "plain_text", "text": "Open"},
                                    "accessibility_label": "Open artifact in browser",
                                    "url": str(overview.viewer_url),
                                    "action_id": SlackBlockAction.ARTIFACT_OPEN.value,
                                },
                            ],
                        },
                    ],
                )
            ],
            channel_thread_id=(
                address.channel_thread_id or address.chat_id or address.user_id
            ),
        )

ink instance-attribute

ink: Ink[SlackOutboundMessage] = ink

send async

send(
    address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None

Source code in octomate/tentacles/slack/feelers/artifacts.py

@slack_logfire.instrument("slack.artifact.send", extract_args=False)
async def send(
    self, address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None:
    title = overview.title[:150]
    provider = f"{overview.provider_name} · Interactive artifact"[:150]
    text = (
        f"{escape(title, quote=False)}\n{escape(provider, quote=False)}"
        f"\n{overview.viewer_url}"
    )
    return await self.ink.send_message(
        address.chat_id or address.user_id,
        address.chat_type,
        [
            SlackOutboundMessage(
                text=text,
                blocks=[
                    {
                        "type": "card",
                        "slack_icon": {"type": "icon", "name": "sparkle"},
                        "title": {"type": "plain_text", "text": title},
                        "subtitle": {"type": "plain_text", "text": provider},
                        "actions": [
                            {
                                "type": "button",
                                "text": {"type": "plain_text", "text": "Open"},
                                "accessibility_label": "Open artifact in browser",
                                "url": str(overview.viewer_url),
                                "action_id": SlackBlockAction.ARTIFACT_OPEN.value,
                            },
                        ],
                    },
                ],
            )
        ],
        channel_thread_id=(
            address.channel_thread_id or address.chat_id or address.user_id
        ),
    )

octomate.tentacles.lark.feelers.artifacts

Lark artifact cards linking to the authenticated web viewer.

LarkArtifactFeeler

Bases: ArtifactFeeler

Show a blue title header, provider subtitle, and a native viewer button.

Titles and provider names remain plain text. The JSON 2.0 button opens the viewer directly, without a card callback or another upstream tool call.

Source code in octomate/tentacles/lark/feelers/artifacts.py

class LarkArtifactFeeler(ArtifactFeeler):
    """Show a blue title header, provider subtitle, and a native viewer button.

    Titles and provider names remain plain text. The JSON 2.0 button opens the
    viewer directly, without a card callback or another upstream tool call.
    """

    def __init__(
        self,
        *,
        artifacts: ArtifactManager,
        users: UserManager,
        ink: Ink[LarkOutboundMessage],
    ) -> None:
        super().__init__(artifacts=artifacts, users=users)
        self.ink: Ink[LarkOutboundMessage] = ink

    @lark_logfire.instrument("lark.artifact.send", extract_args=False)
    async def send(
        self, address: ChannelAddress, overview: McpAppOverview
    ) -> IMMessageID | None:
        header = cards.header(overview.title, template="blue")
        header["subtitle"] = {
            "tag": "plain_text",
            "content": f"{overview.provider_name} · Interactive artifact",
        }
        button = cards.button("Open artifact", button_type="primary")
        button["behaviors"] = [
            {"type": "open_url", "default_url": str(overview.viewer_url)}
        ]
        card = LarkInteractiveCard.model_validate(
            cards.card_v2([button], header=header)
        )
        chat_id = address.chat_id or address.user_id
        return await self.ink.send_message(
            chat_id,
            address.chat_type,
            [
                LarkOutboundMessage(
                    msg_type="interactive",
                    content=card.model_dump_json(by_alias=True, exclude_unset=True),
                )
            ],
            channel_thread_id=address.channel_thread_id or chat_id,
            reply_in_thread=address.chat_type == "thread",
        )

ink instance-attribute

ink: Ink[LarkOutboundMessage] = ink

send async

send(
    address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None

Source code in octomate/tentacles/lark/feelers/artifacts.py

@lark_logfire.instrument("lark.artifact.send", extract_args=False)
async def send(
    self, address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None:
    header = cards.header(overview.title, template="blue")
    header["subtitle"] = {
        "tag": "plain_text",
        "content": f"{overview.provider_name} · Interactive artifact",
    }
    button = cards.button("Open artifact", button_type="primary")
    button["behaviors"] = [
        {"type": "open_url", "default_url": str(overview.viewer_url)}
    ]
    card = LarkInteractiveCard.model_validate(
        cards.card_v2([button], header=header)
    )
    chat_id = address.chat_id or address.user_id
    return await self.ink.send_message(
        chat_id,
        address.chat_type,
        [
            LarkOutboundMessage(
                msg_type="interactive",
                content=card.model_dump_json(by_alias=True, exclude_unset=True),
            )
        ],
        channel_thread_id=address.channel_thread_id or chat_id,
        reply_in_thread=address.chat_type == "thread",
    )

octomate.tentacles.discord.feelers.artifacts

Discord artifact cards linking to the authenticated web viewer.

DiscordArtifactFeeler

Bases: ArtifactFeeler

A Components V2 container with a title, provider, and viewer link button.

Saved metadata is bounded and escaped for Discord's Markdown text displays. The link opens the viewer directly without an interaction callback.

Source code in octomate/tentacles/discord/feelers/artifacts.py

class DiscordArtifactFeeler(ArtifactFeeler):
    """A Components V2 container with a title, provider, and viewer link button.

    Saved metadata is bounded and escaped for Discord's Markdown text displays.
    The link opens the viewer directly without an interaction callback.
    """

    def __init__(
        self,
        *,
        artifacts: ArtifactManager,
        users: UserManager,
        ink: Ink[DiscordOutboundMessage],
    ) -> None:
        super().__init__(artifacts=artifacts, users=users)
        self.ink: Ink[DiscordOutboundMessage] = ink

    @channel_logfire.instrument("discord.artifact.send", extract_args=False)
    async def send(
        self, address: ChannelAddress, overview: McpAppOverview
    ) -> IMMessageID | None:
        title = discord.utils.escape_markdown(
            discord.utils.escape_mentions(" ".join(overview.title.split())[:256]),
            ignore_links=False,
        )
        provider = discord.utils.escape_markdown(
            discord.utils.escape_mentions(
                " ".join(overview.provider_name.split())[:256]
            ),
            ignore_links=False,
        )
        view = discord.ui.LayoutView(timeout=None)
        view.add_item(
            discord.ui.Container(
                discord.ui.Section(
                    discord.ui.TextDisplay(f"### ✨ {title}"),
                    discord.ui.TextDisplay(f"-# {provider} · Interactive artifact"),
                    accessory=discord.ui.Button(
                        label="Open artifact",
                        style=discord.ButtonStyle.link,
                        url=str(overview.viewer_url),
                    ),
                ),
                accent_colour=discord.Colour.blurple(),
            )
        )
        chat_id = address.chat_id or address.user_id
        return await self.ink.send_message(
            chat_id,
            address.chat_type,
            [DiscordOutboundMessage(view=view)],
            channel_thread_id=address.channel_thread_id or chat_id,
        )

ink instance-attribute

ink: Ink[DiscordOutboundMessage] = ink

send async

send(
    address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None

Source code in octomate/tentacles/discord/feelers/artifacts.py

@channel_logfire.instrument("discord.artifact.send", extract_args=False)
async def send(
    self, address: ChannelAddress, overview: McpAppOverview
) -> IMMessageID | None:
    title = discord.utils.escape_markdown(
        discord.utils.escape_mentions(" ".join(overview.title.split())[:256]),
        ignore_links=False,
    )
    provider = discord.utils.escape_markdown(
        discord.utils.escape_mentions(
            " ".join(overview.provider_name.split())[:256]
        ),
        ignore_links=False,
    )
    view = discord.ui.LayoutView(timeout=None)
    view.add_item(
        discord.ui.Container(
            discord.ui.Section(
                discord.ui.TextDisplay(f"### ✨ {title}"),
                discord.ui.TextDisplay(f"-# {provider} · Interactive artifact"),
                accessory=discord.ui.Button(
                    label="Open artifact",
                    style=discord.ButtonStyle.link,
                    url=str(overview.viewer_url),
                ),
            ),
            accent_colour=discord.Colour.blurple(),
        )
    )
    chat_id = address.chat_id or address.user_id
    return await self.ink.send_message(
        chat_id,
        address.chat_type,
        [DiscordOutboundMessage(view=view)],
        channel_thread_id=address.channel_thread_id or chat_id,
    )

octomate.tentacles.feelers.deferred

Plain-text fallbacks for human-in-the-loop deferred action feelers.

QUESTION_PROGRESS_SUFFIX module-attribute

QUESTION_PROGRESS_SUFFIX = re.compile(
    "\\s*\\(?\\s*Question\\s+\\d+\\s+of\\s+\\d+\\s*\\)?\\s*$",
    re.IGNORECASE,
)

ApprovalFeeler

Bases: ABC

Presents approval actions for one response target.

Source code in octomate/tentacles/feelers/deferred.py

class ApprovalFeeler(ABC):
    """Presents approval actions for one response target."""

    @abstractmethod
    async def present(
        self,
        address: ChannelAddress,
        actions: list[DeferredApproval],
    ) -> dict[UUID7, IMMessageID | None]: ...

present abstractmethod async

present(
    address: ChannelAddress, actions: list[DeferredApproval]
) -> dict[UUID7, IMMessageID | None]

Source code in octomate/tentacles/feelers/deferred.py

@abstractmethod
async def present(
    self,
    address: ChannelAddress,
    actions: list[DeferredApproval],
) -> dict[UUID7, IMMessageID | None]: ...

QuestionFeeler

Bases: ABC

Presents a card-based question wizard for one response target.

Source code in octomate/tentacles/feelers/deferred.py

class QuestionFeeler(ABC):
    """Presents a card-based question wizard for one response target."""

    @abstractmethod
    async def present(
        self,
        address: ChannelAddress,
        actions: list[DeferredQuestion],
    ) -> dict[UUID7, IMMessageID | None]: ...

present abstractmethod async

present(
    address: ChannelAddress, actions: list[DeferredQuestion]
) -> dict[UUID7, IMMessageID | None]

Source code in octomate/tentacles/feelers/deferred.py

@abstractmethod
async def present(
    self,
    address: ChannelAddress,
    actions: list[DeferredQuestion],
) -> dict[UUID7, IMMessageID | None]: ...

PlainTextApprovalFeeler

Bases: ApprovalFeeler

Presents approvals as markdown text, for channels without interactive cards.

Source code in octomate/tentacles/feelers/deferred.py

class PlainTextApprovalFeeler(ApprovalFeeler):
    """Presents approvals as markdown text, for channels without interactive
    cards."""

    def __init__(self, markdown: MarkdownFeeler) -> None:
        self.markdown = markdown

    @channel_logfire.instrument("plaintext.approvals.present", extract_args=False)
    async def present(
        self,
        address: ChannelAddress,
        actions: list[DeferredApproval],
    ) -> dict[UUID7, IMMessageID | None]:
        message_ids: dict[UUID7, IMMessageID | None] = {}
        for action in actions:
            message_ids[action.id] = await self.markdown.present(
                address,
                (
                    f"Octomate needs approval for `{action.tool_name}` "
                    f"({action.id}). This channel can show the request, but does "
                    "not support interactive approval cards yet."
                ),
            )
        return message_ids

markdown instance-attribute

markdown = markdown

present async

present(
    address: ChannelAddress, actions: list[DeferredApproval]
) -> dict[UUID7, IMMessageID | None]

Source code in octomate/tentacles/feelers/deferred.py

@channel_logfire.instrument("plaintext.approvals.present", extract_args=False)
async def present(
    self,
    address: ChannelAddress,
    actions: list[DeferredApproval],
) -> dict[UUID7, IMMessageID | None]:
    message_ids: dict[UUID7, IMMessageID | None] = {}
    for action in actions:
        message_ids[action.id] = await self.markdown.present(
            address,
            (
                f"Octomate needs approval for `{action.tool_name}` "
                f"({action.id}). This channel can show the request, but does "
                "not support interactive approval cards yet."
            ),
        )
    return message_ids

PlainTextAskQuestionFeeler

Bases: QuestionFeeler

Presents questions as markdown text, for channels without interactive cards.

Source code in octomate/tentacles/feelers/deferred.py

class PlainTextAskQuestionFeeler(QuestionFeeler):
    """Presents questions as markdown text, for channels without interactive
    cards."""

    def __init__(self, markdown: MarkdownFeeler) -> None:
        self.markdown = markdown

    @channel_logfire.instrument("plaintext.ask_questions.present", extract_args=False)
    async def present(
        self,
        address: ChannelAddress,
        actions: list[DeferredQuestion],
    ) -> dict[UUID7, IMMessageID | None]:
        message_ids: dict[UUID7, IMMessageID | None] = {}
        for action in actions:
            choices_text = ""
            if choices := action.args.get("choices"):
                choices_text = "\nChoices:\n" + "\n".join(
                    f"- {choice}" for choice in choices
                )
            hint = action.args.get("hint") or ""
            hint_text = f"\nHint: {hint}" if hint else ""
            message_ids[action.id] = await self.markdown.present(
                address,
                (
                    f"Octomate needs an answer: {question_text(action)}"
                    f"{hint_text}{choices_text}\nDeferred action: `{action.id}`"
                ),
            )
        return message_ids

markdown instance-attribute

markdown = markdown

present async

present(
    address: ChannelAddress, actions: list[DeferredQuestion]
) -> dict[UUID7, IMMessageID | None]

Source code in octomate/tentacles/feelers/deferred.py

@channel_logfire.instrument("plaintext.ask_questions.present", extract_args=False)
async def present(
    self,
    address: ChannelAddress,
    actions: list[DeferredQuestion],
) -> dict[UUID7, IMMessageID | None]:
    message_ids: dict[UUID7, IMMessageID | None] = {}
    for action in actions:
        choices_text = ""
        if choices := action.args.get("choices"):
            choices_text = "\nChoices:\n" + "\n".join(
                f"- {choice}" for choice in choices
            )
        hint = action.args.get("hint") or ""
        hint_text = f"\nHint: {hint}" if hint else ""
        message_ids[action.id] = await self.markdown.present(
            address,
            (
                f"Octomate needs an answer: {question_text(action)}"
                f"{hint_text}{choices_text}\nDeferred action: `{action.id}`"
            ),
        )
    return message_ids

question_text

question_text(action: DeferredQuestion) -> str

Source code in octomate/tentacles/feelers/deferred.py

def question_text(action: DeferredQuestion) -> str:
    text = str(action.args["question"]).strip()
    cleaned = QUESTION_PROGRESS_SUFFIX.sub("", text).strip()
    return cleaned or text

answer_text

answer_text(answer: QuestionAnswer | None) -> str

An answer as one line of text, for a card to show: a multi-select question's picks are listed.

Source code in octomate/tentacles/feelers/deferred.py

def answer_text(answer: QuestionAnswer | None) -> str:
    """An answer as one line of text, for a card to show: a multi-select
    question's picks are listed."""
    if isinstance(answer, list):
        return ", ".join(answer)
    return answer or ""

octomate.tentacles.feelers.oauth

Presenting a pending OAuth authorization, and the plain-text fallback.

An authorization is a two-step errand for the user: open the provider's page with a one-time code, then tell the agent so it can finish the connection. The event reaches this feeler as the authorization itself rather than as rendered text, so a channel with interactive cards can put the link on a button; channels without cards fall back to the markdown feeler.

Where it lands is this layer's decision, not each channel's: present resolves the private address and a channel only fills in send, so there is no way for one to render an authorization onto an address it chose itself.

AuthorizationEvent module-attribute

AuthorizationEvent = (
    OAuthAuthorizationEvent | LinkProfileAuthorizationEvent
)

OAuthFeeler

Bases: ABC

Presents one pending authorization for one response target.

Source code in octomate/tentacles/feelers/oauth.py

class OAuthFeeler[MessageT](ABC):
    """Presents one pending authorization for one response target."""

    ink: Ink[MessageT]

    def __init__(self, ink: Ink[MessageT]) -> None:
        self.ink = ink

    async def present(
        self,
        address: ChannelAddress,
        event: AuthorizationEvent,
    ) -> IMMessageID | None:
        """The only way an authorization reaches a channel.

        Concrete so the private address is settled here for every channel; a
        subclass fills in `send` and never gets to pick where it sends.
        """
        return await self.send(await self.deliver_to(address), event)

    async def deliver_to(self, address: ChannelAddress) -> ChannelAddress:
        """Where this authorization has to land.

        The code authorizes one person's account, so a request made from a group
        must not be read out to the group: it goes to that person's direct
        messages instead. A platform with nowhere private to reach them gets an
        error rather than the group, which is the whole point of asking.

        What makes a surface private is `shared`, not the chat type: an assistant
        pane is a thread only its own user can read, and moving the link out of it
        would strand "return here and tell me to confirm" in another surface.
        """
        if not address.shared:
            return address
        chat_id = await self.ink.open_dm(address.user_id)
        if chat_id is None:
            raise RuntimeError(
                f"no direct message to reach {address.user_id} on "
                f"{address.channel_tentacle_id}; a one-time code is not going to a group"
            )
        return replace(
            address,
            chat_type="dm",
            chat_id=chat_id,
            channel_thread_id=None,
            shared=False,
        )

    @abstractmethod
    async def send(
        self,
        address: ChannelAddress,
        event: AuthorizationEvent,
    ) -> IMMessageID | None:
        """Render this authorization onto `address`, which `present` has already
        made private. Never call this with an address of your own."""

ink instance-attribute

ink: Ink[MessageT] = ink

present async

present(
    address: ChannelAddress, event: AuthorizationEvent
) -> IMMessageID | None

The only way an authorization reaches a channel.

Concrete so the private address is settled here for every channel; a subclass fills in send and never gets to pick where it sends.

Source code in octomate/tentacles/feelers/oauth.py

async def present(
    self,
    address: ChannelAddress,
    event: AuthorizationEvent,
) -> IMMessageID | None:
    """The only way an authorization reaches a channel.

    Concrete so the private address is settled here for every channel; a
    subclass fills in `send` and never gets to pick where it sends.
    """
    return await self.send(await self.deliver_to(address), event)

deliver_to async

deliver_to(address: ChannelAddress) -> ChannelAddress

Where this authorization has to land.

The code authorizes one person's account, so a request made from a group must not be read out to the group: it goes to that person's direct messages instead. A platform with nowhere private to reach them gets an error rather than the group, which is the whole point of asking.

What makes a surface private is shared, not the chat type: an assistant pane is a thread only its own user can read, and moving the link out of it would strand "return here and tell me to confirm" in another surface.

Source code in octomate/tentacles/feelers/oauth.py

async def deliver_to(self, address: ChannelAddress) -> ChannelAddress:
    """Where this authorization has to land.

    The code authorizes one person's account, so a request made from a group
    must not be read out to the group: it goes to that person's direct
    messages instead. A platform with nowhere private to reach them gets an
    error rather than the group, which is the whole point of asking.

    What makes a surface private is `shared`, not the chat type: an assistant
    pane is a thread only its own user can read, and moving the link out of it
    would strand "return here and tell me to confirm" in another surface.
    """
    if not address.shared:
        return address
    chat_id = await self.ink.open_dm(address.user_id)
    if chat_id is None:
        raise RuntimeError(
            f"no direct message to reach {address.user_id} on "
            f"{address.channel_tentacle_id}; a one-time code is not going to a group"
        )
    return replace(
        address,
        chat_type="dm",
        chat_id=chat_id,
        channel_thread_id=None,
        shared=False,
    )

send abstractmethod async

send(
    address: ChannelAddress, event: AuthorizationEvent
) -> IMMessageID | None

Render this authorization onto address, which present has already made private. Never call this with an address of your own.

Source code in octomate/tentacles/feelers/oauth.py

@abstractmethod
async def send(
    self,
    address: ChannelAddress,
    event: AuthorizationEvent,
) -> IMMessageID | None:
    """Render this authorization onto `address`, which `present` has already
    made private. Never call this with an address of your own."""

PlainTextOAuthFeeler

Bases: OAuthFeeler[MessageT]

Presents an authorization as a markdown link, with the device code when there is one.

Source code in octomate/tentacles/feelers/oauth.py

class PlainTextOAuthFeeler[MessageT](OAuthFeeler[MessageT]):
    """Presents an authorization as a markdown link, with the device code when
    there is one."""

    def __init__(self, ink: Ink[MessageT], markdown: MarkdownFeeler) -> None:
        super().__init__(ink)
        self.markdown = markdown

    @channel_logfire.instrument("plaintext.oauth.send", extract_args=False)
    async def send(
        self,
        address: ChannelAddress,
        event: AuthorizationEvent,
    ) -> IMMessageID | None:
        if isinstance(event, LinkProfileAuthorizationEvent):
            return await self.markdown.present(
                address,
                f"[{event.host} authorization]({event.authorization.authorization_uri})"
                f"\n\n{link_profile_body(event)}",
            )
        link = f"[Connect {event.label}]({event.authorization_uri})"
        if isinstance(event, OAuthDeviceAuthorizationEvent):
            body = (
                f"{link}\n\n"
                f"Code: `{event.user_code}`\n\n"
                f"After {event.label} accepts the code, return here and tell me "
                "to confirm."
            )
        else:
            body = f"{link}\n\nOpen the link and approve it — that finishes it."
        return await self.markdown.present(address, body)

markdown instance-attribute

markdown = markdown

send async

send(
    address: ChannelAddress, event: AuthorizationEvent
) -> IMMessageID | None

Source code in octomate/tentacles/feelers/oauth.py

@channel_logfire.instrument("plaintext.oauth.send", extract_args=False)
async def send(
    self,
    address: ChannelAddress,
    event: AuthorizationEvent,
) -> IMMessageID | None:
    if isinstance(event, LinkProfileAuthorizationEvent):
        return await self.markdown.present(
            address,
            f"[{event.host} authorization]({event.authorization.authorization_uri})"
            f"\n\n{link_profile_body(event)}",
        )
    link = f"[Connect {event.label}]({event.authorization_uri})"
    if isinstance(event, OAuthDeviceAuthorizationEvent):
        body = (
            f"{link}\n\n"
            f"Code: `{event.user_code}`\n\n"
            f"After {event.label} accepts the code, return here and tell me "
            "to confirm."
        )
    else:
        body = f"{link}\n\nOpen the link and approve it — that finishes it."
    return await self.markdown.present(address, body)

link_profile_body

link_profile_body(
    event: LinkProfileAuthorizationEvent,
) -> str

Source code in octomate/tentacles/feelers/oauth.py

def link_profile_body(event: LinkProfileAuthorizationEvent) -> str:
    profile = event.authorization.profile
    return (
        f"Requested through: {profile.channel_tentacle_id}\n"
        f"Profile ID: {profile.channel_user_id}\n\n"
        f"Link this profile to your {event.host} account. "
        "Review the profile and signed-in account, then confirm in the browser. "
        "Your existing sign-in is reused. No access token is sent to the channel."
    )