Skip to content

Conversations

octomate.schemas.thread

The thread ledger: threads, their messages, message bindings and handoffs.

ATTRIBUTABLE_KINDS module-attribute 

ATTRIBUTABLE_KINDS: frozenset[ThreadKind] = frozenset(
    {"thread", "native_thread"}
)

ChannelThreadId module-attribute 

ChannelThreadId = Annotated[
    str | None,
    AfterValidator(lambda thread_id: thread_id or None),
]

ThreadKey

The channel's identity for a thread: channel, chat type, chat id, and the platform's thread id.

Source code in octomate/schemas/thread.py 
 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
@dataclass(frozen=True)
class ThreadKey:
    """The channel's identity for a thread: channel, chat type, chat id, and the
    platform's thread id."""

    channel_tentacle_id: str
    chat_type: ChatType
    chat_id: str
    channel_thread_id: ChannelThreadId = None

    @classmethod
    def from_address(cls, address: ChannelAddress) -> ThreadKey:
        """The address with the person and the visibility stripped off."""
        return cls(
            channel_tentacle_id=address.channel_tentacle_id,
            chat_type=address.chat_type,
            chat_id=address.chat_id,
            channel_thread_id=address.channel_thread_id,
        )

    def address(self, user_id: str) -> ChannelAddress:
        """The surface this key names, spoken to `user_id` — `from_address` back."""
        return ChannelAddress(
            channel_tentacle_id=self.channel_tentacle_id,
            chat_type=self.chat_type,
            chat_id=self.chat_id,
            channel_thread_id=self.channel_thread_id,
            user_id=user_id,
        )

    @property
    def kind(self) -> ThreadKind:
        """What this key names — the chat type, unless a native client owns it."""
        if self.channel_tentacle_id in NATIVE_TENTACLE_IDS:
            return "native_thread"
        return self.chat_type

    def __str__(self) -> str:
        return (
            f"{self.channel_tentacle_id}/{self.chat_type}/"
            f"{self.chat_id}/{self.channel_thread_id or '-'}"
        )

channel_tentacle_id instance-attribute 

channel_tentacle_id: str

chat_type instance-attribute 

chat_type: ChatType

chat_id instance-attribute 

chat_id: str

channel_thread_id class-attribute instance-attribute 

channel_thread_id: ChannelThreadId = None

kind property 

kind: ThreadKind

What this key names — the chat type, unless a native client owns it.

from_address classmethod 

from_address(address: ChannelAddress) -> ThreadKey

The address with the person and the visibility stripped off.

Source code in octomate/schemas/thread.py 
82
83
84
85
86
87
88
89
90
@classmethod
def from_address(cls, address: ChannelAddress) -> ThreadKey:
    """The address with the person and the visibility stripped off."""
    return cls(
        channel_tentacle_id=address.channel_tentacle_id,
        chat_type=address.chat_type,
        chat_id=address.chat_id,
        channel_thread_id=address.channel_thread_id,
    )

address 

address(user_id: str) -> ChannelAddress

The surface this key names, spoken to user_id — from_address back.

Source code in octomate/schemas/thread.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
def address(self, user_id: str) -> ChannelAddress:
    """The surface this key names, spoken to `user_id` — `from_address` back."""
    return ChannelAddress(
        channel_tentacle_id=self.channel_tentacle_id,
        chat_type=self.chat_type,
        chat_id=self.chat_id,
        channel_thread_id=self.channel_thread_id,
        user_id=user_id,
    )

ThreadMessageFTS

Bases: BaseTransmuter

A row of the ledger's full-text index, with its search rank.

Source code in octomate/schemas/thread.py 
116
117
118
119
120
121
122
123
124
125
@sqlalchemy_materia.bless(thread_models.ThreadMessageFTS)
class ThreadMessageFTS(BaseTransmuter):
    """A row of the ledger's full-text index, with its search rank."""

    model_config = ConfigDict(from_attributes=True)

    rowid: Annotated[int, Identity]
    message_id: UUID7
    message_text: str
    rank: float | None = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

rowid instance-attribute 

rowid: Annotated[int, Identity]

message_id instance-attribute 

message_id: UUID7

message_text instance-attribute 

message_text: str

rank class-attribute instance-attribute 

rank: float | None = None

ThreadLedger

Bases: BaseTransmuter

A ledger a thread shows: the messages delivered to one thread, up to a cut.

Source code in octomate/schemas/thread.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@sqlalchemy_materia.bless(thread_models.ThreadLedger)
class ThreadLedger(BaseTransmuter):
    """A ledger a thread shows: the messages delivered to one thread, up to a cut."""

    model_config = ConfigDict(from_attributes=True)

    thread_id: Annotated[UUID7, Identity]
    ledger_id: Annotated[UUID7, Identity] = Field(
        description="The thread whose delivered messages make up this ledger."
    )
    cut_message_id: UUID7 | None = Field(
        default=None,
        description=(
            "The ledger's last message the thread shows, frozen at the fork that "
            "shared it; None for the thread's own ledger, shown whole as it grows."
        ),
    )

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

thread_id instance-attribute 

thread_id: Annotated[UUID7, Identity]

ledger_id class-attribute instance-attribute 

ledger_id: Annotated[UUID7, Identity] = Field(
    description="The thread whose delivered messages make up this ledger."
)

cut_message_id class-attribute instance-attribute 

cut_message_id: UUID7 | None = Field(
    default=None,
    description="The ledger's last message the thread shows, frozen at the fork that shared it; None for the thread's own ledger, shown whole as it grows.",
)

ThreadMessage

Bases: BaseTransmuter

One line of the thread ledger: who said what, when, and the model messages it is bound to.

