Skip to content

The graph’s vocabulary

octomate.schemas.awakes

AwakeSignal 

AwakeSignal = (
    UserMessageSignal
    | DeferredActionBatchResponse
    | CommandSignal
    | NativeGatewaySignal
    | DrivenGatewaySignal
)

UserMessageSignal dataclass

Source code in octomate/schemas/awakes.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@dataclass(frozen=True)
class UserMessageSignal:
    messages: list[MessageEvent]
    trigger_thread_message_id: UUID7 | None = None

    def __bool__(self) -> bool:
        return bool(self.messages)

    @cached_property
    def address(self) -> ChannelAddress:
        if not self.messages:
            raise ValueError("empty user message signal has no conversation address")
        last_event = self.messages[-1]
        return ChannelAddress(
            channel_tentacle_id=last_event.tentacle_id,
            chat_type=last_event.chat_type,
            chat_id=last_event.chat_id,
            user_id=last_event.user_id,
            channel_thread_id=last_event.channel_thread_id,
            shared=last_event.shared,
        )

messages instance-attribute 

messages: list[MessageEvent]

trigger_thread_message_id class-attribute instance-attribute 

trigger_thread_message_id: UUID7 | None = None

address cached property 

address: ChannelAddress

DeferredActionBatchResponse pydantic-model

Bases: BaseModel

Fields:

Source code in octomate/schemas/awakes.py 
39
40
41
42
43
44
class DeferredActionBatchResponse(BaseModel):
    batch_id: UUID7
    responder_id: str = ""
    answers: dict[UUID7, QuestionAnswer] = Field(default_factory=dict)
    approvals: dict[UUID7, bool] = Field(default_factory=dict)
    allow_session: bool = False

batch_id pydantic-field 

batch_id: UUID7

responder_id pydantic-field 

responder_id: str = ''

answers pydantic-field 

answers: dict[UUID7, QuestionAnswer]

approvals pydantic-field 

approvals: dict[UUID7, bool]

allow_session pydantic-field 

allow_session: bool = False

NativeGatewaySignal dataclass

A native session's own spell, kicked as its own turn.

A driven turn's decision is read off its Octomate session when the run ends; a native session has no run in the graph, so the served spell hands its validated decision straight to the graph instead.

Source code in octomate/schemas/awakes.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass(frozen=True)
class NativeGatewaySignal:
    """A native session's own spell, kicked as its own turn.

    A driven turn's decision is read off its Octomate session when the run ends; a
    native session has no run in the graph, so the served spell hands its
    validated decision straight to the graph instead.
    """

    # A native session cannot be summoned: there is nothing here to hand over.
    decision: SchemeDecision | TeleportDecision
    # The native pseudo-channel the handoff is attributed to — its ledger `from` side.
    agent_id: str
    # The registry profile the native id is linked to; None when nobody claims it.
    user_profile: UserProfile | None
    # Where the spell came from: the session's own thread for a teleport, else a
    # pseudo-address on the native id, for the crossing announce to speak to and
    # any failure to land back against.
    source: ChannelAddress | None = None

decision instance-attribute 

decision: SchemeDecision | TeleportDecision

agent_id instance-attribute 

agent_id: str

user_profile instance-attribute 

user_profile: UserProfile | None

source class-attribute instance-attribute 

source: ChannelAddress | None = None

CommandSignal dataclass

Explicit command intent on an authenticated, resolved conversation surface.

Source code in octomate/schemas/awakes.py 
68
69
70
71
72
73
74
@dataclass(frozen=True)
class CommandSignal:
    """Explicit command intent on an authenticated, resolved conversation surface."""

    context: CommandContext
    invocation: CommandInvocation
    delivery_id: str  # Stable across retries of the same explicit request.

context instance-attribute 

context: CommandContext

invocation instance-attribute 

invocation: CommandInvocation

delivery_id instance-attribute 

delivery_id: str

DrivenGatewaySignal dataclass

An authenticated, validated operation on an existing thread.

Source code in octomate/schemas/awakes.py 
77
78
79
80
81
82
83
84
85
@dataclass(frozen=True)
class DrivenGatewaySignal:
    """An authenticated, validated operation on an existing thread."""

    thread_id: UUID7
    decision: SummonDecision | TeleportDecision
    agent_id: str
    user_profile: UserProfile
    source: ChannelAddress

