Skip to content

Nerve and siphon

What a kick does travels on the nerve, and each channel draws what reaches it through its siphon. Feelers explains how a channel draws a run.

octomate.nerve

The nerve: what a thread does, as it happens, for whoever is watching.

In-process publish and subscribe, keyed by channel, thread or user topics, carrying wire events: the run's stream as a wire consumer sees it, and what the graph says around a run — its start and end, a move, a failure, the reflex waking and falling asleep, an authorization. Every way in and out is a stream end the nerve hands over. A kick publishes on the send end of a stream opened on the thread it was asked on, as impulses: an event and the address it happens at. The nerve pumps the other end, delivering each impulse to the channel it happens on and to the subscribers of its thread and of the asking thread, so a console follows the kick wherever a move carries it. A channel holds one subscription for all its threads and draws every run on them; it misses nothing and holds nobody up, since what it has yet to read is kept for it. A thread's subscriber — the console's live socket — reads the receive end of a stream of its own, for as long as its block runs, and one that falls further behind than its backlog reads what was held for it and is cut off, never quietly skipped ahead. Each such stream is a fiber: a pair of ends in memory today, and the one thing to replace for a stream that outlives the process. Publishing is one way and knows nothing of who subscribed, and nothing is kept for a reader that is not there: an event published to a thread nobody subscribes to goes nowhere, and a console re-reads the thread ledger when it opens.

Nerve.deliver reaches every subscriber of exactly one topic. A publisher sends user notifications by delivering separately to each authorized user's topic. Nerve.publish owns channel and thread routing for kicks and native tails.

FIBER_BUFFER module-attribute 

FIBER_BUFFER = 1024

Topic 

Topic = ChannelTopic | ThreadTopic | UserTopic

Impulse dataclass

One wire event with an optional channel address and the routing metadata used by Nerve.publish. Direct topic publication carries it unchanged.

Source code in octomate/nerve.py 
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class Impulse:
    """One wire event with an optional channel address and the routing metadata
    used by `Nerve.publish`. Direct topic publication carries it unchanged."""

    event: WireEvent
    address: ChannelAddress | None = None
    # Whether Nerve.publish excludes the origin thread for a private delivery.
    private: bool = False
    # The thread Nerve.publish stamps on the event: the asking thread
    # for a kick, the native thread for a tail, or None when neither names a thread.
    origin: ThreadKey | None = None

event instance-attribute 

event: WireEvent

address class-attribute instance-attribute 

address: ChannelAddress | None = None

private class-attribute instance-attribute 

private: bool = False

origin class-attribute instance-attribute 

origin: ThreadKey | None = None

ChannelTopic dataclass

Every impulse at one channel's addresses, each once.

Source code in octomate/nerve.py 
66
67
68
69
70
@dataclass(frozen=True)
class ChannelTopic:
    """Every impulse at one channel's addresses, each once."""

    channel_tentacle_id: str

channel_tentacle_id instance-attribute 

channel_tentacle_id: str

ThreadTopic dataclass

The impulses published to one thread.

Source code in octomate/nerve.py 
73
74
75
76
77
@dataclass(frozen=True)
class ThreadTopic:
    """The impulses published to one thread."""

    key: ThreadKey

key instance-attribute 

key: ThreadKey

UserTopic dataclass

Notifications explicitly published to one registered user.

Source code in octomate/nerve.py 
80
81
82
83
84
@dataclass(frozen=True)
class UserTopic:
    """Notifications explicitly published to one registered user."""

    user_id: UUID7

user_id instance-attribute 

user_id: UUID7

Fiber

One stream through the nerve, both ends: a kick's or a subscriber's, holding up to buffer items its reader has yet to take.

Source code in octomate/nerve.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Fiber[ItemT]:
    """One stream through the nerve, both ends: a kick's or a subscriber's,
    holding up to `buffer` items its reader has yet to take."""

    send: MemoryObjectSendStream[ItemT]
    receive: MemoryObjectReceiveStream[ItemT]

    def __init__(self, buffer: float = FIBER_BUFFER) -> None:
        self.send, self.receive = anyio.create_memory_object_stream[ItemT](buffer)

    def close(self) -> None:
        self.send.close()
        self.receive.close()

send instance-attribute 