Source code in octomate/schemas/thread.py 
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
@sqlalchemy_materia.bless(thread_models.ThreadMessage)
class ThreadMessage(BaseTransmuter):
    """One line of the thread ledger: who said what, when, and the model messages
    it is bound to."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    kind: Literal["message"] = Field(
        default="message",
        frozen=True,
        description="Whether this ledger row is a chat message, explicit command, or artifact reference.",
    )
    thread_id: UUID7
    platform_message_id: str | None = None
    reply_id: str = ""
    # When the message happened, and what the ledger orders on: a platform's or a
    # transcript's clock where one is known, else when Octomate recorded it. Never unset
    # — the order is only total if every row has one. `created_at` stays the bookkeeping
    # answer to "when was this written", which for replayed history is a different
    # instant entirely.
    happened_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    direction: ThreadMessageDirection
    actor_kind: ChannelActorKind
    user_id: str = ""
    agent_tentacle_id: str | None = None
    sender_id: UUID7 = Field(
        description=(
            "The sender's registry profile row (user_profiles); `sender` "
            "resolves it. Inbound: the platform account; outbound: the channel "
            "bot or a native session's pseudo-user."
        ),
    )
    sender: Relation[UserProfile] = Field(default_factory=Relation, frozen=True)
    segments: list[MessageSegment] = Field(default_factory=list)
    message_text: str | None = None
    raw: str = ""
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

    model_messages: RelationCollection[ModelRequest | ModelResponse] = Relationships()

    def __str__(self) -> str:
        """One line of a page: the row id, the `#msg:<id>` handle a brief cites,
        when, who, and what was said."""
        handle = f" #msg:{self.platform_message_id}" if self.platform_message_id else ""
        who = self.agent_tentacle_id or self.user_id or self.actor_kind
        return (
            f"{self.id}{handle} {self.happened_at:%Y-%m-%d %H:%M} "
            f"{self.actor_kind} {who}: {self.message_text or ''}"
        )

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
)

kind class-attribute instance-attribute 

kind: Literal["message"] = Field(
    default="message",
    frozen=True,
    description="Whether this ledger row is a chat message, explicit command, or artifact reference.",
)

thread_id instance-attribute 

thread_id: UUID7

platform_message_id class-attribute instance-attribute 

platform_message_id: str | None = None

reply_id class-attribute instance-attribute 

reply_id: str = ''

happened_at class-attribute instance-attribute 

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

direction instance-attribute 

direction: ThreadMessageDirection

actor_kind instance-attribute 

actor_kind: ChannelActorKind

user_id class-attribute instance-attribute 

user_id: str = ''

agent_tentacle_id class-attribute instance-attribute 

agent_tentacle_id: str | None = None

sender_id class-attribute instance-attribute 

sender_id: UUID7 = Field(
    description="The sender's registry profile row (user_profiles); `sender` resolves it. Inbound: the platform account; outbound: the channel bot or a native session's pseudo-user."
)

sender class-attribute instance-attribute 

sender: Relation[UserProfile] = Field(
    default_factory=Relation, frozen=True
)

segments class-attribute instance-attribute 

segments: list[MessageSegment] = Field(default_factory=list)

message_text class-attribute instance-attribute 

message_text: str | None = None

raw class-attribute instance-attribute 

raw: str = ''

created_at class-attribute instance-attribute 

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

model_messages class-attribute instance-attribute 

model_messages: RelationCollection[
    ModelRequest | ModelResponse
] = Relationships()

ThreadCommand

Bases: ThreadMessage

A durable command receipt and its outcome, never ordinary prompt input.

The inherited delivery key identifies retries. Record the receipt before invoking the runtime; an absent outcome does not make replay safe.

Source code in octomate/schemas/thread.py 
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
@sqlalchemy_materia.bless(thread_models.ThreadCommand)
class ThreadCommand(ThreadMessage):
    """A durable command receipt and its outcome, never ordinary prompt input.

    The inherited delivery key identifies retries. Record the receipt before
    invoking the runtime; an absent outcome does not make replay safe.
    """

    kind: Literal["command"] = Field(default="command", frozen=True)
    conversation_id: UUID7 | None = Field(
        description="The command's target conversation; null after that conversation is deleted."
    )
    invocation: CommandInvocation = Field(
        description="Explicit command intent with raw arguments and resolved attachments."
    )
    outcome: CommandOutcome | None = Field(
        default=None,
        description="Recorded command outcome; null means no terminal outcome was recorded, not permission to retry.",
    )

    @model_validator(mode="after")
    def has_delivery_id(self) -> Self:
        """A durable receipt must identify the delivery it protects from replay."""
        if not self.platform_message_id:
            raise ValueError("command deliveries require a platform_message_id")
        return self

kind class-attribute instance-attribute 

kind: Literal["command"] = Field(
    default="command", frozen=True
)

conversation_id class-attribute instance-attribute 

conversation_id: UUID7 | None = Field(
    description="The command's target conversation; null after that conversation is deleted."
)

invocation class-attribute instance-attribute 

invocation: CommandInvocation = Field(
    description="Explicit command intent with raw arguments and resolved attachments."
)

outcome class-attribute instance-attribute 

outcome: CommandOutcome | None = Field(
    default=None,
    description="Recorded command outcome; null means no terminal outcome was recorded, not permission to retry.",
)

has_delivery_id 

has_delivery_id() -> Self

A durable receipt must identify the delivery it protects from replay.

Source code in octomate/schemas/thread.py 
219
220
221
222
223
224
@model_validator(mode="after")
def has_delivery_id(self) -> Self:
    """A durable receipt must identify the delivery it protects from replay."""
    if not self.platform_message_id:
        raise ValueError("command deliveries require a platform_message_id")
    return self

ThreadArtifact

Bases: ThreadMessage

A ledger card referencing separately owned artifact content.

Generic ledger responses contain artifact_id only. The row's own id identifies this appearance for live/history deduplication and fork cuts. Artifact data requires an independently authorized read.

Source code in octomate/schemas/thread.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@sqlalchemy_materia.bless(thread_models.ThreadArtifact)
class ThreadArtifact(ThreadMessage):
    """A ledger card referencing separately owned artifact content.

    Generic ledger responses contain `artifact_id` only. The row's own `id`
    identifies this appearance for live/history deduplication and fork cuts.
    Artifact data requires an independently authorized read.
    """

    kind: Literal["artifact"] = Field(default="artifact", frozen=True)
    artifact_id: UUID7 = Field(
        frozen=True,
        description="Referenced artifact UUID, distinct from this ledger row's UUID.",
    )
    artifact: Relation[Artifact] = Field(
        default_factory=Relation,
        exclude=True,
        repr=False,
        description="Content relation for persistence, never included in ledger serialization.",
    )

kind class-attribute instance-attribute 

kind: Literal["artifact"] = Field(
    default="artifact", frozen=True
)

artifact_id class-attribute instance-attribute 

artifact_id: UUID7 = Field(
    frozen=True,
    description="Referenced artifact UUID, distinct from this ledger row's UUID.",
)

artifact class-attribute instance-attribute 

artifact: Relation[Artifact] = Field(
    default_factory=Relation,
    exclude=True,
    repr=False,
    description="Content relation for persistence, never included in ledger serialization.",
)

MessageBinding

Bases: BaseTransmuter

A link between a ledger message and a model message, by kind and run.

Source code in octomate/schemas/thread.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
@sqlalchemy_materia.bless(thread_models.MessageBinding)
class MessageBinding(BaseTransmuter):
    """A link between a ledger message and a model message, by kind and run."""

    model_config = ConfigDict(from_attributes=True)

    thread_message_id: Annotated[UUID7, Identity]
    model_message_id: Annotated[UUID7, Identity]
    kind: Annotated[MessageBindingKind, Identity]
    run_id: str
    tool_call_id: str | None = None
    position: int = 0
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

thread_message_id instance-attribute 

thread_message_id: Annotated[UUID7, Identity]

model_message_id instance-attribute 

model_message_id: Annotated[UUID7, Identity]

kind instance-attribute 

kind: Annotated[MessageBindingKind, Identity]

run_id instance-attribute 

run_id: str

tool_call_id class-attribute instance-attribute 

tool_call_id: str | None = None

position class-attribute instance-attribute 

position: int = 0

created_at class-attribute instance-attribute 

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

Handoff

Bases: BaseTransmuter

A recorded change of the agent owning a thread, and where it came from.

Source code in octomate/schemas/thread.py 
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
@sqlalchemy_materia.bless(thread_models.Handoff)
class Handoff(BaseTransmuter):
    """A recorded change of the agent owning a thread, and where it came from."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    thread_id: UUID7
    source_agent_tentacle_id: str | None = None
    to_agent_tentacle_id: str
    to_model: AgentRouteModelName | None = None
    reason: str = ""
    hint: str = ""
    brief: str = ""
    source_conversation_id: UUID7 | None = None
    target_conversation_id: UUID7 | None = None
    source_run_id: str | None = None
    source_model_message_id: UUID7 | None = None
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

    def __lt__(self, other: Handoff) -> bool:
        return self.id < other.id

    def __gt__(self, other: Handoff) -> bool:
        return self.id > other.id

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
)

thread_id instance-attribute 

thread_id: UUID7

source_agent_tentacle_id class-attribute instance-attribute 

source_agent_tentacle_id: str | None = None

to_agent_tentacle_id instance-attribute 

to_agent_tentacle_id: str

to_model class-attribute instance-attribute 

to_model: AgentRouteModelName | None = None

reason class-attribute instance-attribute 

reason: str = ''

hint class-attribute instance-attribute 

hint: str = ''

brief class-attribute instance-attribute 

brief: str = ''

source_conversation_id class-attribute instance-attribute 

source_conversation_id: UUID7 | None = None

target_conversation_id class-attribute instance-attribute 

target_conversation_id: UUID7 | None = None

source_run_id class-attribute instance-attribute 

source_run_id: str | None = None

source_model_message_id class-attribute instance-attribute 

source_model_message_id: UUID7 | None = None

created_at class-attribute instance-attribute 

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

Thread

Bases: BaseTransmuter

One chat surface in the ledger: its kind and channel key, the project its work is in, and the handoffs that say which agent owns it.