thread_id instance-attribute 

thread_id: UUID7

decision instance-attribute 

decision: SummonDecision | TeleportDecision

agent_id instance-attribute 

agent_id: str

user_profile instance-attribute 

user_profile: UserProfile

source instance-attribute 

source: ChannelAddress

StillAwake

Bases: RuntimeError

A kick refused because a reflex is still awake on the thread it was asked on: a thread holds one at a time.

Source code in octomate/schemas/awakes.py 
 97
 98
 99
100
101
102
class StillAwake(RuntimeError):
    """A kick refused because a reflex is still awake on the thread it was asked
    on: a thread holds one at a time."""

    def __init__(self) -> None:
        super().__init__("A turn is still running in this thread.")

octomate.schemas.triage

Gateway decisions a run leaves.

ResponseTargetMode module-attribute 

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

RunName module-attribute 

RunName = Literal[
    "react", "summon", "teleport", "resume", "command"
]

InspectFacet module-attribute 

InspectFacet = Literal["routes", "destinations", "projects"]

TELEPORT_DEFER_KIND module-attribute 

TELEPORT_DEFER_KIND = 'teleport'

HERE_TARGET module-attribute 

HERE_TARGET = HereTarget()

DIRECT_TARGET module-attribute 

DIRECT_TARGET = DirectTarget()

SchemeTarget 

SchemeTarget = DirectTarget

SendTarget 

SendTarget = Annotated[
    HereTarget | DirectTarget, Field(discriminator="kind")
]

GatewayDecision 

GatewayDecision = Annotated[
    SummonDecision | SchemeDecision | TeleportDecision,
    Field(discriminator="action"),
]

GatewayAction 

GatewayAction = Literal['summon', 'scheme', 'teleport']

AgentRouteKey

Bases: NamedTuple

An agent and a model, or None for the agent's native default.

Source code in octomate/schemas/triage.py 
37
38
39
40
41
class AgentRouteKey(NamedTuple):
    """An agent and a model, or None for the agent's native default."""

    agent_id: str
    model: AgentRouteModelName | None  # None preserves the harness's native default.

agent_id instance-attribute 

agent_id: str

model instance-attribute 

model: AgentRouteModelName | None

SpellTarget pydantic-model

Bases: BaseModel

Base of the places a spell's destination argument can name.

A variant per place rather than one free string, so a tool definition says what it accepts instead of documenting it in prose. Channel IDs are runtime values: the schema stays stable while the tool validates reachability and access.

Config:

  • frozen: True
Source code in octomate/schemas/triage.py 
44
45
46
47
48
49
50
51
52
class SpellTarget(BaseModel):
    """Base of the places a spell's `destination` argument can name.

    A variant per place rather than one free string, so a tool definition says what
    it accepts instead of documenting it in prose. Channel IDs are runtime values:
    the schema stays stable while the tool validates reachability and access.
    """

    model_config = ConfigDict(frozen=True)

model_config class-attribute instance-attribute 

model_config = ConfigDict(frozen=True)

HereTarget pydantic-model

Bases: SpellTarget

This conversation, as it stands.

Fields:

  • kind (Literal['here'])
Source code in octomate/schemas/triage.py 
55
56
57
58
class HereTarget(SpellTarget):
    """This conversation, as it stands."""

    kind: Literal["here"] = "here"

kind pydantic-field 

kind: Literal['here'] = 'here'

DirectTarget pydantic-model

Bases: SpellTarget

The asking user's direct messages on a connected channel.

Fields:

Source code in octomate/schemas/triage.py 
61
62
63
64
65
66
67
68
69
class DirectTarget(SpellTarget):
    """The asking user's direct messages on a connected channel."""

    kind: Literal["dm"] = "dm"

    channel: str | None = Field(
        default=None,
        description="The connected channel ID; omit for this conversation's channel.",
    )

kind pydantic-field 

kind: Literal['dm'] = 'dm'

channel pydantic-field 

channel: str | None = None

The connected channel ID; omit for this conversation's channel.

SummonDecision pydantic-model

Bases: BaseModel

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

Fields:

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

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

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