send: MemoryObjectSendStream[ItemT]

receive instance-attribute 

receive: MemoryObjectReceiveStream[ItemT]

close 

close() -> None
Source code in octomate/nerve.py 
100
101
102
def close(self) -> None:
    self.send.close()
    self.receive.close()

Nerve

The subscribers of each channel, thread and user, and what is published to them.

Source code in octomate/nerve.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class Nerve:
    """The subscribers of each channel, thread and user, and what is
    published to them."""

    def __init__(self) -> None:
        # Each subscriber's fiber, by the topic it hears.
        self.subscribers: defaultdict[Topic, set[Fiber[Impulse]]] = defaultdict(set)

    @contextmanager
    def subscribe(
        self, topic: Topic, *, backlog: float = math.inf
    ) -> Generator[ObjectReceiveStream[Impulse]]:
        """The receive end of a stream of the topic's impulses, from now until
        the block ends. Nothing is dropped: a reader that falls more than
        `backlog` impulses behind reads what was held for it, and then its
        stream ends.

        `ChannelTopic` names a channel's addresses, `ThreadTopic` wraps one
        `ThreadKey`, and `UserTopic` wraps a registered user's UUID.
        Publishers choose authorized topics before publishing; the nerve does
        not filter subscribers within a topic.
        """
        fiber = Fiber[Impulse](backlog)
        self.subscribers[topic].add(fiber)
        try:
            yield fiber.receive
        finally:
            self.subscribers[topic].discard(fiber)
            if not self.subscribers[topic]:
                del self.subscribers[topic]
            fiber.close()

    @asynccontextmanager
    async def publish(
        self, origin: ChannelAddress | None
    ) -> AsyncGenerator[ObjectSendStream[Impulse]]:
        """A buffered publisher for kicks and native tails, open for the block.

        A kick names the thread it was asked on as `origin`; a native tail names
        its native thread and scopes the stream to its connection. The publisher
        stamps that origin on each impulse and publishes separately to its channel,
        its destination thread, and its origin thread unless private. Each distinct
        topic receives one publication.

        Sending may wait for the pump, never for a subscriber. Everything sent before
        the block ends is published; closing emits no lifecycle events of its own.
        """
        fiber = Fiber[Impulse]()
        asked = ThreadKey.from_address(origin) if origin is not None else None

        async def pump() -> None:
            async for sent in fiber.receive:
                impulse = replace(sent, origin=asked)
                topics: set[Topic] = set()
                if impulse.address is not None:
                    topics.add(ChannelTopic(impulse.address.channel_tentacle_id))
                    topics.add(ThreadTopic(ThreadKey.from_address(impulse.address)))
                if not impulse.private and asked is not None:
                    topics.add(ThreadTopic(asked))
                for topic in topics:
                    self.deliver(topic, impulse)

        pumping = asyncio.create_task(pump())
        try:
            yield fiber.send
        finally:
            fiber.send.close()
            await pumping
            fiber.close()

    def deliver(self, topic: Topic, impulse: Impulse) -> None:
        """Deliver one impulse to every subscriber of exactly `topic`, waiting
        for none. The impulse is delivered unchanged; its address, origin and
        privacy flag do not select or exclude subscribers."""
        for fiber in self.subscribers.get(topic, set()):
            try:
                fiber.send.send_nowait(impulse)
            except anyio.WouldBlock:
                # Further behind than it asked to be held for: its stream ends.
                fiber.send.close()
            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
                pass  # Already leaving, or already cut off.

subscribers instance-attribute 

subscribers: defaultdict[Topic, set[Fiber[Impulse]]] = (
    defaultdict(set)
)

subscribe 

subscribe(
    topic: Topic, *, backlog: float = math.inf
) -> Generator[ObjectReceiveStream[Impulse]]

The receive end of a stream of the topic's impulses, from now until the block ends. Nothing is dropped: a reader that falls more than backlog impulses behind reads what was held for it, and then its stream ends.

ChannelTopic names a channel's addresses, ThreadTopic wraps one ThreadKey, and UserTopic wraps a registered user's UUID. Publishers choose authorized topics before publishing; the nerve does not filter subscribers within a topic.