Source code in octomate/schemas/thread.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
@sqlalchemy_materia.bless(thread_models.Thread)
class Thread(BaseTransmuter):
    """One chat surface in the ledger: its kind and channel key, the project its
    work is in, and the handoffs that say which agent owns it."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    kind: ThreadKind = Field(
        frozen=True,
        description=(
            "Which surface this thread is. Set from `ThreadKey.kind` when the row is "
            "created; only a `thread` and a `native_thread` may carry a project."
        ),
    )

    chat_type: ChatType
    chat_id: str
    channel_tentacle_id: str
    channel_thread_id: str | None = Field(
        default=None,
        description=("The platform's own thread id; None unless `kind` is `thread`."),
    )
    parent_thread_id: UUID7 | None = Field(
        default=None,
        description=(
            "The thread this one was opened inside — a kick in a chat room works "
            "in a thread of its own, while the chat everyone sees stays on the chat "
            "room. None for a thread nobody opened; one level deep, never a chain."
        ),
    )
    title: str | None = Field(
        default=None,
        description=(
            "What this thread goes by in a listing. Taken from the first thing a "
            "person said in it, and replaced by a name the runtime grabbed for "
            "itself — a Claude session's own ai-title. None until the thread has "
            "been spoken in, and a listing then falls back to the surface."
        ),
    )
    project_id: UUID7 | None = Field(
        default=None,
        description=(
            "The declared project this thread's work is in; None is unattributed, "
            "which is what a chat thread and a directory no project claims both "
            "produce. Set once — when the row is created, or later by "
            "`ThreadManager.bind` for a thread that had none — and never moved "
            "after that: an external session's history is full of absolute paths, "
            "so a thread that changed project would resume its sessions somewhere "
            "they do not fit. Not frozen, because settable once from None is a rule "
            "the manager can state and a field cannot."
        ),
    )
    source_cursor_message_id: UUID7 | None = None
    status: ThreadStatus = "active"
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

    project: Relation[Project | None] = Field(
        default_factory=Relation,
        frozen=True,
        exclude=True,
        description="The project this thread's work is in, eagerly loaded.",
    )
    parent: Relation[Thread | None] = Field(default_factory=Relation, frozen=True)

    messages: RelationCollection[ThreadMessage | ThreadCommand | ThreadArtifact] = (
        Relationships()
    )
    handoffs: RelationCollection[Handoff] = Relationships()
    conversations: RelationCollection[Conversation] = Relationships(exclude=True)

    @model_validator(mode="after")
    def kind_agrees_with_the_key(self) -> Self:
        """The channel's key is the fact; this row is our copy of it.

        A copy that contradicts the fact is corrupt, and it would answer questions
        — chiefly whether a project may be attached — with the wrong surface.
        """
        if self.kind != self.key.kind:
            raise ValueError(
                f"thread {self.key} is a {self.key.kind}, "
                f"but the row calls itself a {self.kind}"
            )
        return self

    @property
    def key(self) -> ThreadKey:
        return ThreadKey(
            channel_tentacle_id=self.channel_tentacle_id,
            chat_type=self.chat_type,
            chat_id=self.chat_id,
            channel_thread_id=self.channel_thread_id,
        )

    @property
    def latest_handoff(self) -> Handoff | None:
        return self.handoffs[-1] if self.handoffs else None

    @computed_field
    @property
    def active_agent_tentacle_id(self) -> str | None:
        handoff = self.latest_handoff
        if handoff is not None:
            return handoff.to_agent_tentacle_id
        conversations = [
            conversation
            for conversation in self.conversations
            if not conversation.subagent_id
        ]
        return conversations[-1].agent_tentacle_id if conversations else None

    @property
    def active_model(self) -> AgentRouteModelName | None:
        handoff = self.latest_handoff
        if handoff is None:
            return None
        return handoff.to_model

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
)

kind class-attribute instance-attribute 

kind: ThreadKind = Field(
    frozen=True,
    description="Which surface this thread is. Set from `ThreadKey.kind` when the row is created; only a `thread` and a `native_thread` may carry a project.",
)

chat_type instance-attribute 

chat_type: ChatType

chat_id instance-attribute 

chat_id: str

channel_tentacle_id instance-attribute 

channel_tentacle_id: str

channel_thread_id class-attribute instance-attribute 

channel_thread_id: str | None = Field(
    default=None,
    description="The platform's own thread id; None unless `kind` is `thread`.",
)

parent_thread_id class-attribute instance-attribute 

parent_thread_id: UUID7 | None = Field(
    default=None,
    description="The thread this one was opened inside — a kick in a chat room works in a thread of its own, while the chat everyone sees stays on the chat room. None for a thread nobody opened; one level deep, never a chain.",
)

title class-attribute instance-attribute 

title: str | None = Field(
    default=None,
    description="What this thread goes by in a listing. Taken from the first thing a person said in it, and replaced by a name the runtime grabbed for itself — a Claude session's own ai-title. None until the thread has been spoken in, and a listing then falls back to the surface.",
)

project_id class-attribute instance-attribute 

project_id: UUID7 | None = Field(
    default=None,
    description="The declared project this thread's work is in; None is unattributed, which is what a chat thread and a directory no project claims both produce. Set once — when the row is created, or later by `ThreadManager.bind` for a thread that had none — and never moved after that: an external session's history is full of absolute paths, so a thread that changed project would resume its sessions somewhere they do not fit. Not frozen, because settable once from None is a rule the manager can state and a field cannot.",
)

source_cursor_message_id class-attribute instance-attribute 

source_cursor_message_id: UUID7 | None = None

status class-attribute instance-attribute 

status: ThreadStatus = 'active'

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

project class-attribute instance-attribute 

project: Relation[Project | None] = Field(
    default_factory=Relation,
    frozen=True,
    exclude=True,
    description="The project this thread's work is in, eagerly loaded.",
)

parent class-attribute instance-attribute 

parent: Relation[Thread | None] = Field(
    default_factory=Relation, frozen=True
)

messages class-attribute instance-attribute 

messages: RelationCollection[
    ThreadMessage | ThreadCommand | ThreadArtifact
] = Relationships()

handoffs class-attribute instance-attribute 

handoffs: RelationCollection[Handoff] = Relationships()

conversations class-attribute instance-attribute 

conversations: RelationCollection[Conversation] = (
    Relationships(exclude=True)
)

key property 

key: ThreadKey

latest_handoff property 

latest_handoff: Handoff | None

active_agent_tentacle_id property 

active_agent_tentacle_id: str | None

active_model property 

active_model: AgentRouteModelName | None

kind_agrees_with_the_key 

kind_agrees_with_the_key() -> Self

The channel's key is the fact; this row is our copy of it.

A copy that contradicts the fact is corrupt, and it would answer questions — chiefly whether a project may be attached — with the wrong surface.

Source code in octomate/schemas/thread.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
@model_validator(mode="after")
def kind_agrees_with_the_key(self) -> Self:
    """The channel's key is the fact; this row is our copy of it.

    A copy that contradicts the fact is corrupt, and it would answer questions
    — chiefly whether a project may be attached — with the wrong surface.
    """
    if self.kind != self.key.kind:
        raise ValueError(
            f"thread {self.key} is a {self.key.kind}, "
            f"but the row calls itself a {self.kind}"
        )
    return self

octomate.schemas.conversation

Channel addresses and agent conversations.

AddressMetadata

Bases: TypedDict

What a channel adds to an address to show it; each channel types its own.

Source code in octomate/schemas/conversation.py 
25
26
27
28
29
30
31
32
33
@with_config(ConfigDict(extra="allow"))
class AddressMetadata(TypedDict, total=False):
    """What a channel adds to an address to show it; each channel types its own."""

    name: str
    # Lists what is inside; set on a place to open, absent on one to land in.
    inside: str
    # Why a thread cannot land here; absent on an address that can be used.
    barred: str

name instance-attribute 

name: str

inside instance-attribute 

inside: str

barred instance-attribute 

barred: str

ChannelAddress

One surface on a channel — a DM, a group, a thread — and the user spoken to on it.

Source code in octomate/schemas/conversation.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
@dataclass(frozen=True, eq=False)
class ChannelAddress:
    """One surface on a channel — a DM, a group, a thread — and the user spoken to
    on it."""

    channel_tentacle_id: str
    chat_type: ChatType
    chat_id: str
    user_id: str
    # The platform's own thread id — a Slack `thread_ts`, a Lark sub-thread's root
    # message. None on a surface that is not a thread. Never `Thread.id`, which is
    # the row a ledger message points at.
    channel_thread_id: str | None = None
    # Whether anyone besides `user_id` can read this surface, carried from the
    # message that resolved it. `chat_type` cannot answer it: a thread overwrites
    # the type it sits in, so a thread in a Slack DM and a group thread both arrive
    # as "thread". False is the safe default — a spell that moves work somewhere
    # private then refuses rather than moving it out of a surface already private.
    shared: bool = False
    # A channel's extras for showing this surface; never part of its identity.
    metadata: AddressMetadata = field(default_factory=AddressMetadata, compare=False)

    # A channel's subclass only types `metadata`, so it is still the same address.
    def __eq__(self, other: object) -> bool:
        return isinstance(other, ChannelAddress) and all(
            getattr(self, field.name) == getattr(other, field.name)
            for field in fields(self)
            if field.compare
        )

    def __hash__(self) -> int:
        return hash(
            tuple(getattr(self, field.name) for field in fields(self) if field.compare)
        )

    @cached_property
    def group_id(self) -> str:
        return self.chat_id if self.chat_type == "group" else ""

    @cached_property
    def user_memory_key(self) -> str:
        return f"user:{self.channel_tentacle_id}:{self.user_id}"

    def __str__(self) -> str:
        return (
            f"{self.channel_tentacle_id}/{self.chat_type}/{self.chat_id}"
            f"/{self.channel_thread_id or '-'}/{self.user_id}"
        )

channel_tentacle_id instance-attribute 

channel_tentacle_id: str

chat_type instance-attribute 

chat_type: ChatType

chat_id instance-attribute 

chat_id: str

user_id instance-attribute 

user_id: str

channel_thread_id class-attribute instance-attribute 

channel_thread_id: str | None = None

shared class-attribute instance-attribute 

shared: bool = False

metadata class-attribute instance-attribute 

metadata: AddressMetadata = field(
    default_factory=AddressMetadata, compare=False
)

group_id cached property 

group_id: str

user_memory_key cached property 

user_memory_key: str

ConversationKey

Bases: NamedTuple

Identity of an agent conversation: the owning thread, the agent, and — for a context spawned by one of that agent's runs — the subagent. The empty subagent_id is the agent's own conversation (the Thread.channel_thread_id sentinel convention): a thread owns one bare conversation per agent, so every sender in a group thread keys to that one, independent of who woke it; each subagent keys to its own.

Source code in octomate/schemas/conversation.py 
86
87
88
89
90
91
92
93
94
95
96
97
class ConversationKey(NamedTuple):
    """Identity of an agent conversation: the owning thread, the agent, and —
    for a context spawned by one of that agent's runs — the subagent. The
    empty subagent_id is the agent's own conversation (the
    `Thread.channel_thread_id` sentinel convention): a thread owns one bare
    conversation per agent, so
    every sender in a group thread keys to that one, independent of who woke
    it; each subagent keys to its own."""

    thread_id: UUID7
    agent_id: str
    subagent_id: str = ""

thread_id instance-attribute 

thread_id: UUID7

agent_id instance-attribute 

agent_id: str

subagent_id class-attribute instance-attribute 

subagent_id: str = ''

ConversationRun

Bases: BaseTransmuter

A run in a conversation's history: one it ran, or one it was forked with.

Source code in octomate/schemas/conversation.py 
100
101
102
103
104
105
106
107
@sqlalchemy_materia.bless(ConversationRunModel)
class ConversationRun(BaseTransmuter):
    """A run in a conversation's history: one it ran, or one it was forked with."""

    model_config = ConfigDict(from_attributes=True)

    conversation_id: Annotated[UUID7, Identity]
    run_id: Annotated[str, Identity]

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