action pydantic-field 

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

reason pydantic-field 

reason: str

agent_id pydantic-field 

agent_id: str

model pydantic-field 

model: AgentRouteModelName | None

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

effort pydantic-field 

effort: str | None = None

hint pydantic-field 

hint: str

brief pydantic-field 

brief: str

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

key property 

key: AgentRouteKey

SchemeDecision pydantic-model

Bases: BaseModel

Take this turn to the asking user's DM, from a brief.

No agent is named: whoever already handles that user's DM picks the work up, so a group can never point someone's private assistant at an agent it chose. The receiver is resolved against the DM's own thread, which is why this carries a brief rather than a route.

Fields:

Source code in octomate/schemas/triage.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class SchemeDecision(BaseModel):
    """Take this turn to the asking user's DM, from a brief.

    No agent is named: whoever already handles that user's DM picks the work up, so a
    group can never point someone's private assistant at an agent it chose. The
    receiver is resolved against the DM's own thread, which is why this carries a brief
    rather than a route.
    """

    action: Literal["scheme"] = "scheme"
    hint: str = Field(
        description="The line that opens the conversation over there. Nothing is "
        "posted where the request came from — the run's own reply closes that out."
    )
    brief: str = Field(max_length=8_000)
    destination: ChannelAddress = Field(
        description="Which direct messages, resolved by the gateway using the "
        "requesting user's linked identity, never a model-supplied user ID."
    )

action pydantic-field 

action: Literal['scheme'] = 'scheme'

hint pydantic-field 

hint: str

The line that opens the conversation over there. Nothing is posted where the request came from — the run's own reply closes that out.

brief pydantic-field 

brief: str

destination pydantic-field 

destination: ChannelAddress

Which direct messages, resolved by the gateway using the requesting user's linked identity, never a model-supplied user ID.

TeleportDecision pydantic-model

Bases: BaseModel

The same agent continues somewhere else, its history with it; Reflex performs the move. A run ends on it — Inkling by deferring the call, a runtime a tool result cannot suspend by being interrupted on the recorded decision and ending its turn as the same deferral — and the graph resumes the agent in the new place: a sub-thread, a crossing, or this very thread when the move is only into a project's workspace.

Fields:

Source code in octomate/schemas/triage.py 
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
class TeleportDecision(BaseModel):
    """The same agent continues somewhere else, its history with it; Reflex
    performs the move. A run ends on it — Inkling by deferring the call, a runtime
    a tool result cannot suspend by being interrupted on the recorded decision and
    ending its turn as the same deferral — and the graph resumes the agent in the
    new place: a sub-thread, a crossing, or this very thread when the move is only
    into a project's workspace.
    """

    action: Literal["teleport"] = "teleport"
    agent_id: str = Field(
        description="The driven agent selected to resume the history at the destination."
    )
    hint: str = Field(description="The short, user-facing thread-starter message.")
    destination: ChannelAddress = Field(
        description="Where the move goes: the chat a new thread opens in, or this "
        "conversation when the move only binds it to a project. A spell that names "
        "none means the current conversation, resolved before the decision exists."
    )
    new_thread: bool = Field(
        default=True,
        description="Create a thread at the address; false only binds the current thread to a project.",
    )
    project: str | None = Field(
        default=None,
        description="The project the thread landed in is bound to, and whose "
        "workspace the agent resumes in; None carries the conversation only.",
    )
    ref: str | None = Field(
        default=None,
        description="The branch, tag or commit that workspace starts from; None "
        "for the project's default branch.",
    )
    resume: bool = Field(
        default=False,
        description="Whether the agent carries on at once where it lands; false "
        "leaves the conversation there for the next message.",
    )
    prompt: str | None = Field(
        default=None,
        description="The user's next message, sent where the move lands; only the "
        "console's own Teleport has one.",
    )

    def metadata(self) -> JsonObject:
        """What the deferral carries, as plain values: enough for the graph to
        rebuild this decision at its boundary."""
        return {
            "kind": TELEPORT_DEFER_KIND,
            "hint": self.hint,
            "destination": asdict(self.destination),
            "new_thread": self.new_thread,
            "project": self.project or "",
            "ref": self.ref or "",
            "resume": self.resume,
        }

    def deferral(self, tool_call_id: str) -> DeferredToolRequests:
        """This move as the deferral the graph performs — what an interrupted
        runtime's turn ends with, shaped as Inkling's own `CallDeferred` is."""
        return DeferredToolRequests(
            calls=[
                ToolCallPart(
                    tool_name=GatewayTool.TELEPORT,
                    args={
                        "hint": self.hint,
                        "destination": asdict(self.destination),
                        "new_thread": self.new_thread,
                        "project": self.project,
                        "ref": self.ref,
                        "resume": self.resume,
                    },
                    tool_call_id=tool_call_id,
                )
            ],
            metadata={tool_call_id: self.metadata()},
        )