Source code in octomate/nerve.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@contextmanager
def subscribe(
    self, topic: Topic, *, backlog: float = math.inf
) -> Generator[ObjectReceiveStream[Impulse]]:
    """The receive end of a stream of the topic's impulses, from now until
    the block ends. Nothing is dropped: a reader that falls more than
    `backlog` impulses behind reads what was held for it, and then its
    stream ends.

    `ChannelTopic` names a channel's addresses, `ThreadTopic` wraps one
    `ThreadKey`, and `UserTopic` wraps a registered user's UUID.
    Publishers choose authorized topics before publishing; the nerve does
    not filter subscribers within a topic.
    """
    fiber = Fiber[Impulse](backlog)
    self.subscribers[topic].add(fiber)
    try:
        yield fiber.receive
    finally:
        self.subscribers[topic].discard(fiber)
        if not self.subscribers[topic]:
            del self.subscribers[topic]
        fiber.close()

publish async 

publish(
    origin: ChannelAddress | None,
) -> AsyncGenerator[ObjectSendStream[Impulse]]

A buffered publisher for kicks and native tails, open for the block.

A kick names the thread it was asked on as origin; a native tail names its native thread and scopes the stream to its connection. The publisher stamps that origin on each impulse and publishes separately to its channel, its destination thread, and its origin thread unless private. Each distinct topic receives one publication.

Sending may wait for the pump, never for a subscriber. Everything sent before the block ends is published; closing emits no lifecycle events of its own.

Source code in octomate/nerve.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@asynccontextmanager
async def publish(
    self, origin: ChannelAddress | None
) -> AsyncGenerator[ObjectSendStream[Impulse]]:
    """A buffered publisher for kicks and native tails, open for the block.

    A kick names the thread it was asked on as `origin`; a native tail names
    its native thread and scopes the stream to its connection. The publisher
    stamps that origin on each impulse and publishes separately to its channel,
    its destination thread, and its origin thread unless private. Each distinct
    topic receives one publication.

    Sending may wait for the pump, never for a subscriber. Everything sent before
    the block ends is published; closing emits no lifecycle events of its own.
    """
    fiber = Fiber[Impulse]()
    asked = ThreadKey.from_address(origin) if origin is not None else None

    async def pump() -> None:
        async for sent in fiber.receive:
            impulse = replace(sent, origin=asked)
            topics: set[Topic] = set()
            if impulse.address is not None:
                topics.add(ChannelTopic(impulse.address.channel_tentacle_id))
                topics.add(ThreadTopic(ThreadKey.from_address(impulse.address)))
            if not impulse.private and asked is not None:
                topics.add(ThreadTopic(asked))
            for topic in topics:
                self.deliver(topic, impulse)

    pumping = asyncio.create_task(pump())
    try:
        yield fiber.send
    finally:
        fiber.send.close()
        await pumping
        fiber.close()

deliver 

deliver(topic: Topic, impulse: Impulse) -> None

Deliver one impulse to every subscriber of exactly topic, waiting for none. The impulse is delivered unchanged; its address, origin and privacy flag do not select or exclude subscribers.

Source code in octomate/nerve.py 
175
176
177
178
179
180
181
182
183
184
185
186
def deliver(self, topic: Topic, impulse: Impulse) -> None:
    """Deliver one impulse to every subscriber of exactly `topic`, waiting
    for none. The impulse is delivered unchanged; its address, origin and
    privacy flag do not select or exclude subscribers."""
    for fiber in self.subscribers.get(topic, set()):
        try:
            fiber.send.send_nowait(impulse)
        except anyio.WouldBlock:
            # Further behind than it asked to be held for: its stream ends.
            fiber.send.close()
        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
            pass  # Already leaving, or already cut off.

octomate.tentacles.siphon

A channel's siphon: where what the nerve carries leaves for the platform.

An octopus pushes its ink out through its siphon. A channel's siphon pushes out what the nerve carries to it, through the channel's feelers — a run as a timeline when the channel streams, or its sends, batches and reply when it does not, and around runs a move's line, a failure, a send, an authorization. It holds the channel's subscription to its own addresses, and draws each thread in order in a lane of its own, so a slow thread holds up no other. A row the graph recorded comes with its id, and is marked with the platform id it was rendered as.

logger module-attribute 

logger = logging.getLogger(__name__)

Siphon

Draws what the nerve carries to one channel, thread by thread.