conversation_id instance-attribute 

conversation_id: Annotated[UUID7, Identity]

run_id instance-attribute 

run_id: Annotated[str, Identity]

Conversation

Bases: BaseTransmuter

One agent's model context within a thread, keyed by thread, agent and subagent.

Source code in octomate/schemas/conversation.py 
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
@sqlalchemy_materia.bless(ConversationModel)
class Conversation(BaseTransmuter):
    """One agent's model context within a thread, keyed by thread, agent and
    subagent."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    external_id: str | None = None
    transcript_file_id: UUID7 | None = Field(
        default=None,
        description="Live native transcript, or the independent starting copy of an imported fork.",
    )

    thread_id: UUID7 = Field(
        frozen=True,
        description=(
            "The owning thread; with agent_tentacle_id it is the conversation's "
            "identity."
        ),
    )
    agent_tentacle_id: str = Field(
        frozen=True,
        description=(
            "The owning agent; with thread_id it is the conversation's identity."
        ),
    )
    subagent_id: str = Field(
        default="",
        frozen=True,
        description=(
            "Empty for the agent's own conversation in the thread; a subagent's "
            "stable identity (Claude agentId, Codex child thread id, a "
            "commission's name) for a context spawned by one run."
        ),
    )
    parent_conversation_id: UUID7 | None = Field(
        default=None,
        frozen=True,
        description=(
            "The conversation whose run spawned this subagent's context — set "
            "iff subagent_id is. Which parent turn drove each child run stays "
            "on the run (parent_run_id)."
        ),
    )

    name: str | None = None
    status: str = "active"
    permission_mode: AgentPermissionMode | None = Field(
        default=None,
        description=(
            "Approval posture this conversation's agent works under, in that agent's "
            "own vocabulary. None declares nothing, and the agent's configured "
            "default decides. Seeded from the project at creation and owned here "
            "after, so a change made mid-thread is never revoked by a later ensure."
        ),
    )
    effort: str | None = Field(
        default=None,
        description=(
            "Reasoning effort this conversation's runs ask for, one of the levels its "
            "agent's route claims for the model. None declares nothing, and the "
            "runtime's own default decides. A run given an effort of its own, as a "
            "summon is, uses that one instead."
        ),
    )
    allowed_tools: list[str] = Field(
        default_factory=list,
        description=(
            "Tools the user allowed for the life of this conversation ('allow for "
            "session'), auto-approved without raising a card again."
        ),
    )

    runs: RelationCollection[AgentRun | ExternalAgentRun] = Relationships()
    messages: RelationCollection[ModelRequest | ModelResponse] = Relationships()

    @property
    def native_run_ids(self) -> set[str]:
        """Native runs with a completed transcript range, safe to skip on replay.

        Empty hook reservations and incomplete imports still need their transcript.
        """
        return {
            run.id
            for run in self.runs
            if isinstance(run, ExternalAgentRun) and run.end_offset is not None
        }

    @property
    def latest_run(self) -> AgentRun | ExternalAgentRun | None:
        """The last run in the loaded, ordered history, or None before the first run."""
        return self.runs[-1] if self.runs else None

    @property
    def key(self) -> ConversationKey:
        return ConversationKey(self.thread_id, self.agent_tentacle_id, self.subagent_id)

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
)

external_id class-attribute instance-attribute 

external_id: str | None = None

transcript_file_id class-attribute instance-attribute 

transcript_file_id: UUID7 | None = Field(
    default=None,
    description="Live native transcript, or the independent starting copy of an imported fork.",
)

thread_id class-attribute instance-attribute 

thread_id: UUID7 = Field(
    frozen=True,
    description="The owning thread; with agent_tentacle_id it is the conversation's identity.",
)

agent_tentacle_id class-attribute instance-attribute 

agent_tentacle_id: str = Field(
    frozen=True,
    description="The owning agent; with thread_id it is the conversation's identity.",
)

subagent_id class-attribute instance-attribute 

subagent_id: str = Field(
    default="",
    frozen=True,
    description="Empty for the agent's own conversation in the thread; a subagent's stable identity (Claude agentId, Codex child thread id, a commission's name) for a context spawned by one run.",
)

parent_conversation_id class-attribute instance-attribute 

parent_conversation_id: UUID7 | None = Field(
    default=None,
    frozen=True,
    description="The conversation whose run spawned this subagent's context — set iff subagent_id is. Which parent turn drove each child run stays on the run (parent_run_id).",
)

name class-attribute instance-attribute 

name: str | None = None

status class-attribute instance-attribute 

status: str = 'active'

permission_mode class-attribute instance-attribute 

permission_mode: AgentPermissionMode | None = Field(
    default=None,
    description="Approval posture this conversation's agent works under, in that agent's own vocabulary. None declares nothing, and the agent's configured default decides. Seeded from the project at creation and owned here after, so a change made mid-thread is never revoked by a later ensure.",
)

effort class-attribute instance-attribute 

effort: str | None = Field(
    default=None,
    description="Reasoning effort this conversation's runs ask for, one of the levels its agent's route claims for the model. None declares nothing, and the runtime's own default decides. A run given an effort of its own, as a summon is, uses that one instead.",
)

allowed_tools class-attribute instance-attribute 

allowed_tools: list[str] = Field(
    default_factory=list,
    description="Tools the user allowed for the life of this conversation ('allow for session'), auto-approved without raising a card again.",
)

runs class-attribute instance-attribute 

runs: RelationCollection[AgentRun | ExternalAgentRun] = (
    Relationships()
)

messages class-attribute instance-attribute 

messages: RelationCollection[
    ModelRequest | ModelResponse
] = Relationships()

native_run_ids property 

native_run_ids: set[str]

Native runs with a completed transcript range, safe to skip on replay.

Empty hook reservations and incomplete imports still need their transcript.

latest_run property 

latest_run: AgentRun | ExternalAgentRun | None

The last run in the loaded, ordered history, or None before the first run.

key property 

key: ConversationKey

octomate.schemas.runs

Agent runs: one per agent.run(), driven by Octomate or replayed from an external transcript.

AgentRun

Bases: BaseTransmuter

One agent.run() invocation. PK is the pydantic-ai run_id (a uuid7 string).

The octomate variant of the polymorphic run — driven by Octomate. A native run replayed from an external transcript is the ExternalAgentRun variant.

Source code in octomate/schemas/runs.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@sqlalchemy_materia.bless(AgentRunModel)
class AgentRun(BaseTransmuter):
    """One agent.run() invocation. PK is the pydantic-ai run_id (a uuid7 string).

    The `octomate` variant of the polymorphic run — driven by Octomate. A native
    run replayed from an external transcript is the `ExternalAgentRun` variant.
    """

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[str, Identity]
    # The polymorphic discriminator: a distinct Literal per variant is what lets the
    # `AgentRun | ExternalAgentRun` union (e.g. `Conversation.runs`) resolve each row to
    # its type — mirroring pydantic-ai's own `ModelRequest | ModelResponse`. The
    # `external` variant narrows it; the override warns but is intended.
    kind: Literal["octomate"] = "octomate"
    conversation_id: UUID7
    name: str | None = None
    model_name: str | None = None
    permission_mode: str | None = None
    cwd: Path | None = Field(
        default=None,
        description=(
            "The directory this run happened in, as its source reported it — spelled "
            "by `Path`, so a trailing slash or a `.` segment is normalized away. None "
            "when the source reported none, which is never a guess."
        ),
    )
    # The parent link of a subagent's turn, on the base so both variants carry it:
    # the run whose tool call spawned this one, and which ToolCallPart it answers.
    parent_run_id: str | None = None
    parent_tool_call_id: str | None = None
    started_at: AwareDatetime | None = None
    native_id: str | None = Field(
        default=None,
        description="The native runtime owning the native session and turn IDs.",
    )
    native_turn_id: str | None = Field(
        default=None,
        description="The native ingest turn key used to recognize a replayed driven run.",
    )
    native_session_id: str | None = None

    messages: RelationCollection[ModelRequest | ModelResponse] = Relationships()

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id instance-attribute 

id: Annotated[str, Identity]

kind class-attribute instance-attribute 

kind: Literal['octomate'] = 'octomate'

conversation_id instance-attribute 

conversation_id: UUID7

name class-attribute instance-attribute 

name: str | None = None

model_name class-attribute instance-attribute 

model_name: str | None = None

permission_mode class-attribute instance-attribute 

permission_mode: str | None = None

cwd class-attribute instance-attribute 

cwd: Path | None = Field(
    default=None,
    description="The directory this run happened in, as its source reported it — spelled by `Path`, so a trailing slash or a `.` segment is normalized away. None when the source reported none, which is never a guess.",
)

parent_run_id class-attribute instance-attribute 

parent_run_id: str | None = None

parent_tool_call_id class-attribute instance-attribute 

parent_tool_call_id: str | None = None

started_at class-attribute instance-attribute 

started_at: AwareDatetime | None = None

native_id class-attribute instance-attribute 

native_id: str | None = Field(
    default=None,
    description="The native runtime owning the native session and turn IDs.",
)

native_turn_id class-attribute instance-attribute 

native_turn_id: str | None = Field(
    default=None,
    description="The native ingest turn key used to recognize a replayed driven run.",
)

native_session_id class-attribute instance-attribute 

native_session_id: str | None = None

messages class-attribute instance-attribute 

messages: RelationCollection[
    ModelRequest | ModelResponse
] = Relationships()

ExternalAgentRun

Bases: AgentRun

A run rebuilt from an external runtime's transcript (native Claude session). Carries the transcript coordinates its offsets check-point ingest against.

Source code in octomate/schemas/runs.py 
65
66
67
68
69
70
71
72
73
74
@sqlalchemy_materia.bless(ExternalAgentRunModel)
class ExternalAgentRun(AgentRun):
    """A run rebuilt from an external runtime's transcript (native Claude session).
    Carries the transcript coordinates its offsets check-point ingest against."""

    kind: Literal["external"] = "external"  # pyright: ignore[reportIncompatibleVariableOverride]
    source: str | None = None
    start_offset: int | None = None
    end_offset: int | None = None
    last_line_uuid: str | None = None

kind class-attribute instance-attribute 

kind: Literal['external'] = 'external'

source class-attribute instance-attribute 

source: str | None = None

start_offset class-attribute instance-attribute 

start_offset: int | None = None

end_offset class-attribute instance-attribute 

end_offset: int | None = None

last_line_uuid class-attribute instance-attribute 

last_line_uuid: str | None = None

octomate.schemas.messages

Persisted model messages: pydantic-ai's request and response, blessed to the polymorphic model_messages table.

dataclass_config module-attribute 

dataclass_config = ConfigDict(
    populate_by_name=True,
    validate_by_name=True,
    validate_by_alias=True,
)

ModelMessage

Bases: BaseTransmuter, ABC

Polymorphic parent of the message transmuters — just the shared, queryable columns. Blessed to the abstract model_messages base so arcanus can run one polymorphic select that adapts each row to its ModelRequest/ModelResponse subtype (keyed on kind). Abstract and never instantiated; the subtypes can't subclass it (each already subclasses its pydantic-ai message type), so it is a sibling registration that the polymorphic result adapter discovers by kind.

Source code in octomate/schemas/messages.py 
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
@sqlalchemy_materia.bless(ModelMessageModel)
class ModelMessage(BaseTransmuter, ABC):
    """Polymorphic parent of the message transmuters — just the shared, queryable
    columns. Blessed to the abstract `model_messages` base so arcanus can run one
    polymorphic select that adapts each row to its `ModelRequest`/`ModelResponse`
    subtype (keyed on `kind`). Abstract and never instantiated; the subtypes can't
    subclass it (each already subclasses its pydantic-ai message type), so it is a
    sibling registration that the polymorphic result adapter discovers by `kind`."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    run_id: str | None = None
    conversation_id: str | None = None
    timestamp: AwareDatetime | None = None
    role: Literal["user", "assistant"] = "assistant"
    message_text: str | None = None
    thread_messages: RelationCollection[
        ThreadMessage | ThreadCommand | ThreadArtifact
    ] = Relationships()

    @abstractmethod
    def _concrete_message(self) -> None:
        """Marks `ModelMessage` abstract: it is a query anchor only. Every row
        materializes as a concrete sibling (`ModelRequest`/`ModelResponse`) via the
        polymorphic adapter, so the base itself is never instantiated."""

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
)

