Skip to content

Graph, state and dependencies

octomate.reflex.graph

Wiring for the reflex graph.

The nodes live one per module under nodes/; this is where they are assembled into a runnable graph. Names the rest of the project imports from octomate.reflex.graph are re-exported here so the split stays invisible to callers.

ReflexEntryT module-attribute 

ReflexEntryT = TypeVar(
    "ReflexEntryT",
    bound="BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]",
)

ResponseTargetMode module-attribute 

ResponseTargetMode = Literal['main', 'sub']

reflex_graph module-attribute 

reflex_graph = build_reflex_graph()

ReflexGraphResult 

ReflexGraphResult = (
    ReflexResult | DeferredResult | CommandOutcome
)

Awake dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

The entry node: explicit commands dispatch, batch replies resume, native handoffs land, and user messages resolve their thread and go to Route.

Source code in octomate/reflex/nodes/awake.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@dataclass
class Awake(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """The entry node: explicit commands dispatch, batch replies resume, native
    handoffs land, and user messages resolve their thread and go to `Route`."""

    signal: AwakeSignal

    @reflex_logfire.instrument("reflex.awake", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> (
        Command
        | Route
        | ResumeDeferred
        | Summon
        | Teleport
        | Scheme
        | End[ReflexGraphResult]
    ):
        if isinstance(self.signal, CommandSignal):
            return Command(signal=self.signal)

        if isinstance(self.signal, DeferredActionBatchResponse):
            return ResumeDeferred(awake=self.signal)

        if isinstance(self.signal, DrivenGatewaySignal):
            signal = self.signal
            return await self.thread_operation(
                ctx,
                signal.thread_id,
                signal.decision,
                agent_id=signal.agent_id,
                sender=signal.user_profile,
                source=signal.source,
                requested_from="Trunkline",
            )

        if isinstance(self.signal, NativeGatewaySignal):
            # A native session's spell, already validated at the gateway: enter
            # the graph where React's post-run consumption would have, with the
            # native pseudo-channel — which nobody serves, so no lookup — as the
            # source.
            source = self.signal.source
            if source is None:
                raise ValueError("a gateway handoff signal needs its source address")
            decision = self.signal.decision
            if isinstance(decision, TeleportDecision):
                if self.signal.user_profile is None:
                    raise ValueError("a native teleport needs its thread's owner")
                return await self.thread_operation(
                    ctx,
                    ThreadKey.from_address(source),
                    decision,
                    agent_id=self.signal.agent_id,
                    sender=self.signal.user_profile,
                    source=source,
                    requested_from=self.signal.agent_id,
                )
            source_target = ResponseTarget(
                channel_id=source.channel_tentacle_id, address=source
            )
            ctx.state.source_target = source_target
            ctx.state.user_profile = self.signal.user_profile
            ctx.state.handoff = PendingHandoff(
                source_agent_tentacle_id=self.signal.agent_id
            )
            return Scheme(
                request=decision,
                origin=source_target,
                agent_id=self.signal.agent_id,
            )

        if not self.signal:
            reflex_logfire.info("awake short-circuit: empty signal")
            return End(
                ReflexResult(
                    decision=None,
                    target=ResponseTarget(channel_id=""),
                )
            )

        address = self.signal.address
        channel = ctx.deps.channels.get(address.channel_tentacle_id)
        if channel is None:
            raise ValueError(f"unknown channel {address.channel_tentacle_id!r}")

        source_target = ResponseTarget(
            channel_id=address.channel_tentacle_id,
            address=address,
            thread_strategy=channel.thread_strategy,
            mode="main",
        )
        ctx.state.source_target = source_target
        ctx.state.thread = await ctx.deps.thread_manager.enter(address)
        ctx.state.trigger_thread_message_id = self.signal.trigger_thread_message_id
        ctx.state.user_profile = self.signal.messages[-1].sender

        user_prompt = "\n\n".join(str(event) for event in self.signal.messages).strip()
        ctx.state.user_prompt = user_prompt
        if not user_prompt and self.signal.trigger_thread_message_id is None:
            reflex_logfire.info(
                "awake short-circuit: empty prompt",
                channel_id=address.channel_tentacle_id,
                conversation_address=str(address),
            )
            return End(
                ReflexResult(
                    decision=None,
                    target=source_target,
                )
            )
        return Route()

    async def thread_operation(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        thread_ref: uuid.UUID | ThreadKey,
        decision: SummonDecision | TeleportDecision,
        *,
        agent_id: str,
        sender: UserProfile,
        source: ChannelAddress,
        requested_from: str,
    ) -> Summon | Teleport:
        """Enter an existing thread its owner acts on: the console's operation on
        any thread, or a native session's teleport of its own."""
        origin = ResponseTarget(
            channel_id=source.channel_tentacle_id,
            address=source,
            mode="sub" if source.channel_thread_id else "main",
        )
        # Set first, so a refusal below is still reported.
        ctx.state.source_target = origin
        thread = await ctx.deps.thread_manager.get(
            thread_ref, with_messages=False, user_id=sender.user_id
        )
        if thread is None or thread.active_agent_tentacle_id != agent_id:
            raise ValueError("The source thread changed; reload its operations.")
        ctx.state.thread = thread
        ctx.state.user_profile = sender
        ctx.state.target = origin
        if isinstance(decision, TeleportDecision):
            agent = ctx.deps.agent(decision.agent_id)
            conversation = next(
                c
                for c in thread.conversations
                if c.agent_tentacle_id == agent_id and not c.subagent_id
            )
            stored = await ctx.deps.conversation_manager.get(conversation.id)
            if thread.kind == "native_thread":
                await agent.validate_fork(stored, sender=sender)
            model = stored.runs[-1].model_name if stored.runs else thread.active_model
            ctx.state.decision = SummonDecision(
                agent_id=agent.id,
                model=None if thread.kind == "native_thread" else model,
                reason=f"Teleport requested from {requested_from}",
                hint=decision.hint,
                brief="",
            )
            return Teleport(
                request=TeleportRequest(
                    tool_call_id=None,
                    hint=decision.hint,
                    destination=decision.destination,
                    new_thread=decision.new_thread,
                    project=decision.project,
                    ref=decision.ref,
                    resume=decision.resume,
                    prompt=decision.prompt,
                ),
                origin=origin,
                agent_id=agent.id,
            )
        ctx.state.decision = decision
        ctx.state.run_name = "summon"
        ctx.state.handoff = PendingHandoff(
            source_agent_tentacle_id=agent_id,
            source_conversation_id=next(
                (
                    c.id
                    for c in thread.conversations
                    if c.agent_tentacle_id == agent_id and not c.subagent_id
                ),
                None,
            ),
        )
        return Summon()

signal instance-attribute 

signal: AwakeSignal

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> (
    Command
    | Route
    | ResumeDeferred
    | Summon
    | Teleport
    | Scheme
    | End[ReflexGraphResult]
)
Source code in octomate/reflex/nodes/awake.py 
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.awake", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> (
    Command
    | Route
    | ResumeDeferred
    | Summon
    | Teleport
    | Scheme
    | End[ReflexGraphResult]
):
    if isinstance(self.signal, CommandSignal):
        return Command(signal=self.signal)

    if isinstance(self.signal, DeferredActionBatchResponse):
        return ResumeDeferred(awake=self.signal)

    if isinstance(self.signal, DrivenGatewaySignal):
        signal = self.signal
        return await self.thread_operation(
            ctx,
            signal.thread_id,
            signal.decision,
            agent_id=signal.agent_id,
            sender=signal.user_profile,
            source=signal.source,
            requested_from="Trunkline",
        )

    if isinstance(self.signal, NativeGatewaySignal):
        # A native session's spell, already validated at the gateway: enter
        # the graph where React's post-run consumption would have, with the
        # native pseudo-channel — which nobody serves, so no lookup — as the
        # source.
        source = self.signal.source
        if source is None:
            raise ValueError("a gateway handoff signal needs its source address")
        decision = self.signal.decision
        if isinstance(decision, TeleportDecision):
            if self.signal.user_profile is None:
                raise ValueError("a native teleport needs its thread's owner")
            return await self.thread_operation(
                ctx,
                ThreadKey.from_address(source),
                decision,
                agent_id=self.signal.agent_id,
                sender=self.signal.user_profile,
                source=source,
                requested_from=self.signal.agent_id,
            )
        source_target = ResponseTarget(
            channel_id=source.channel_tentacle_id, address=source
        )
        ctx.state.source_target = source_target
        ctx.state.user_profile = self.signal.user_profile
        ctx.state.handoff = PendingHandoff(
            source_agent_tentacle_id=self.signal.agent_id
        )
        return Scheme(
            request=decision,
            origin=source_target,
            agent_id=self.signal.agent_id,
        )

    if not self.signal:
        reflex_logfire.info("awake short-circuit: empty signal")
        return End(
            ReflexResult(
                decision=None,
                target=ResponseTarget(channel_id=""),
            )
        )

    address = self.signal.address
    channel = ctx.deps.channels.get(address.channel_tentacle_id)
    if channel is None:
        raise ValueError(f"unknown channel {address.channel_tentacle_id!r}")

    source_target = ResponseTarget(
        channel_id=address.channel_tentacle_id,
        address=address,
        thread_strategy=channel.thread_strategy,
        mode="main",
    )
    ctx.state.source_target = source_target
    ctx.state.thread = await ctx.deps.thread_manager.enter(address)
    ctx.state.trigger_thread_message_id = self.signal.trigger_thread_message_id
    ctx.state.user_profile = self.signal.messages[-1].sender

    user_prompt = "\n\n".join(str(event) for event in self.signal.messages).strip()
    ctx.state.user_prompt = user_prompt
    if not user_prompt and self.signal.trigger_thread_message_id is None:
        reflex_logfire.info(
            "awake short-circuit: empty prompt",
            channel_id=address.channel_tentacle_id,
            conversation_address=str(address),
        )
        return End(
            ReflexResult(
                decision=None,
                target=source_target,
            )
        )
    return Route()

thread_operation async 

thread_operation(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    thread_ref: UUID | ThreadKey,
    decision: SummonDecision | TeleportDecision,
    *,
    agent_id: str,
    sender: UserProfile,
    source: ChannelAddress,
    requested_from: str,
) -> Summon | Teleport

Enter an existing thread its owner acts on: the console's operation on any thread, or a native session's teleport of its own.

Source code in octomate/reflex/nodes/awake.py 
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
async def thread_operation(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    thread_ref: uuid.UUID | ThreadKey,
    decision: SummonDecision | TeleportDecision,
    *,
    agent_id: str,
    sender: UserProfile,
    source: ChannelAddress,
    requested_from: str,
) -> Summon | Teleport:
    """Enter an existing thread its owner acts on: the console's operation on
    any thread, or a native session's teleport of its own."""
    origin = ResponseTarget(
        channel_id=source.channel_tentacle_id,
        address=source,
        mode="sub" if source.channel_thread_id else "main",
    )
    # Set first, so a refusal below is still reported.
    ctx.state.source_target = origin
    thread = await ctx.deps.thread_manager.get(
        thread_ref, with_messages=False, user_id=sender.user_id
    )
    if thread is None or thread.active_agent_tentacle_id != agent_id:
        raise ValueError("The source thread changed; reload its operations.")
    ctx.state.thread = thread
    ctx.state.user_profile = sender
    ctx.state.target = origin
    if isinstance(decision, TeleportDecision):
        agent = ctx.deps.agent(decision.agent_id)
        conversation = next(
            c
            for c in thread.conversations
            if c.agent_tentacle_id == agent_id and not c.subagent_id
        )
        stored = await ctx.deps.conversation_manager.get(conversation.id)
        if thread.kind == "native_thread":
            await agent.validate_fork(stored, sender=sender)
        model = stored.runs[-1].model_name if stored.runs else thread.active_model
        ctx.state.decision = SummonDecision(
            agent_id=agent.id,
            model=None if thread.kind == "native_thread" else model,
            reason=f"Teleport requested from {requested_from}",
            hint=decision.hint,
            brief="",
        )
        return Teleport(
            request=TeleportRequest(
                tool_call_id=None,
                hint=decision.hint,
                destination=decision.destination,
                new_thread=decision.new_thread,
                project=decision.project,
                ref=decision.ref,
                resume=decision.resume,
                prompt=decision.prompt,
            ),
            origin=origin,
            agent_id=agent.id,
        )
    ctx.state.decision = decision
    ctx.state.run_name = "summon"
    ctx.state.handoff = PendingHandoff(
        source_agent_tentacle_id=agent_id,
        source_conversation_id=next(
            (
                c.id
                for c in thread.conversations
                if c.agent_tentacle_id == agent_id and not c.subagent_id
            ),
            None,
        ),
    )
    return Summon()

Command dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Own one command invocation through direct feedback or an ordinary agent stream.

Source code in octomate/reflex/nodes/command.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@dataclass
class Command(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Own one command invocation through direct feedback or an ordinary agent stream."""

    signal: CommandSignal

    @reflex_logfire.instrument("reflex.command", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> Summon | Scheme | Teleport | End[ReflexGraphResult]:
        context = self.signal.context
        agent = ctx.deps.agent(context.agent_id)
        channel = ctx.deps.channel(context.address.channel_tentacle_id)
        state = ctx.state
        target = ResponseTarget(
            channel_id=channel.id,
            address=context.address,
            thread_strategy=channel.thread_strategy,
        )
        state.source_target = state.target = target
        state.user_profile = None
        state.run_name = "command"
        state.decision = None
        state.selection = AgentRouteKey(context.agent_id, context.model)
        conversation = context.conversation
        state.conversation_id = conversation.id if conversation else None
        state.thread = (
            await ctx.deps.thread_manager.get(
                conversation.thread_id, with_messages=False
            )
            if conversation is not None
            else None
        )
        runtime = ctx.deps.runtime
        session, suspender, capabilities = await runtime.resources(ctx)
        try:
            async with AsyncExitStack() as stack:
                validated = await stack.enter_async_context(
                    agent.commands.validate(
                        agent,
                        context,
                        self.signal.invocation,
                        delivery_id=self.signal.delivery_id,
                        session=session,
                    )
                )
                if isinstance(validated, CommandResult | CommandError):
                    await runtime.present_command(ctx, suspender, validated)
                    return End(validated)
                if validated.profile is not None:
                    await runtime.prepare_user(
                        ctx,
                        validated.profile,
                        session=session,
                        suspender=suspender,
                        capabilities=capabilities,
                    )
                output = await stack.enter_async_context(
                    agent.commands.execute(
                        agent,
                        context,
                        self.signal.invocation,
                        validated,
                        delivery_id=self.signal.delivery_id,
                        deferred_suspender=suspender,
                        capabilities=capabilities,
                    )
                )
                if isinstance(output, CommandResult | CommandError):
                    await runtime.present_command(ctx, suspender, output)
                    return End(output)
                state.decision = suspender.continuation_decision()
                result = await runtime.drive(ctx, suspender, output)
                return await runtime.finish(ctx, session, suspender, result)
        finally:
            if state.thread is not None and state.user_profile is not None:
                await ctx.deps.workspaces.save(state.thread)

signal instance-attribute 

signal: CommandSignal

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> Summon | Scheme | Teleport | End[ReflexGraphResult]
Source code in octomate/reflex/nodes/command.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.command", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> Summon | Scheme | Teleport | End[ReflexGraphResult]:
    context = self.signal.context
    agent = ctx.deps.agent(context.agent_id)
    channel = ctx.deps.channel(context.address.channel_tentacle_id)
    state = ctx.state
    target = ResponseTarget(
        channel_id=channel.id,
        address=context.address,
        thread_strategy=channel.thread_strategy,
    )
    state.source_target = state.target = target
    state.user_profile = None
    state.run_name = "command"
    state.decision = None
    state.selection = AgentRouteKey(context.agent_id, context.model)
    conversation = context.conversation
    state.conversation_id = conversation.id if conversation else None
    state.thread = (
        await ctx.deps.thread_manager.get(
            conversation.thread_id, with_messages=False
        )
        if conversation is not None
        else None
    )
    runtime = ctx.deps.runtime
    session, suspender, capabilities = await runtime.resources(ctx)
    try:
        async with AsyncExitStack() as stack:
            validated = await stack.enter_async_context(
                agent.commands.validate(
                    agent,
                    context,
                    self.signal.invocation,
                    delivery_id=self.signal.delivery_id,
                    session=session,
                )
            )
            if isinstance(validated, CommandResult | CommandError):
                await runtime.present_command(ctx, suspender, validated)
                return End(validated)
            if validated.profile is not None:
                await runtime.prepare_user(
                    ctx,
                    validated.profile,
                    session=session,
                    suspender=suspender,
                    capabilities=capabilities,
                )
            output = await stack.enter_async_context(
                agent.commands.execute(
                    agent,
                    context,
                    self.signal.invocation,
                    validated,
                    delivery_id=self.signal.delivery_id,
                    deferred_suspender=suspender,
                    capabilities=capabilities,
                )
            )
            if isinstance(output, CommandResult | CommandError):
                await runtime.present_command(ctx, suspender, output)
                return End(output)
            state.decision = suspender.continuation_decision()
            result = await runtime.drive(ctx, suspender, output)
            return await runtime.finish(ctx, session, suspender, result)
    finally:
        if state.thread is not None and state.user_profile is not None:
            await ctx.deps.workspaces.save(state.thread)

React dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Runs the summoned agent on the target channel and acts on how the run ends: a reply, a spell to perform, or a deferral to park.

Source code in octomate/reflex/nodes/react.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@dataclass
class React(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Runs the summoned agent on the target channel and acts on how the run ends:
    a reply, a spell to perform, or a deferral to park."""

    resume_batch_id: UUID7 | None = None
    # Set by Teleport to resume the same agent where it landed, with its pending
    # call resolved — against the forked history, or in place.
    resume_results: DeferredToolResults | None = None

    @reflex_logfire.instrument("reflex.react", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React | Summon | Scheme | Teleport | End[ReflexGraphResult]:
        """React, and leave the turn's workspace in the mirror however it ends.

        In a `finally` because a turn that raised still did whatever it did on
        disk. The files an agent wrote before its provider dropped the connection
        are work, and until this runs the workspace is the only copy of them —
        which also means the sweep can never reclaim that workspace, since it
        refuses anything the mirror has not seen. A failed turn on a thread nobody
        resumes would otherwise hold its disk for good.

        A cancelled turn is the one case this does not cover: the save's first
        await raises straight back out, so its work stays in the workspace and the
        sweep keeps it, which is the safe direction.
        """
        try:
            return await self.react(ctx)
        finally:
            if ctx.state.thread is not None:
                await ctx.deps.workspaces.save(ctx.state.thread)

    async def react(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React | Summon | Scheme | Teleport | End[ReflexGraphResult]:
        state = ctx.state
        decision = state.decision
        target = state.target
        source_target = state.source_target
        if (
            decision is None
            or target is None
            or target.address is None
            or source_target is None
            or source_target.address is None
        ):
            raise ValueError("React requires a decision and resolved target")
        if not isinstance(decision, SummonDecision):
            raise ValueError("React requires a summon decision")
        target_address = target.address
        # Against the channel the run will happen on, not the one it came from. A
        # spell that moves the turn to another channel leaves those two different,
        # and which agents serve a channel is that channel's own config — resolving
        # against the origin swaps in whatever the origin happens to list first.
        resolved = ctx.deps.resolve_agent(
            target.channel_id, decision.agent_id, decision.model
        )
        model = resolved.model
        decision = decision.model_copy(
            update={"agent_id": resolved.agent, "model": model}
        )
        state.decision = decision
        state.selection = decision.key
        state.conversation_id = None
        agent = ctx.deps.agent(resolved.agent)
        run_model = agent.models[model] if model is not None else None
        thread_id = state.thread.id if state.thread else None
        claim = state.handoff
        if state.thread is not None and claim is not None:
            await claim.land(ctx.deps, state.thread, decision)
            state.handoff = None
        runtime = ctx.deps.runtime
        session, suspender, capabilities = await runtime.resources(ctx)
        # A summon names the level it hands over at; every other turn runs at the
        # one the conversation was set to.
        effort = decision.effort
        if effort is None and state.conversation_id is not None:
            conversation = await ctx.deps.conversation_manager.get(
                state.conversation_id, with_history=False
            )
            effort = await agent.resolve_effort(conversation, model=model)
        if state.user_profile is not None:
            await runtime.prepare_user(
                ctx,
                state.user_profile,
                session=session,
                suspender=suspender,
                capabilities=capabilities,
            )
        deferred_results = self.resume_results
        if self.resume_batch_id is not None:
            batch = await ctx.deps.action_manager.get_batch(self.resume_batch_id)
            deferred_results = batch.build_results()
        user_prompt: str | Sequence[UserContent] | None = (
            None
            if deferred_results is not None
            else decision.brief or str(state.user_prompt or "")
        )

        async def events() -> AsyncGenerator[ReactStreamEvent[ChannelOutput], None]:
            async with agent.run_stream_events(
                user_prompt,
                conversation_address=target_address,
                thread_id=thread_id,
                source_thread_address=state.source_thread_address,
                source_thread_message_ids=state.source_thread_message_ids,
                run_name=state.run_name,
                model=run_model,
                effort=effort,
                deferred_tool_results=deferred_results,
                deferred_suspender=suspender,
                capabilities=capabilities,
            ) as stream:
                async for event in stream:
                    yield event

        with reflex_logfire.span(
            "react",
            channel_id=target_address.channel_tentacle_id,
            agent_id=agent.id,
            conversation_address=str(target_address),
            streaming=suspender.channel.config.stream.enabled,
            resume_batch_id=str(self.resume_batch_id) if self.resume_batch_id else None,
        ) as span:
            async with ctx.deps.gateway.driving(
                session, conversation_id=state.conversation_id
            ):
                result = await runtime.drive(ctx, suspender, events())
                span.set_attribute("react.run_id", result.run_id)
                span.set_attribute(
                    "react.deferred", isinstance(result.output, DeferredToolRequests)
                )
                if self.resume_batch_id is not None:
                    await ctx.deps.action_manager.mark_batch(
                        self.resume_batch_id, "completed", completed=True
                    )
                return await runtime.finish(ctx, session, suspender, result)

resume_batch_id class-attribute instance-attribute 

resume_batch_id: UUID7 | None = None

resume_results class-attribute instance-attribute 

resume_results: DeferredToolResults | None = None

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> (
    React
    | Summon
    | Scheme
    | Teleport
    | End[ReflexGraphResult]
)

React, and leave the turn's workspace in the mirror however it ends.

In a finally because a turn that raised still did whatever it did on disk. The files an agent wrote before its provider dropped the connection are work, and until this runs the workspace is the only copy of them — which also means the sweep can never reclaim that workspace, since it refuses anything the mirror has not seen. A failed turn on a thread nobody resumes would otherwise hold its disk for good.

A cancelled turn is the one case this does not cover: the save's first await raises straight back out, so its work stays in the workspace and the sweep keeps it, which is the safe direction.

Source code in octomate/reflex/nodes/react.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@reflex_logfire.instrument("reflex.react", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | Summon | Scheme | Teleport | End[ReflexGraphResult]:
    """React, and leave the turn's workspace in the mirror however it ends.

    In a `finally` because a turn that raised still did whatever it did on
    disk. The files an agent wrote before its provider dropped the connection
    are work, and until this runs the workspace is the only copy of them —
    which also means the sweep can never reclaim that workspace, since it
    refuses anything the mirror has not seen. A failed turn on a thread nobody
    resumes would otherwise hold its disk for good.

    A cancelled turn is the one case this does not cover: the save's first
    await raises straight back out, so its work stays in the workspace and the
    sweep keeps it, which is the safe direction.
    """
    try:
        return await self.react(ctx)
    finally:
        if ctx.state.thread is not None:
            await ctx.deps.workspaces.save(ctx.state.thread)

react async 

react(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> (
    React
    | Summon
    | Scheme
    | Teleport
    | End[ReflexGraphResult]
)
Source code in octomate/reflex/nodes/react.py 
 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
async def react(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | Summon | Scheme | Teleport | End[ReflexGraphResult]:
    state = ctx.state
    decision = state.decision
    target = state.target
    source_target = state.source_target
    if (
        decision is None
        or target is None
        or target.address is None
        or source_target is None
        or source_target.address is None
    ):
        raise ValueError("React requires a decision and resolved target")
    if not isinstance(decision, SummonDecision):
        raise ValueError("React requires a summon decision")
    target_address = target.address
    # Against the channel the run will happen on, not the one it came from. A
    # spell that moves the turn to another channel leaves those two different,
    # and which agents serve a channel is that channel's own config — resolving
    # against the origin swaps in whatever the origin happens to list first.
    resolved = ctx.deps.resolve_agent(
        target.channel_id, decision.agent_id, decision.model
    )
    model = resolved.model
    decision = decision.model_copy(
        update={"agent_id": resolved.agent, "model": model}
    )
    state.decision = decision
    state.selection = decision.key
    state.conversation_id = None
    agent = ctx.deps.agent(resolved.agent)
    run_model = agent.models[model] if model is not None else None
    thread_id = state.thread.id if state.thread else None
    claim = state.handoff
    if state.thread is not None and claim is not None:
        await claim.land(ctx.deps, state.thread, decision)
        state.handoff = None
    runtime = ctx.deps.runtime
    session, suspender, capabilities = await runtime.resources(ctx)
    # A summon names the level it hands over at; every other turn runs at the
    # one the conversation was set to.
    effort = decision.effort
    if effort is None and state.conversation_id is not None:
        conversation = await ctx.deps.conversation_manager.get(
            state.conversation_id, with_history=False
        )
        effort = await agent.resolve_effort(conversation, model=model)
    if state.user_profile is not None:
        await runtime.prepare_user(
            ctx,
            state.user_profile,
            session=session,
            suspender=suspender,
            capabilities=capabilities,
        )
    deferred_results = self.resume_results
    if self.resume_batch_id is not None:
        batch = await ctx.deps.action_manager.get_batch(self.resume_batch_id)
        deferred_results = batch.build_results()
    user_prompt: str | Sequence[UserContent] | None = (
        None
        if deferred_results is not None
        else decision.brief or str(state.user_prompt or "")
    )

    async def events() -> AsyncGenerator[ReactStreamEvent[ChannelOutput], None]:
        async with agent.run_stream_events(
            user_prompt,
            conversation_address=target_address,
            thread_id=thread_id,
            source_thread_address=state.source_thread_address,
            source_thread_message_ids=state.source_thread_message_ids,
            run_name=state.run_name,
            model=run_model,
            effort=effort,
            deferred_tool_results=deferred_results,
            deferred_suspender=suspender,
            capabilities=capabilities,
        ) as stream:
            async for event in stream:
                yield event

    with reflex_logfire.span(
        "react",
        channel_id=target_address.channel_tentacle_id,
        agent_id=agent.id,
        conversation_address=str(target_address),
        streaming=suspender.channel.config.stream.enabled,
        resume_batch_id=str(self.resume_batch_id) if self.resume_batch_id else None,
    ) as span:
        async with ctx.deps.gateway.driving(
            session, conversation_id=state.conversation_id
        ):
            result = await runtime.drive(ctx, suspender, events())
            span.set_attribute("react.run_id", result.run_id)
            span.set_attribute(
                "react.deferred", isinstance(result.output, DeferredToolRequests)
            )
            if self.resume_batch_id is not None:
                await ctx.deps.action_manager.mark_batch(
                    self.resume_batch_id, "completed", completed=True
                )
            return await runtime.finish(ctx, session, suspender, result)

ResumeDeferred dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Resolves a batch from the user's reply and, once it is whole, resumes the suspended run through React.

Source code in octomate/reflex/nodes/resume_deferred.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@dataclass
class ResumeDeferred(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Resolves a batch from the user's reply and, once it is whole, resumes the
    suspended run through `React`."""

    awake: DeferredActionBatchResponse

    @reflex_logfire.instrument("reflex.resume_deferred", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React | End[ReflexGraphResult]:
        state = ctx.state
        with reflex_logfire.span(
            "resume_deferred",
            batch_id=str(self.awake.batch_id),
        ) as span:
            batch = await ctx.deps.action_manager.get_batch(self.awake.batch_id)
            if batch.response_mode != "resume":
                raise ValueError("A live request cannot resume a suspended run")
            batch = await ctx.deps.action_manager.resolve_batch(self.awake)
            span.set_attribute("run_name", batch.run_name)
            span.set_attribute("batch_status", batch.status)
            span.set_attribute("completed", batch.completed)
        # For every reader still showing the batch's cards, before the run resumes.
        await ctx.deps.impulses.send(
            Impulse(ActionBatchAnsweredEvent.of(self.awake), batch.target_address)
        )
        target_channel = ctx.deps.channels.get(batch.target_address.channel_tentacle_id)
        if target_channel is None:
            raise ValueError(
                f"unknown channel {batch.target_address.channel_tentacle_id!r}"
            )
        ctx.deps.agent(batch.agent_tentacle_id)
        target = ResponseTarget(
            channel_id=batch.target_address.channel_tentacle_id,
            address=batch.target_address,
            thread_strategy=target_channel.thread_strategy,
            mode=batch.target_mode,
        )
        source_channel = ctx.deps.channels.get(batch.source_address.channel_tentacle_id)
        state.target = target
        state.source_target = ResponseTarget(
            channel_id=batch.source_address.channel_tentacle_id,
            address=batch.source_address,
            thread_strategy=(
                source_channel.thread_strategy
                if source_channel is not None
                else target_channel.thread_strategy
            ),
            mode="main",
        )

        if isinstance(batch.decision, SummonDecision):
            decision = batch.decision
        elif batch.decision:
            decision = SummonDecision.model_validate(batch.decision)
        else:
            resolved = ctx.deps.resolve_agent(
                batch.source_address.channel_tentacle_id,
                batch.agent_tentacle_id,
                None,
            )
            model = resolved.model
            decision = SummonDecision(
                action="summon",
                agent_id=resolved.agent,
                model=model,
                reason="Resuming deferred human input.",
                hint="Resuming deferred human input.",
                brief="",
            )
        if not isinstance(decision, SummonDecision):
            raise ValueError("react deferred batch requires a summon decision")
        decision = decision.model_copy(update={"agent_id": batch.agent_tentacle_id})
        state.decision = decision
        if batch.status in {"completed", "resuming"}:
            reflex_logfire.info("resume already completed", batch_id=str(batch.id))
            return End(ReflexResult(decision=decision, target=target))
        if not batch.completed:
            return End(
                DeferredResult(
                    requests=batch.requests,
                    target=target,
                    run_name=batch.run_name or "react",
                    result=AgentRunResult(batch.requests),
                    batch_id=batch.id,
                )
            )

        await ctx.deps.action_manager.mark_batch(batch.id, "resuming")
        # The resumed agent run continues the batch's conversation, so bind the
        # thread that owns it (React reads state.thread for the run's thread_id).
        # Through the conversation and not the address: a chat room's kick ran in a
        # sub-thread, and the address names the chat it was seen on, which would
        # answer the deferral in a context the suspended run never had.
        thread_id = await ctx.deps.conversation_manager.thread_id(batch.conversation_id)
        if thread_id is None:
            raise ValueError(
                f"deferred batch {batch.id} names conversation "
                f"{batch.conversation_id}, which does not exist"
            )
        thread = await ctx.deps.thread_manager.get(thread_id, with_messages=False)
        if thread is None:
            raise ValueError(
                f"conversation {batch.conversation_id} names thread {thread_id}, "
                "which does not exist"
            )
        state.thread = thread
        # The resumed run must serve the user the suspended run served: React
        # mounts their per-user capabilities (and the gate's identity) from
        # state.user_profile, and without it the run loses tools mid-conversation
        # — which also breaks the provider prompt cache both ways.
        state.user_profile = await ctx.deps.thread_manager.users.profile(
            batch.source_address.channel_tentacle_id,
            batch.source_address.user_id,
        )
        state.user_prompt = None
        state.run_name = "resume"
        reflex_logfire.info("resume routes to React", batch_id=str(batch.id))
        return React(resume_batch_id=batch.id)

awake instance-attribute 

awake: DeferredActionBatchResponse

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]
Source code in octomate/reflex/nodes/resume_deferred.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.resume_deferred", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]:
    state = ctx.state
    with reflex_logfire.span(
        "resume_deferred",
        batch_id=str(self.awake.batch_id),
    ) as span:
        batch = await ctx.deps.action_manager.get_batch(self.awake.batch_id)
        if batch.response_mode != "resume":
            raise ValueError("A live request cannot resume a suspended run")
        batch = await ctx.deps.action_manager.resolve_batch(self.awake)
        span.set_attribute("run_name", batch.run_name)
        span.set_attribute("batch_status", batch.status)
        span.set_attribute("completed", batch.completed)
    # For every reader still showing the batch's cards, before the run resumes.
    await ctx.deps.impulses.send(
        Impulse(ActionBatchAnsweredEvent.of(self.awake), batch.target_address)
    )
    target_channel = ctx.deps.channels.get(batch.target_address.channel_tentacle_id)
    if target_channel is None:
        raise ValueError(
            f"unknown channel {batch.target_address.channel_tentacle_id!r}"
        )
    ctx.deps.agent(batch.agent_tentacle_id)
    target = ResponseTarget(
        channel_id=batch.target_address.channel_tentacle_id,
        address=batch.target_address,
        thread_strategy=target_channel.thread_strategy,
        mode=batch.target_mode,
    )
    source_channel = ctx.deps.channels.get(batch.source_address.channel_tentacle_id)
    state.target = target
    state.source_target = ResponseTarget(
        channel_id=batch.source_address.channel_tentacle_id,
        address=batch.source_address,
        thread_strategy=(
            source_channel.thread_strategy
            if source_channel is not None
            else target_channel.thread_strategy
        ),
        mode="main",
    )

    if isinstance(batch.decision, SummonDecision):
        decision = batch.decision
    elif batch.decision:
        decision = SummonDecision.model_validate(batch.decision)
    else:
        resolved = ctx.deps.resolve_agent(
            batch.source_address.channel_tentacle_id,
            batch.agent_tentacle_id,
            None,
        )
        model = resolved.model
        decision = SummonDecision(
            action="summon",
            agent_id=resolved.agent,
            model=model,
            reason="Resuming deferred human input.",
            hint="Resuming deferred human input.",
            brief="",
        )
    if not isinstance(decision, SummonDecision):
        raise ValueError("react deferred batch requires a summon decision")
    decision = decision.model_copy(update={"agent_id": batch.agent_tentacle_id})
    state.decision = decision
    if batch.status in {"completed", "resuming"}:
        reflex_logfire.info("resume already completed", batch_id=str(batch.id))
        return End(ReflexResult(decision=decision, target=target))
    if not batch.completed:
        return End(
            DeferredResult(
                requests=batch.requests,
                target=target,
                run_name=batch.run_name or "react",
                result=AgentRunResult(batch.requests),
                batch_id=batch.id,
            )
        )

    await ctx.deps.action_manager.mark_batch(batch.id, "resuming")
    # The resumed agent run continues the batch's conversation, so bind the
    # thread that owns it (React reads state.thread for the run's thread_id).
    # Through the conversation and not the address: a chat room's kick ran in a
    # sub-thread, and the address names the chat it was seen on, which would
    # answer the deferral in a context the suspended run never had.
    thread_id = await ctx.deps.conversation_manager.thread_id(batch.conversation_id)
    if thread_id is None:
        raise ValueError(
            f"deferred batch {batch.id} names conversation "
            f"{batch.conversation_id}, which does not exist"
        )
    thread = await ctx.deps.thread_manager.get(thread_id, with_messages=False)
    if thread is None:
        raise ValueError(
            f"conversation {batch.conversation_id} names thread {thread_id}, "
            "which does not exist"
        )
    state.thread = thread
    # The resumed run must serve the user the suspended run served: React
    # mounts their per-user capabilities (and the gate's identity) from
    # state.user_profile, and without it the run loses tools mid-conversation
    # — which also breaks the provider prompt cache both ways.
    state.user_profile = await ctx.deps.thread_manager.users.profile(
        batch.source_address.channel_tentacle_id,
        batch.source_address.user_id,
    )
    state.user_prompt = None
    state.run_name = "resume"
    reflex_logfire.info("resume routes to React", batch_id=str(batch.id))
    return React(resume_batch_id=batch.id)

Route dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Picks the agent for an inbound message — the chat's pinned owner, the flat-thread agent, or the channel's entry agent — and enters React.

Source code in octomate/reflex/nodes/route.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@dataclass
class Route(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Picks the agent for an inbound message — the chat's pinned owner, the
    flat-thread agent, or the channel's entry agent — and enters `React`."""

    @reflex_logfire.instrument("reflex.route", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React:
        state = ctx.state
        source_target = state.source_target
        if source_target is None or source_target.address is None:
            raise ValueError("Route requires a resolved source target")

        source_address = source_target.address

        # Who owns the chat, not who owns this kick: a chat room's sub-thread is
        # opened fresh every time, and the owner a handoff pinned is the chat's.
        chat = (
            await ctx.deps.thread_manager.surface(state.thread)
            if state.thread is not None
            else None
        )
        active_agent_id = chat.active_agent_tentacle_id if chat is not None else None
        if chat is not None and active_agent_id is not None:
            active_model = chat.active_model
            ctx.deps.agent(active_agent_id)
            resolved = ctx.deps.resolve_agent(
                source_address.channel_tentacle_id,
                active_agent_id,
                active_model,
            )
            await ctx.deps.load_pending_prompt(state, resolved.agent)
            model = resolved.model
            state.decision = SummonDecision(
                action="summon",
                agent_id=resolved.agent,
                model=model,
                reason="Continuing with the active thread owner.",
                hint="Continuing with the active thread owner.",
                brief="",
            )
            state.target = replace(
                source_target,
                mode="sub" if source_address.channel_thread_id else "main",
            )
            state.handoff = None
            return React()

        if (
            source_address.channel_thread_id
            and source_target.thread_strategy == "flat_thread"
        ):
            reflex_logfire.info(
                "route: flat-thread, skipping triage",
                channel_id=source_address.channel_tentacle_id,
                conversation_address=str(source_address),
            )
            resolved = ctx.deps.resolve_agent(
                source_address.channel_tentacle_id,
                None,
                None,
            )
            await ctx.deps.load_pending_prompt(state, resolved.agent)
            model = resolved.model
            state.decision = SummonDecision(
                action="summon",
                agent_id=resolved.agent,
                model=model,
                reason="Continuing in the current thread.",
                hint="Continuing in the current thread.",
                brief="",
            )
            state.target = replace(source_target, mode="sub")
            state.handoff = PendingHandoff()
            return React()

        # No active owner and not already in a flat thread: run the channel's default
        # agent directly in this conversation — no separate screening pass. It
        # self-routes via the gate toolset (summon / teleport) if it wants to hand off
        # or relocate. No handoff is recorded, so a group main is never pinned to an
        # owner (Case 1).
        resolved = ctx.deps.resolve_agent(
            source_address.channel_tentacle_id, None, None
        )
        await ctx.deps.load_pending_prompt(state, resolved.agent)
        state.decision = SummonDecision(
            action="summon",
            agent_id=resolved.agent,
            model=resolved.model,
            reason="Entry agent.",
            hint="",
            brief="",
        )
        state.target = replace(
            source_target, mode="sub" if source_address.channel_thread_id else "main"
        )
        state.handoff = None
        return React()

run async 

run(ctx: GraphRunContext[ReflexState, ReflexDeps]) -> React
Source code in octomate/reflex/nodes/route.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.route", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React:
    state = ctx.state
    source_target = state.source_target
    if source_target is None or source_target.address is None:
        raise ValueError("Route requires a resolved source target")

    source_address = source_target.address

    # Who owns the chat, not who owns this kick: a chat room's sub-thread is
    # opened fresh every time, and the owner a handoff pinned is the chat's.
    chat = (
        await ctx.deps.thread_manager.surface(state.thread)
        if state.thread is not None
        else None
    )
    active_agent_id = chat.active_agent_tentacle_id if chat is not None else None
    if chat is not None and active_agent_id is not None:
        active_model = chat.active_model
        ctx.deps.agent(active_agent_id)
        resolved = ctx.deps.resolve_agent(
            source_address.channel_tentacle_id,
            active_agent_id,
            active_model,
        )
        await ctx.deps.load_pending_prompt(state, resolved.agent)
        model = resolved.model
        state.decision = SummonDecision(
            action="summon",
            agent_id=resolved.agent,
            model=model,
            reason="Continuing with the active thread owner.",
            hint="Continuing with the active thread owner.",
            brief="",
        )
        state.target = replace(
            source_target,
            mode="sub" if source_address.channel_thread_id else "main",
        )
        state.handoff = None
        return React()

    if (
        source_address.channel_thread_id
        and source_target.thread_strategy == "flat_thread"
    ):
        reflex_logfire.info(
            "route: flat-thread, skipping triage",
            channel_id=source_address.channel_tentacle_id,
            conversation_address=str(source_address),
        )
        resolved = ctx.deps.resolve_agent(
            source_address.channel_tentacle_id,
            None,
            None,
        )
        await ctx.deps.load_pending_prompt(state, resolved.agent)
        model = resolved.model
        state.decision = SummonDecision(
            action="summon",
            agent_id=resolved.agent,
            model=model,
            reason="Continuing in the current thread.",
            hint="Continuing in the current thread.",
            brief="",
        )
        state.target = replace(source_target, mode="sub")
        state.handoff = PendingHandoff()
        return React()

    # No active owner and not already in a flat thread: run the channel's default
    # agent directly in this conversation — no separate screening pass. It
    # self-routes via the gate toolset (summon / teleport) if it wants to hand off
    # or relocate. No handoff is recorded, so a group main is never pinned to an
    # owner (Case 1).
    resolved = ctx.deps.resolve_agent(
        source_address.channel_tentacle_id, None, None
    )
    await ctx.deps.load_pending_prompt(state, resolved.agent)
    state.decision = SummonDecision(
        action="summon",
        agent_id=resolved.agent,
        model=resolved.model,
        reason="Entry agent.",
        hint="",
        brief="",
    )
    state.target = replace(
        source_target, mode="sub" if source_address.channel_thread_id else "main"
    )
    state.handoff = None
    return React()

Scheme dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

A scheme call: hand this turn's brief to the asking user's direct messages.

The receiver is whoever already owns that DM, or the channel's default agent when nobody does — never an agent this run picked, so a group can't point someone's private assistant somewhere. From there it is an ordinary handoff: the brief becomes the receiving agent's prompt and the handoff is recorded on the DM's own thread, the same way Route re-enters React with a decision it resolved itself.

Source code in octomate/reflex/nodes/scheme.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@dataclass
class Scheme(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """A `scheme` call: hand this turn's brief to the asking user's direct messages.

    The receiver is whoever already owns that DM, or the channel's default agent when
    nobody does — never an agent this run picked, so a group can't point someone's
    private assistant somewhere. From there it is an ordinary handoff: the brief becomes
    the receiving agent's prompt and the handoff is recorded on the DM's own thread, the
    same way `Route` re-enters `React` with a decision it resolved itself.
    """

    request: SchemeDecision
    origin: ResponseTarget
    agent_id: str

    @reflex_logfire.instrument("reflex.scheme", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React | End[ReflexGraphResult]:
        state = ctx.state
        origin = self.origin
        if origin.address is None:
            raise ValueError("Scheme requires a resolved origin")
        # Where the gate resolved this to: this channel's direct messages, or another
        # channel's, with the account there taken from the identity registry.
        target = self.request.destination
        channel = ctx.deps.channel(target.channel_tentacle_id)

        # The hint opens it, the way a teleport's does the sub-thread it lands in: on a
        # channel that can only be run inside a thread it is the message that thread
        # hangs from, and it is written to be read. The brief is not — that goes to
        # whoever answers, as their prompt.
        dm_address = await channel.open_dm(target.user_id, self.request.hint)
        if dm_address is None:
            # The gate refused the cases we can know in advance, so this is the platform
            # failing at the moment of asking. Nothing has moved: leave the turn where
            # it is, with the origin agent's own reply already delivered.
            logger.warning(
                "Channel %s could not open a DM with %s; leaving the turn here",
                target.channel_tentacle_id,
                target.user_id,
            )
            return End(ReflexResult(decision=None, target=origin))

        dm_thread = await ctx.deps.thread_manager.ensure(dm_address)
        receiver = dm_thread.active_agent_tentacle_id
        # Resolved against the *target* channel: which agents serve a channel is that
        # channel's own config, so a cross-channel move lands on an agent configured
        # there rather than one carried over from where the request came from.
        resolved = ctx.deps.resolve_agent(
            target.channel_tentacle_id,
            receiver,
            dm_thread.active_model if receiver else None,
        )
        state.run_name = "summon"
        state.thread = await ctx.deps.thread_manager.enter(
            dm_address, current=state.thread
        )
        state.target = ResponseTarget(
            channel_id=target.channel_tentacle_id,
            address=dm_address,
            thread_strategy=channel.thread_strategy,
            mode="main",
        )
        await ctx.deps.impulses.send(
            Impulse(
                GatewayEvent(action="scheme", destination=dm_address), origin.address
            )
        )
        state.decision = SummonDecision(
            action="summon",
            agent_id=resolved.agent,
            model=resolved.model,
            reason="Continuing with this user privately.",
            hint=self.request.hint,
            brief=self.request.brief,
        )
        return React()

request instance-attribute 

request: SchemeDecision

origin instance-attribute 

origin: ResponseTarget

agent_id instance-attribute 

agent_id: str

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]
Source code in octomate/reflex/nodes/scheme.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.scheme", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]:
    state = ctx.state
    origin = self.origin
    if origin.address is None:
        raise ValueError("Scheme requires a resolved origin")
    # Where the gate resolved this to: this channel's direct messages, or another
    # channel's, with the account there taken from the identity registry.
    target = self.request.destination
    channel = ctx.deps.channel(target.channel_tentacle_id)

    # The hint opens it, the way a teleport's does the sub-thread it lands in: on a
    # channel that can only be run inside a thread it is the message that thread
    # hangs from, and it is written to be read. The brief is not — that goes to
    # whoever answers, as their prompt.
    dm_address = await channel.open_dm(target.user_id, self.request.hint)
    if dm_address is None:
        # The gate refused the cases we can know in advance, so this is the platform
        # failing at the moment of asking. Nothing has moved: leave the turn where
        # it is, with the origin agent's own reply already delivered.
        logger.warning(
            "Channel %s could not open a DM with %s; leaving the turn here",
            target.channel_tentacle_id,
            target.user_id,
        )
        return End(ReflexResult(decision=None, target=origin))

    dm_thread = await ctx.deps.thread_manager.ensure(dm_address)
    receiver = dm_thread.active_agent_tentacle_id
    # Resolved against the *target* channel: which agents serve a channel is that
    # channel's own config, so a cross-channel move lands on an agent configured
    # there rather than one carried over from where the request came from.
    resolved = ctx.deps.resolve_agent(
        target.channel_tentacle_id,
        receiver,
        dm_thread.active_model if receiver else None,
    )
    state.run_name = "summon"
    state.thread = await ctx.deps.thread_manager.enter(
        dm_address, current=state.thread
    )
    state.target = ResponseTarget(
        channel_id=target.channel_tentacle_id,
        address=dm_address,
        thread_strategy=channel.thread_strategy,
        mode="main",
    )
    await ctx.deps.impulses.send(
        Impulse(
            GatewayEvent(action="scheme", destination=dm_address), origin.address
        )
    )
    state.decision = SummonDecision(
        action="summon",
        agent_id=resolved.agent,
        model=resolved.model,
        reason="Continuing with this user privately.",
        hint=self.request.hint,
        brief=self.request.brief,
    )
    return React()

Summon dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Performs a summon: takes the surface over in place for the agent the decision names, then re-enters React there.

Source code in octomate/reflex/nodes/summon.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@dataclass
class Summon(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Performs a summon: takes the surface over in place for the agent the decision
    names, then re-enters `React` there."""

    @reflex_logfire.instrument("reflex.summon", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React:
        state = ctx.state
        decision = state.decision
        target = state.target
        source_target = state.source_target
        if (
            decision is None
            or target is None
            or source_target is None
            or source_target.address is None
        ):
            raise ValueError("Summon requires a decision and source target")
        if not isinstance(decision, SummonDecision):
            raise ValueError("Summon requires a summon decision")
        source_address = source_target.address
        target_address = target.address
        if target_address is None:
            target_address = ChannelAddress(
                channel_tentacle_id=target.channel_id,
                chat_type=source_address.chat_type,
                chat_id=source_address.chat_id,
                user_id=source_address.user_id,
                channel_thread_id=None,
                shared=source_address.shared,
            )
            target = replace(target, address=target_address)

        state.target = target
        if state.thread is not None:
            state.thread = await ctx.deps.thread_manager.enter(
                target_address, current=state.thread
            )
        await ctx.deps.impulses.send(
            Impulse(
                GatewayEvent(action="summon", destination=target_address),
                target_address,
            )
        )
        return React()

run async 

run(ctx: GraphRunContext[ReflexState, ReflexDeps]) -> React
Source code in octomate/reflex/nodes/summon.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@reflex_logfire.instrument("reflex.summon", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React:
    state = ctx.state
    decision = state.decision
    target = state.target
    source_target = state.source_target
    if (
        decision is None
        or target is None
        or source_target is None
        or source_target.address is None
    ):
        raise ValueError("Summon requires a decision and source target")
    if not isinstance(decision, SummonDecision):
        raise ValueError("Summon requires a summon decision")
    source_address = source_target.address
    target_address = target.address
    if target_address is None:
        target_address = ChannelAddress(
            channel_tentacle_id=target.channel_id,
            chat_type=source_address.chat_type,
            chat_id=source_address.chat_id,
            user_id=source_address.user_id,
            channel_thread_id=None,
            shared=source_address.shared,
        )
        target = replace(target, address=target_address)

    state.target = target
    if state.thread is not None:
        state.thread = await ctx.deps.thread_manager.enter(
            target_address, current=state.thread
        )
    await ctx.deps.impulses.send(
        Impulse(
            GatewayEvent(action="summon", destination=target_address),
            target_address,
        )
    )
    return React()

Teleport dataclass

Bases: BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]

Create a thread at the prepared address, copy history and resume the agent.

An in-place move binds a project without creating a thread. A failed creation refuses the move before any history is copied.

Source code in octomate/reflex/nodes/teleport.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
@dataclass
class Teleport(BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]):
    """Create a thread at the prepared address, copy history and resume the agent.

    An in-place move binds a project without creating a thread. A failed creation
    refuses the move before any history is copied.
    """

    request: TeleportRequest
    origin: ResponseTarget
    agent_id: str

    @reflex_logfire.instrument("reflex.teleport", extract_args=False)
    async def run(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> React | End[ReflexGraphResult]:
        state = ctx.state
        state.run_name = "teleport"
        origin = self.origin
        if origin.address is None or state.thread is None:
            raise ValueError("Teleport requires a resolved origin and thread")
        origin_address = origin.address
        source = state.thread
        hint = self.request.hint or "Octomate is continuing this request here."
        sentence = "Continuing the conversation here."

        new_target = origin
        opened = None
        if self.request.new_thread:
            opened = await open_crossing(
                ctx, self.request.destination, origin_address, hint, self.agent_id
            )
            if opened is not None:
                channel = ctx.deps.channel(opened.channel_tentacle_id)
                new_target = ResponseTarget(
                    channel_id=channel.id,
                    address=opened,
                    thread_strategy=channel.thread_strategy,
                    mode="sub",
                )

        new_address = new_target.address
        if self.request.new_thread and new_address == origin_address:
            raise ValueError(
                "The destination could not create a thread; nothing was teleported."
            )
        source_conversation = None
        if new_address is None or new_address == origin_address:
            # The current conversation already holds the trailing teleport
            # deferral. An explicit workspace move may still fork its runtime.
            state.target = origin
            state.handoff = None
            landed = state.thread
            conversation = await ctx.deps.conversation_manager.ensure(
                landed.id, agent_tentacle_id=self.agent_id
            )
        else:
            # Move: fork the origin conversation into the new sub-thread, claim it
            # for the same agent so follow-ups continue there, and resume against
            # the fork. The runtime prepares its destination handle after the
            # destination workspace is ready.
            if state.thread.kind == "native_thread":
                source_agent_id = state.thread.active_agent_tentacle_id
                if (
                    state.user_profile is None
                    or state.decision is None
                    or source_agent_id is None
                ):
                    raise ValueError(
                        "Native teleport requires its requesting user and agent."
                    )
                source = await ctx.deps.conversation_manager.ensure(
                    state.thread.id,
                    agent_tentacle_id=source_agent_id,
                )
                agent = ctx.deps.agent(self.agent_id)
                landed = await agent.fork(
                    source,
                    ThreadKey.from_address(new_address),
                    sender=state.user_profile,
                )
                copied = await ctx.deps.conversation_manager.ensure(
                    landed.id, agent_tentacle_id=agent.id
                )
                state.decision = state.decision.model_copy(
                    update={"model": copied.runs[-1].model_name}
                )
                state.thread = landed
                state.target = new_target
                state.handoff = None
                await announce_crossing(
                    ctx,
                    self.request.destination,
                    new_address,
                    origin_address,
                    hint,
                    self.agent_id,
                )
                if self.request.prompt is None and not self.request.resume:
                    return await self.land(ctx)
                return await self.carry_on(ctx, sentence)
            landed = await ctx.deps.thread_manager.enter(
                new_address, current=state.thread
            )
            source_conversation = await ctx.deps.conversation_manager.ensure(
                state.thread.id, agent_tentacle_id=self.agent_id
            )
            target_conversation = await ctx.deps.conversation_manager.ensure(
                landed.id, agent_tentacle_id=self.agent_id
            )
            conversation = target_conversation
            state.thread = landed
            state.target = new_target
            state.handoff = PendingHandoff(
                source_agent_tentacle_id=self.agent_id,
                source_conversation_id=source_conversation.id,
            )

        project = self.request.project
        carried = (
            await ctx.deps.workspaces.projects.of(source)
            if project is None and landed.id != source.id
            else None
        )
        if project is not None:
            state.thread = await self.bind(ctx, landed, project)
            sentence = (
                f"Continuing the conversation here, in the workspace of {project!r}."
            )
        elif carried is not None:
            # The conversation was about a project, and its history is of work in
            # that tree: the thread it lands in is about the same one, as it stands.
            state.thread = await ctx.deps.thread_manager.bind(landed.id, carried)
            await ctx.deps.workspaces.carry(source, state.thread)
            sentence = (
                "Continuing the conversation here, in a workspace of "
                f"{carried.name!r} that holds your work as you left it."
            )
        # The agent resumes in another directory either way — a new thread's own
        # workspace, or the project's — and a runtime session may be filed under
        # the one it ran in. The tentacle knows how to move its own; this is when.
        cwd = ctx.deps.workspaces.open(
            state.thread.id, await ctx.deps.workspaces.projects.of(state.thread)
        ).path
        agent = ctx.deps.agent(self.agent_id)
        if source_conversation is not None:
            external_id = await agent.fork_session(source_conversation, cwd=cwd)
            carry = (
                external_id is not None
                and external_id == source_conversation.external_id
            )
            await ctx.deps.conversation_manager.fork(
                source_conversation,
                conversation,
                carry_external_id=carry,
                external_id=None if carry else external_id,
            )
            await agent.relocate(conversation, cwd=cwd)
            state.thread = await ctx.deps.thread_manager.rename(
                state.thread,
                ctx.deps.thread_manager.fork_title(source_conversation, source),
            )
        elif not self.request.new_thread:
            external_id = await agent.fork_session(conversation, cwd=cwd)
            if external_id is not None and external_id != conversation.external_id:
                await ctx.deps.conversation_manager.set_external_id(
                    conversation, external_id
                )
            await agent.relocate(conversation, cwd=cwd)
        if opened is not None:
            await announce_crossing(
                ctx,
                self.request.destination,
                opened,
                origin_address,
                hint,
                self.agent_id,
            )
        if self.request.prompt is None and not self.request.resume:
            return await self.land(ctx)
        return await self.carry_on(ctx, sentence)

    async def carry_on(
        self, ctx: GraphRunContext[ReflexState, ReflexDeps], sentence: str
    ) -> React:
        """Run the agent where the move landed: on the user's prompt, recorded there
        as their message, or on its own call answered by where it now is."""
        state = ctx.state
        address = state.target.address if state.target is not None else None
        if address is None:
            raise ValueError("Teleport carries on only where the move landed")
        prompt = self.request.prompt
        if prompt is None:
            # The pending call resolves into the resumed run, whichever runtime cast it.
            if self.request.tool_call_id is None:
                state.user_prompt = f"{sentence}\nCurrent channel address: {address}"
                return React()
            return React(
                resume_results=DeferredToolResults(
                    calls={self.request.tool_call_id: sentence}
                )
            )
        if state.user_profile is None:
            raise ValueError("A prompt after a teleport needs its user")
        channel = ctx.deps.channel(address.channel_tentacle_id)
        message = await ctx.deps.thread_manager.record_inbound(
            MessageEvent(
                tentacle_id=channel.id,
                chat_type=address.chat_type,
                chat_id=address.chat_id,
                user_id=address.user_id,
                channel_thread_id=address.channel_thread_id,
                shared=address.shared,
                self_id=channel.self_profile.channel_user_id,
                sender=state.user_profile,
                segments=[TextSegment(data={"text": prompt})],
                raw=prompt,
            )
        )
        state.source_target = state.target
        state.trigger_thread_message_id = message.id
        await ctx.deps.load_pending_prompt(state, self.agent_id)
        return React()

    async def land(
        self, ctx: GraphRunContext[ReflexState, ReflexDeps]
    ) -> End[ReflexGraphResult]:
        """End the move where it landed, for the next message there to carry on:
        the handoff is recorded and the workspace saved, as a turn's end would."""
        state = ctx.state
        if state.decision is None or state.target is None or state.thread is None:
            raise ValueError("Teleport lands with a decision, a target and a thread")
        resolved = ctx.deps.resolve_agent(
            state.target.channel_id, state.decision.agent_id, state.decision.model
        )
        state.decision = state.decision.model_copy(
            update={"agent_id": resolved.agent, "model": resolved.model}
        )
        if state.handoff is not None:
            await state.handoff.land(ctx.deps, state.thread, state.decision)
            state.handoff = None
        await ctx.deps.workspaces.save(state.thread)
        return End(ReflexResult(decision=state.decision, target=state.target))

    async def bind(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        thread: Thread,
        name: str,
    ) -> Thread:
        """Bind the thread landed in to the project the request names, and fork
        its workspace, so the resumed run starts in the project's code. The gate
        validated the project and the ref; a project missing here is a wiring
        bug. Answers the bound thread, re-read: the state's copy predates the
        binding, and the turn-end save reads the project off it."""
        project = ctx.deps.workspaces.projects.get(name)
        if project is None:
            raise RuntimeError(
                f"project {name!r} vanished between the gate and the move"
            )
        mirror = await ctx.deps.workspaces.mirrors.sync(project)
        bound = await ctx.deps.thread_manager.bind(thread.id, project)
        await ctx.deps.workspaces.materialize(
            ctx.deps.workspaces.open(thread.id, project), mirror, self.request.ref
        )
        return bound

request instance-attribute 

request: TeleportRequest

origin instance-attribute 

origin: ResponseTarget

agent_id instance-attribute 

agent_id: str

run async 

run(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]
Source code in octomate/reflex/nodes/teleport.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
@reflex_logfire.instrument("reflex.teleport", extract_args=False)
async def run(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> React | End[ReflexGraphResult]:
    state = ctx.state
    state.run_name = "teleport"
    origin = self.origin
    if origin.address is None or state.thread is None:
        raise ValueError("Teleport requires a resolved origin and thread")
    origin_address = origin.address
    source = state.thread
    hint = self.request.hint or "Octomate is continuing this request here."
    sentence = "Continuing the conversation here."

    new_target = origin
    opened = None
    if self.request.new_thread:
        opened = await open_crossing(
            ctx, self.request.destination, origin_address, hint, self.agent_id
        )
        if opened is not None:
            channel = ctx.deps.channel(opened.channel_tentacle_id)
            new_target = ResponseTarget(
                channel_id=channel.id,
                address=opened,
                thread_strategy=channel.thread_strategy,
                mode="sub",
            )

    new_address = new_target.address
    if self.request.new_thread and new_address == origin_address:
        raise ValueError(
            "The destination could not create a thread; nothing was teleported."
        )
    source_conversation = None
    if new_address is None or new_address == origin_address:
        # The current conversation already holds the trailing teleport
        # deferral. An explicit workspace move may still fork its runtime.
        state.target = origin
        state.handoff = None
        landed = state.thread
        conversation = await ctx.deps.conversation_manager.ensure(
            landed.id, agent_tentacle_id=self.agent_id
        )
    else:
        # Move: fork the origin conversation into the new sub-thread, claim it
        # for the same agent so follow-ups continue there, and resume against
        # the fork. The runtime prepares its destination handle after the
        # destination workspace is ready.
        if state.thread.kind == "native_thread":
            source_agent_id = state.thread.active_agent_tentacle_id
            if (
                state.user_profile is None
                or state.decision is None
                or source_agent_id is None
            ):
                raise ValueError(
                    "Native teleport requires its requesting user and agent."
                )
            source = await ctx.deps.conversation_manager.ensure(
                state.thread.id,
                agent_tentacle_id=source_agent_id,
            )
            agent = ctx.deps.agent(self.agent_id)
            landed = await agent.fork(
                source,
                ThreadKey.from_address(new_address),
                sender=state.user_profile,
            )
            copied = await ctx.deps.conversation_manager.ensure(
                landed.id, agent_tentacle_id=agent.id
            )
            state.decision = state.decision.model_copy(
                update={"model": copied.runs[-1].model_name}
            )
            state.thread = landed
            state.target = new_target
            state.handoff = None
            await announce_crossing(
                ctx,
                self.request.destination,
                new_address,
                origin_address,
                hint,
                self.agent_id,
            )
            if self.request.prompt is None and not self.request.resume:
                return await self.land(ctx)
            return await self.carry_on(ctx, sentence)
        landed = await ctx.deps.thread_manager.enter(
            new_address, current=state.thread
        )
        source_conversation = await ctx.deps.conversation_manager.ensure(
            state.thread.id, agent_tentacle_id=self.agent_id
        )
        target_conversation = await ctx.deps.conversation_manager.ensure(
            landed.id, agent_tentacle_id=self.agent_id
        )
        conversation = target_conversation
        state.thread = landed
        state.target = new_target
        state.handoff = PendingHandoff(
            source_agent_tentacle_id=self.agent_id,
            source_conversation_id=source_conversation.id,
        )

    project = self.request.project
    carried = (
        await ctx.deps.workspaces.projects.of(source)
        if project is None and landed.id != source.id
        else None
    )
    if project is not None:
        state.thread = await self.bind(ctx, landed, project)
        sentence = (
            f"Continuing the conversation here, in the workspace of {project!r}."
        )
    elif carried is not None:
        # The conversation was about a project, and its history is of work in
        # that tree: the thread it lands in is about the same one, as it stands.
        state.thread = await ctx.deps.thread_manager.bind(landed.id, carried)
        await ctx.deps.workspaces.carry(source, state.thread)
        sentence = (
            "Continuing the conversation here, in a workspace of "
            f"{carried.name!r} that holds your work as you left it."
        )
    # The agent resumes in another directory either way — a new thread's own
    # workspace, or the project's — and a runtime session may be filed under
    # the one it ran in. The tentacle knows how to move its own; this is when.
    cwd = ctx.deps.workspaces.open(
        state.thread.id, await ctx.deps.workspaces.projects.of(state.thread)
    ).path
    agent = ctx.deps.agent(self.agent_id)
    if source_conversation is not None:
        external_id = await agent.fork_session(source_conversation, cwd=cwd)
        carry = (
            external_id is not None
            and external_id == source_conversation.external_id
        )
        await ctx.deps.conversation_manager.fork(
            source_conversation,
            conversation,
            carry_external_id=carry,
            external_id=None if carry else external_id,
        )
        await agent.relocate(conversation, cwd=cwd)
        state.thread = await ctx.deps.thread_manager.rename(
            state.thread,
            ctx.deps.thread_manager.fork_title(source_conversation, source),
        )
    elif not self.request.new_thread:
        external_id = await agent.fork_session(conversation, cwd=cwd)
        if external_id is not None and external_id != conversation.external_id:
            await ctx.deps.conversation_manager.set_external_id(
                conversation, external_id
            )
        await agent.relocate(conversation, cwd=cwd)
    if opened is not None:
        await announce_crossing(
            ctx,
            self.request.destination,
            opened,
            origin_address,
            hint,
            self.agent_id,
        )
    if self.request.prompt is None and not self.request.resume:
        return await self.land(ctx)
    return await self.carry_on(ctx, sentence)

carry_on async 

carry_on(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    sentence: str,
) -> React

Run the agent where the move landed: on the user's prompt, recorded there as their message, or on its own call answered by where it now is.

Source code in octomate/reflex/nodes/teleport.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
async def carry_on(
    self, ctx: GraphRunContext[ReflexState, ReflexDeps], sentence: str
) -> React:
    """Run the agent where the move landed: on the user's prompt, recorded there
    as their message, or on its own call answered by where it now is."""
    state = ctx.state
    address = state.target.address if state.target is not None else None
    if address is None:
        raise ValueError("Teleport carries on only where the move landed")
    prompt = self.request.prompt
    if prompt is None:
        # The pending call resolves into the resumed run, whichever runtime cast it.
        if self.request.tool_call_id is None:
            state.user_prompt = f"{sentence}\nCurrent channel address: {address}"
            return React()
        return React(
            resume_results=DeferredToolResults(
                calls={self.request.tool_call_id: sentence}
            )
        )
    if state.user_profile is None:
        raise ValueError("A prompt after a teleport needs its user")
    channel = ctx.deps.channel(address.channel_tentacle_id)
    message = await ctx.deps.thread_manager.record_inbound(
        MessageEvent(
            tentacle_id=channel.id,
            chat_type=address.chat_type,
            chat_id=address.chat_id,
            user_id=address.user_id,
            channel_thread_id=address.channel_thread_id,
            shared=address.shared,
            self_id=channel.self_profile.channel_user_id,
            sender=state.user_profile,
            segments=[TextSegment(data={"text": prompt})],
            raw=prompt,
        )
    )
    state.source_target = state.target
    state.trigger_thread_message_id = message.id
    await ctx.deps.load_pending_prompt(state, self.agent_id)
    return React()

land async 

land(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> End[ReflexGraphResult]

End the move where it landed, for the next message there to carry on: the handoff is recorded and the workspace saved, as a turn's end would.

Source code in octomate/reflex/nodes/teleport.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
async def land(
    self, ctx: GraphRunContext[ReflexState, ReflexDeps]
) -> End[ReflexGraphResult]:
    """End the move where it landed, for the next message there to carry on:
    the handoff is recorded and the workspace saved, as a turn's end would."""
    state = ctx.state
    if state.decision is None or state.target is None or state.thread is None:
        raise ValueError("Teleport lands with a decision, a target and a thread")
    resolved = ctx.deps.resolve_agent(
        state.target.channel_id, state.decision.agent_id, state.decision.model
    )
    state.decision = state.decision.model_copy(
        update={"agent_id": resolved.agent, "model": resolved.model}
    )
    if state.handoff is not None:
        await state.handoff.land(ctx.deps, state.thread, state.decision)
        state.handoff = None
    await ctx.deps.workspaces.save(state.thread)
    return End(ReflexResult(decision=state.decision, target=state.target))

bind async 

bind(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    thread: Thread,
    name: str,
) -> Thread

Bind the thread landed in to the project the request names, and fork its workspace, so the resumed run starts in the project's code. The gate validated the project and the ref; a project missing here is a wiring bug. Answers the bound thread, re-read: the state's copy predates the binding, and the turn-end save reads the project off it.

Source code in octomate/reflex/nodes/teleport.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
async def bind(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    thread: Thread,
    name: str,
) -> Thread:
    """Bind the thread landed in to the project the request names, and fork
    its workspace, so the resumed run starts in the project's code. The gate
    validated the project and the ref; a project missing here is a wiring
    bug. Answers the bound thread, re-read: the state's copy predates the
    binding, and the turn-end save reads the project off it."""
    project = ctx.deps.workspaces.projects.get(name)
    if project is None:
        raise RuntimeError(
            f"project {name!r} vanished between the gate and the move"
        )
    mirror = await ctx.deps.workspaces.mirrors.sync(project)
    bound = await ctx.deps.thread_manager.bind(thread.id, project)
    await ctx.deps.workspaces.materialize(
        ctx.deps.workspaces.open(thread.id, project), mirror, self.request.ref
    )
    return bound

DeferredResult dataclass

A turn that stopped on deferred requests, and the batch they were parked as.

Source code in octomate/reflex/state.py 
164
165
166
167
168
169
170
171
172
173
174
175
@dataclass
class DeferredResult:
    """A turn that stopped on deferred requests, and the batch they were parked
    as."""

    requests: DeferredToolRequests
    target: ResponseTarget
    # The name of the run that deferred — carried for observability. `str`, not
    # `RunName`, because on a re-present it is read back from the persisted batch.
    run_name: str
    result: AgentRunResult[Any]
    batch_id: UUID7 | None = None

requests instance-attribute 

requests: DeferredToolRequests

target instance-attribute 

target: ResponseTarget

run_name instance-attribute 

run_name: str

result instance-attribute 

result: AgentRunResult[Any]

batch_id class-attribute instance-attribute 

batch_id: UUID7 | None = None

ReflexDeps dataclass

The host's registries and managers a reflex run reaches through.

Source code in octomate/reflex/state.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
@dataclass
class ReflexDeps:
    """The host's registries and managers a reflex run reaches through."""

    channels: dict[str, ChannelTentacle]
    # No defaults for the managers: a deps object must carry the host's own — the
    # ledger, the conversations, the deferred actions, the gateway's live-session
    # registry — never private ones with their own identity or state.
    thread_manager: ThreadManager
    # For the bookkeeping a finished turn owes and the agent has no part in:
    # leaving the thread's workspace somewhere losing the directory cannot cost it.
    workspaces: WorkspaceManager
    conversation_manager: ConversationManager
    action_manager: DeferredActionManager
    gateway: GatewayManager
    # The graph's end of the nerve: what the kick does, sent as impulses.
    impulses: ObjectSendStream[Impulse]
    agents: dict[str, AgentTentacle] = field(default_factory=dict)

    @cached_property
    def runtime(self) -> ReflexRuntime:
        """Stateless orchestration shared by this graph's execution entries."""
        return ReflexRuntime()

    @overload
    def channel(self, target: ResponseTarget) -> ChannelTentacle: ...

    @overload
    def channel(self, target: str) -> ChannelTentacle: ...

    def channel(self, target: ResponseTarget | str) -> ChannelTentacle:
        channel_id = target.channel_id if isinstance(target, ResponseTarget) else target
        channel = self.channels.get(channel_id)
        if channel is None:
            raise ValueError(f"unknown channel {channel_id!r}")
        return channel

    def agent(self, agent_id: str) -> AgentTentacle:
        agent = self.agents.get(agent_id)
        if agent is None:
            raise ValueError(f"unknown agent {agent_id!r}")
        return agent

    @cached_property
    def gateway_agents(self) -> frozenset[str]:
        """The agents whose driven turns offer the gateway spells: each agent's own
        flag, settled at startup and read once. A turn of any other agent builds no
        session, so external callers find nothing live for it."""
        return frozenset(id for id, agent in self.agents.items() if agent.gateway)

    async def octomate_session(
        self,
        agent: AgentTentacle,
        *,
        user_profile: UserProfile | None,
        thread_id: UUID7 | None,
        conversation_address: ChannelAddress,
        conversation_id: uuid.UUID | None = None,
    ) -> OctomateSession | None:
        """One turn's gateway for `agent`, or None for an agent whose flag is off.

        Built from the host's own registries, every channel's and not just this
        one's: a spell that crosses lands where another channel's config decides who
        runs, so the gateway offers — and checks against — that channel's routes,
        and reads `surfaces` off it to know whether `scheme` can land. The agents
        are what the accomplice spells run; without a thread there is nowhere for
        a child conversation to live, and the gateway then does not offer them.
        """
        if agent.id not in self.gateway_agents:
            return None
        session = OctomateSession(
            channel_routes=self.available_routes,
            current_agent_id=agent.id,
            channels=self.channels,
            users=self.thread_manager.users,
            user_profile=user_profile,
            agents=self.agents,
            thread_id=thread_id,
            conversation_address=conversation_address,
            threads=self.thread_manager,
            workspaces=self.workspaces,
            impulses=self.impulses,
        )
        if conversation_id is not None:
            session.conversation_id = conversation_id
        elif thread_id is not None:
            # The same (thread, agent) key the run resolves internally, so an
            # external runtime's tool call finds this turn's session by the
            # conversation it already knows.
            conversation = await self.conversation_manager.ensure(
                thread_id, agent_tentacle_id=agent.id, with_history=False
            )
            session.conversation_id = conversation.id
        return session

    @property
    def available_routes(self) -> dict[str, list[AgentRoute]]:
        return self.gateway.available_routes(self.channels, self.agents)

    def resolve_agent(
        self,
        channel_id: str,
        agent_id: str | None,
        model: AgentRouteModelName | None,
    ) -> AgentModelConfig:
        """Select an exposed agent and resolve a model from its catalog."""
        agent_ids = self.channel(channel_id).agent_ids
        if agent_id is None:
            agent_id = agent_ids[0]
        if agent_id not in agent_ids:
            raise ValueError(
                f"agent {agent_id!r} is not bound to channel {channel_id!r}"
            )

        agent = self.agent(agent_id)
        if not agent.models:
            raise ValueError(f"agent {agent_id!r} has no available model catalog")
        if model is None:
            return AgentModelConfig(agent=agent_id, model=agent.resolve_model())
        served = agent.served_model(model)
        if served is None:
            raise ValueError(f"agent {agent_id!r} does not serve model {model!r}")
        return AgentModelConfig(agent=agent_id, model=served)

    async def render_chat(
        self, messages: list[ThreadMessage], *, ceiling: int = 0
    ) -> str:
        """The ledger rows as one transcript: who spoke, under which identities, and
        the `#msg:<id>` handle a brief would cite them by.

        Each message is its segments, as they render themselves — the same body an
        inbound event shows. `message_text` would be the shorter route and the wrong
        one: it keeps only text and markdown, so a message that was a picture or a
        quoted reply arrives as nothing at all.

        `ceiling` cuts each message. It is 0 for the messages a turn must answer —
        those are the work — and set for the chat behind them, where one pasted log
        would otherwise be the whole slice.
        """
        parts: list[str] = []
        for message in messages:
            if message.kind != "message":
                continue
            text = "\n".join(str(segment) for segment in message.segments)
            if not text:
                continue
            if ceiling and len(text) > ceiling:
                text = f"{text[:ceiling]}…"
            sender = await message.sender
            display_name = (
                (sender.name or sender.nickname or "anonymous")
                if sender is not None
                else "anonymous"
            )
            owner = sender.user.peek() if sender is not None else None
            ids = (
                f"{message.user_id}, user:{owner.username}"
                if owner is not None
                else message.user_id
            )
            platform_id = (
                f" #msg:{message.platform_message_id}"
                if message.platform_message_id
                else ""
            )
            parts.append(f"{display_name} ({ids}){platform_id}:\n{text}")
        return "\n\n".join(parts)

    async def report(self, state: ReflexState, error: Exception) -> None:
        """Publish that the turn failed, on the thread it came from. A turn that
        failed before it knew its source has nobody to tell."""
        source = state.source_target.address if state.source_target else None
        if source is None:
            return
        trace_id = format(trace.get_current_span().get_span_context().trace_id, "032x")
        failure = RunErrorEvent(message=str(error), trace_id=trace_id)
        await self.impulses.send(Impulse(failure, source))

    async def carry(
        self,
        address: ChannelAddress,
        stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
    ) -> None:
        """Send a run's stream on the nerve at `address` as wire events, item by
        item, until it ends."""
        async for item in stream:
            wire = to_wire(item)
            if wire is not None:
                await self.impulses.send(Impulse(wire, address))

    async def record_move(
        self,
        address: ChannelAddress,
        text: str,
        *,
        agent_tentacle_id: str,
        platform_message_id: str | None = None,
    ) -> ThreadMessage:
        """Write the line a move leaves behind to the ledger of the chat it left.

        A person already sees it — the channel posted it, and a thread opened here
        hangs off it. The turn that comes next does not: a chat room's recap is built
        from the ledger, and a move nobody recorded leaves a chat in which the work
        simply stops. An agent reading that answers what was carried away ten minutes
        ago.

        The row carries the opener's own platform id, so the ledger and the platform
        name the same message rather than two; a line still to be posted gets its id
        from the channel that posts it.
        """
        return await self.thread_manager.record_outbound(
            address,
            agent_tentacle_id=agent_tentacle_id,
            segments=[MarkdownSegment(data={"text": text})],
            sender=self.channel(address.channel_tentacle_id).self_profile,
            platform_message_id=platform_message_id,
            message_text=text,
            raw=text,
        )

    async def load_pending_prompt(
        self,
        state: ReflexState,
        active_agent_id: str,
    ) -> None:
        source_target = state.source_target
        if (
            state.thread is None
            or state.trigger_thread_message_id is None
            or source_target is None
            or source_target.address is None
        ):
            return
        # Pull every recorded chat-ledger row that has not been bound into a
        # model request yet: rule-gated group messages, sleeping/not-kicked
        # messages, and messages that stacked up behind an already-running turn.
        messages = await self.thread_manager.pending_prompt_messages(
            state.thread,
            state.trigger_thread_message_id,
            active_agent_id,
        )
        if not messages:
            return
        state.source_thread_address = source_target.address
        state.source_thread_message_ids = [message.id for message in messages]

        asked = await self.render_chat(messages)
        if not asked:
            return
        recap = await self.chat_recap(state, messages[0])
        # The messages the turn must answer are the body, unmarked. What was said
        # before them is the system's addition and stands in a tag.
        state.user_prompt = f"{recap}\n\n{asked}" if recap else asked

    async def chat_recap(self, state: ReflexState, earliest: ThreadMessage) -> str:
        """What was said in this chat room before the messages the turn must answer.

        Only a kick in a dm or a group chat needs it: that runs in a thread of its
        own, so its model context starts empty every time and the agent would answer
        a chat it cannot see. A thread carries its own history in the conversation
        and gets none of this — its context is already the whole of the work.

        The chat is the parent's, so the rows come from there; how many, and how much
        of each, is the channel's to say.
        """
        thread = state.thread
        source_target = state.source_target
        if thread is None or thread.parent_thread_id is None or source_target is None:
            return ""
        recap = self.channel(source_target).config.recap
        if not recap.messages:
            return ""
        messages = await self.thread_manager.chat_messages_before(
            thread.parent_thread_id, earliest.id, limit=recap.messages
        )
        rendered = await self.render_chat(messages, ceiling=recap.characters)
        return tagged("chat_recap", f"{RECAP_HEADER}\n\n{rendered}" if rendered else "")

channels instance-attribute 

channels: dict[str, ChannelTentacle]

thread_manager instance-attribute 

thread_manager: ThreadManager

workspaces instance-attribute 

workspaces: WorkspaceManager

conversation_manager instance-attribute 

conversation_manager: ConversationManager

action_manager instance-attribute 

action_manager: DeferredActionManager

gateway instance-attribute 

gateway: GatewayManager

impulses instance-attribute 

impulses: ObjectSendStream[Impulse]

agents class-attribute instance-attribute 

agents: dict[str, AgentTentacle] = field(
    default_factory=dict
)

runtime cached property 

runtime: ReflexRuntime

Stateless orchestration shared by this graph's execution entries.

gateway_agents cached property 

gateway_agents: frozenset[str]

The agents whose driven turns offer the gateway spells: each agent's own flag, settled at startup and read once. A turn of any other agent builds no session, so external callers find nothing live for it.

available_routes property 

available_routes: dict[str, list[AgentRoute]]

channel 

channel(target: ResponseTarget) -> ChannelTentacle

channel(target: str) -> ChannelTentacle

channel(target: ResponseTarget | str) -> ChannelTentacle
Source code in octomate/reflex/state.py 
243
244
245
246
247
248
def channel(self, target: ResponseTarget | str) -> ChannelTentacle:
    channel_id = target.channel_id if isinstance(target, ResponseTarget) else target
    channel = self.channels.get(channel_id)
    if channel is None:
        raise ValueError(f"unknown channel {channel_id!r}")
    return channel

agent 

agent(agent_id: str) -> AgentTentacle
Source code in octomate/reflex/state.py 
250
251
252
253
254
def agent(self, agent_id: str) -> AgentTentacle:
    agent = self.agents.get(agent_id)
    if agent is None:
        raise ValueError(f"unknown agent {agent_id!r}")
    return agent

octomate_session async 

octomate_session(
    agent: AgentTentacle,
    *,
    user_profile: UserProfile | None,
    thread_id: UUID7 | None,
    conversation_address: ChannelAddress,
    conversation_id: UUID | None = None,
) -> OctomateSession | None

One turn's gateway for agent, or None for an agent whose flag is off.

Built from the host's own registries, every channel's and not just this one's: a spell that crosses lands where another channel's config decides who runs, so the gateway offers — and checks against — that channel's routes, and reads surfaces off it to know whether scheme can land. The agents are what the accomplice spells run; without a thread there is nowhere for a child conversation to live, and the gateway then does not offer them.

Source code in octomate/reflex/state.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
async def octomate_session(
    self,
    agent: AgentTentacle,
    *,
    user_profile: UserProfile | None,
    thread_id: UUID7 | None,
    conversation_address: ChannelAddress,
    conversation_id: uuid.UUID | None = None,
) -> OctomateSession | None:
    """One turn's gateway for `agent`, or None for an agent whose flag is off.

    Built from the host's own registries, every channel's and not just this
    one's: a spell that crosses lands where another channel's config decides who
    runs, so the gateway offers — and checks against — that channel's routes,
    and reads `surfaces` off it to know whether `scheme` can land. The agents
    are what the accomplice spells run; without a thread there is nowhere for
    a child conversation to live, and the gateway then does not offer them.
    """
    if agent.id not in self.gateway_agents:
        return None
    session = OctomateSession(
        channel_routes=self.available_routes,
        current_agent_id=agent.id,
        channels=self.channels,
        users=self.thread_manager.users,
        user_profile=user_profile,
        agents=self.agents,
        thread_id=thread_id,
        conversation_address=conversation_address,
        threads=self.thread_manager,
        workspaces=self.workspaces,
        impulses=self.impulses,
    )
    if conversation_id is not None:
        session.conversation_id = conversation_id
    elif thread_id is not None:
        # The same (thread, agent) key the run resolves internally, so an
        # external runtime's tool call finds this turn's session by the
        # conversation it already knows.
        conversation = await self.conversation_manager.ensure(
            thread_id, agent_tentacle_id=agent.id, with_history=False
        )
        session.conversation_id = conversation.id
    return session

resolve_agent 

resolve_agent(
    channel_id: str,
    agent_id: str | None,
    model: AgentRouteModelName | None,
) -> AgentModelConfig

Select an exposed agent and resolve a model from its catalog.

Source code in octomate/reflex/state.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def resolve_agent(
    self,
    channel_id: str,
    agent_id: str | None,
    model: AgentRouteModelName | None,
) -> AgentModelConfig:
    """Select an exposed agent and resolve a model from its catalog."""
    agent_ids = self.channel(channel_id).agent_ids
    if agent_id is None:
        agent_id = agent_ids[0]
    if agent_id not in agent_ids:
        raise ValueError(
            f"agent {agent_id!r} is not bound to channel {channel_id!r}"
        )

    agent = self.agent(agent_id)
    if not agent.models:
        raise ValueError(f"agent {agent_id!r} has no available model catalog")
    if model is None:
        return AgentModelConfig(agent=agent_id, model=agent.resolve_model())
    served = agent.served_model(model)
    if served is None:
        raise ValueError(f"agent {agent_id!r} does not serve model {model!r}")
    return AgentModelConfig(agent=agent_id, model=served)

render_chat async 

render_chat(
    messages: list[ThreadMessage], *, ceiling: int = 0
) -> str

The ledger rows as one transcript: who spoke, under which identities, and the #msg:<id> handle a brief would cite them by.

Each message is its segments, as they render themselves — the same body an inbound event shows. message_text would be the shorter route and the wrong one: it keeps only text and markdown, so a message that was a picture or a quoted reply arrives as nothing at all.

ceiling cuts each message. It is 0 for the messages a turn must answer — those are the work — and set for the chat behind them, where one pasted log would otherwise be the whole slice.

Source code in octomate/reflex/state.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
async def render_chat(
    self, messages: list[ThreadMessage], *, ceiling: int = 0
) -> str:
    """The ledger rows as one transcript: who spoke, under which identities, and
    the `#msg:<id>` handle a brief would cite them by.

    Each message is its segments, as they render themselves — the same body an
    inbound event shows. `message_text` would be the shorter route and the wrong
    one: it keeps only text and markdown, so a message that was a picture or a
    quoted reply arrives as nothing at all.

    `ceiling` cuts each message. It is 0 for the messages a turn must answer —
    those are the work — and set for the chat behind them, where one pasted log
    would otherwise be the whole slice.
    """
    parts: list[str] = []
    for message in messages:
        if message.kind != "message":
            continue
        text = "\n".join(str(segment) for segment in message.segments)
        if not text:
            continue
        if ceiling and len(text) > ceiling:
            text = f"{text[:ceiling]}…"
        sender = await message.sender
        display_name = (
            (sender.name or sender.nickname or "anonymous")
            if sender is not None
            else "anonymous"
        )
        owner = sender.user.peek() if sender is not None else None
        ids = (
            f"{message.user_id}, user:{owner.username}"
            if owner is not None
            else message.user_id
        )
        platform_id = (
            f" #msg:{message.platform_message_id}"
            if message.platform_message_id
            else ""
        )
        parts.append(f"{display_name} ({ids}){platform_id}:\n{text}")
    return "\n\n".join(parts)

report async 

report(state: ReflexState, error: Exception) -> None

Publish that the turn failed, on the thread it came from. A turn that failed before it knew its source has nobody to tell.

Source code in octomate/reflex/state.py 
381
382
383
384
385
386
387
388
389
async def report(self, state: ReflexState, error: Exception) -> None:
    """Publish that the turn failed, on the thread it came from. A turn that
    failed before it knew its source has nobody to tell."""
    source = state.source_target.address if state.source_target else None
    if source is None:
        return
    trace_id = format(trace.get_current_span().get_span_context().trace_id, "032x")
    failure = RunErrorEvent(message=str(error), trace_id=trace_id)
    await self.impulses.send(Impulse(failure, source))

carry async 

carry(
    address: ChannelAddress,
    stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
) -> None

Send a run's stream on the nerve at address as wire events, item by item, until it ends.

Source code in octomate/reflex/state.py 
391
392
393
394
395
396
397
398
399
400
401
async def carry(
    self,
    address: ChannelAddress,
    stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
) -> None:
    """Send a run's stream on the nerve at `address` as wire events, item by
    item, until it ends."""
    async for item in stream:
        wire = to_wire(item)
        if wire is not None:
            await self.impulses.send(Impulse(wire, address))

record_move async 

record_move(
    address: ChannelAddress,
    text: str,
    *,
    agent_tentacle_id: str,
    platform_message_id: str | None = None,
) -> ThreadMessage

Write the line a move leaves behind to the ledger of the chat it left.

A person already sees it — the channel posted it, and a thread opened here hangs off it. The turn that comes next does not: a chat room's recap is built from the ledger, and a move nobody recorded leaves a chat in which the work simply stops. An agent reading that answers what was carried away ten minutes ago.

The row carries the opener's own platform id, so the ledger and the platform name the same message rather than two; a line still to be posted gets its id from the channel that posts it.

Source code in octomate/reflex/state.py 
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
async def record_move(
    self,
    address: ChannelAddress,
    text: str,
    *,
    agent_tentacle_id: str,
    platform_message_id: str | None = None,
) -> ThreadMessage:
    """Write the line a move leaves behind to the ledger of the chat it left.

    A person already sees it — the channel posted it, and a thread opened here
    hangs off it. The turn that comes next does not: a chat room's recap is built
    from the ledger, and a move nobody recorded leaves a chat in which the work
    simply stops. An agent reading that answers what was carried away ten minutes
    ago.

    The row carries the opener's own platform id, so the ledger and the platform
    name the same message rather than two; a line still to be posted gets its id
    from the channel that posts it.
    """
    return await self.thread_manager.record_outbound(
        address,
        agent_tentacle_id=agent_tentacle_id,
        segments=[MarkdownSegment(data={"text": text})],
        sender=self.channel(address.channel_tentacle_id).self_profile,
        platform_message_id=platform_message_id,
        message_text=text,
        raw=text,
    )

load_pending_prompt async 

load_pending_prompt(
    state: ReflexState, active_agent_id: str
) -> None
Source code in octomate/reflex/state.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
async def load_pending_prompt(
    self,
    state: ReflexState,
    active_agent_id: str,
) -> None:
    source_target = state.source_target
    if (
        state.thread is None
        or state.trigger_thread_message_id is None
        or source_target is None
        or source_target.address is None
    ):
        return
    # Pull every recorded chat-ledger row that has not been bound into a
    # model request yet: rule-gated group messages, sleeping/not-kicked
    # messages, and messages that stacked up behind an already-running turn.
    messages = await self.thread_manager.pending_prompt_messages(
        state.thread,
        state.trigger_thread_message_id,
        active_agent_id,
    )
    if not messages:
        return
    state.source_thread_address = source_target.address
    state.source_thread_message_ids = [message.id for message in messages]

    asked = await self.render_chat(messages)
    if not asked:
        return
    recap = await self.chat_recap(state, messages[0])
    # The messages the turn must answer are the body, unmarked. What was said
    # before them is the system's addition and stands in a tag.
    state.user_prompt = f"{recap}\n\n{asked}" if recap else asked

chat_recap async 

chat_recap(
    state: ReflexState, earliest: ThreadMessage
) -> str

What was said in this chat room before the messages the turn must answer.

Only a kick in a dm or a group chat needs it: that runs in a thread of its own, so its model context starts empty every time and the agent would answer a chat it cannot see. A thread carries its own history in the conversation and gets none of this — its context is already the whole of the work.

The chat is the parent's, so the rows come from there; how many, and how much of each, is the channel's to say.

Source code in octomate/reflex/state.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
async def chat_recap(self, state: ReflexState, earliest: ThreadMessage) -> str:
    """What was said in this chat room before the messages the turn must answer.

    Only a kick in a dm or a group chat needs it: that runs in a thread of its
    own, so its model context starts empty every time and the agent would answer
    a chat it cannot see. A thread carries its own history in the conversation
    and gets none of this — its context is already the whole of the work.

    The chat is the parent's, so the rows come from there; how many, and how much
    of each, is the channel's to say.
    """
    thread = state.thread
    source_target = state.source_target
    if thread is None or thread.parent_thread_id is None or source_target is None:
        return ""
    recap = self.channel(source_target).config.recap
    if not recap.messages:
        return ""
    messages = await self.thread_manager.chat_messages_before(
        thread.parent_thread_id, earliest.id, limit=recap.messages
    )
    rendered = await self.render_chat(messages, ceiling=recap.characters)
    return tagged("chat_recap", f"{RECAP_HEADER}\n\n{rendered}" if rendered else "")

ReflexResult dataclass

A finished turn: the decision it ran on, where it answered, and the agent's result.

Source code in octomate/reflex/state.py 
154
155
156
157
158
159
160
161
@dataclass
class ReflexResult:
    """A finished turn: the decision it ran on, where it answered, and the agent's
    result."""

    decision: SummonDecision | None
    target: ResponseTarget
    result: AgentRunResult[ChannelOutput] | None = None

decision instance-attribute 

decision: SummonDecision | None

target instance-attribute 

target: ResponseTarget

result class-attribute instance-attribute 

result: AgentRunResult[ChannelOutput] | None = None

ReflexState dataclass

All run-wide context for one reflex graph run.

Awake resolves the source context once and writes it here; downstream nodes read from state and carry only transition discriminators.

Source code in octomate/reflex/state.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@dataclass
class ReflexState:
    """All run-wide context for one reflex graph run.

    Awake resolves the source context once and writes it here; downstream nodes
    read from state and carry only transition discriminators.
    """

    source_target: ResponseTarget | None = None
    target: ResponseTarget | None = None
    run_name: RunName = "react"
    decision: SummonDecision | None = None
    selection: AgentRouteKey | None = None  # Selected agent and model.
    conversation_id: uuid.UUID | None = None  # Current invocation's conversation.
    targets: dict[str, ResponseTarget] = field(default_factory=dict)
    summon_routes: list[AgentRoute] = field(default_factory=list)
    thread: Thread | None = None
    trigger_thread_message_id: UUID7 | None = None
    source_thread_address: ChannelAddress | None = None
    source_thread_message_ids: list[UUID7] = field(default_factory=list)
    # The handoff the next React records when it lands, or None when it records
    # none.
    handoff: PendingHandoff | None = None
    user_prompt: str | Sequence[UserContent] | None = None
    user_profile: UserProfile | None = None

source_target class-attribute instance-attribute 

source_target: ResponseTarget | None = None

target class-attribute instance-attribute 

target: ResponseTarget | None = None

run_name class-attribute instance-attribute 

run_name: RunName = 'react'

decision class-attribute instance-attribute 

decision: SummonDecision | None = None

selection class-attribute instance-attribute 

selection: AgentRouteKey | None = None

conversation_id class-attribute instance-attribute 

conversation_id: UUID | None = None

targets class-attribute instance-attribute 

targets: dict[str, ResponseTarget] = field(
    default_factory=dict
)

summon_routes class-attribute instance-attribute 

summon_routes: list[AgentRoute] = field(
    default_factory=list
)

thread class-attribute instance-attribute 

thread: Thread | None = None

trigger_thread_message_id class-attribute instance-attribute 

trigger_thread_message_id: UUID7 | None = None

source_thread_address class-attribute instance-attribute 

source_thread_address: ChannelAddress | None = None

source_thread_message_ids class-attribute instance-attribute 

source_thread_message_ids: list[UUID7] = field(
    default_factory=list
)

handoff class-attribute instance-attribute 

handoff: PendingHandoff | None = None

user_prompt class-attribute instance-attribute 

user_prompt: str | Sequence[UserContent] | None = None

user_profile class-attribute instance-attribute 

user_profile: UserProfile | None = None

ResponseTarget dataclass

Where a turn answers: the channel, the address on it, and how the reply is placed.

Source code in octomate/reflex/state.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class ResponseTarget:
    """Where a turn answers: the channel, the address on it, and how the reply is
    placed."""

    channel_id: str
    address: ChannelAddress | None = None
    # Routing only — how an inbound threaded message is handled (`Route`). What this
    # channel can actually open lives on the channel itself, as `surfaces`.
    thread_strategy: ThreadStrategy = "main_only"
    mode: ResponseTargetMode = "main"

    def __str__(self) -> str:
        chat_type = self.address.chat_type if self.address else "unresolved"
        return (
            f"- {self.channel_id}: chat_type={chat_type}, mode={self.mode}, "
            f"thread_strategy={self.thread_strategy}"
        )

channel_id instance-attribute 

channel_id: str

address class-attribute instance-attribute 

address: ChannelAddress | None = None

thread_strategy class-attribute instance-attribute 

thread_strategy: ThreadStrategy = 'main_only'

mode class-attribute instance-attribute 

mode: ResponseTargetMode = 'main'

SummonDecision pydantic-model

Bases: BaseModel

A handoff decision: continue this turn with another agent, from a brief, in the conversation where it already is.

Fields:

Source code in octomate/schemas/triage.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class SummonDecision(BaseModel):
    """A handoff decision: continue this turn with another agent, from a brief, in the
    conversation where it already is."""

    action: Literal["summon"] = "summon"
    reason: str
    agent_id: str
    model: AgentRouteModelName | None = Field(
        description="Selected model, or null to use the harness's native default."
    )
    effort: str | None = None
    hint: str
    brief: str = Field(
        max_length=8_000,
        description="The receiver's whole opening prompt, refused over the cap rather "
        "than trimmed: what it leaves out, the receiver reads back through the "
        "handoff. Eight thousand characters is roughly two thousand tokens — the "
        "size a distilled handoff runs to, and well short of a pasted transcript.",
    )

    @property
    def key(self) -> AgentRouteKey:
        return AgentRouteKey(agent_id=self.agent_id, model=self.model)

action pydantic-field 

action: Literal['summon'] = 'summon'

reason pydantic-field 

reason: str

agent_id pydantic-field 

agent_id: str

model pydantic-field 

model: AgentRouteModelName | None

Selected model, or null to use the harness's native default.

effort pydantic-field 

effort: str | None = None

hint pydantic-field 

hint: str

brief pydantic-field 

brief: str

The receiver's whole opening prompt, refused over the cap rather than trimmed: what it leaves out, the receiver reads back through the handoff. Eight thousand characters is roughly two thousand tokens — the size a distilled handoff runs to, and well short of a pasted transcript.

key property 

key: AgentRouteKey

build_reflex_graph 

build_reflex_graph(
    entry: type[ReflexEntryT] = Awake,
) -> Graph[
    ReflexState, ReflexDeps, ReflexEntryT, ReflexGraphResult
]

Wire the reflex nodes into a runnable graph, entered at entry.

Every edge comes from the nodes' own run return annotations, so the shape stays declared where the transition is written rather than in a second list here. A graph declares the one node it is entered at: a signal wakes the whole reflex, so that is Awake, and only a test wires the same nodes with a different door to exercise a stretch of them on its own.

Source code in octomate/reflex/graph.py 
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
def build_reflex_graph(
    entry: type[ReflexEntryT] = Awake,
) -> Graph[ReflexState, ReflexDeps, ReflexEntryT, ReflexGraphResult]:
    """Wire the reflex nodes into a runnable graph, entered at `entry`.

    Every edge comes from the nodes' own `run` return annotations, so the shape
    stays declared where the transition is written rather than in a second list
    here. A graph declares the one node it is entered at: a signal wakes the
    whole reflex, so that is `Awake`, and only a test wires the same nodes with
    a different door to exercise a stretch of them on its own.
    """
    builder = GraphBuilder(
        name="reflex",
        state_type=ReflexState,
        deps_type=ReflexDeps,
        input_type=entry,
        # `TypeExpression` is pydantic-graph's stand-in for a union in a
        # `type[...]` position — the result may also be a direct command outcome.
        output_type=TypeExpression[ReflexGraphResult],
    )
    builder.add(
        builder.edge_from(builder.start_node).to(entry),
        builder.node(Awake),
        builder.node(Command),
        builder.node(Route),
        builder.node(Summon),
        builder.node(React),
        builder.node(Scheme),
        builder.node(Teleport),
        builder.node(ResumeDeferred),
    )
    # Entering anywhere but the top necessarily strands the nodes above it, so the
    # reachability check only means something for the real entry.
    return builder.build(validate_graph_structure=entry is Awake)

run_reflex async 

run_reflex(
    signal: AwakeSignal,
    *,
    state: ReflexState,
    deps: ReflexDeps,
) -> ReflexGraphResult

Wake the reflex with signal, and say so on the nerve before anything else. A failure is reported where the kick came from, and the reflex falls asleep last whatever happened.

Source code in octomate/reflex/graph.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def run_reflex(
    signal: AwakeSignal, *, state: ReflexState, deps: ReflexDeps
) -> ReflexGraphResult:
    """Wake the reflex with `signal`, and say so on the nerve before anything
    else. A failure is reported where the kick came from, and the reflex falls
    asleep last whatever happened."""
    await deps.impulses.send(Impulse(ReflexAwakeEvent()))
    try:
        return await reflex_graph.run(
            inputs=Awake(signal=signal), state=state, deps=deps
        )
    except Exception as error:
        await deps.report(state, error)
        raise
    finally:
        await deps.impulses.send(Impulse(ReflexAsleepEvent()))

octomate.reflex.state

Graph state, dependencies and runtime orchestration.

The state and deps objects a node is handed, the runtime shared by execution entries, and the result variants a run ends in.

logger module-attribute 

logger = logging.getLogger(__name__)

RECAP_HEADER module-attribute 

RECAP_HEADER = "Recent messages in this chat. They are context and have been answered already — do not reply to them."

ReflexEntryT module-attribute 

ReflexEntryT = TypeVar(
    "ReflexEntryT",
    bound="BaseNode[ReflexState, ReflexDeps, ReflexGraphResult]",
)

ReflexGraphResult 

ReflexGraphResult = (
    ReflexResult | DeferredResult | CommandOutcome
)

ResponseTarget dataclass

Where a turn answers: the channel, the address on it, and how the reply is placed.

Source code in octomate/reflex/state.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class ResponseTarget:
    """Where a turn answers: the channel, the address on it, and how the reply is
    placed."""

    channel_id: str
    address: ChannelAddress | None = None
    # Routing only — how an inbound threaded message is handled (`Route`). What this
    # channel can actually open lives on the channel itself, as `surfaces`.
    thread_strategy: ThreadStrategy = "main_only"
    mode: ResponseTargetMode = "main"

    def __str__(self) -> str:
        chat_type = self.address.chat_type if self.address else "unresolved"
        return (
            f"- {self.channel_id}: chat_type={chat_type}, mode={self.mode}, "
            f"thread_strategy={self.thread_strategy}"
        )

channel_id instance-attribute 

channel_id: str

address class-attribute instance-attribute 

address: ChannelAddress | None = None

thread_strategy class-attribute instance-attribute 

thread_strategy: ThreadStrategy = 'main_only'

mode class-attribute instance-attribute 

mode: ResponseTargetMode = 'main'

PendingHandoff dataclass

A handoff one node decided, as far as it is known before it lands: the source side. The target is what landing resolves — the decision's route, as the agent is actually mounted — so it is read there when the row is recorded, and nothing here could name it without risking a second, drifting copy.

The source is who handed the conversation over, and where from as the row names it: the conversation the deciding turn ran in, the last model message in it and the run that left that message, read off the ledger rather than the run result because the row's foreign keys hold only for what was persisted. A native summon names its agent and nothing else, since the gateway cannot name the terminal session it came from; a route that pins a thread's owner names nobody.

Source code in octomate/reflex/state.py 
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
@dataclass(frozen=True)
class PendingHandoff:
    """A handoff one node decided, as far as it is known before it lands: the source
    side. The target is what landing resolves — the decision's route, as the agent
    is actually mounted — so it is read there when the row is recorded, and
    nothing here could name it without risking a second, drifting copy.

    The source is who handed the conversation over, and where from as the row names
    it: the conversation the deciding turn ran in, the last model message in it and
    the run that left that message, read off the ledger rather than the run result
    because the row's foreign keys hold only for what was persisted. A native summon
    names its agent and nothing else, since the gateway cannot name the terminal
    session it came from; a route that pins a thread's owner names nobody."""

    source_agent_tentacle_id: str | None = None
    source_conversation_id: UUID7 | None = None
    source_run_id: str | None = None
    source_model_message_id: UUID7 | None = None

    async def land(
        self, deps: ReflexDeps, thread: Thread, decision: SummonDecision
    ) -> None:
        """Record this handoff on the chat `thread` belongs to, naming the agent
        and model `decision` resolved to, unless that chat already names them."""
        target_conversation = await deps.conversation_manager.ensure(
            thread.id, agent_tentacle_id=decision.agent_id, with_history=False
        )
        # A handoff pins who owns the chat, so it is read and written there: a
        # chat room's sub-thread is new every kick and would forget the owner.
        chat = await deps.thread_manager.surface(thread)
        latest = chat.latest_handoff
        if (
            latest is not None
            and latest.to_agent_tentacle_id == decision.agent_id
            and latest.to_model == decision.model
        ):
            return
        await deps.thread_manager.record_handoff(
            chat,
            source_agent_tentacle_id=self.source_agent_tentacle_id,
            to_agent_tentacle_id=decision.agent_id,
            to_model=decision.model,
            reason=decision.reason,
            hint=decision.hint,
            brief=decision.brief,
            source_conversation_id=self.source_conversation_id,
            target_conversation_id=target_conversation.id,
            source_run_id=self.source_run_id,
            source_model_message_id=self.source_model_message_id,
        )

source_agent_tentacle_id class-attribute instance-attribute 

source_agent_tentacle_id: str | None = None

source_conversation_id class-attribute instance-attribute 

source_conversation_id: UUID7 | None = None

source_run_id class-attribute instance-attribute 

source_run_id: str | None = None

source_model_message_id class-attribute instance-attribute 

source_model_message_id: UUID7 | None = None

land async 

land(
    deps: ReflexDeps,
    thread: Thread,
    decision: SummonDecision,
) -> None

Record this handoff on the chat thread belongs to, naming the agent and model decision resolved to, unless that chat already names them.

Source code in octomate/reflex/state.py 
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
async def land(
    self, deps: ReflexDeps, thread: Thread, decision: SummonDecision
) -> None:
    """Record this handoff on the chat `thread` belongs to, naming the agent
    and model `decision` resolved to, unless that chat already names them."""
    target_conversation = await deps.conversation_manager.ensure(
        thread.id, agent_tentacle_id=decision.agent_id, with_history=False
    )
    # A handoff pins who owns the chat, so it is read and written there: a
    # chat room's sub-thread is new every kick and would forget the owner.
    chat = await deps.thread_manager.surface(thread)
    latest = chat.latest_handoff
    if (
        latest is not None
        and latest.to_agent_tentacle_id == decision.agent_id
        and latest.to_model == decision.model
    ):
        return
    await deps.thread_manager.record_handoff(
        chat,
        source_agent_tentacle_id=self.source_agent_tentacle_id,
        to_agent_tentacle_id=decision.agent_id,
        to_model=decision.model,
        reason=decision.reason,
        hint=decision.hint,
        brief=decision.brief,
        source_conversation_id=self.source_conversation_id,
        target_conversation_id=target_conversation.id,
        source_run_id=self.source_run_id,
        source_model_message_id=self.source_model_message_id,
    )

ReflexResult dataclass

A finished turn: the decision it ran on, where it answered, and the agent's result.

Source code in octomate/reflex/state.py 
154
155
156
157
158
159
160
161
@dataclass
class ReflexResult:
    """A finished turn: the decision it ran on, where it answered, and the agent's
    result."""

    decision: SummonDecision | None
    target: ResponseTarget
    result: AgentRunResult[ChannelOutput] | None = None

decision instance-attribute 

decision: SummonDecision | None

target instance-attribute 

target: ResponseTarget

result class-attribute instance-attribute 

result: AgentRunResult[ChannelOutput] | None = None

DeferredResult dataclass

A turn that stopped on deferred requests, and the batch they were parked as.

Source code in octomate/reflex/state.py 
164
165
166
167
168
169
170
171
172
173
174
175
@dataclass
class DeferredResult:
    """A turn that stopped on deferred requests, and the batch they were parked
    as."""

    requests: DeferredToolRequests
    target: ResponseTarget
    # The name of the run that deferred — carried for observability. `str`, not
    # `RunName`, because on a re-present it is read back from the persisted batch.
    run_name: str
    result: AgentRunResult[Any]
    batch_id: UUID7 | None = None

requests instance-attribute 

requests: DeferredToolRequests

target instance-attribute 

target: ResponseTarget

run_name instance-attribute 

run_name: str

result instance-attribute 

result: AgentRunResult[Any]

batch_id class-attribute instance-attribute 

batch_id: UUID7 | None = None

ReflexState dataclass

All run-wide context for one reflex graph run.

Awake resolves the source context once and writes it here; downstream nodes read from state and carry only transition discriminators.

Source code in octomate/reflex/state.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@dataclass
class ReflexState:
    """All run-wide context for one reflex graph run.

    Awake resolves the source context once and writes it here; downstream nodes
    read from state and carry only transition discriminators.
    """

    source_target: ResponseTarget | None = None
    target: ResponseTarget | None = None
    run_name: RunName = "react"
    decision: SummonDecision | None = None
    selection: AgentRouteKey | None = None  # Selected agent and model.
    conversation_id: uuid.UUID | None = None  # Current invocation's conversation.
    targets: dict[str, ResponseTarget] = field(default_factory=dict)
    summon_routes: list[AgentRoute] = field(default_factory=list)
    thread: Thread | None = None
    trigger_thread_message_id: UUID7 | None = None
    source_thread_address: ChannelAddress | None = None
    source_thread_message_ids: list[UUID7] = field(default_factory=list)
    # The handoff the next React records when it lands, or None when it records
    # none.
    handoff: PendingHandoff | None = None
    user_prompt: str | Sequence[UserContent] | None = None
    user_profile: UserProfile | None = None

source_target class-attribute instance-attribute 

source_target: ResponseTarget | None = None

target class-attribute instance-attribute 

target: ResponseTarget | None = None

run_name class-attribute instance-attribute 

run_name: RunName = 'react'

decision class-attribute instance-attribute 

decision: SummonDecision | None = None

selection class-attribute instance-attribute 

selection: AgentRouteKey | None = None

conversation_id class-attribute instance-attribute 

conversation_id: UUID | None = None

targets class-attribute instance-attribute 

targets: dict[str, ResponseTarget] = field(
    default_factory=dict
)

summon_routes class-attribute instance-attribute 

summon_routes: list[AgentRoute] = field(
    default_factory=list
)

thread class-attribute instance-attribute 

thread: Thread | None = None

trigger_thread_message_id class-attribute instance-attribute 

trigger_thread_message_id: UUID7 | None = None

source_thread_address class-attribute instance-attribute 

source_thread_address: ChannelAddress | None = None

source_thread_message_ids class-attribute instance-attribute 

source_thread_message_ids: list[UUID7] = field(
    default_factory=list
)

handoff class-attribute instance-attribute 

handoff: PendingHandoff | None = None

user_prompt class-attribute instance-attribute 

user_prompt: str | Sequence[UserContent] | None = None

user_profile class-attribute instance-attribute 

user_profile: UserProfile | None = None

ReflexDeps dataclass

The host's registries and managers a reflex run reaches through.

Source code in octomate/reflex/state.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
@dataclass
class ReflexDeps:
    """The host's registries and managers a reflex run reaches through."""

    channels: dict[str, ChannelTentacle]
    # No defaults for the managers: a deps object must carry the host's own — the
    # ledger, the conversations, the deferred actions, the gateway's live-session
    # registry — never private ones with their own identity or state.
    thread_manager: ThreadManager
    # For the bookkeeping a finished turn owes and the agent has no part in:
    # leaving the thread's workspace somewhere losing the directory cannot cost it.
    workspaces: WorkspaceManager
    conversation_manager: ConversationManager
    action_manager: DeferredActionManager
    gateway: GatewayManager
    # The graph's end of the nerve: what the kick does, sent as impulses.
    impulses: ObjectSendStream[Impulse]
    agents: dict[str, AgentTentacle] = field(default_factory=dict)

    @cached_property
    def runtime(self) -> ReflexRuntime:
        """Stateless orchestration shared by this graph's execution entries."""
        return ReflexRuntime()

    @overload
    def channel(self, target: ResponseTarget) -> ChannelTentacle: ...

    @overload
    def channel(self, target: str) -> ChannelTentacle: ...

    def channel(self, target: ResponseTarget | str) -> ChannelTentacle:
        channel_id = target.channel_id if isinstance(target, ResponseTarget) else target
        channel = self.channels.get(channel_id)
        if channel is None:
            raise ValueError(f"unknown channel {channel_id!r}")
        return channel

    def agent(self, agent_id: str) -> AgentTentacle:
        agent = self.agents.get(agent_id)
        if agent is None:
            raise ValueError(f"unknown agent {agent_id!r}")
        return agent

    @cached_property
    def gateway_agents(self) -> frozenset[str]:
        """The agents whose driven turns offer the gateway spells: each agent's own
        flag, settled at startup and read once. A turn of any other agent builds no
        session, so external callers find nothing live for it."""
        return frozenset(id for id, agent in self.agents.items() if agent.gateway)

    async def octomate_session(
        self,
        agent: AgentTentacle,
        *,
        user_profile: UserProfile | None,
        thread_id: UUID7 | None,
        conversation_address: ChannelAddress,
        conversation_id: uuid.UUID | None = None,
    ) -> OctomateSession | None:
        """One turn's gateway for `agent`, or None for an agent whose flag is off.

        Built from the host's own registries, every channel's and not just this
        one's: a spell that crosses lands where another channel's config decides who
        runs, so the gateway offers — and checks against — that channel's routes,
        and reads `surfaces` off it to know whether `scheme` can land. The agents
        are what the accomplice spells run; without a thread there is nowhere for
        a child conversation to live, and the gateway then does not offer them.
        """
        if agent.id not in self.gateway_agents:
            return None
        session = OctomateSession(
            channel_routes=self.available_routes,
            current_agent_id=agent.id,
            channels=self.channels,
            users=self.thread_manager.users,
            user_profile=user_profile,
            agents=self.agents,
            thread_id=thread_id,
            conversation_address=conversation_address,
            threads=self.thread_manager,
            workspaces=self.workspaces,
            impulses=self.impulses,
        )
        if conversation_id is not None:
            session.conversation_id = conversation_id
        elif thread_id is not None:
            # The same (thread, agent) key the run resolves internally, so an
            # external runtime's tool call finds this turn's session by the
            # conversation it already knows.
            conversation = await self.conversation_manager.ensure(
                thread_id, agent_tentacle_id=agent.id, with_history=False
            )
            session.conversation_id = conversation.id
        return session

    @property
    def available_routes(self) -> dict[str, list[AgentRoute]]:
        return self.gateway.available_routes(self.channels, self.agents)

    def resolve_agent(
        self,
        channel_id: str,
        agent_id: str | None,
        model: AgentRouteModelName | None,
    ) -> AgentModelConfig:
        """Select an exposed agent and resolve a model from its catalog."""
        agent_ids = self.channel(channel_id).agent_ids
        if agent_id is None:
            agent_id = agent_ids[0]
        if agent_id not in agent_ids:
            raise ValueError(
                f"agent {agent_id!r} is not bound to channel {channel_id!r}"
            )

        agent = self.agent(agent_id)
        if not agent.models:
            raise ValueError(f"agent {agent_id!r} has no available model catalog")
        if model is None:
            return AgentModelConfig(agent=agent_id, model=agent.resolve_model())
        served = agent.served_model(model)
        if served is None:
            raise ValueError(f"agent {agent_id!r} does not serve model {model!r}")
        return AgentModelConfig(agent=agent_id, model=served)

    async def render_chat(
        self, messages: list[ThreadMessage], *, ceiling: int = 0
    ) -> str:
        """The ledger rows as one transcript: who spoke, under which identities, and
        the `#msg:<id>` handle a brief would cite them by.

        Each message is its segments, as they render themselves — the same body an
        inbound event shows. `message_text` would be the shorter route and the wrong
        one: it keeps only text and markdown, so a message that was a picture or a
        quoted reply arrives as nothing at all.

        `ceiling` cuts each message. It is 0 for the messages a turn must answer —
        those are the work — and set for the chat behind them, where one pasted log
        would otherwise be the whole slice.
        """
        parts: list[str] = []
        for message in messages:
            if message.kind != "message":
                continue
            text = "\n".join(str(segment) for segment in message.segments)
            if not text:
                continue
            if ceiling and len(text) > ceiling:
                text = f"{text[:ceiling]}…"
            sender = await message.sender
            display_name = (
                (sender.name or sender.nickname or "anonymous")
                if sender is not None
                else "anonymous"
            )
            owner = sender.user.peek() if sender is not None else None
            ids = (
                f"{message.user_id}, user:{owner.username}"
                if owner is not None
                else message.user_id
            )
            platform_id = (
                f" #msg:{message.platform_message_id}"
                if message.platform_message_id
                else ""
            )
            parts.append(f"{display_name} ({ids}){platform_id}:\n{text}")
        return "\n\n".join(parts)

    async def report(self, state: ReflexState, error: Exception) -> None:
        """Publish that the turn failed, on the thread it came from. A turn that
        failed before it knew its source has nobody to tell."""
        source = state.source_target.address if state.source_target else None
        if source is None:
            return
        trace_id = format(trace.get_current_span().get_span_context().trace_id, "032x")
        failure = RunErrorEvent(message=str(error), trace_id=trace_id)
        await self.impulses.send(Impulse(failure, source))

    async def carry(
        self,
        address: ChannelAddress,
        stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
    ) -> None:
        """Send a run's stream on the nerve at `address` as wire events, item by
        item, until it ends."""
        async for item in stream:
            wire = to_wire(item)
            if wire is not None:
                await self.impulses.send(Impulse(wire, address))

    async def record_move(
        self,
        address: ChannelAddress,
        text: str,
        *,
        agent_tentacle_id: str,
        platform_message_id: str | None = None,
    ) -> ThreadMessage:
        """Write the line a move leaves behind to the ledger of the chat it left.

        A person already sees it — the channel posted it, and a thread opened here
        hangs off it. The turn that comes next does not: a chat room's recap is built
        from the ledger, and a move nobody recorded leaves a chat in which the work
        simply stops. An agent reading that answers what was carried away ten minutes
        ago.

        The row carries the opener's own platform id, so the ledger and the platform
        name the same message rather than two; a line still to be posted gets its id
        from the channel that posts it.
        """
        return await self.thread_manager.record_outbound(
            address,
            agent_tentacle_id=agent_tentacle_id,
            segments=[MarkdownSegment(data={"text": text})],
            sender=self.channel(address.channel_tentacle_id).self_profile,
            platform_message_id=platform_message_id,
            message_text=text,
            raw=text,
        )

    async def load_pending_prompt(
        self,
        state: ReflexState,
        active_agent_id: str,
    ) -> None:
        source_target = state.source_target
        if (
            state.thread is None
            or state.trigger_thread_message_id is None
            or source_target is None
            or source_target.address is None
        ):
            return
        # Pull every recorded chat-ledger row that has not been bound into a
        # model request yet: rule-gated group messages, sleeping/not-kicked
        # messages, and messages that stacked up behind an already-running turn.
        messages = await self.thread_manager.pending_prompt_messages(
            state.thread,
            state.trigger_thread_message_id,
            active_agent_id,
        )
        if not messages:
            return
        state.source_thread_address = source_target.address
        state.source_thread_message_ids = [message.id for message in messages]

        asked = await self.render_chat(messages)
        if not asked:
            return
        recap = await self.chat_recap(state, messages[0])
        # The messages the turn must answer are the body, unmarked. What was said
        # before them is the system's addition and stands in a tag.
        state.user_prompt = f"{recap}\n\n{asked}" if recap else asked

    async def chat_recap(self, state: ReflexState, earliest: ThreadMessage) -> str:
        """What was said in this chat room before the messages the turn must answer.

        Only a kick in a dm or a group chat needs it: that runs in a thread of its
        own, so its model context starts empty every time and the agent would answer
        a chat it cannot see. A thread carries its own history in the conversation
        and gets none of this — its context is already the whole of the work.

        The chat is the parent's, so the rows come from there; how many, and how much
        of each, is the channel's to say.
        """
        thread = state.thread
        source_target = state.source_target
        if thread is None or thread.parent_thread_id is None or source_target is None:
            return ""
        recap = self.channel(source_target).config.recap
        if not recap.messages:
            return ""
        messages = await self.thread_manager.chat_messages_before(
            thread.parent_thread_id, earliest.id, limit=recap.messages
        )
        rendered = await self.render_chat(messages, ceiling=recap.characters)
        return tagged("chat_recap", f"{RECAP_HEADER}\n\n{rendered}" if rendered else "")

channels instance-attribute 

channels: dict[str, ChannelTentacle]

thread_manager instance-attribute 

thread_manager: ThreadManager

workspaces instance-attribute 

workspaces: WorkspaceManager

conversation_manager instance-attribute 

conversation_manager: ConversationManager

action_manager instance-attribute 

action_manager: DeferredActionManager

gateway instance-attribute 

gateway: GatewayManager

impulses instance-attribute 

impulses: ObjectSendStream[Impulse]

agents class-attribute instance-attribute 

agents: dict[str, AgentTentacle] = field(
    default_factory=dict
)

runtime cached property 

runtime: ReflexRuntime

Stateless orchestration shared by this graph's execution entries.

gateway_agents cached property 

gateway_agents: frozenset[str]

The agents whose driven turns offer the gateway spells: each agent's own flag, settled at startup and read once. A turn of any other agent builds no session, so external callers find nothing live for it.

available_routes property 

available_routes: dict[str, list[AgentRoute]]

channel 

channel(target: ResponseTarget) -> ChannelTentacle

channel(target: str) -> ChannelTentacle

channel(target: ResponseTarget | str) -> ChannelTentacle
Source code in octomate/reflex/state.py 
243
244
245
246
247
248
def channel(self, target: ResponseTarget | str) -> ChannelTentacle:
    channel_id = target.channel_id if isinstance(target, ResponseTarget) else target
    channel = self.channels.get(channel_id)
    if channel is None:
        raise ValueError(f"unknown channel {channel_id!r}")
    return channel

agent 

agent(agent_id: str) -> AgentTentacle
Source code in octomate/reflex/state.py 
250
251
252
253
254
def agent(self, agent_id: str) -> AgentTentacle:
    agent = self.agents.get(agent_id)
    if agent is None:
        raise ValueError(f"unknown agent {agent_id!r}")
    return agent

octomate_session async 

octomate_session(
    agent: AgentTentacle,
    *,
    user_profile: UserProfile | None,
    thread_id: UUID7 | None,
    conversation_address: ChannelAddress,
    conversation_id: UUID | None = None,
) -> OctomateSession | None

One turn's gateway for agent, or None for an agent whose flag is off.

Built from the host's own registries, every channel's and not just this one's: a spell that crosses lands where another channel's config decides who runs, so the gateway offers — and checks against — that channel's routes, and reads surfaces off it to know whether scheme can land. The agents are what the accomplice spells run; without a thread there is nowhere for a child conversation to live, and the gateway then does not offer them.

Source code in octomate/reflex/state.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
async def octomate_session(
    self,
    agent: AgentTentacle,
    *,
    user_profile: UserProfile | None,
    thread_id: UUID7 | None,
    conversation_address: ChannelAddress,
    conversation_id: uuid.UUID | None = None,
) -> OctomateSession | None:
    """One turn's gateway for `agent`, or None for an agent whose flag is off.

    Built from the host's own registries, every channel's and not just this
    one's: a spell that crosses lands where another channel's config decides who
    runs, so the gateway offers — and checks against — that channel's routes,
    and reads `surfaces` off it to know whether `scheme` can land. The agents
    are what the accomplice spells run; without a thread there is nowhere for
    a child conversation to live, and the gateway then does not offer them.
    """
    if agent.id not in self.gateway_agents:
        return None
    session = OctomateSession(
        channel_routes=self.available_routes,
        current_agent_id=agent.id,
        channels=self.channels,
        users=self.thread_manager.users,
        user_profile=user_profile,
        agents=self.agents,
        thread_id=thread_id,
        conversation_address=conversation_address,
        threads=self.thread_manager,
        workspaces=self.workspaces,
        impulses=self.impulses,
    )
    if conversation_id is not None:
        session.conversation_id = conversation_id
    elif thread_id is not None:
        # The same (thread, agent) key the run resolves internally, so an
        # external runtime's tool call finds this turn's session by the
        # conversation it already knows.
        conversation = await self.conversation_manager.ensure(
            thread_id, agent_tentacle_id=agent.id, with_history=False
        )
        session.conversation_id = conversation.id
    return session

resolve_agent 

resolve_agent(
    channel_id: str,
    agent_id: str | None,
    model: AgentRouteModelName | None,
) -> AgentModelConfig

Select an exposed agent and resolve a model from its catalog.

Source code in octomate/reflex/state.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def resolve_agent(
    self,
    channel_id: str,
    agent_id: str | None,
    model: AgentRouteModelName | None,
) -> AgentModelConfig:
    """Select an exposed agent and resolve a model from its catalog."""
    agent_ids = self.channel(channel_id).agent_ids
    if agent_id is None:
        agent_id = agent_ids[0]
    if agent_id not in agent_ids:
        raise ValueError(
            f"agent {agent_id!r} is not bound to channel {channel_id!r}"
        )

    agent = self.agent(agent_id)
    if not agent.models:
        raise ValueError(f"agent {agent_id!r} has no available model catalog")
    if model is None:
        return AgentModelConfig(agent=agent_id, model=agent.resolve_model())
    served = agent.served_model(model)
    if served is None:
        raise ValueError(f"agent {agent_id!r} does not serve model {model!r}")
    return AgentModelConfig(agent=agent_id, model=served)

render_chat async 

render_chat(
    messages: list[ThreadMessage], *, ceiling: int = 0
) -> str

The ledger rows as one transcript: who spoke, under which identities, and the #msg:<id> handle a brief would cite them by.

Each message is its segments, as they render themselves — the same body an inbound event shows. message_text would be the shorter route and the wrong one: it keeps only text and markdown, so a message that was a picture or a quoted reply arrives as nothing at all.

ceiling cuts each message. It is 0 for the messages a turn must answer — those are the work — and set for the chat behind them, where one pasted log would otherwise be the whole slice.

Source code in octomate/reflex/state.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
async def render_chat(
    self, messages: list[ThreadMessage], *, ceiling: int = 0
) -> str:
    """The ledger rows as one transcript: who spoke, under which identities, and
    the `#msg:<id>` handle a brief would cite them by.

    Each message is its segments, as they render themselves — the same body an
    inbound event shows. `message_text` would be the shorter route and the wrong
    one: it keeps only text and markdown, so a message that was a picture or a
    quoted reply arrives as nothing at all.

    `ceiling` cuts each message. It is 0 for the messages a turn must answer —
    those are the work — and set for the chat behind them, where one pasted log
    would otherwise be the whole slice.
    """
    parts: list[str] = []
    for message in messages:
        if message.kind != "message":
            continue
        text = "\n".join(str(segment) for segment in message.segments)
        if not text:
            continue
        if ceiling and len(text) > ceiling:
            text = f"{text[:ceiling]}…"
        sender = await message.sender
        display_name = (
            (sender.name or sender.nickname or "anonymous")
            if sender is not None
            else "anonymous"
        )
        owner = sender.user.peek() if sender is not None else None
        ids = (
            f"{message.user_id}, user:{owner.username}"
            if owner is not None
            else message.user_id
        )
        platform_id = (
            f" #msg:{message.platform_message_id}"
            if message.platform_message_id
            else ""
        )
        parts.append(f"{display_name} ({ids}){platform_id}:\n{text}")
    return "\n\n".join(parts)

report async 

report(state: ReflexState, error: Exception) -> None

Publish that the turn failed, on the thread it came from. A turn that failed before it knew its source has nobody to tell.

Source code in octomate/reflex/state.py 
381
382
383
384
385
386
387
388
389
async def report(self, state: ReflexState, error: Exception) -> None:
    """Publish that the turn failed, on the thread it came from. A turn that
    failed before it knew its source has nobody to tell."""
    source = state.source_target.address if state.source_target else None
    if source is None:
        return
    trace_id = format(trace.get_current_span().get_span_context().trace_id, "032x")
    failure = RunErrorEvent(message=str(error), trace_id=trace_id)
    await self.impulses.send(Impulse(failure, source))

carry async 

carry(
    address: ChannelAddress,
    stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
) -> None

Send a run's stream on the nerve at address as wire events, item by item, until it ends.

Source code in octomate/reflex/state.py 
391
392
393
394
395
396
397
398
399
400
401
async def carry(
    self,
    address: ChannelAddress,
    stream: AsyncIterator[ReactStreamEvent[ChannelOutput]],
) -> None:
    """Send a run's stream on the nerve at `address` as wire events, item by
    item, until it ends."""
    async for item in stream:
        wire = to_wire(item)
        if wire is not None:
            await self.impulses.send(Impulse(wire, address))

record_move async 

record_move(
    address: ChannelAddress,
    text: str,
    *,
    agent_tentacle_id: str,
    platform_message_id: str | None = None,
) -> ThreadMessage

Write the line a move leaves behind to the ledger of the chat it left.

A person already sees it — the channel posted it, and a thread opened here hangs off it. The turn that comes next does not: a chat room's recap is built from the ledger, and a move nobody recorded leaves a chat in which the work simply stops. An agent reading that answers what was carried away ten minutes ago.

The row carries the opener's own platform id, so the ledger and the platform name the same message rather than two; a line still to be posted gets its id from the channel that posts it.

Source code in octomate/reflex/state.py 
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
async def record_move(
    self,
    address: ChannelAddress,
    text: str,
    *,
    agent_tentacle_id: str,
    platform_message_id: str | None = None,
) -> ThreadMessage:
    """Write the line a move leaves behind to the ledger of the chat it left.

    A person already sees it — the channel posted it, and a thread opened here
    hangs off it. The turn that comes next does not: a chat room's recap is built
    from the ledger, and a move nobody recorded leaves a chat in which the work
    simply stops. An agent reading that answers what was carried away ten minutes
    ago.

    The row carries the opener's own platform id, so the ledger and the platform
    name the same message rather than two; a line still to be posted gets its id
    from the channel that posts it.
    """
    return await self.thread_manager.record_outbound(
        address,
        agent_tentacle_id=agent_tentacle_id,
        segments=[MarkdownSegment(data={"text": text})],
        sender=self.channel(address.channel_tentacle_id).self_profile,
        platform_message_id=platform_message_id,
        message_text=text,
        raw=text,
    )

load_pending_prompt async 

load_pending_prompt(
    state: ReflexState, active_agent_id: str
) -> None
Source code in octomate/reflex/state.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
async def load_pending_prompt(
    self,
    state: ReflexState,
    active_agent_id: str,
) -> None:
    source_target = state.source_target
    if (
        state.thread is None
        or state.trigger_thread_message_id is None
        or source_target is None
        or source_target.address is None
    ):
        return
    # Pull every recorded chat-ledger row that has not been bound into a
    # model request yet: rule-gated group messages, sleeping/not-kicked
    # messages, and messages that stacked up behind an already-running turn.
    messages = await self.thread_manager.pending_prompt_messages(
        state.thread,
        state.trigger_thread_message_id,
        active_agent_id,
    )
    if not messages:
        return
    state.source_thread_address = source_target.address
    state.source_thread_message_ids = [message.id for message in messages]

    asked = await self.render_chat(messages)
    if not asked:
        return
    recap = await self.chat_recap(state, messages[0])
    # The messages the turn must answer are the body, unmarked. What was said
    # before them is the system's addition and stands in a tag.
    state.user_prompt = f"{recap}\n\n{asked}" if recap else asked

chat_recap async 

chat_recap(
    state: ReflexState, earliest: ThreadMessage
) -> str

What was said in this chat room before the messages the turn must answer.

Only a kick in a dm or a group chat needs it: that runs in a thread of its own, so its model context starts empty every time and the agent would answer a chat it cannot see. A thread carries its own history in the conversation and gets none of this — its context is already the whole of the work.

The chat is the parent's, so the rows come from there; how many, and how much of each, is the channel's to say.

Source code in octomate/reflex/state.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
async def chat_recap(self, state: ReflexState, earliest: ThreadMessage) -> str:
    """What was said in this chat room before the messages the turn must answer.

    Only a kick in a dm or a group chat needs it: that runs in a thread of its
    own, so its model context starts empty every time and the agent would answer
    a chat it cannot see. A thread carries its own history in the conversation
    and gets none of this — its context is already the whole of the work.

    The chat is the parent's, so the rows come from there; how many, and how much
    of each, is the channel's to say.
    """
    thread = state.thread
    source_target = state.source_target
    if thread is None or thread.parent_thread_id is None or source_target is None:
        return ""
    recap = self.channel(source_target).config.recap
    if not recap.messages:
        return ""
    messages = await self.thread_manager.chat_messages_before(
        thread.parent_thread_id, earliest.id, limit=recap.messages
    )
    rendered = await self.render_chat(messages, ceiling=recap.characters)
    return tagged("chat_recap", f"{RECAP_HEADER}\n\n{rendered}" if rendered else "")

ReflexRuntime

Stateless orchestration in graph deps; invocation resources stay with callers.

Source code in octomate/reflex/state.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
class ReflexRuntime:
    """Stateless orchestration in graph deps; invocation resources stay with callers."""

    async def resources(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
    ) -> tuple[OctomateSession | None, ReflexSuspender, list[AgentCapability[None]]]:
        """Build fresh gateway, approval and tool context for the selected invocation."""
        state = ctx.state
        target = state.target
        source = state.source_target
        selection = state.selection
        if (
            target is None
            or target.address is None
            or source is None
            or source.address is None
            or selection is None
        ):
            raise ValueError(
                "a Reflex run requires a selected agent and resolved source and target addresses"
            )
        agent = ctx.deps.agent(selection.agent_id)
        channel = ctx.deps.channel(target)
        thread_id = state.thread.id if state.thread else None
        state.summon_routes = [
            route
            for route in ctx.deps.available_routes[target.channel_id]
            if route.agent_id != agent.id
        ]
        if state.conversation_id is None and thread_id is not None:
            conversation = await ctx.deps.conversation_manager.ensure(
                thread_id, agent_tentacle_id=agent.id, with_history=False
            )
            state.conversation_id = conversation.id
        session = await ctx.deps.octomate_session(
            agent,
            user_profile=state.user_profile,
            thread_id=thread_id,
            conversation_address=target.address,
            conversation_id=state.conversation_id,
        )
        capabilities: list[AgentCapability[None]] = []
        if session is not None:
            capabilities.append(
                GatewayCapability(
                    session=session, conversations=ctx.deps.conversation_manager
                )
            )
        suspender = ReflexSuspender(
            channel=channel,
            action_manager=ctx.deps.action_manager,
            conversation_manager=ctx.deps.conversation_manager,
            agent_tentacle_id=agent.id,
            run_name=state.run_name,
            source_address=source.address,
            target_address=target.address,
            target_mode=target.mode,
            decision=state.decision,
            model=selection.model,
            thread_id=thread_id,
        )
        return session, suspender, capabilities

    async def prepare_user(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        profile: UserProfile,
        *,
        session: OctomateSession | None,
        suspender: ReflexSuspender,
        capabilities: list[AgentCapability[None]],
    ) -> None:
        """Mount capabilities belonging to the user authorized for this invocation."""
        ctx.state.user_profile = profile
        if session is not None:
            session.user_profile = profile
        agent = ctx.deps.agent(suspender.agent_tentacle_id)
        capabilities.extend(await agent.user_capabilities(profile))

    async def send(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        suspender: ReflexSuspender,
        event: MessageSentEvent,
    ) -> bool:
        """Deliver a gateway send to a person's direct messages, when possible:
        recorded there first, then published for that thread alone."""
        destination = event.destination
        if destination is None:
            return False
        channel = ctx.deps.channel(destination.channel_tentacle_id)
        dm = await channel.open_dm(destination.user_id)
        if dm is None:
            logger.warning(
                "Channel %s could not open a DM with %s; delivering the send to %s instead",
                destination.channel_tentacle_id,
                destination.user_id,
                suspender.target_address,
            )
            return False
        message = await ctx.deps.thread_manager.record_outbound(
            dm,
            agent_tentacle_id=suspender.agent_tentacle_id,
            segments=event.segments,
            sender=channel.self_profile,
        )
        sent = MessageSentEvent(
            segments=event.segments, destination=dm, thread_message_id=message.id
        )
        await ctx.deps.impulses.send(Impulse(sent, dm, private=True))
        return True

    async def drive(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        suspender: ReflexSuspender,
        source: AsyncGenerator[ReactStreamEvent[ChannelOutput], None],
    ) -> AgentRunResult[ChannelOutput]:
        """Run the native invocation onto the nerve, closing it before recording
        its reply: the run's start before anything it streams, and its end after,
        carrying the reply's ledger row, whatever happened."""
        results: list[AgentRunResult[ChannelOutput]] = []
        address = suspender.target_address

        async def events() -> AsyncGenerator[ReactStreamEvent[ChannelOutput], None]:
            async for event in source:
                if isinstance(event, AgentRunResultEvent):
                    results.append(event.result)
                if isinstance(event, MessageSentEvent) and await self.send(
                    ctx, suspender, event
                ):
                    continue
                yield event

        await ctx.deps.impulses.send(Impulse(RunStartedEvent(address=address), address))
        reply: ThreadMessage | None = None
        try:
            async with aclosing(source), aclosing(events()) as stream:
                await ctx.deps.carry(address, stream)
            if not results:
                raise RuntimeError(
                    f"react stream for {address} completed without a result"
                )
            reply = await self.record_reply(ctx, suspender, results[-1])
            return results[-1]
        finally:
            ended = RunEndedEvent(
                thread_message_id=reply.id if reply is not None else None
            )
            await ctx.deps.impulses.send(Impulse(ended, address))

    async def record_reply(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        suspender: ReflexSuspender,
        result: AgentRunResult[ChannelOutput],
    ) -> ThreadMessage | None:
        """Record and bind the run's reply, answering its row for the channel to
        mark with the platform id it renders the reply as."""
        output = result.output
        segments: list[MessageSegment]
        if isinstance(output, str):
            segments = [MarkdownSegment(data={"text": output})] if output else []
        elif isinstance(output, Iterable):
            segments = list(output)
        else:
            return None
        reply_to, body = split_reply(segments)
        message = None
        if body:
            message = await ctx.deps.thread_manager.record_outbound(
                suspender.target_address,
                agent_tentacle_id=suspender.agent_tentacle_id,
                segments=segments,
                reply_id=reply_to or "",
                sender=suspender.channel.self_profile,
                raw=output
                if isinstance(output, str)
                else "\n\n".join(str(segment) for segment in body),
                message_text=output if isinstance(output, str) else None,
            )
        await ctx.deps.thread_manager.bind_assistant_replies(
            [message.id] if message is not None else [],
            run_id=result.run_id,
        )
        return message

    async def present_command(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        suspender: ReflexSuspender,
        outcome: CommandOutcome,
    ) -> None:
        """Publish direct feedback; its command receipt already owns the output."""
        segments: list[MessageSegment] = (
            [TextSegment(data={"text": outcome.message})]
            if isinstance(outcome, CommandError)
            else outcome.segments
        )
        if not segments:
            return
        await ctx.deps.impulses.send(
            Impulse(MessageSentEvent(segments=segments), suspender.target_address)
        )

    async def finish(
        self,
        ctx: GraphRunContext[ReflexState, ReflexDeps],
        session: OctomateSession | None,
        suspender: ReflexSuspender,
        run_result: AgentRunResult[ChannelOutput],
    ) -> Summon | Teleport | Scheme | End[ReflexGraphResult]:
        """Follow the gateway decision or deferral left by an actual agent run."""
        # These nodes import the graph types consumed by this runtime.
        from octomate.reflex.nodes.scheme import Scheme
        from octomate.reflex.nodes.summon import Summon
        from octomate.reflex.nodes.teleport import Teleport

        state = ctx.state
        target = state.target
        if target is None:
            raise ValueError("a Reflex run requires a resolved target")
        decision = state.decision
        agent_id = suspender.agent_tentacle_id
        thread_id = state.thread.id if state.thread else None
        output = run_result.output
        with reflex_logfire.span("reflex.finish", run_id=run_result.run_id) as span:
            if session is not None and session.dismissing and state.thread is not None:
                # The agent said this thread's work is done: its tree goes now
                # that the run is out of it, saved first and kept if that failed.
                result = await ctx.deps.workspaces.dismiss(state.thread)
                span.set_attribute(
                    "react.dismissed",
                    result,
                )
            if isinstance(output, DeferredToolRequests):
                # `teleport` is resolved by the graph (fork + resume), not a human. The
                # suspender classified it by its declared metadata kind and stashed it,
                # so route on the typed request instead of re-scanning tool names.
                if suspender.teleport is not None:
                    return Teleport(
                        request=suspender.teleport, origin=target, agent_id=agent_id
                    )
                return End(
                    DeferredResult(
                        requests=output,
                        target=target,
                        run_name=state.run_name,
                        result=run_result,
                        batch_id=suspender.suspended_batch_id,
                    )
                )

            gateway_decision = session.decision if session else None
            if isinstance(gateway_decision, SchemeDecision | SummonDecision):
                # Where this handoff came from, as the row records it: the
                # conversation this turn ran in, as of its last message.
                conversation = (
                    await ctx.deps.conversation_manager.ensure(
                        thread_id, agent_tentacle_id=agent_id
                    )
                    if thread_id is not None
                    else None
                )
                last = (
                    max(
                        conversation.messages,
                        key=lambda message: message.id,
                        default=None,
                    )
                    if conversation is not None
                    else None
                )
                state.handoff = PendingHandoff(
                    source_agent_tentacle_id=agent_id,
                    source_conversation_id=conversation.id if conversation else None,
                    source_run_id=last.run_id if last else None,
                    source_model_message_id=last.id if last else None,
                )
            if isinstance(gateway_decision, SchemeDecision):
                span.set_attribute("react.action", gateway_decision.action)
                reflex_logfire.info(
                    "react -> scheme into the asker's dm",
                    destination=str(gateway_decision.destination),
                )
                return Scheme(
                    request=gateway_decision,
                    origin=target,
                    agent_id=agent_id,
                )
            if isinstance(gateway_decision, SummonDecision):
                state.decision = gateway_decision
                state.target = target
                state.run_name = "summon"
                span.set_attribute("react.action", gateway_decision.action)
                span.set_attribute("react.next_agent_id", gateway_decision.agent_id)
                reflex_logfire.info(
                    "react -> {action} agent={agent_id}",
                    action=gateway_decision.action,
                    agent_id=gateway_decision.agent_id,
                    reason=gateway_decision.reason,
                )
                return Summon()

            return End(
                ReflexResult(
                    decision=decision,
                    target=target,
                    result=run_result,
                )
            )

resources async 

resources(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> tuple[
    OctomateSession | None,
    ReflexSuspender,
    list[AgentCapability[None]],
]

Build fresh gateway, approval and tool context for the selected invocation.

Source code in octomate/reflex/state.py 
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
async def resources(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
) -> tuple[OctomateSession | None, ReflexSuspender, list[AgentCapability[None]]]:
    """Build fresh gateway, approval and tool context for the selected invocation."""
    state = ctx.state
    target = state.target
    source = state.source_target
    selection = state.selection
    if (
        target is None
        or target.address is None
        or source is None
        or source.address is None
        or selection is None
    ):
        raise ValueError(
            "a Reflex run requires a selected agent and resolved source and target addresses"
        )
    agent = ctx.deps.agent(selection.agent_id)
    channel = ctx.deps.channel(target)
    thread_id = state.thread.id if state.thread else None
    state.summon_routes = [
        route
        for route in ctx.deps.available_routes[target.channel_id]
        if route.agent_id != agent.id
    ]
    if state.conversation_id is None and thread_id is not None:
        conversation = await ctx.deps.conversation_manager.ensure(
            thread_id, agent_tentacle_id=agent.id, with_history=False
        )
        state.conversation_id = conversation.id
    session = await ctx.deps.octomate_session(
        agent,
        user_profile=state.user_profile,
        thread_id=thread_id,
        conversation_address=target.address,
        conversation_id=state.conversation_id,
    )
    capabilities: list[AgentCapability[None]] = []
    if session is not None:
        capabilities.append(
            GatewayCapability(
                session=session, conversations=ctx.deps.conversation_manager
            )
        )
    suspender = ReflexSuspender(
        channel=channel,
        action_manager=ctx.deps.action_manager,
        conversation_manager=ctx.deps.conversation_manager,
        agent_tentacle_id=agent.id,
        run_name=state.run_name,
        source_address=source.address,
        target_address=target.address,
        target_mode=target.mode,
        decision=state.decision,
        model=selection.model,
        thread_id=thread_id,
    )
    return session, suspender, capabilities

prepare_user async 

prepare_user(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    profile: UserProfile,
    *,
    session: OctomateSession | None,
    suspender: ReflexSuspender,
    capabilities: list[AgentCapability[None]],
) -> None

Mount capabilities belonging to the user authorized for this invocation.

Source code in octomate/reflex/state.py 
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
async def prepare_user(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    profile: UserProfile,
    *,
    session: OctomateSession | None,
    suspender: ReflexSuspender,
    capabilities: list[AgentCapability[None]],
) -> None:
    """Mount capabilities belonging to the user authorized for this invocation."""
    ctx.state.user_profile = profile
    if session is not None:
        session.user_profile = profile
    agent = ctx.deps.agent(suspender.agent_tentacle_id)
    capabilities.extend(await agent.user_capabilities(profile))

send async 

send(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    event: MessageSentEvent,
) -> bool

Deliver a gateway send to a person's direct messages, when possible: recorded there first, then published for that thread alone.

Source code in octomate/reflex/state.py 
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
async def send(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    event: MessageSentEvent,
) -> bool:
    """Deliver a gateway send to a person's direct messages, when possible:
    recorded there first, then published for that thread alone."""
    destination = event.destination
    if destination is None:
        return False
    channel = ctx.deps.channel(destination.channel_tentacle_id)
    dm = await channel.open_dm(destination.user_id)
    if dm is None:
        logger.warning(
            "Channel %s could not open a DM with %s; delivering the send to %s instead",
            destination.channel_tentacle_id,
            destination.user_id,
            suspender.target_address,
        )
        return False
    message = await ctx.deps.thread_manager.record_outbound(
        dm,
        agent_tentacle_id=suspender.agent_tentacle_id,
        segments=event.segments,
        sender=channel.self_profile,
    )
    sent = MessageSentEvent(
        segments=event.segments, destination=dm, thread_message_id=message.id
    )
    await ctx.deps.impulses.send(Impulse(sent, dm, private=True))
    return True

drive async 

drive(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    source: AsyncGenerator[
        ReactStreamEvent[ChannelOutput], None
    ],
) -> AgentRunResult[ChannelOutput]

Run the native invocation onto the nerve, closing it before recording its reply: the run's start before anything it streams, and its end after, carrying the reply's ledger row, whatever happened.

Source code in octomate/reflex/state.py 
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
async def drive(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    source: AsyncGenerator[ReactStreamEvent[ChannelOutput], None],
) -> AgentRunResult[ChannelOutput]:
    """Run the native invocation onto the nerve, closing it before recording
    its reply: the run's start before anything it streams, and its end after,
    carrying the reply's ledger row, whatever happened."""
    results: list[AgentRunResult[ChannelOutput]] = []
    address = suspender.target_address

    async def events() -> AsyncGenerator[ReactStreamEvent[ChannelOutput], None]:
        async for event in source:
            if isinstance(event, AgentRunResultEvent):
                results.append(event.result)
            if isinstance(event, MessageSentEvent) and await self.send(
                ctx, suspender, event
            ):
                continue
            yield event

    await ctx.deps.impulses.send(Impulse(RunStartedEvent(address=address), address))
    reply: ThreadMessage | None = None
    try:
        async with aclosing(source), aclosing(events()) as stream:
            await ctx.deps.carry(address, stream)
        if not results:
            raise RuntimeError(
                f"react stream for {address} completed without a result"
            )
        reply = await self.record_reply(ctx, suspender, results[-1])
        return results[-1]
    finally:
        ended = RunEndedEvent(
            thread_message_id=reply.id if reply is not None else None
        )
        await ctx.deps.impulses.send(Impulse(ended, address))

record_reply async 

record_reply(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    result: AgentRunResult[ChannelOutput],
) -> ThreadMessage | None

Record and bind the run's reply, answering its row for the channel to mark with the platform id it renders the reply as.

Source code in octomate/reflex/state.py 
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
async def record_reply(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    result: AgentRunResult[ChannelOutput],
) -> ThreadMessage | None:
    """Record and bind the run's reply, answering its row for the channel to
    mark with the platform id it renders the reply as."""
    output = result.output
    segments: list[MessageSegment]
    if isinstance(output, str):
        segments = [MarkdownSegment(data={"text": output})] if output else []
    elif isinstance(output, Iterable):
        segments = list(output)
    else:
        return None
    reply_to, body = split_reply(segments)
    message = None
    if body:
        message = await ctx.deps.thread_manager.record_outbound(
            suspender.target_address,
            agent_tentacle_id=suspender.agent_tentacle_id,
            segments=segments,
            reply_id=reply_to or "",
            sender=suspender.channel.self_profile,
            raw=output
            if isinstance(output, str)
            else "\n\n".join(str(segment) for segment in body),
            message_text=output if isinstance(output, str) else None,
        )
    await ctx.deps.thread_manager.bind_assistant_replies(
        [message.id] if message is not None else [],
        run_id=result.run_id,
    )
    return message

present_command async 

present_command(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    outcome: CommandOutcome,
) -> None

Publish direct feedback; its command receipt already owns the output.

Source code in octomate/reflex/state.py 
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
async def present_command(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    suspender: ReflexSuspender,
    outcome: CommandOutcome,
) -> None:
    """Publish direct feedback; its command receipt already owns the output."""
    segments: list[MessageSegment] = (
        [TextSegment(data={"text": outcome.message})]
        if isinstance(outcome, CommandError)
        else outcome.segments
    )
    if not segments:
        return
    await ctx.deps.impulses.send(
        Impulse(MessageSentEvent(segments=segments), suspender.target_address)
    )

finish async 

finish(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    session: OctomateSession | None,
    suspender: ReflexSuspender,
    run_result: AgentRunResult[ChannelOutput],
) -> Summon | Teleport | Scheme | End[ReflexGraphResult]

Follow the gateway decision or deferral left by an actual agent run.

Source code in octomate/reflex/state.py 
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
async def finish(
    self,
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    session: OctomateSession | None,
    suspender: ReflexSuspender,
    run_result: AgentRunResult[ChannelOutput],
) -> Summon | Teleport | Scheme | End[ReflexGraphResult]:
    """Follow the gateway decision or deferral left by an actual agent run."""
    # These nodes import the graph types consumed by this runtime.
    from octomate.reflex.nodes.scheme import Scheme
    from octomate.reflex.nodes.summon import Summon
    from octomate.reflex.nodes.teleport import Teleport

    state = ctx.state
    target = state.target
    if target is None:
        raise ValueError("a Reflex run requires a resolved target")
    decision = state.decision
    agent_id = suspender.agent_tentacle_id
    thread_id = state.thread.id if state.thread else None
    output = run_result.output
    with reflex_logfire.span("reflex.finish", run_id=run_result.run_id) as span:
        if session is not None and session.dismissing and state.thread is not None:
            # The agent said this thread's work is done: its tree goes now
            # that the run is out of it, saved first and kept if that failed.
            result = await ctx.deps.workspaces.dismiss(state.thread)
            span.set_attribute(
                "react.dismissed",
                result,
            )
        if isinstance(output, DeferredToolRequests):
            # `teleport` is resolved by the graph (fork + resume), not a human. The
            # suspender classified it by its declared metadata kind and stashed it,
            # so route on the typed request instead of re-scanning tool names.
            if suspender.teleport is not None:
                return Teleport(
                    request=suspender.teleport, origin=target, agent_id=agent_id
                )
            return End(
                DeferredResult(
                    requests=output,
                    target=target,
                    run_name=state.run_name,
                    result=run_result,
                    batch_id=suspender.suspended_batch_id,
                )
            )

        gateway_decision = session.decision if session else None
        if isinstance(gateway_decision, SchemeDecision | SummonDecision):
            # Where this handoff came from, as the row records it: the
            # conversation this turn ran in, as of its last message.
            conversation = (
                await ctx.deps.conversation_manager.ensure(
                    thread_id, agent_tentacle_id=agent_id
                )
                if thread_id is not None
                else None
            )
            last = (
                max(
                    conversation.messages,
                    key=lambda message: message.id,
                    default=None,
                )
                if conversation is not None
                else None
            )
            state.handoff = PendingHandoff(
                source_agent_tentacle_id=agent_id,
                source_conversation_id=conversation.id if conversation else None,
                source_run_id=last.run_id if last else None,
                source_model_message_id=last.id if last else None,
            )
        if isinstance(gateway_decision, SchemeDecision):
            span.set_attribute("react.action", gateway_decision.action)
            reflex_logfire.info(
                "react -> scheme into the asker's dm",
                destination=str(gateway_decision.destination),
            )
            return Scheme(
                request=gateway_decision,
                origin=target,
                agent_id=agent_id,
            )
        if isinstance(gateway_decision, SummonDecision):
            state.decision = gateway_decision
            state.target = target
            state.run_name = "summon"
            span.set_attribute("react.action", gateway_decision.action)
            span.set_attribute("react.next_agent_id", gateway_decision.agent_id)
            reflex_logfire.info(
                "react -> {action} agent={agent_id}",
                action=gateway_decision.action,
                agent_id=gateway_decision.agent_id,
                reason=gateway_decision.reason,
            )
            return Summon()

        return End(
            ReflexResult(
                decision=decision,
                target=target,
                result=run_result,
            )
        )

octomate.reflex.suspender

The reflex graph's deferred suspender: where a run's deferrals go — a teleport back to the graph, everything else to a human as a persisted batch.

TeleportRequest dataclass

A teleport for the graph to perform: carry the history and resume the agent in the new place. One shape for every runtime — an Inkling run defers the teleport call itself; a runtime a tool result cannot suspend is interrupted on the recorded decision and ends its turn as the same deferral — read off the deferred call by its metadata kind.

Source code in octomate/reflex/suspender.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@dataclass(frozen=True)
class TeleportRequest:
    """A teleport for the graph to perform: carry the history and resume the agent
    in the new place. One shape for every runtime — an Inkling run defers the
    `teleport` call itself; a runtime a tool result cannot suspend is interrupted
    on the recorded decision and ends its turn as the same deferral — read off
    the deferred call by its metadata kind."""

    hint: str
    # The deferred call to resolve into the resumed run.
    tool_call_id: str | None
    # Where the move goes: the chat a new thread opens in, or this conversation.
    destination: ChannelAddress
    new_thread: bool = True
    # The project the thread landed in is bound to, and the ref its workspace starts
    # from; None carries the conversation only.
    project: str | None = None
    ref: str | None = None
    # Whether the agent carries on at once where it lands.
    resume: bool = False
    # The user's next message where it lands, which the agent answers there.
    prompt: str | None = None

    @classmethod
    def of(cls, requests: DeferredToolRequests) -> TeleportRequest | None:
        """The teleport these requests carry."""
        for call in requests.calls:
            meta = requests.metadata.get(call.tool_call_id, {})
            if meta.get("kind") != TELEPORT_DEFER_KIND:
                continue
            destination = meta.get("destination")
            if not destination:
                raise ValueError("a teleport deferral names no destination")
            return cls(
                tool_call_id=call.tool_call_id,
                hint=str(meta.get("hint") or ""),
                new_thread=bool(meta.get("new_thread", True)),
                project=str(meta.get("project") or "") or None,
                ref=str(meta.get("ref") or "") or None,
                resume=bool(meta.get("resume", False)),
                # The address validates itself, being a pydantic dataclass.
                destination=ChannelAddress(**destination),
            )
        return None

hint instance-attribute 

hint: str

tool_call_id instance-attribute 

tool_call_id: str | None

destination instance-attribute 

destination: ChannelAddress

new_thread class-attribute instance-attribute 

new_thread: bool = True

project class-attribute instance-attribute 

project: str | None = None

ref class-attribute instance-attribute 

ref: str | None = None

resume class-attribute instance-attribute 

resume: bool = False

prompt class-attribute instance-attribute 

prompt: str | None = None

of classmethod 

of(
    requests: DeferredToolRequests,
) -> TeleportRequest | None

The teleport these requests carry.

Source code in octomate/reflex/suspender.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@classmethod
def of(cls, requests: DeferredToolRequests) -> TeleportRequest | None:
    """The teleport these requests carry."""
    for call in requests.calls:
        meta = requests.metadata.get(call.tool_call_id, {})
        if meta.get("kind") != TELEPORT_DEFER_KIND:
            continue
        destination = meta.get("destination")
        if not destination:
            raise ValueError("a teleport deferral names no destination")
        return cls(
            tool_call_id=call.tool_call_id,
            hint=str(meta.get("hint") or ""),
            new_thread=bool(meta.get("new_thread", True)),
            project=str(meta.get("project") or "") or None,
            ref=str(meta.get("ref") or "") or None,
            resume=bool(meta.get("resume", False)),
            # The address validates itself, being a pydantic dataclass.
            destination=ChannelAddress(**destination),
        )
    return None

ReflexSuspender dataclass

The reflex graph's DeferredSuspender: every deferral a run ends on comes through here once, and each kind goes where it is resolved — a teleport to the graph, which performs it and resumes the agent; anything else to a human, persisted as a batch whose event rides the run's stream, for the channel to present wherever it draws the run from. React builds it with the run's context; Inkling reaches it through ResolveDeferred, a runtime a tool result cannot suspend through the deferred_suspender its run was handed. A runtime that asks a human mid-run pauses here instead, and stays live.

Source code in octomate/reflex/suspender.py 
 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
@dataclass
class ReflexSuspender:
    """The reflex graph's `DeferredSuspender`: every deferral a run ends on comes
    through here once, and each kind goes where it is resolved — a `teleport` to
    the graph, which performs it and resumes the agent; anything else to a human,
    persisted as a batch whose event rides the run's stream, for the channel to
    present wherever it draws the run from. React builds it with the
    run's context; Inkling reaches it through `ResolveDeferred`, a runtime a tool
    result cannot suspend through the `deferred_suspender` its run was handed.
    A runtime that asks a human mid-run pauses here instead, and stays live.
    """

    channel: ChannelTentacle
    action_manager: DeferredActionManager
    conversation_manager: ConversationManager
    agent_tentacle_id: str
    run_name: RunName
    source_address: ChannelAddress
    target_address: ChannelAddress
    target_mode: ResponseTargetMode
    decision: SummonDecision | None
    model: str | None = None  # Selected model when a command has no routing decision.
    thread_id: UUID7 | None = None
    suspended_batch_id: UUID7 | None = field(default=None, init=False)
    # The deferred `teleport`, for the graph to perform instead of a batch.
    teleport: TeleportRequest | None = field(default=None, init=False)

    def continuation_decision(self) -> SummonDecision:
        """Preserve an actual agent run's selection for resume and teleport."""
        if self.decision is None:
            self.decision = SummonDecision(
                agent_id=self.agent_tentacle_id,
                model=self.model,
                reason="Continue the selected agent run.",
                hint="",
                brief="",
            )
        return self.decision

    async def suspend(self, requests: DeferredToolRequests) -> ActionBatchEvent | None:
        self.continuation_decision()
        teleport = TeleportRequest.of(requests)
        if teleport is not None:
            # The agent's own permission check already let the call through, so
            # nothing here asks: the graph moves it once the run ends.
            if len(requests.calls) + len(requests.approvals) > 1:
                raise RuntimeError("a teleport was deferred beside other calls")
            self.teleport = teleport
            return None
        with reflex_logfire.span(
            "suspend_for_review",
            run_name=self.run_name,
            agent_id=self.agent_tentacle_id,
            target_address=str(self.target_address),
            source_address=str(self.source_address),
        ) as span:
            batch = await self.persist(requests, response_mode="resume")
            self.suspended_batch_id = batch.id
            span.set_attribute("batch_id", str(batch.id))
            return ActionBatchEvent.from_batch(batch)

    async def pause(
        self, requests: DeferredToolRequests, *, batch_id: UUID7
    ) -> ActionBatchEvent:
        with reflex_logfire.span(
            "pause_for_review",
            run_name=self.run_name,
            agent_id=self.agent_tentacle_id,
            target_address=str(self.target_address),
        ) as span:
            batch = await self.persist(
                requests, response_mode="live", batch_id=batch_id
            )
            span.set_attribute("batch_id", str(batch.id))
            return ActionBatchEvent.from_batch(batch)

    async def persist(
        self,
        requests: DeferredToolRequests,
        *,
        response_mode: DeferredResponseMode,
        batch_id: UUID7 | None = None,
    ) -> DeferredActionBatch:
        if self.thread_id is None:
            raise ValueError("deferred review requires a thread_id")
        conversation = await self.conversation_manager.ensure(
            self.thread_id,
            agent_tentacle_id=self.agent_tentacle_id,
        )
        return await self.action_manager.create_batch(
            response_mode=response_mode,
            batch_id=batch_id,
            conversation=conversation,
            agent_tentacle_id=self.agent_tentacle_id,
            run_name=self.run_name,
            source_address=self.source_address,
            target_address=self.target_address,
            target_mode=self.target_mode,
            decision=self.decision,
            requests=requests,
        )

channel instance-attribute 

channel: ChannelTentacle

action_manager instance-attribute 

action_manager: DeferredActionManager

conversation_manager instance-attribute 

conversation_manager: ConversationManager

agent_tentacle_id instance-attribute 

agent_tentacle_id: str

run_name instance-attribute 

run_name: RunName

source_address instance-attribute 

source_address: ChannelAddress

target_address instance-attribute 

target_address: ChannelAddress

target_mode instance-attribute 

target_mode: ResponseTargetMode

decision instance-attribute 

decision: SummonDecision | None

model class-attribute instance-attribute 

model: str | None = None

thread_id class-attribute instance-attribute 

thread_id: UUID7 | None = None

suspended_batch_id class-attribute instance-attribute 

suspended_batch_id: UUID7 | None = field(
    default=None, init=False
)

teleport class-attribute instance-attribute 

teleport: TeleportRequest | None = field(
    default=None, init=False
)

continuation_decision 

continuation_decision() -> SummonDecision

Preserve an actual agent run's selection for resume and teleport.

Source code in octomate/reflex/suspender.py 
100
101
102
103
104
105
106
107
108
109
110
def continuation_decision(self) -> SummonDecision:
    """Preserve an actual agent run's selection for resume and teleport."""
    if self.decision is None:
        self.decision = SummonDecision(
            agent_id=self.agent_tentacle_id,
            model=self.model,
            reason="Continue the selected agent run.",
            hint="",
            brief="",
        )
    return self.decision

suspend async 

suspend(
    requests: DeferredToolRequests,
) -> ActionBatchEvent | None
Source code in octomate/reflex/suspender.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def suspend(self, requests: DeferredToolRequests) -> ActionBatchEvent | None:
    self.continuation_decision()
    teleport = TeleportRequest.of(requests)
    if teleport is not None:
        # The agent's own permission check already let the call through, so
        # nothing here asks: the graph moves it once the run ends.
        if len(requests.calls) + len(requests.approvals) > 1:
            raise RuntimeError("a teleport was deferred beside other calls")
        self.teleport = teleport
        return None
    with reflex_logfire.span(
        "suspend_for_review",
        run_name=self.run_name,
        agent_id=self.agent_tentacle_id,
        target_address=str(self.target_address),
        source_address=str(self.source_address),
    ) as span:
        batch = await self.persist(requests, response_mode="resume")
        self.suspended_batch_id = batch.id
        span.set_attribute("batch_id", str(batch.id))
        return ActionBatchEvent.from_batch(batch)

pause async 

pause(
    requests: DeferredToolRequests, *, batch_id: UUID7
) -> ActionBatchEvent
Source code in octomate/reflex/suspender.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def pause(
    self, requests: DeferredToolRequests, *, batch_id: UUID7
) -> ActionBatchEvent:
    with reflex_logfire.span(
        "pause_for_review",
        run_name=self.run_name,
        agent_id=self.agent_tentacle_id,
        target_address=str(self.target_address),
    ) as span:
        batch = await self.persist(
            requests, response_mode="live", batch_id=batch_id
        )
        span.set_attribute("batch_id", str(batch.id))
        return ActionBatchEvent.from_batch(batch)

persist async 

persist(
    requests: DeferredToolRequests,
    *,
    response_mode: DeferredResponseMode,
    batch_id: UUID7 | None = None,
) -> DeferredActionBatch
Source code in octomate/reflex/suspender.py 
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
async def persist(
    self,
    requests: DeferredToolRequests,
    *,
    response_mode: DeferredResponseMode,
    batch_id: UUID7 | None = None,
) -> DeferredActionBatch:
    if self.thread_id is None:
        raise ValueError("deferred review requires a thread_id")
    conversation = await self.conversation_manager.ensure(
        self.thread_id,
        agent_tentacle_id=self.agent_tentacle_id,
    )
    return await self.action_manager.create_batch(
        response_mode=response_mode,
        batch_id=batch_id,
        conversation=conversation,
        agent_tentacle_id=self.agent_tentacle_id,
        run_name=self.run_name,
        source_address=self.source_address,
        target_address=self.target_address,
        target_mode=self.target_mode,
        decision=self.decision,
        requests=requests,
    )

octomate.reflex.crossing

Create a channel-owned destination and announce the move at its origin.

logger module-attribute 

logger = logging.getLogger(__name__)

open_crossing async 

open_crossing(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    destination: ChannelAddress,
    source_address: ChannelAddress,
    hint_text: str,
    agent_tentacle_id: str,
) -> ChannelAddress | None

Ask the destination channel to create an isolated thread.

The existing opener is recorded with its requesting user so the destination remains visible in their console. No fallback may reuse an existing chat.

Source code in octomate/reflex/crossing.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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
async def open_crossing(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    destination: ChannelAddress,
    source_address: ChannelAddress,
    hint_text: str,
    agent_tentacle_id: str,
) -> ChannelAddress | None:
    """Ask the destination channel to create an isolated thread.

    The existing opener is recorded with its requesting user so the destination
    remains visible in their console. No fallback may reuse an existing chat.
    """
    channel = ctx.deps.channel(destination.channel_tentacle_id)
    profile = ctx.state.user_profile
    if profile is not None:
        if profile.channel_tentacle_id != channel.id:
            linked = await ctx.deps.thread_manager.users.profile(
                channel.id, destination.user_id
            )
            if linked is not None and linked.user_id == profile.user_id:
                profile = linked
            elif channel.thread_user_id(profile) != destination.user_id:
                raise ValueError(
                    "The destination profile is no longer linked to this user."
                )
    try:
        opened = await channel.start_thread(destination, hint_text)
    except Exception:
        logger.warning(
            "Channel %s could not create the destination thread",
            channel.id,
            exc_info=True,
        )
        return None
    if destination == source_address:
        return opened
    if profile is not None:
        if profile.channel_tentacle_id != channel.id:
            profile = UserProfile(
                channel_tentacle_id=channel.id,
                channel_user_id=opened.user_id,
                user_id=profile.user_id,
                name=profile.name,
                nickname=profile.nickname,
            )
        ctx.state.user_profile = profile
        # Attribute the existing opener so the requesting user can find the destination.
        # Native teleport's transcript fork already publishes its attributed fork notice.
        native_fork = (
            ctx.state.run_name == "teleport"
            and ctx.state.thread is not None
            and ctx.state.thread.kind == "native_thread"
        )
        if not native_fork:
            await ctx.deps.thread_manager.record_outbound(
                opened,
                agent_tentacle_id=agent_tentacle_id,
                sender=ctx.state.user_profile,
                actor_kind="system",
                segments=[MarkdownSegment(data={"text": hint_text})],
            )
    return opened

announce_crossing async 

announce_crossing(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    destination: ChannelAddress,
    opened: ChannelAddress,
    source_address: ChannelAddress,
    hint_text: str,
    agent_tentacle_id: str,
) -> None

Leave the move's line in the chat it left, and say where it landed.

Called once the history is carried: a fork shows the source's ledger as it stands, and the line is the source's own, not part of what moved.

Source code in octomate/reflex/crossing.py 
 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
async def announce_crossing(
    ctx: GraphRunContext[ReflexState, ReflexDeps],
    destination: ChannelAddress,
    opened: ChannelAddress,
    source_address: ChannelAddress,
    hint_text: str,
    agent_tentacle_id: str,
) -> None:
    """Leave the move's line in the chat it left, and say where it landed.

    Called once the history is carried: a fork shows the source's ledger as it
    stands, and the line is the source's own, not part of what moved.
    """
    if destination == source_address:
        await ctx.deps.record_move(
            source_address,
            hint_text,
            agent_tentacle_id=agent_tentacle_id,
            platform_message_id=opened.channel_thread_id,
        )
        # The thread hangs from its opener there, which is the move's line.
        await ctx.deps.impulses.send(
            Impulse(GatewayEvent(action="teleport", destination=opened), source_address)
        )
        return
    # A native session's pseudo-channel posts no line, and has no one to post it as.
    line = (
        await ctx.deps.record_move(
            source_address, hint_text, agent_tentacle_id=agent_tentacle_id
        )
        if source_address.channel_tentacle_id in ctx.deps.channels
        else None
    )
    await ctx.deps.impulses.send(
        Impulse(
            GatewayEvent(
                action="teleport",
                destination=opened,
                announcement=hint_text,
                thread_message_id=line.id if line is not None else None,
            ),
            source_address,
        )
    )