Skip to content

Nodes

octomate.reflex.nodes.awake

The graph's entry node: reads the signal that woke it and picks the first transition.

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()

octomate.reflex.nodes.command

Execute explicit command intent on its selected conversation, without routing text.

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)

octomate.reflex.nodes.route

The node that decides which agent takes an inbound message: the thread's owner, the flat-thread agent, or the channel's entry agent.

logger module-attribute 

logger = logging.getLogger(__name__)

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()

octomate.reflex.nodes.react

The node that runs the agent: records the handoff, drives the run onto the channel's timeline, and acts on whatever decision or deferral the run left.

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)

octomate.reflex.nodes.summon

The node that performs a summon: hands the conversation, where it is, to the agent the decision names, then re-enters React there.

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()

octomate.reflex.nodes.scheme

The node that performs a scheme: hands this turn's brief to a person's direct messages.

logger module-attribute 

logger = logging.getLogger(__name__)

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()

octomate.reflex.nodes.teleport

The node that performs a teleport: carries the running agent's history somewhere else and resumes it there.

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

octomate.reflex.nodes.resume_deferred

The node that continues a run after a human answered its deferred batch.

logger module-attribute 

logger = logging.getLogger(__name__)

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)