run_id class-attribute instance-attribute 

run_id: str | None = None

conversation_id class-attribute instance-attribute 

conversation_id: str | None = None

timestamp class-attribute instance-attribute 

timestamp: AwareDatetime | None = None

role class-attribute instance-attribute 

role: Literal['user', 'assistant'] = 'assistant'

message_text class-attribute instance-attribute 

message_text: str | None = None

thread_messages class-attribute instance-attribute 

thread_messages: RelationCollection[
    ThreadMessage | ThreadCommand | ThreadArtifact
] = Relationships()

ModelRequest

Bases: Transmuter, ModelRequest

A persisted pydantic-ai request; role and message_text are derived from its parts.

Source code in octomate/schemas/messages.py 
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
@sqlalchemy_materia.bless(ModelRequestModel)
@arcanus_dataclass(config=dataclass_config)
class ModelRequest(Transmuter, PydanticModelRequest):
    """A persisted pydantic-ai request; `role` and `message_text` are derived from
    its parts."""

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    timestamp: AwareDatetime | None = None
    metadata: Annotated[JsonObject | None, Field(alias="meta")] = None
    role: Literal["user", "assistant"] = "assistant"
    message_text: str | None = None
    # The chat rows this request was built from: who said them, and what they
    # said before the prompt dressed it.
    thread_messages: RelationCollection[
        ThreadMessage | ThreadCommand | ThreadArtifact
    ] = Relationships()

    @model_validator(mode="after")
    def _derive_message_fields(self) -> Self:
        # A request carrying a user prompt is the user's turn; one carrying only
        # tool returns is assistant-side. Text comes from the user prompt(s).
        self.role = (
            "user"
            if any(isinstance(part, UserPromptPart) for part in self.parts)
            else "assistant"
        )
        if self.message_text is None:
            text = "\n\n".join(
                t
                for part in self.parts
                if isinstance(part, UserPromptPart)
                and (t := _user_prompt_text(part.content))
            )
            self.message_text = text or None
        return self