Source code in octomate/tentacles/siphon.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class Siphon:
    """Draws what the nerve carries to one channel, thread by thread."""

    def __init__(self, channel: ChannelTentacle) -> None:
        self.channel = channel
        # Each thread's lane: the impulses it has yet to draw.
        self.lanes: dict[ThreadKey, MemoryObjectSendStream[Impulse]] = {}
        self.subscription = ExitStack()
        self.reading: asyncio.Task[None] | None = None

    def start(self) -> None:
        """Subscribe to the channel's addresses now, so nothing published from
        here on is missed, and read in the background until `stop`."""
        nerve = self.channel.octomate.nerve
        impulses = self.subscription.enter_context(
            nerve.subscribe(ChannelTopic(self.channel.id))
        )
        self.reading = asyncio.create_task(self.read(impulses))

    async def stop(self) -> None:
        if self.reading is not None:
            self.reading.cancel()
            with suppress(asyncio.CancelledError):
                await self.reading
        self.subscription.close()

    async def read(self, impulses: ObjectReceiveStream[Impulse]) -> None:
        """Hand each impulse to its thread's lane, opening one where none is."""
        async with asyncio.TaskGroup() as lanes:
            async for impulse in impulses:
                if impulse.address is None:
                    continue  # The nerve hands a channel only what has an address.
                key = ThreadKey.from_address(impulse.address)
                lane = self.lanes.get(key)
                if lane is None:
                    lane, waiting = anyio.create_memory_object_stream[Impulse](math.inf)
                    self.lanes[key] = lane
                    lanes.create_task(self.draw(key, waiting))
                lane.send_nowait(impulse)

    async def draw(
        self, key: ThreadKey, impulses: MemoryObjectReceiveStream[Impulse]
    ) -> None:
        """Draw one thread's impulses in order, and close the lane once nothing
        is left in it. A rendering that fails is logged and never ends the lane."""
        with impulses:
            async for impulse in impulses:
                try:
                    await self.feel(impulse, impulses)
                except Exception:
                    logger.warning(
                        "Channel %s: rendering %s failed",
                        self.channel.id,
                        type(impulse.event).__name__,
                        exc_info=True,
                    )
                if impulses.statistics().current_buffer_used == 0:
                    self.lanes.pop(key).close()
                    return

    async def feel(
        self, impulse: Impulse, impulses: MemoryObjectReceiveStream[Impulse]
    ) -> None:
        """Draw one impulse; a run's start draws the whole run, through its end."""
        feelers = self.channel.feelers
        address = impulse.address
        if address is None:
            return
        match impulse.event:
            case RunStartedEvent():
                await self.run(address, impulses)
            case GatewayEvent() as event:
                line = await feelers.present(address, event)
                await self.mark(event.thread_message_id, line)
            case RunErrorEvent() as event:
                await feelers.present(address, event)
            case MessageSentEvent() as event:
                sent = await feelers.segments.present(
                    event.destination or address, event.segments
                )
                await self.mark(event.thread_message_id, sent)
            case OAuthAuthorizationEvent() as event:
                await feelers.oauth.present(address, event)
            case ArtifactAvailableEvent() as event:
                await feelers.artifacts.present(address, event)
            case _:
                pass

    async def run(
        self, address: ChannelAddress, impulses: MemoryObjectReceiveStream[Impulse]
    ) -> None:
        """Draw one agent run from the lane, up to its end: as a timeline when the
        channel streams, else quietly. What a failed rendering left unread is
        drained, so the lane resumes after the run."""
        ended: list[RunEndedEvent] = []

        async def events() -> AsyncGenerator[WireEvent]:
            async for impulse in impulses:
                if isinstance(impulse.event, RunEndedEvent):
                    ended.append(impulse.event)
                    return
                yield impulse.event

        stream = events()
        try:
            if self.channel.config.stream.enabled:
                async with self.channel.feelers.timeline.open(address) as timeline:
                    timeline.artifacts = self.channel.feelers.artifacts
                    await timeline.drive(stream)
                rendered = timeline.message_id
            else:
                rendered = await self.quietly(address, stream)
        except Exception:
            async for _ in stream:
                pass
            raise
        if ended:
            await self.mark(ended[0].thread_message_id, rendered)

    async def quietly(
        self, address: ChannelAddress, events: AsyncGenerator[WireEvent]
    ) -> IMMessageID | None:
        """A run on a channel that does not stream: its sends, batches and
        authorizations as they come, then its reply once. Answers the platform id
        the reply was posted as."""
        feelers = self.channel.feelers
        output: str | list[MessageSegment] | None = None
        async for event in events:
            match event:
                case MessageSentEvent():
                    await feelers.segments.present(
                        event.destination or address, event.segments
                    )
                case ActionBatchEvent():
                    await feelers.present_actions(
                        address,
                        event,
                        action_manager=self.channel.octomate.deferred_actions,
                    )
                case OAuthAuthorizationEvent():
                    await feelers.oauth.present(address, event)
                case ArtifactAvailableEvent():
                    await feelers.artifacts.present(address, event)
                case RunResultEvent():
                    output = event.output
                case _:
                    pass
        if isinstance(output, str):
            return await feelers.markdown.present(address, output) if output else None
        return await feelers.segments.present(address, output) if output else None

    async def mark(
        self, thread_message_id: UUID7 | None, platform_message_id: str | None
    ) -> None:
        """Name the platform message a recorded row was rendered as."""
        if thread_message_id is not None:
            await self.channel.octomate.threads.mark_presented(
                thread_message_id, platform_message_id
            )