action pydantic-field 

action: Literal['teleport'] = 'teleport'

agent_id pydantic-field 

agent_id: str

The driven agent selected to resume the history at the destination.

hint pydantic-field 

hint: str

The short, user-facing thread-starter message.

destination pydantic-field 

destination: ChannelAddress

Where the move goes: the chat a new thread opens in, or this conversation when the move only binds it to a project. A spell that names none means the current conversation, resolved before the decision exists.

new_thread pydantic-field 

new_thread: bool = True

Create a thread at the address; false only binds the current thread to a project.

project pydantic-field 

project: str | None = None

The project the thread landed in is bound to, and whose workspace the agent resumes in; None carries the conversation only.

ref pydantic-field 

ref: str | None = None

The branch, tag or commit that workspace starts from; None for the project's default branch.

resume pydantic-field 

resume: bool = False

Whether the agent carries on at once where it lands; false leaves the conversation there for the next message.

prompt pydantic-field 

prompt: str | None = None

The user's next message, sent where the move lands; only the console's own Teleport has one.

metadata 

metadata() -> JsonObject

What the deferral carries, as plain values: enough for the graph to rebuild this decision at its boundary.

Source code in octomate/schemas/triage.py 
169
170
171
172
173
174
175
176
177
178
179
180
def metadata(self) -> JsonObject:
    """What the deferral carries, as plain values: enough for the graph to
    rebuild this decision at its boundary."""
    return {
        "kind": TELEPORT_DEFER_KIND,
        "hint": self.hint,
        "destination": asdict(self.destination),
        "new_thread": self.new_thread,
        "project": self.project or "",
        "ref": self.ref or "",
        "resume": self.resume,
    }

deferral 

deferral(tool_call_id: str) -> DeferredToolRequests

This move as the deferral the graph performs — what an interrupted runtime's turn ends with, shaped as Inkling's own CallDeferred is.

Source code in octomate/schemas/triage.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def deferral(self, tool_call_id: str) -> DeferredToolRequests:
    """This move as the deferral the graph performs — what an interrupted
    runtime's turn ends with, shaped as Inkling's own `CallDeferred` is."""
    return DeferredToolRequests(
        calls=[
            ToolCallPart(
                tool_name=GatewayTool.TELEPORT,
                args={
                    "hint": self.hint,
                    "destination": asdict(self.destination),
                    "new_thread": self.new_thread,
                    "project": self.project,
                    "ref": self.ref,
                    "resume": self.resume,
                },
                tool_call_id=tool_call_id,
            )
        ],
        metadata={tool_call_id: self.metadata()},
    )

AgentRoute dataclass

A summonable (agent, model) pair and the claim it advertises. Agents advertise; the caller requests — the claim publishes the space this route supports, and a caller picks a point in it.

Source code in octomate/schemas/triage.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
@dataclass(frozen=True)
class AgentRoute:
    """A summonable (agent, model) pair and the claim it advertises. Agents
    advertise; the caller requests — the claim publishes the space this route
    supports, and a caller picks a point in it."""

    agent_id: str
    model: AgentRouteModelName
    claim: Claim

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

    def __str__(self) -> str:
        return f"- agent_id={self.agent_id}, model={self.model!r}: {self.claim}"

agent_id instance-attribute 

agent_id: str

model instance-attribute 

model: AgentRouteModelName

claim instance-attribute 

claim: Claim

key property 

key: AgentRouteKey

ProjectSummary dataclass

One registered project, as a model choosing between them needs it.