id class-attribute instance-attribute 

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

timestamp class-attribute instance-attribute 

timestamp: AwareDatetime | None = None

metadata class-attribute instance-attribute 

metadata: Annotated[
    JsonObject | None, Field(alias="meta")
] = None

role class-attribute instance-attribute 

role: Literal['user', 'assistant'] = 'assistant'

message_text class-attribute instance-attribute 

message_text: str | None = None

thread_messages class-attribute instance-attribute 

thread_messages: RelationCollection[
    ThreadMessage | ThreadCommand | ThreadArtifact
] = Relationships()

ModelResponse

Bases: Transmuter, ModelResponse

A persisted pydantic-ai response; message_text is its spoken text, including what it delivered through send.

Source code in octomate/schemas/messages.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@sqlalchemy_materia.bless(ModelResponseModel)
@arcanus_dataclass(config=dataclass_config)
class ModelResponse(Transmuter, PydanticModelResponse):
    """A persisted pydantic-ai response; `message_text` is its spoken text,
    including what it delivered through `send`."""

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    timestamp: AwareDatetime = Field(default_factory=now_utc)
    metadata: Annotated[JsonObject | None, Field(alias="meta")] = None
    workspace_ref: WorkspaceRef | None = Field(
        default=None,
        description="The provider and identifier of the response's workspace, without credentials.",
    )
    role: Literal["user", "assistant"] = "assistant"
    message_text: str | None = None
    # The chat rows that delivered this response's answer.
    thread_messages: RelationCollection[
        ThreadMessage | ThreadCommand | ThreadArtifact
    ] = Relationships()

    @model_validator(mode="after")
    def _derive_message_fields(self) -> Self:
        # A response is always assistant-side (the default). Its spoken text is the
        # answer parts plus anything delivered via `send`; thinking and other
        # tool calls are excluded.
        if self.message_text is None:
            fragments: list[str] = []
            for part in self.parts:
                if isinstance(part, TextPart):
                    if part.content:
                        fragments.append(part.content)
                elif (
                    isinstance(part, ToolCallPart)
                    and part.tool_name == GatewayTool.SEND
                ):
                    fragments.append(str(part.args_as_dict()))
            self.message_text = "\n\n".join(fragments) or None
        return self

id class-attribute instance-attribute 

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

timestamp class-attribute instance-attribute 

timestamp: AwareDatetime = Field(default_factory=now_utc)

metadata class-attribute instance-attribute 

metadata: Annotated[
    JsonObject | None, Field(alias="meta")
] = None

workspace_ref class-attribute instance-attribute 

workspace_ref: WorkspaceRef | None = Field(
    default=None,
    description="The provider and identifier of the response's workspace, without credentials.",
)

role class-attribute instance-attribute 

role: Literal['user', 'assistant'] = 'assistant'

message_text class-attribute instance-attribute 

message_text: str | None = None

thread_messages class-attribute instance-attribute 

thread_messages: RelationCollection[
    ThreadMessage | ThreadCommand | ThreadArtifact
] = Relationships()

octomate.schemas.events

The inbound message event a channel's chromo produces.

MessageEvent pydantic-model

Bases: BaseModel

An inbound platform message as a channel's chromo translated it: who sent it, on which surface, and the segments it carries.

Config:

  • extra: ignore
  • coerce_numbers_to_str: True

Fields:

Source code in octomate/schemas/events.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class MessageEvent(BaseModel):
    """An inbound platform message as a channel's chromo translated it: who sent
    it, on which surface, and the segments it carries."""

    model_config = ConfigDict(extra="ignore", coerce_numbers_to_str=True)

    tentacle_id: str = ""
    message_id: str = ""
    channel_thread_id: str | None = None
    reply_id: str = ""
    timestamp: float = 0.0
    user_id: str = ""
    chat_id: str = ""
    chat_type: ChatType = "dm"
    chat_name: str | None = Field(
        default=None,
        description="The platform's chat name, when supplied by the inbound event.",
    )
    shared: bool = Field(
        default=False,
        description=(
            "Whether anyone besides the sender can read this surface. Separate from "
            "`chat_type`, which a thread overwrites and which therefore cannot tell a "
            "thread in a group chat from one inside a DM. Only the channel knows: it "
            "is what the mention gate asks before it drops an unaddressed message."
        ),
    )
    self_id: str = ""
    sender: UserProfile = Field(default_factory=UserProfile)
    segments: list[MessageSegment] = Field(default_factory=list)
    raw: str = ""

    @property
    def display_name(self) -> str:
        return self.sender.name or self.sender.nickname or "anonymous"

    @property
    def sender_label(self) -> str:
        """The parenthesized ids of the sender line: the raw platform id, plus
        the `user:{username}` marker (the owning user) that equates one human
        across channels. The owner is present once the sender is a registry row."""
        owner = self.sender.user.peek()
        if owner is not None:
            return f"{self.user_id}, user:{owner.username}"
        return self.user_id

    def is_at(self, user_id: str | None = None) -> bool:
        return user_id is not None and any(
            isinstance(seg, AtSegment) and seg.data.user_id == user_id
            for seg in self.segments
        )

    def replies_to(self, user_id: str | None = None) -> bool:
        return user_id is not None and any(
            isinstance(seg, ReplySegment) and seg.data.get("user_id") == user_id
            for seg in self.segments
        )

    def text_parts(self) -> list[str]:
        return [
            seg.data["text"] for seg in self.segments if isinstance(seg, TextSegment)
        ]

    def __str__(self) -> str:
        body = "\n".join(f"  {seg}" for seg in self.segments)
        return (
            f"{self.display_name} ({self.sender_label}) #msg:{self.message_id}:\n{body}"
        )

    def to_content_parts(self) -> list[UserContent]:
        header = f"{self.display_name} ({self.sender_label}) #msg:{self.message_id}: "
        return [header] + [seg.to_content() for seg in self.segments]

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    extra="ignore", coerce_numbers_to_str=True
)

tentacle_id pydantic-field 

tentacle_id: str = ''

message_id pydantic-field 

message_id: str = ''

channel_thread_id pydantic-field 

channel_thread_id: str | None = None

reply_id pydantic-field 

reply_id: str = ''

timestamp pydantic-field 

timestamp: float = 0.0

user_id pydantic-field 

user_id: str = ''

chat_id pydantic-field 

chat_id: str = ''

chat_type pydantic-field 

chat_type: ChatType = 'dm'

chat_name pydantic-field 