channel instance-attribute 

channel = channel

lanes instance-attribute 

lanes: dict[ThreadKey, MemoryObjectSendStream[Impulse]] = {}

subscription instance-attribute 

subscription = ExitStack()

reading instance-attribute 

reading: Task[None] | None = None

start 

start() -> None

Subscribe to the channel's addresses now, so nothing published from here on is missed, and read in the background until stop.

Source code in octomate/tentacles/siphon.py 
61
62
63
64
65
66
67
68
def start(self) -> None:
    """Subscribe to the channel's addresses now, so nothing published from
    here on is missed, and read in the background until `stop`."""
    nerve = self.channel.octomate.nerve
    impulses = self.subscription.enter_context(
        nerve.subscribe(ChannelTopic(self.channel.id))
    )
    self.reading = asyncio.create_task(self.read(impulses))

stop async 

stop() -> None
Source code in octomate/tentacles/siphon.py 
70
71
72
73
74
75
async def stop(self) -> None:
    if self.reading is not None:
        self.reading.cancel()
        with suppress(asyncio.CancelledError):
            await self.reading
    self.subscription.close()

read async 

read(impulses: ObjectReceiveStream[Impulse]) -> None

Hand each impulse to its thread's lane, opening one where none is.

Source code in octomate/tentacles/siphon.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
async def read(self, impulses: ObjectReceiveStream[Impulse]) -> None:
    """Hand each impulse to its thread's lane, opening one where none is."""
    async with asyncio.TaskGroup() as lanes:
        async for impulse in impulses:
            if impulse.address is None:
                continue  # The nerve hands a channel only what has an address.
            key = ThreadKey.from_address(impulse.address)
            lane = self.lanes.get(key)
            if lane is None:
                lane, waiting = anyio.create_memory_object_stream[Impulse](math.inf)
                self.lanes[key] = lane
                lanes.create_task(self.draw(key, waiting))
            lane.send_nowait(impulse)

draw async 

draw(
    key: ThreadKey,
    impulses: MemoryObjectReceiveStream[Impulse],
) -> None

Draw one thread's impulses in order, and close the lane once nothing is left in it. A rendering that fails is logged and never ends the lane.

Source code in octomate/tentacles/siphon.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def draw(
    self, key: ThreadKey, impulses: MemoryObjectReceiveStream[Impulse]
) -> None:
    """Draw one thread's impulses in order, and close the lane once nothing
    is left in it. A rendering that fails is logged and never ends the lane."""
    with impulses:
        async for impulse in impulses:
            try:
                await self.feel(impulse, impulses)
            except Exception:
                logger.warning(
                    "Channel %s: rendering %s failed",
                    self.channel.id,
                    type(impulse.event).__name__,
                    exc_info=True,
                )
            if impulses.statistics().current_buffer_used == 0:
                self.lanes.pop(key).close()
                return

feel async 

feel(
    impulse: Impulse,
    impulses: MemoryObjectReceiveStream[Impulse],
) -> None