Source code in octomate/schemas/triage.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
@dataclass(frozen=True)
class ProjectSummary:
    """One registered project, as a model choosing between them needs it."""

    # Deliberately not the root: which absolute path a project is on the server is
    # the operator's business, and naming it in a chat thread is how it leaks.
    name: str
    description: str | None

    def __str__(self) -> str:
        if self.description is None:
            return f"- {self.name}"
        return f"- {self.name}: {self.description}"

name instance-attribute 

name: str

description instance-attribute 

description: str | None

octomate.schemas.deferred

Deferred actions: the questions and approvals a run parks for a human, and the batch they are presented and resolved as.

MAX_QUESTION_CHOICES module-attribute 

MAX_QUESTION_CHOICES = 5

DeferredActionVariantAdapter module-attribute 

DeferredActionVariantAdapter = TypeAdapter(
    DeferredActionVariant
)

DeferredActionCollection module-attribute 

DeferredActionCollection = TypeAdapter(
    Annotated[
        list[DeferredActionVariant],
        BeforeValidator(from_deferred_requests),
    ]
)

QuestionAnswer 

QuestionAnswer = str | list[str]

DeferredActionPayload 

DeferredActionPayload = (
    DeferredQuestionPayload | DeferredApprovalPayload
)

DeferredActionCollectionInput 

DeferredActionCollectionInput = (
    DeferredToolRequests
    | list[DeferredActionPayload | JsonObject]
)

DeferredQuestionResult 

DeferredQuestionResult = QuestionAnswer | None

DeferredApprovalResult 

DeferredApprovalResult = bool | None

DeferredActionVariant 

DeferredActionVariant = Annotated[
    DeferredQuestion | DeferredApproval,
    Field(discriminator="kind"),
]

QuestionRequest

Bases: TypedDict

One question's arguments: the text, optional choices, and a hint.

Source code in octomate/schemas/deferred.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
class QuestionRequest(TypedDict):
    """One question's arguments: the text, optional choices, and a hint."""

    question: str
    choices: NotRequired[
        Annotated[
            list[str] | None,
            Field(
                max_length=MAX_QUESTION_CHOICES,
                description=(
                    f"Up to {MAX_QUESTION_CHOICES} suggested choices. The user can "
                    "still answer with text."
                ),
            ),
        ]
    ]
    hint: NotRequired[str]
    multi_select: NotRequired[
        Annotated[
            bool,
            Field(
                description="Whether several choices may be picked together, in "
                "which case the answer lists the picks."
            ),
        ]
    ]

question instance-attribute 

question: str

choices instance-attribute 

choices: NotRequired[
    Annotated[
        list[str] | None,
        Field(
            max_length=MAX_QUESTION_CHOICES,
            description=f"Up to {MAX_QUESTION_CHOICES} suggested choices. The user can still answer with text.",
        ),
    ]
]

hint instance-attribute 

hint: NotRequired[str]

multi_select instance-attribute 

multi_select: NotRequired[
    Annotated[
        bool,
        Field(
            description="Whether several choices may be picked together, in which case the answer lists the picks."
        ),
    ]
]

ApprovalRequest pydantic-model

Bases: BaseModel

What an approval card shows: the tool, its arguments, a title and a description.

Fields:

Source code in octomate/schemas/deferred.py 
87
88
89
90
91
92
93
94
class ApprovalRequest(BaseModel):
    """What an approval card shows: the tool, its arguments, a title and a
    description."""

    tool_name: str
    args: JsonObject = Field(default_factory=dict)
    title: str = "Permission Required"
    description: str = ""

tool_name pydantic-field 

tool_name: str

args pydantic-field 

args: JsonObject

title pydantic-field 

title: str = 'Permission Required'

description pydantic-field 

description: str = ''

ApprovalRequestPayload

Bases: TypedDict

Source code in octomate/schemas/deferred.py 
 97
 98
 99
100
class ApprovalRequestPayload(TypedDict):
    tool_name: str
    args: JsonObject
    description: str

tool_name instance-attribute 

tool_name: str

args instance-attribute 

args: JsonObject

description instance-attribute 

description: str

DeferredQuestionPayload

Bases: TypedDict