chat_name: str | None = None

The platform's chat name, when supplied by the inbound event.

shared pydantic-field 

shared: bool = False

Whether anyone besides the sender can read this surface. Separate from chat_type, which a thread overwrites and which therefore cannot tell a thread in a group chat from one inside a DM. Only the channel knows: it is what the mention gate asks before it drops an unaddressed message.

self_id pydantic-field 

self_id: str = ''

sender pydantic-field 

sender: UserProfile

segments pydantic-field 

segments: list[MessageSegment]

raw pydantic-field 

raw: str = ''

display_name property 

display_name: str

sender_label property 

sender_label: str

The parenthesized ids of the sender line: the raw platform id, plus the user:{username} marker (the owning user) that equates one human across channels. The owner is present once the sender is a registry row.

is_at 

is_at(user_id: str | None = None) -> bool
Source code in octomate/schemas/events.py 
64
65
66
67
68
def is_at(self, user_id: str | None = None) -> bool:
    return user_id is not None and any(
        isinstance(seg, AtSegment) and seg.data.user_id == user_id
        for seg in self.segments
    )

replies_to 

replies_to(user_id: str | None = None) -> bool
Source code in octomate/schemas/events.py 
70
71
72
73
74
def replies_to(self, user_id: str | None = None) -> bool:
    return user_id is not None and any(
        isinstance(seg, ReplySegment) and seg.data.get("user_id") == user_id
        for seg in self.segments
    )

text_parts 

text_parts() -> list[str]
Source code in octomate/schemas/events.py 
76
77
78
79
def text_parts(self) -> list[str]:
    return [
        seg.data["text"] for seg in self.segments if isinstance(seg, TextSegment)
    ]

to_content_parts 

to_content_parts() -> list[UserContent]
Source code in octomate/schemas/events.py 
87
88
89
def to_content_parts(self) -> list[UserContent]:
    header = f"{self.display_name} ({self.sender_label}) #msg:{self.message_id}: "
    return [header] + [seg.to_content() for seg in self.segments]

octomate.schemas.segments

Message segments: the typed pieces — text, mentions, images, markdown, replies, files, cards — a message is made of on every platform.

CARD_TEXT_KEYS module-attribute 

CARD_TEXT_KEYS = frozenset(
    {"text", "content", "title", "alt_text"}
)

MessageSegment module-attribute 

MessageSegment = Annotated[
    TextSegment
    | AtSegment
    | ImageSegment
    | MarkdownSegment
    | ReplySegment
    | FileSegment
    | CardSegment,
    Discriminator("type"),
]

TextData

Bases: TypedDict

The text of a text segment.

Source code in octomate/schemas/segments.py 
18
19
20
21
class TextData(TypedDict):
    """The text of a text segment."""

    text: str

text instance-attribute 

text: str

AtData pydantic-model

Bases: BaseModel

The mentioned user's platform id, and a display name when known.

Fields:

Source code in octomate/schemas/segments.py 
24
25
26
27
28
class AtData(BaseModel):
    """The mentioned user's platform id, and a display name when known."""

    user_id: str
    name: str | None = None

user_id pydantic-field 

user_id: str

name pydantic-field 

name: str | None = None

ImageData pydantic-model

Bases: BaseModel

Image payload. file is always a local file path (str).

Inbound images are downloaded to local storage by the tentacle layer. Outbound images are read from local files and uploaded by the tentacle. Use the path property to get it as a Path object.

Fields:

Validators:

  • _coerce_path → file
Source code in octomate/schemas/segments.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class ImageData(BaseModel):
    """Image payload. `file` is always a local file path (str).

    Inbound images are downloaded to local storage by the tentacle layer.
    Outbound images are read from local files and uploaded by the tentacle.
    Use the `path` property to get it as a `Path` object.
    """

    file: str
    url: str | None = None
    name: str | None = None
    summary: str | None = None

    @field_validator("file", mode="before")
    @classmethod
    def _coerce_path(cls, value: str | Path) -> str:
        if isinstance(value, Path):
            return str(value.resolve())
        return value

    @property
    def path(self) -> Path:
        return Path(self.file)

file pydantic-field 

file: str

url pydantic-field 

url: str | None = None

name pydantic-field 

name: str | None = None

summary pydantic-field 

summary: str | None = None

path property 

path: Path

MarkdownData

Bases: TypedDict

The markdown of a markdown segment.

Source code in octomate/schemas/segments.py 
56
57
58
59
class MarkdownData(TypedDict):
    """The markdown of a markdown segment."""

    text: str

text instance-attribute 

text: str

ReplyData

Bases: TypedDict

The replied-to message's id, with its content and author when known.

Source code in octomate/schemas/segments.py 
62
63
64
65
66
67
68
69
class ReplyData(TypedDict):
    """The replied-to message's id, with its content and author when known."""

    id: str
    content: NotRequired[str]
    # Resolved platform author, when known; lets a reply address that user without
    # manufacturing a visible mention in the conversation.
    user_id: NotRequired[str]

id instance-attribute 

id: str

content instance-attribute 

content: NotRequired[str]

user_id instance-attribute 

user_id: NotRequired[str]

FileData pydantic-model

Bases: BaseModel

A local file path, the name it is shown as, and its size in bytes.

Fields:

Source code in octomate/schemas/segments.py 
72
73
74
75
76
77
class FileData(BaseModel):
    """A local file path, the name it is shown as, and its size in bytes."""

    file: str
    name: str = ""
    size: int = 0

file pydantic-field 

file: str

name pydantic-field 

name: str = ''

size pydantic-field 

size: int = 0

CardData pydantic-model

Bases: BaseModel

A platform-native card payload as JSON.

Fields:

Source code in octomate/schemas/segments.py 
80
81
82
83
class CardData(BaseModel):
    """A platform-native card payload as JSON."""

    payload: JsonObject

payload pydantic-field 

payload: JsonObject

Segment pydantic-model

Bases: BaseModel

Base class for all message segments.

Source code in octomate/schemas/segments.py 
86
87
88
89
90
91
92
93
class Segment(BaseModel):
    """Base class for all message segments."""

    def __str__(self) -> str:
        return f"[{getattr(self, 'type', 'unknown')}]"

    def to_content(self) -> UserContent:
        return str(self)

to_content 

to_content() -> UserContent
Source code in octomate/schemas/segments.py 
92
93
def to_content(self) -> UserContent:
    return str(self)

TextSegment pydantic-model

Bases: Segment

Plain text content. Set data.text to the message text.

Fields:

Source code in octomate/schemas/segments.py 
 96
 97
 98
 99
100
101
102
103
class TextSegment(Segment):
    """Plain text content. Set data.text to the message text."""

    type: Literal["text"] = "text"
    data: TextData

    def __str__(self) -> str:
        return self.data["text"]

type pydantic-field 

type: Literal['text'] = 'text'

data pydantic-field 

data: TextData

AtSegment pydantic-model

Bases: Segment

Mention/at a user. Set data.user_id to the user's platform ID (as string). Optionally set data.name.

Fields:

Source code in octomate/schemas/segments.py 
106
107
108
109
110
111
112
113
class AtSegment(Segment):
    """Mention/at a user. Set data.user_id to the user's platform ID (as string). Optionally set data.name."""

    type: Literal["at"] = "at"
    data: AtData

    def __str__(self) -> str:
        return f"@{self.data.name or self.data.user_id}"

type pydantic-field 

type: Literal['at'] = 'at'

data pydantic-field 

data: AtData

ImageSegment pydantic-model

Bases: Segment

Send an image. Set data.file to a local file path. Never use http/https URLs — the system manages file storage automatically.

Fields:

Source code in octomate/schemas/segments.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class ImageSegment(Segment):
    """Send an image. Set data.file to a local file path.
    Never use http/https URLs — the system manages file storage automatically."""

    type: Literal["image"] = "image"
    data: ImageData

    def __str__(self) -> str:
        label = self.data.summary or self.data.name or "image"
        return f"[image: {label} | {self.data.file}]"

    def to_content(self) -> UserContent:
        path = self.data.path
        if path.exists():
            media_type = mimetypes.guess_type(path.name)[0] or "image/png"
            return BinaryContent(data=path.read_bytes(), media_type=media_type)
        return str(self)