Draw one impulse; a run's start draws the whole run, through its end.

Source code in octomate/tentacles/siphon.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def feel(
    self, impulse: Impulse, impulses: MemoryObjectReceiveStream[Impulse]
) -> None:
    """Draw one impulse; a run's start draws the whole run, through its end."""
    feelers = self.channel.feelers
    address = impulse.address
    if address is None:
        return
    match impulse.event:
        case RunStartedEvent():
            await self.run(address, impulses)
        case GatewayEvent() as event:
            line = await feelers.present(address, event)
            await self.mark(event.thread_message_id, line)
        case RunErrorEvent() as event:
            await feelers.present(address, event)
        case MessageSentEvent() as event:
            sent = await feelers.segments.present(
                event.destination or address, event.segments
            )
            await self.mark(event.thread_message_id, sent)
        case OAuthAuthorizationEvent() as event:
            await feelers.oauth.present(address, event)
        case ArtifactAvailableEvent() as event:
            await feelers.artifacts.present(address, event)
        case _:
            pass

run async 

run(
    address: ChannelAddress,
    impulses: MemoryObjectReceiveStream[Impulse],
) -> None

Draw one agent run from the lane, up to its end: as a timeline when the channel streams, else quietly. What a failed rendering left unread is drained, so the lane resumes after the run.

Source code in octomate/tentacles/siphon.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def run(
    self, address: ChannelAddress, impulses: MemoryObjectReceiveStream[Impulse]
) -> None:
    """Draw one agent run from the lane, up to its end: as a timeline when the
    channel streams, else quietly. What a failed rendering left unread is
    drained, so the lane resumes after the run."""
    ended: list[RunEndedEvent] = []

    async def events() -> AsyncGenerator[WireEvent]:
        async for impulse in impulses:
            if isinstance(impulse.event, RunEndedEvent):
                ended.append(impulse.event)
                return
            yield impulse.event

    stream = events()
    try:
        if self.channel.config.stream.enabled:
            async with self.channel.feelers.timeline.open(address) as timeline:
                timeline.artifacts = self.channel.feelers.artifacts
                await timeline.drive(stream)
            rendered = timeline.message_id
        else:
            rendered = await self.quietly(address, stream)
    except Exception:
        async for _ in stream:
            pass
        raise
    if ended:
        await self.mark(ended[0].thread_message_id, rendered)

quietly async 

quietly(
    address: ChannelAddress,
    events: AsyncGenerator[WireEvent],
) -> IMMessageID | None

A run on a channel that does not stream: its sends, batches and authorizations as they come, then its reply once. Answers the platform id the reply was posted as.

Source code in octomate/tentacles/siphon.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def quietly(
    self, address: ChannelAddress, events: AsyncGenerator[WireEvent]
) -> IMMessageID | None:
    """A run on a channel that does not stream: its sends, batches and
    authorizations as they come, then its reply once. Answers the platform id
    the reply was posted as."""
    feelers = self.channel.feelers
    output: str | list[MessageSegment] | None = None
    async for event in events:
        match event:
            case MessageSentEvent():
                await feelers.segments.present(
                    event.destination or address, event.segments
                )
            case ActionBatchEvent():
                await feelers.present_actions(
                    address,
                    event,
                    action_manager=self.channel.octomate.deferred_actions,
                )
            case OAuthAuthorizationEvent():
                await feelers.oauth.present(address, event)
            case ArtifactAvailableEvent():
                await feelers.artifacts.present(address, event)
            case RunResultEvent():
                output = event.output
            case _:
                pass
    if isinstance(output, str):
        return await feelers.markdown.present(address, output) if output else None
    return await feelers.segments.present(address, output) if output else None

mark async 

mark(
    thread_message_id: UUID7 | None,
    platform_message_id: str | None,
) -> None

Name the platform message a recorded row was rendered as.

Source code in octomate/tentacles/siphon.py 
202
203
204
205
206
207
208
209
async def mark(
    self, thread_message_id: UUID7 | None, platform_message_id: str | None
) -> None:
    """Name the platform message a recorded row was rendered as."""
    if thread_message_id is not None:
        await self.channel.octomate.threads.mark_presented(
            thread_message_id, platform_message_id
        )