Source code in octomate/schemas/deferred.py 
103
104
105
106
107
108
109
class DeferredQuestionPayload(TypedDict):
    kind: Literal["question"]
    tool_name: str
    tool_call_id: str
    position: int
    args: QuestionRequest
    metadata: JsonObject

kind instance-attribute 

kind: Literal['question']

tool_name instance-attribute 

tool_name: str

tool_call_id instance-attribute 

tool_call_id: str

position instance-attribute 

position: int

args instance-attribute 

args: QuestionRequest

metadata instance-attribute 

metadata: JsonObject

DeferredApprovalPayload

Bases: TypedDict

Source code in octomate/schemas/deferred.py 
112
113
114
115
116
117
class DeferredApprovalPayload(TypedDict):
    kind: Literal["approval"]
    tool_name: str
    tool_call_id: str
    args: ApprovalRequest | ApprovalRequestPayload
    metadata: JsonObject

kind instance-attribute 

kind: Literal['approval']

tool_name instance-attribute 

tool_name: str

tool_call_id instance-attribute 

tool_call_id: str

args instance-attribute 

args: ApprovalRequest | ApprovalRequestPayload

metadata instance-attribute 

metadata: JsonObject

DeferredAction

Bases: BaseTransmuter

One pending action in a batch — a question or an approval — and how it resolved.

Source code in octomate/schemas/deferred.py 
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
@sqlalchemy_materia.bless(deferred_models.DeferredAction)
class DeferredAction(BaseTransmuter):
    """One pending action in a batch — a question or an approval — and how it
    resolved."""

    model_config = ConfigDict(
        from_attributes=True,
        populate_by_name=True,
    )

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    batch_id: UUID7 | None = None
    status: DeferredActionStatus = "pending"
    tool_name: str
    tool_call_id: str
    position: int = 0
    metadata: JsonObject = Field(
        default_factory=dict,
        validation_alias=AliasChoices("meta", "metadata"),
    )
    result: DeferredQuestionResult | DeferredApprovalResult = None
    platform_message_id: str | None = None
    responder_id: str | None = None
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    resolved_at: AwareDatetime | None = None

    batch: Relation[DeferredActionBatch] = Relationship()

    @property
    def resolved(self) -> bool:
        return self.status in {"answered", "approved", "denied", "expired", "failed"}

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    from_attributes=True, populate_by_name=True
)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

batch_id class-attribute instance-attribute 

batch_id: UUID7 | None = None

status class-attribute instance-attribute 

status: DeferredActionStatus = 'pending'

tool_name instance-attribute 

tool_name: str

tool_call_id instance-attribute 

tool_call_id: str

position class-attribute instance-attribute 

position: int = 0

metadata class-attribute instance-attribute 

metadata: JsonObject = Field(
    default_factory=dict,
    validation_alias=AliasChoices("meta", "metadata"),
)

result class-attribute instance-attribute 

result: DeferredQuestionResult | DeferredApprovalResult = (
    None
)

platform_message_id class-attribute instance-attribute 

platform_message_id: str | None = None

responder_id class-attribute instance-attribute 

responder_id: str | None = None

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

updated_at class-attribute instance-attribute 

updated_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

resolved_at class-attribute instance-attribute 

resolved_at: AwareDatetime | None = None

batch class-attribute instance-attribute 

batch: Relation[DeferredActionBatch] = Relationship()

resolved property 

resolved: bool

DeferredQuestion

Bases: DeferredAction

A question the run asked, answered with text.

Source code in octomate/schemas/deferred.py 
166
167
168
169
170
171
172
173
174
175
176
177
@sqlalchemy_materia.bless(deferred_models.DeferredQuestionAction)
class DeferredQuestion(DeferredAction):
    """A question the run asked, answered with text."""

    kind: Literal["question"] = "question"
    args: QuestionRequest

    def __lt__(self, other: DeferredQuestion) -> bool:
        return self.position < other.position

    def __gt__(self, other: DeferredQuestion) -> bool:
        return self.position > other.position

kind class-attribute instance-attribute 

kind: Literal['question'] = 'question'

args instance-attribute 

args: QuestionRequest

DeferredApproval

Bases: DeferredAction

A tool call awaiting approval, resolved to allow or deny.