type pydantic-field 

type: Literal['image'] = 'image'

data pydantic-field 

data: ImageData

to_content 

to_content() -> UserContent
Source code in octomate/schemas/segments.py 
127
128
129
130
131
132
def to_content(self) -> UserContent:
    path = self.data.path
    if path.exists():
        media_type = mimetypes.guess_type(path.name)[0] or "image/png"
        return BinaryContent(data=path.read_bytes(), media_type=media_type)
    return str(self)

MarkdownSegment pydantic-model

Bases: Segment

Markdown-formatted text. Set data.text to the markdown content. Platforms that support markdown render it natively; others strip it to plain text. Images can be embedded inline with ![alt](image_key); no separate ImageSegment is needed.

Fields:

Source code in octomate/schemas/segments.py 
135
136
137
138
139
140
141
142
143
144
class MarkdownSegment(Segment):
    """Markdown-formatted text. Set data.text to the markdown content.
    Platforms that support markdown render it natively; others strip it to plain text.
    Images can be embedded inline with `![alt](image_key)`; no separate ImageSegment is needed."""

    type: Literal["markdown"] = "markdown"
    data: MarkdownData

    def __str__(self) -> str:
        return self.data["text"]

type pydantic-field 

type: Literal['markdown'] = 'markdown'

data pydantic-field 

data: MarkdownData

ReplySegment pydantic-model

Bases: Segment

Quote/reply to a previous message so the answer threads onto it. Set data.id to the target's message id — the #msg:<id> handle shown beside each message in the conversation. Put it first; the message threads onto the first reply it carries.

Fields:

Source code in octomate/schemas/segments.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
class ReplySegment(Segment):
    """Quote/reply to a previous message so the answer threads onto it. Set data.id
    to the target's message id — the `#msg:<id>` handle shown beside each message in
    the conversation. Put it first; the message threads onto the first reply it carries."""

    type: Literal["reply"] = "reply"
    data: ReplyData

    def __str__(self) -> str:
        content = self.data.get("content")
        if content:
            return f"[reply: {self.data['id']}] {content}"
        return f"[reply: {self.data['id']}]"

type pydantic-field 

type: Literal['reply'] = 'reply'

data pydantic-field 

data: ReplyData

FileSegment pydantic-model

Bases: Segment

Send a file. Set data.file to a local file path and data.name to the name it is shown as.

Fields:

Source code in octomate/schemas/segments.py 
162
163
164
165
166
167
168
169
170
171
172
class FileSegment(Segment):
    """Send a file. Set data.file to a local file path and data.name to the name it
    is shown as."""

    type: Literal["file"] = "file"
    data: FileData

    def __str__(self) -> str:
        # The path as well as the name, as an image gives: a reader that wants to
        # open the thing needs where it is, and a named file used to hide it.
        return f"[file: {self.data.name or 'file'} | {self.data.file}]"

type pydantic-field 

type: Literal['file'] = 'file'

data pydantic-field 

data: FileData

CardSegment pydantic-model

Bases: Segment

A platform-native card. Set data.payload to the card JSON the platform renders.

Fields:

Source code in octomate/schemas/segments.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class CardSegment(Segment):
    """A platform-native card. Set data.payload to the card JSON the platform
    renders."""

    type: Literal["card"] = "card"
    data: CardData

    def __str__(self) -> str:
        """`[card]` alone said only that something was there. What was on it is the
        part worth having: a card is where a decision gets asked for, and one nobody
        can read is a hole in the transcript at exactly that point.

        A card is a different shape on every platform, so this looks for the keys
        that carry words rather than for a structure — walked in the order the
        payload lays them out, which is the order they were shown in.
        """
        shown: list[str] = []
        stack: list[tuple[str, JsonValue]] = [("", self.data.payload)]
        while stack:
            key, node = stack.pop()
            if key in CARD_TEXT_KEYS and isinstance(node, str) and node:
                shown.append(node)
            elif isinstance(node, dict):
                stack.extend(reversed(list(node.items())))
            elif isinstance(node, list):
                stack.extend((key, item) for item in reversed(node))
        return "\n".join(["[card]", *shown])

type pydantic-field 

type: Literal['card'] = 'card'

data pydantic-field 

data: CardData

octomate.schemas.actions

Message-level action payloads: the composed agent message, confirmations, questions and todos.

AgentMessage pydantic-model

Bases: BaseModel

A single outgoing message composed of one or more segments.

Fields:

Source code in octomate/schemas/actions.py 
15
16
17
18
19
20
21
class AgentMessage(BaseModel):
    """A single outgoing message composed of one or more segments."""

    segments: list[MessageSegment]

    def __str__(self) -> str:
        return "".join(str(seg) for seg in self.segments)

segments pydantic-field 

segments: list[MessageSegment]

ConfirmAction pydantic-model

Bases: BaseModel

A pending confirmation of a tool call, with its approvers and expiry.

Fields:

Source code in octomate/schemas/actions.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class ConfirmAction(BaseModel):
    """A pending confirmation of a tool call, with its approvers and expiry."""

    confirmation_id: str
    conversation_address: ChannelAddress
    tool_name: str
    tool_call_id: str
    args: JsonObject
    title: str = ""
    description: str = ""
    skill: str = ""
    approvers: list[str] = Field(default_factory=list)
    created_at: float
    expires_at: float
    status: Literal["pending", "approved", "denied", "expired"] = "pending"

confirmation_id pydantic-field 

confirmation_id: str

conversation_address pydantic-field 

conversation_address: ChannelAddress

tool_name pydantic-field 

tool_name: str

tool_call_id pydantic-field 

tool_call_id: str

args pydantic-field 

args: JsonObject

title pydantic-field 

title: str = ''

description pydantic-field 

description: str = ''

skill pydantic-field 

skill: str = ''

approvers pydantic-field 

approvers: list[str]

created_at pydantic-field 

created_at: float

expires_at pydantic-field 

expires_at: float

status pydantic-field 

status: Literal[
    "pending", "approved", "denied", "expired"
] = "pending"

QuestionAction pydantic-model

Bases: BaseModel

A pending question with its options and expiry.

Fields:

Source code in octomate/schemas/actions.py 
41
42
43
44
45
46
47
48
49
50
51
class QuestionAction(BaseModel):
    """A pending question with its options and expiry."""

    question_id: str
    conversation_address: ChannelAddress
    text: str
    options: list[str] = Field(default_factory=list)
    multi_select: bool = False
    created_at: float
    expires_at: float
    status: Literal["pending", "answered", "expired"] = "pending"

question_id pydantic-field 

question_id: str

conversation_address pydantic-field 

conversation_address: ChannelAddress

text pydantic-field 

text: str

options pydantic-field 

options: list[str]

multi_select pydantic-field 

multi_select: bool = False

created_at pydantic-field 

created_at: float

expires_at pydantic-field 

expires_at: float

status pydantic-field 

status: Literal["pending", "answered", "expired"] = (
    "pending"
)

QuestionResponse pydantic-model

Bases: BaseModel

A user's answer to a question.

Fields:

Source code in octomate/schemas/actions.py 
54
55
56
57
58
59
class QuestionResponse(BaseModel):
    """A user's answer to a question."""

    question_id: str
    answer: str
    responder_id: str

question_id pydantic-field 

question_id: str

answer pydantic-field 

answer: str

responder_id pydantic-field 

responder_id: str

TodoAction pydantic-model

Bases: BaseModel

A todo item as a message-level action.

Fields:

Source code in octomate/schemas/actions.py 
62
63
64
65
66
67
68
69
class TodoAction(BaseModel):
    """A todo item as a message-level action."""

    todo_id: str
    title: str
    active_form: str = ""
    assignee: str = ""
    status: Literal["pending", "in_progress", "completed", "cancelled"] = "pending"

todo_id pydantic-field 

todo_id: str

title pydantic-field 

title: str

active_form pydantic-field 

active_form: str = ''

assignee pydantic-field 

assignee: str = ''

status pydantic-field 

status: Literal[
    "pending", "in_progress", "completed", "cancelled"
] = "pending"