Source code in octomate/schemas/deferred.py 
180
181
182
183
184
185
@sqlalchemy_materia.bless(deferred_models.DeferredApprovalAction)
class DeferredApproval(DeferredAction):
    """A tool call awaiting approval, resolved to allow or deny."""

    kind: Literal["approval"] = "approval"
    args: ApprovalRequest

kind class-attribute instance-attribute 

kind: Literal['approval'] = 'approval'

args instance-attribute 

args: ApprovalRequest

DeferredActionBatch

Bases: BaseTransmuter

A run's deferred requests as one unit: persisted, presented, and rebuilt into results once every action is resolved.

Source code in octomate/schemas/deferred.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
@sqlalchemy_materia.bless(deferred_models.DeferredActionBatch)
class DeferredActionBatch(BaseTransmuter):
    """A run's deferred requests as one unit: persisted, presented, and rebuilt
    into results once every action is resolved."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    conversation_id: UUID7
    agent_tentacle_id: str
    run_name: str | None = "react"
    status: DeferredBatchStatus = "pending"
    response_mode: DeferredResponseMode = Field(
        description="Whether a reply answers a live request or resumes a suspended run."
    )
    source_address: ChannelAddress
    target_address: ChannelAddress
    target_mode: ResponseTargetMode = "main"
    decision: SummonDecision | None = None
    requests: DeferredToolRequests
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    completed_at: AwareDatetime | None = None

    questions: RelationCollection[DeferredQuestion] = Relationships()
    approvals: RelationCollection[DeferredApproval] = Relationships()

    @property
    def completed(self) -> bool:
        return all(action.resolved for action in self.questions) and all(
            action.resolved for action in self.approvals
        )

    def build_results(self) -> DeferredToolResults:
        results = DeferredToolResults()
        question_actions: dict[str, list[DeferredQuestion]] = {}
        for action in self.questions:
            question_actions.setdefault(action.tool_call_id, []).append(action)
            if action.metadata:
                results.metadata[action.tool_call_id] = action.metadata
        for action in self.approvals:
            results.approvals[action.tool_call_id] = bool(action.result)
            if action.metadata:
                results.metadata[action.tool_call_id] = action.metadata
        for tool_call_id, actions in question_actions.items():
            results.calls[tool_call_id] = [
                "" if action.result is None else action.result
                for action in sorted(actions)
            ]
        return results

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

conversation_id instance-attribute 

conversation_id: UUID7

agent_tentacle_id instance-attribute 

agent_tentacle_id: str

run_name class-attribute instance-attribute 

run_name: str | None = 'react'

status class-attribute instance-attribute 

status: DeferredBatchStatus = 'pending'

response_mode class-attribute instance-attribute 

response_mode: DeferredResponseMode = Field(
    description="Whether a reply answers a live request or resumes a suspended run."
)

source_address instance-attribute 

source_address: ChannelAddress

target_address instance-attribute 

target_address: ChannelAddress

target_mode class-attribute instance-attribute 

target_mode: ResponseTargetMode = 'main'

decision class-attribute instance-attribute 

decision: SummonDecision | None = None

requests instance-attribute 

requests: DeferredToolRequests

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

updated_at class-attribute instance-attribute 

updated_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

completed_at class-attribute instance-attribute 

completed_at: AwareDatetime | None = None

questions class-attribute instance-attribute 

questions: RelationCollection[DeferredQuestion] = (
    Relationships()
)

approvals class-attribute instance-attribute 

approvals: RelationCollection[DeferredApproval] = (
    Relationships()
)

completed property 

completed: bool

build_results 

build_results() -> DeferredToolResults
Source code in octomate/schemas/deferred.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def build_results(self) -> DeferredToolResults:
    results = DeferredToolResults()
    question_actions: dict[str, list[DeferredQuestion]] = {}
    for action in self.questions:
        question_actions.setdefault(action.tool_call_id, []).append(action)
        if action.metadata:
            results.metadata[action.tool_call_id] = action.metadata
    for action in self.approvals:
        results.approvals[action.tool_call_id] = bool(action.result)
        if action.metadata:
            results.metadata[action.tool_call_id] = action.metadata
    for tool_call_id, actions in question_actions.items():
        results.calls[tool_call_id] = [
            "" if action.result is None else action.result
            for action in sorted(actions)
        ]
    return results

from_deferred_requests 

from_deferred_requests(
    request: DeferredActionCollectionInput,
) -> DeferredActionCollectionInput
Source code in octomate/schemas/deferred.py 
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
def from_deferred_requests(
    request: DeferredActionCollectionInput,
) -> DeferredActionCollectionInput:
    if not isinstance(request, DeferredToolRequests):
        return request
    action_payloads: list[DeferredActionPayload | JsonObject] = []
    for call in request.calls:
        metadata = cast(JsonObject, request.metadata.get(call.tool_call_id, {}) or {})
        args = cast(JsonObject, call.args_as_dict())
        questions_value = args.get("questions") or []
        if not isinstance(questions_value, list):
            raise ValueError(f"deferred call {call.tool_call_id!r} has no questions")
        questions = cast(list[QuestionRequest], questions_value)
        if not questions:
            raise ValueError(f"deferred call {call.tool_call_id!r} has no questions")
        action_payloads.extend(
            DeferredQuestionPayload(
                kind="question",
                tool_name=call.tool_name,
                tool_call_id=call.tool_call_id,
                position=position,
                args=question,
                metadata=metadata,
            )
            for position, question in enumerate(questions)
        )
    for call in request.approvals:
        metadata = cast(JsonObject, request.metadata.get(call.tool_call_id, {}) or {})
        action_payloads.append(
            DeferredApprovalPayload(
                kind="approval",
                tool_name=call.tool_name,
                tool_call_id=call.tool_call_id,
                args=ApprovalRequestPayload(
                    tool_name=call.tool_name,
                    args=cast(JsonObject, call.args_as_dict()),
                    # What whoever deferred the call wants the card to say about it.
                    description=str(metadata.get("description") or ""),
                ),
                metadata=metadata,
            )
        )
    return action_payloads

octomate.schemas.agent

What the host reports about a registered agent.

AgentInfo pydantic-model

Bases: BaseModel

A registered agent as the host reports it: its routes, permission modes and live sessions.

Fields:

Source code in octomate/schemas/agent.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class AgentInfo(BaseModel):
    """A registered agent as the host reports it: its routes, permission modes and
    live sessions."""

    id: str = Field(description="The registered agent tentacle's id.")
    description: str = Field(description="The agent's own capability blurb.")
    gateway: bool = Field(
        description="Whether this agent's driven turns offer the routing and "
        "project spells; the agent's own flag is the only switch."
    )
    default_model: AgentRouteModelName | None = Field(
        description="The catalog model used to inspect a new conversation's default "
        "capabilities; null when the harness has not reported its default. "
        "Omitting a model at execution still leaves the harness's selection intact."
    )
    permission_modes: tuple[PermissionMode, ...] = Field(
        description="Selectable permission modes in the agent's own vocabulary."
    )
    default_permission_mode: str | None = Field(
        description="The configured permission mode used when a conversation declares none."
    )
    routes: list[AgentRoute]
    driven_sessions: int = Field(
        description="Live runs this instance is driving on the agent's runtime."
    )
    native_sessions: int = Field(
        description="Native transcript streams currently attached to this agent — "
        "sessions someone else started that this instance is reading, not driving."
    )

id pydantic-field 

id: str

The registered agent tentacle's id.

description pydantic-field 

description: str

The agent's own capability blurb.

gateway pydantic-field 

gateway: bool

Whether this agent's driven turns offer the routing and project spells; the agent's own flag is the only switch.

default_model pydantic-field 

default_model: AgentRouteModelName | None

The catalog model used to inspect a new conversation's default capabilities; null when the harness has not reported its default. Omitting a model at execution still leaves the harness's selection intact.

permission_modes pydantic-field 

permission_modes: tuple[PermissionMode, ...]

Selectable permission modes in the agent's own vocabulary.

default_permission_mode pydantic-field 

default_permission_mode: str | None

The configured permission mode used when a conversation declares none.

routes pydantic-field 

routes: list[AgentRoute]

driven_sessions pydantic-field 

driven_sessions: int

Live runs this instance is driving on the agent's runtime.

native_sessions pydantic-field 

native_sessions: int

Native transcript streams currently attached to this agent — sessions someone else started that this instance is reading, not driving.