"""Domain types for agents as persistent backend processes.
Business intent (Execution, Mission, Agent) is separate from runtime
(Thread, Run, Checkpoint). Threads may be ephemeral or attached to an
Execution.
Intents
-------
Each :class:`~agency.models.Agent` has:
- An LLM :class:`~agency.models.Model` with a :class:`~agency.models.Provider` for inference.
- A set of :class:`~agency.models.Tool` for technical capabilities.
- A set of :class:`~agency.models.Skill` for business capabilities.
:class:`~agency.models.Session` is a conversation surface,
:class:`~agency.models.Room` is a conversation space, and
:class:`~agency.models.Message` is a message in that conversation.
Execution flow
--------------
:class:`~agency.models.Task` records Celery job metadata (transport, not
domain lifecycle).
Each :class:`~agency.models.Execution` uses:
- :class:`~agency.models.Mission` for business intent.
- :class:`~agency.models.Agent` for the execution.
- :class:`~agency.models.Thread` as the runtime timeline.
- :class:`~agency.models.Run` as a technical attempt (invoke / resume).
- :class:`~agency.models.Checkpoint` for persisted graph state and
:class:`~agency.models.CheckpointWrite` for graph writes.
- :class:`~agency.models.Interruption` for a human-in-the-loop pause.
"""
from __future__ import annotations
import uuid
from typing import List, Dict, Any, Union, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from django.db.models import manager
from agency.agent import ThreadRunner
from agency.channel import publish_event, channel_log
from agency.photos import photo_upload_to
from agency.utils import app_log, sys_log, lazy_import, serializable
from celery.result import AsyncResult
from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AbstractUser
from django.db import IntegrityError
from django.db.models import (Model as Object, QuerySet, TextChoices, Q,
UUIDField, CharField, ForeignKey, OneToOneField, ManyToManyField, BooleanField, TextField,
JSONField, DateTimeField, IntegerField, PositiveIntegerField, PositiveSmallIntegerField,
BigAutoField, BinaryField, FloatField, ImageField, Index, UniqueConstraint,
CASCADE, SET_NULL, PROTECT)
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from langchain.agents import create_agent
from langchain_core.messages import ( BaseMessage, HumanMessage, SystemMessage, AIMessage, ToolMessage)
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
NB = {"null": True, "blank": True}
def T(name: str) -> str:
"""Translate verbose name."""
return {"verbose_name": _(name)}
User = settings.AUTH_USER_MODEL
"""🧑💻 Only for relationship definitions, use get_user_model() for the actual user model."""
[docs]
class Task(Object):
"""⚙️ A Celery task."""
[docs]
class State(TextChoices):
PENDING = "PENDING", "Pending"
RECEIVED = "RECEIVED", "Received"
STARTED = "STARTED", "Started"
SUCCESS = "SUCCESS", "Success"
FAILURE = "FAILURE", "Failure"
RETRY = "RETRY", "Retry"
REVOKED = "REVOKED", "Revoked"
REJECTED = "REJECTED", "Rejected"
id = UUIDField(primary_key=True, editable=False)
name = CharField(max_length=200)
state = CharField(max_length=20, choices=State.choices, default=State.PENDING)
args = JSONField(default=dict)
kwargs = JSONField(default=dict)
result = JSONField(default=dict)
clock = PositiveIntegerField(**NB)
created_at = DateTimeField(auto_now_add=True)
def __str__(self) -> str:
return self.name
[docs]
class Provider(Object):
"""🔌 Represents a provider configuration for inference services.
Examples: Ollama, OpenAI, FAL, OpenAI-compatible gateways.
Credentials stay in the environment; ``api_key_env`` is only the variable name.
"""
[docs]
class Kind(TextChoices):
OLLAMA = "ollama", "Ollama"
OPENAI = "openai", "OpenAI"
FAL = "fal", "FAL"
OPENAI_COMPATIBLE = "openai_compatible", "OpenAI-compatible"
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = CharField(max_length=255, **T("name"), default="ollama")
url = CharField(max_length=255, **NB, **T("url"), default=settings.OLLAMA_URL)
kind = CharField(
max_length=32,
choices=Kind.choices,
blank=True,
default="",
**T("kind"),
)
api_key_env = CharField(max_length=64, blank=True, default="", **T("api key env"))
if TYPE_CHECKING:
models: QuerySet['Model']
def __str__(self):
return self.name
[docs]
def save(self, *args, **kwargs):
if not (self.kind or "").strip():
from agency.providers import KIND_OLLAMA, infer_kind
self.kind = infer_kind(self.name, self.url) or KIND_OLLAMA
super().save(*args, **kwargs)
[docs]
def update_models(self):
"""Schedule a Celery task to check the provider and reload the models."""
ollama_models_update = lazy_import('agency.tasks', 'ollama_models_update')
return ollama_models_update.delay(str(self.id))
[docs]
def sync_models(self) -> dict:
"""Fetch models from the provider API and upsert local Model rows."""
from agency.providers import sync_models_for
return sync_models_for(self)
[docs]
class Model(Object):
"""🧠 A LLM model that can be used to generate messages."""
class Meta:
verbose_name = _("model")
verbose_name_plural = _("models")
unique_together = ('name', 'provider')
def __str__(self):
return self.name
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = CharField(max_length=255, **T("name"))
provider = ForeignKey(Provider, on_delete=CASCADE, related_name='models', **T("provider"))
notes = TextField(**NB, **T("notes"))
tools = BooleanField(default=False, **T("tools"))
vision = BooleanField(default=False, **T("vision"))
metadata = JSONField(**NB, **T("metadata"))
modified_at = DateTimeField(auto_now=True, **T("modified at"))
size = FloatField(**NB, **T("size"))
parameter_size = CharField(max_length=8, **NB, **T("parameter size"))
digest = CharField(max_length=64, **NB, **T("digest"))
format = CharField(max_length=255, **NB, **T("format"))
family = CharField(max_length=255, **NB, **T("family"))
removed = BooleanField(default=False, **T("removed"))
if TYPE_CHECKING:
agents: QuerySet['Agent']
[docs]
class Skill(Object):
"""🎯 Skill represents a capability or skill."""
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = CharField(max_length=255, unique=True, **T("name"))
description = TextField(**NB, **T("description"))
text = TextField(**NB, **T("text"))
def __str__(self):
return self.name
[docs]
class Voice(Object):
"""🔊 TTS voice (provider + engine-specific key)."""
[docs]
class Provider(TextChoices):
PIPER = "piper", "Piper"
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = CharField(max_length=255, **T("name"))
provider = CharField(
max_length=32,
choices=Provider.choices,
default=Provider.PIPER,
**T("TTS provider"),
)
key = CharField(max_length=64, **T("voice key"))
language = CharField(max_length=8, blank=True, default="", **T("language"))
locale = CharField(max_length=16, blank=True, default="", **T("locale"))
gender = CharField(max_length=16, blank=True, default="", **T("gender"))
quality = CharField(max_length=16, blank=True, default="", **T("quality"))
class Meta:
verbose_name = _("voice")
verbose_name_plural = _("voices")
unique_together = ("provider", "key")
ordering = ["provider", "name"]
def __str__(self) -> str:
return f"{self.name} ({self.provider})"
[docs]
class Agent(Object):
"""😐 Agents are the logical entities that perform missions.
An agent defines *how* missions are executed by specifying:
- the underlying language model
- the set of tools it can use
- the system prompt to use
- the TTS voice used when the agent speaks
"""
id = UUIDField(primary_key=True, default=uuid.uuid4)
name = CharField(max_length=255)
namespace = CharField(max_length=255, blank=True, default="", db_index=True)
system_key = CharField(max_length=64, unique=True, **NB, **T("system key"))
shipped_content_hash = CharField(max_length=64, blank=True, default="")
is_customized = BooleanField(default=False, **T("customized"))
model = ForeignKey(Model, related_name='agents', on_delete=CASCADE)
voice = ForeignKey(
Voice,
on_delete=SET_NULL,
related_name="agents",
**NB,
**T("voice"),
)
tools = ManyToManyField(Tool, related_name='agents', blank=True)
deep = BooleanField(default=False, **T("deep"))
instructions = TextField(blank=True, **T("instructions"))
icon = ImageField(upload_to="agents/", blank=True, null=True, **T("icon"))
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
if TYPE_CHECKING:
executions: QuerySet['Execution']
threads: QuerySet['Thread']
interruptions: QuerySet['Interruption']
def __str__(self):
return self.name
[docs]
@staticmethod
def by_name(name: str) -> 'Agent':
"""Get an agent by its name.
Args:
name: Agent name to search for
Returns:
Agent instance
Raises:
ValueError: If agent not found
"""
found = Agent.objects.filter(name=name).first()
if not found:
raise ValueError(f"Agent {name} not found")
return found
[docs]
def test(self, user_message: str) -> dict:
"""Invoke the agent with a user message and tools.
Args:
user_message: Message to send to the agent
Returns:
Agent output dictionary
"""
from agency.llm import build_chat_model
graph = create_agent(
model=build_chat_model(self.model),
tools=self.get_tools(),
)
messages = [("system", self.instructions), ("human", user_message)]
output = graph.invoke({"messages": messages})
return output
[docs]
def execute(
self,
mission: Union['Mission', str],
user: AbstractUser,
execution: Optional['Execution'] = None,
) -> dict:
"""Execute a mission commanded by a user.
Args:
mission: Mission instance or mission name
user: User executing the mission
execution: Optional pre-created Execution (and Thread). When set,
reuse it instead of creating a new pair (API dispatch path).
Returns:
Execution result dictionary
"""
if isinstance(mission, str):
mission = Mission.by_name(mission)
if execution is None:
execution = Execution.objects.create(
agent=self,
user=user,
mission=mission,
)
thread = Thread.objects.create(
agent=self,
user=user,
execution=execution,
title=mission.name,
)
else:
thread = execution.threads.order_by('created_at').first()
if thread is None:
thread = Thread.objects.create(
agent=self,
user=user,
execution=execution,
title=mission.name,
)
if execution.status == Execution.Status.STARTING:
execution.status = Execution.Status.RUNNING
execution.save(update_fields=['status', 'updated_at'])
executor: 'ThreadRunner' = lazy_import(
'agency.agent', 'ThreadRunner'
)(thread=thread, user=user, deep=bool(self.deep))
app_log.info(f"Agent {self.name} executing mission {mission.name}")
channel_log(f'EXE Mission "{mission.name}" - {self.name} starts execution thread')
meta = execution.metadata if isinstance(execution.metadata, dict) else {}
user_input = (str(meta.get("input") or "")).strip() or mission.prompt
result = serializable(executor.run(agent=self, input=user_input))
execution.metadata = {
**meta,
"agent": self.name,
"mission": mission.name,
"user": user.username,
"thread": str(thread.id),
"execution": str(execution.id),
}
if "__interrupt__" in result.keys():
execution.status = Execution.Status.INTERRUPTED
channel_log(f'INT Mission "{mission.name}" - {self.name} interrupted')
else:
execution.status = Execution.Status.SUCCESS
channel_log(f'END Mission "{mission.name}" - {self.name} completed successfully')
execution.result = result
execution.save()
return result
[docs]
class Mission(Object):
"""🎯 A Mission defines a reusable objective that an agent can perform.
A mission describes *what* should be done, independent of *who* executes it
or *how many times* it is run. It typically encapsulates a prompt, goal,
or task definition that can be executed by different agents.
When a mission is executed:
- an Execution is created to represent the user request
- a Thread is created to hold execution state
- one or more Runs advance the mission through the thread
"""
id = UUIDField(primary_key=True, default=uuid.uuid4)
name = CharField(max_length=255)
namespace = CharField(max_length=255, blank=True, default="", db_index=True)
system_key = CharField(max_length=64, unique=True, **NB, **T("system key"))
shipped_content_hash = CharField(max_length=64, blank=True, default="")
is_customized = BooleanField(default=False, **T("customized"))
prompt = TextField(blank=True)
loop_prompt = TextField(null=True, blank=True, default=None)
max_iterations = PositiveSmallIntegerField(default=10)
loop_delay_seconds = PositiveIntegerField(default=0)
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
if TYPE_CHECKING:
executions: QuerySet['Execution']
@property
def is_loop(self) -> bool:
return self.loop_prompt is not None
[docs]
def clean(self) -> None:
super().clean()
from agency.mission.loop import normalize_loop_prompt, validate_loop_limits
self.loop_prompt = normalize_loop_prompt(self.loop_prompt)
if self.is_loop:
validate_loop_limits(
max_iterations=self.max_iterations,
loop_delay_seconds=self.loop_delay_seconds,
)
[docs]
def save(self, *args, **kwargs):
from agency.mission.loop import normalize_loop_prompt
self.loop_prompt = normalize_loop_prompt(self.loop_prompt)
super().save(*args, **kwargs)
def __str__(self):
return self.name
[docs]
@staticmethod
def by_name(name: str) -> 'Mission':
"""Get a mission by its name.
Args:
name: Mission name to search for
Returns:
Mission instance
Raises:
ValueError: If mission not found
"""
found = Mission.objects.filter(name=name).first()
if not found:
raise ValueError(f"Mission {name} not found")
return found
[docs]
def dispatch(
self,
agent: Agent = None,
user: AbstractUser = None,
execution: Optional['Execution'] = None,
) -> AsyncResult:
"""Create a Celery task to execute the mission.
Args:
agent: Agent to execute the mission (uses first available if not provided)
user: User executing the mission
execution: Optional pre-created Execution to reuse (API dispatch)
Returns:
Celery AsyncResult for the task
"""
if not agent:
from agency.bootstrap import get_default_agent
agent = get_default_agent()
if not agent:
agent = Agent.objects.create(name="default")
start_mission = lazy_import('agency.tasks', 'start_mission')
execution_id = str(execution.id) if execution is not None else None
return start_mission.delay(str(self.id), str(agent.id), user.pk, execution_id)
[docs]
class Execution(Object):
"""⚡ Execution represents a single business-level request to run a Mission
with a specific Agent on behalf of a user.
An Execution coordinates the work required to fulfill that request and
may involve one or more Threads. Each Thread represents an independent
conversational timeline, while the Execution groups them under a single
intent and lifecycle.
An Execution is responsible for:
- tracking the overall status of the mission
- grouping related Threads
- enabling aggregation of results once all threads complete
Executions are independent and repeatable: the same Mission can be run
multiple times, producing separate Executions with their own Threads,
Runs, and Checkpoints.
"""
[docs]
class Status(TextChoices):
STARTING = "starting", "Starting"
RUNNING = "running", "Running"
SUCCESS = "success", "Success"
INTERRUPTED = "interrupted", "Interrupted"
CANCELLED = "cancelled", "Cancelled"
TIMEOUT = "timeout", "Timeout"
ERROR = "error", "Error"
WAITING = "waiting", "Waiting"
BLOCKED = "blocked", "Blocked"
LIMIT_REACHED = "limit_reached", "Limit reached"
[docs]
class LoopStatus(TextChoices):
IDLE = "idle", "Idle"
RUNNING = "running", "Running"
EVALUATING = "evaluating", "Evaluating"
WAITING = "waiting", "Waiting"
BLOCKED = "blocked", "Blocked"
COMPLETED = "completed", "Completed"
FAILED = "failed", "Failed"
LIMIT_REACHED = "limit_reached", "Limit reached"
id = UUIDField(primary_key=True, default=uuid.uuid4)
user = ForeignKey(User, related_name='executions', on_delete=CASCADE)
agent = ForeignKey(Agent, related_name='executions', on_delete=CASCADE)
mission = ForeignKey(Mission, related_name='executions', on_delete=CASCADE)
status = CharField(max_length=200, choices=Status.choices, default=Status.STARTING)
result = JSONField(default=dict, blank=True)
metadata = JSONField(default=dict, blank=True)
iteration = PositiveIntegerField(default=0)
max_iterations = PositiveSmallIntegerField(default=1)
loop_status = CharField(
max_length=32,
choices=LoopStatus.choices,
null=True,
blank=True,
default=None,
)
next_run_at = DateTimeField(null=True, blank=True)
loop_decision = JSONField(default=dict, blank=True)
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
if TYPE_CHECKING:
threads: QuerySet['Thread']
[docs]
class Thread(Object):
"""🧭 Thread represents a single conversational timeline with a human user.
A Thread is the long-lived identity used by LangGraph to scope memory:
all Runs, Checkpoints, and Interruptions are written and retrieved under
the same `thread_id` (this model's UUID).
A thread may exist in two modes:
1) Orchestrated conversation (execution is set)
- The thread is part of a larger Execution (mission + agent + user).
- The Execution groups one or more threads under a single business
request, while each Thread remains an independent conversational
timeline.
2) Personal conversation (no execution associated)
- Durable operator chat with an Agent (``execution_id`` is NULL).
- At most one active personal Thread per (user, agent).
Threads are linear timelines:
each `invoke()` or `resume()` advances the state forward and typically
produces new checkpoints. To continue a conversation, call LangGraph again
using the same thread_id; to start a new conversation, create a new Thread.
"""
id = UUIDField(primary_key=True, default=uuid.uuid4)
user = ForeignKey(User, related_name='threads', on_delete=CASCADE, **NB)
agent = ForeignKey(Agent, related_name='threads', on_delete=CASCADE, **NB)
execution = ForeignKey(
'Execution',
null=True,
blank=True,
related_name="threads",
on_delete=CASCADE,
)
created_at = DateTimeField(auto_now_add=True)
title = CharField(max_length=255, blank=True)
is_active = BooleanField(default=True)
class Meta:
constraints = [
UniqueConstraint(
fields=["user", "agent"],
condition=Q(is_active=True, execution__isnull=True),
name="uniq_active_personal_thread_per_user_agent",
),
]
if TYPE_CHECKING:
runs: QuerySet['Run']
checkpoints: QuerySet['Checkpoint']
[docs]
def get_execution(self) -> 'Execution':
"""Get the execution of the thread.
Returns:
Execution instance
"""
return self.execution
[docs]
def create_run(self, status: 'Run.Status' = None) -> 'Run':
"""Start a new run.
Args:
status: Optional initial status for the run
Returns:
New Run instance
"""
return Run.objects.create(thread=self, status=status or Run.Status.PENDING)
[docs]
class Run(Object):
"""A Run represents a single technical execution attempt of a thread.
A run tracks one concrete invocation of a LangGraph execution. While a
thread represents a continuous execution timeline, runs capture the
individual attempts to advance that timeline.
Runs may:
- complete successfully
- fail due to errors
- pause due to interruptions (human-in-the-loop)
- resume execution in subsequent runs
Multiple runs can belong to the same thread and execution, allowing
retries and resumptions to be represented explicitly.
Run status reflects the outcome of a specific execution attempt, not
the overall mission result.
"""
[docs]
class Status(TextChoices):
PENDING = "pending", "Pending"
RESUMING = "resuming", "Resuming"
STARTING = "starting", "Starting"
RUNNING = "running", "Running"
SUCCESS = "success", "Success"
INTERRUPTED = "interrupted", "Interrupted"
CANCELLED = "cancelled", "Cancelled"
TIMEOUT = "timeout", "Timeout"
ERROR = "error", "Error"
id = BigAutoField(primary_key=True)
config = JSONField(default=dict, blank=True)
thread = ForeignKey(Thread, related_name="runs", on_delete=CASCADE)
status = CharField(max_length=20, choices=Status.choices, default=Status.PENDING)
error = JSONField(default=dict, blank=True)
metadata = JSONField(default=dict, blank=True)
total_tokens = PositiveIntegerField(default=0)
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
if TYPE_CHECKING:
checkpoints: QuerySet['Checkpoint']
def __str__(self):
return f"Run {self.id}"
[docs]
def get_thread(self) -> Thread:
"""Get the thread of the run.
Returns:
Thread instance
"""
return self.thread
[docs]
def save_status(self, status: Status = Status.STARTING):
"""Save the run status.
Args:
status: Status to set (defaults to STARTING)
"""
self.status = status
self.save()
[docs]
def last_checkpoint(self) -> 'Checkpoint' | None:
"""Get the last checkpoint of the run.
Returns:
Last Checkpoint instance or None
"""
return self.checkpoints.order_by("-created_at").first()
[docs]
def has_interruption(self) -> bool:
"""Check if the run has an interruption.
Returns:
True if run has an interruption, False otherwise
"""
last_checkpoint = self.last_checkpoint()
if not last_checkpoint:
return False
return CheckpointWrite.objects.filter(
thread_id=str(self.thread.id),
checkpoint_id=last_checkpoint.checkpoint_id,
channel="__interrupt__",
).exists()
[docs]
class Checkpoint(Object):
"""A Checkpoint is a durable snapshot of a LangGraph execution state.
A checkpoint captures the full logical state of the graph at a specific
point in time, scoped to a thread. It allows execution to be resumed,
retried, or inspected after the fact.
A checkpoint includes:
- the serialized graph state (channels and values)
- metadata produced by the runtime
- a reference to the parent checkpoint
- associated pending writes (CheckpointWrite entries)
Checkpoints are immutable once written. Execution always resumes from
the latest checkpoint in a thread unless an explicit fork or rollback
mechanism is implemented.
Checkpoints enable long-running, fault-tolerant, and human-in-the-loop
workflows by separating state persistence from execution.
"""
id = BigAutoField(primary_key=True)
thread = ForeignKey(Thread, related_name="checkpoints", on_delete=CASCADE)
run = ForeignKey(Run, related_name="checkpoints", on_delete=CASCADE)
namespace = CharField(max_length=255, default="", db_index=True)
checkpoint_id = CharField(max_length=255, db_index=True)
parent = ForeignKey('self', related_name='children', on_delete=CASCADE, **NB)
ts = DateTimeField(**NB)
data = JSONField(default=dict)
metadata = JSONField(default=dict)
new_versions = JSONField(default=dict)
type = CharField(max_length=255)
blob = BinaryField()
"""Contains the fully serialized graph state using serde."""
created_at = DateTimeField(auto_now_add=True)
if TYPE_CHECKING:
interruptions: QuerySet['Interruption']
messages: QuerySet['Message']
children: QuerySet['Checkpoint']
def __str__(self):
return f"Checkpoint {self.id}"
[docs]
def deserialize_blob(self):
"""Deserialize the blob using the same serde as LangGraph.
Returns:
The deserialized checkpoint state dictionary
"""
serde = JsonPlusSerializer()
return serde.loads_typed((self.type, bytes(self.blob)))
[docs]
def get_interruptions(self) -> QuerySet['Interruption']:
"""Get all interruptions for this checkpoint.
Returns:
QuerySet of Interruption instances ordered by creation time
"""
return self.interruptions.select_related(
'checkpoint', 'checkpoint__run', 'checkpoint__run__thread',
'checkpoint__run__thread__user', 'checkpoint__run__thread__agent', 'agent', 'user'
).order_by('created_at')
[docs]
class CheckpointWrite(Object):
"""A CheckpointWrite represents a pending side effect emitted during
LangGraph execution.
When a node produces an effect (a tool result or an interruption), it
is recorded as a write to a channel and applied later by the runtime
when execution is resumed via pending_writes.
CheckpointWrite persists those pending writes so execution can be
safely paused, resumed, or retried.
"""
id = BigAutoField(primary_key=True)
thread_id = CharField(max_length=255)
checkpoint_id = CharField(max_length=255)
checkpoint_ns = CharField(max_length=255, default="")
task_id = CharField(max_length=255)
task_path = CharField(max_length=1024)
channel = CharField(max_length=255)
"""The special '__interrupt__' channel indicates that the graph execution
intentionally paused and is awaiting external input (e.g. human-in-the-loop).
An interrupt write is consumed when execution is resumed."""
value_type = CharField(max_length=255)
value_blob = BinaryField()
created_at = DateTimeField(auto_now_add=True)
[docs]
class Interruption(Object):
"""😀 Represents an explicit interruption raised by an Agent during execution,
typically to request external input (e.g. human-in-the-loop).
An Interruption pauses the execution of a Thread at a specific Checkpoint.
Resuming an interruption advances the thread forward.
The interruption is always triggered by an Agent. The Agent and the
originating user (i.e. the user who dispatched the mission or owns the
thread context) can be derived from the associated Thread and Execution.
The user assigned to handle the interruption may be different from the
originating user. This allows questions to be routed to other users
(e.g. operators, reviewers, or managers).
Optionally, an interruption may also be resolved by another Agent,
enabling agent-to-agent handoffs or automated resolution strategies.
"""
[docs]
class Status(TextChoices):
PENDING = "pending", "Pending"
ANSWERED = "answered", "Answered"
CANCELLED = "cancelled", "Cancelled"
TIMEOUT = "timeout", "Timeout"
ERROR = "error", "Error"
id = UUIDField(primary_key=True, default=uuid.uuid4)
status = CharField(max_length=200, choices=Status.choices, default=Status.PENDING)
checkpoint = ForeignKey(Checkpoint, related_name='interruptions', on_delete=CASCADE)
priority = IntegerField(default=0)
user = ForeignKey(User, related_name='interruptions', **NB, on_delete=PROTECT)
agent = ForeignKey(Agent, related_name='interruptions', on_delete=PROTECT)
question = JSONField(default=dict, blank=True)
answer = JSONField(default=dict, blank=True)
data = JSONField(default=dict, blank=True)
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
answered_at = DateTimeField(**NB)
resumed_at = DateTimeField(**NB)
[docs]
def get_thread(self) -> Thread:
"""Get the thread of the LangGraph execution.
Returns:
Thread instance from checkpoint.run.thread
"""
checkpoint: Checkpoint = self.checkpoint
run: Run = checkpoint.run
thread: Thread = run.thread
return thread
[docs]
def get_to(self) -> str:
"""Get the 'to' representation: Username(Agent.name) from the interruption.
Returns:
String in format '👤 Username(Agent.name)'
"""
user: AbstractUser = self.user
agent: Agent | None = self.agent
return f"👤 {user.username} ({agent.name})"
[docs]
def get_from(self) -> str:
"""Get the 'from' representation: Agent.name(Username) from the thread.
Returns:
String in format '😃 Agent.name(Username)'
"""
thread = self.get_thread()
user: AbstractUser = thread.user
agent: Agent = thread.agent
return f"😃 {agent.name}({user.username})"
[docs]
@staticmethod
def by_id(interruption_id: str) -> 'Interruption':
"""Get an interruption by its ID.
Args:
interruption_id: UUID of the interruption
Returns:
Interruption instance
"""
return Interruption.objects.select_related(
'checkpoint', 'checkpoint__run', 'checkpoint__run__thread'
).get(id=interruption_id)
[docs]
def get_message(self) -> 'Message' | None:
"""Get the messages for this interruption.
"""
checkpoint: Checkpoint = self.checkpoint
return checkpoint.get_last_tool_call_message()
[docs]
class Session(Object):
"""💬 Session represents a user's browser session for real-time communication.
Sessions map Django's request.session cookie (sessionID) to Django Channels'
scope["session"], enabling WebSocket connections to identify and route messages
to the correct user session.
Sessions are typically created when a user connects via WebSocket and are used
to group related messages and events within a single browser tab/window.
The session_key corresponds to Django's session framework session key, allowing
seamless integration between HTTP and WebSocket request handling.
"""
class Meta:
verbose_name = _('session')
verbose_name_plural = _('sessions')
ordering = ['-created_at']
user = ForeignKey(User, on_delete=CASCADE, related_name='agency_sessions', **T('user'))
data = JSONField(default=dict, **T('data'))
session_key = CharField(max_length=40, unique=True, db_index=True, **T('session key'))
created_at = DateTimeField(auto_now_add=True, **T('created at'))
updated_at = DateTimeField(auto_now=True, **T('updated at'))
if TYPE_CHECKING:
rooms: QuerySet['Room']
def __str__(self):
return f"{self.user.username} - {self.created_at}"
def __repr__(self):
return f"<Session: {self.user.username} - {self.created_at}>"
[docs]
def channel_group_name(self) -> str:
"""Get the channel group name for the session.
Returns:
String in format 'user-<user_id>'
"""
return f"user-{self.user.id}"
[docs]
class Room(Object):
"""🏠 Room represents a shared conversation space for multiple user sessions.
Rooms group multiple Sessions together, enabling multi-user conversations or
collaborative chat experiences. The room ID is used as the Django Channels
group name for broadcasting messages to all participants.
Use cases:
- Multi-user chat rooms
- Shared agent conversations
- Collaborative workspaces
Messages sent to a Room are broadcast to all Sessions in that Room via
Django Channels WebSocket groups.
"""
class Meta:
verbose_name = _('room')
verbose_name_plural = _('rooms')
ordering = ['-created_at']
name = CharField(max_length=255, **T('name'))
created_at = DateTimeField(auto_now_add=True, **T('created at'))
updated_at = DateTimeField(auto_now=True, **T('updated at'))
sessions = ManyToManyField(Session, related_name='rooms', **T('sessions'))
if TYPE_CHECKING:
messages: QuerySet['Message']
def __str__(self):
return self.name
[docs]
class Message(Object):
"""💭 Message represents a single message in a conversation thread.
Messages are the persistent record of all communication between users and agents
during a Run execution, they are BaseMessages from LangChain with extra fields.
**When Messages are Created:**
Source of truth:
- The canonical conversation state lives inside LangGraph checkpoints.
- When a checkpoint is persisted, we extract messages from the checkpoint state
and store them here for efficient querying and UI needs.
Creation:
- Messages are created during checkpoint persistence (CheckpointSaver),
not during middleware/model/tool execution. This ensures durability and
prevents duplication across retries/resumes.
Notes:
- Message.id matches the LangChain/LangGraph message id to keep referential
stability across checkpoints.
- A message is linked to the checkpoint where it first appears (or the
checkpoint that persisted it)
"""
MAX_CONTENT = 2048
class Meta:
verbose_name = _('message')
verbose_name_plural = _('messages')
ordering = ['-created_at']
[docs]
class Role(TextChoices):
HUMAN = 'human', 'Human'
SYSTEM = 'system', 'System'
ASSISTANT = 'assistant', 'Assistant'
TOOL = 'tool', 'Tool'
[docs]
class AlreadyExists(Exception):
"""Raised when attempting to create a Message that already exists.
"""
pass
id = CharField(max_length=255, primary_key=True, **T('id'))
role = CharField(max_length=16, choices=Role.choices, **T('role'))
user = ForeignKey(User, on_delete=CASCADE, related_name='agency_messages', **NB, **T('user'))
room = ForeignKey(Room, on_delete=CASCADE, related_name='messages', **NB, **T('room'))
agent = ForeignKey('Agent', on_delete=CASCADE, related_name='messages', **T('agent'))
checkpoint = ForeignKey(Checkpoint, on_delete=SET_NULL, related_name='messages', **NB, **T('checkpoint'))
"""The checkpoint where this message was created. Used to track which messages
existed at the time of an interruption, enabling efficient querying when resuming."""
content = TextField(**NB, **T('content'))
think = TextField(**NB, **T('think'))
created_at = DateTimeField(default=timezone.now, **T('created at'))
importance = FloatField(default=0.5, **T('importance'))
reply_to = ForeignKey('self', on_delete=SET_NULL, related_name='replies', **NB, **T('reply to'))
metadata = JSONField(**NB, **T('metadata'))
tool_calls = JSONField(**NB, **T('tool calls'))
__ROLES= [Role.HUMAN, Role.SYSTEM, Role.ASSISTANT, Role.TOOL]
__TYPES= [HumanMessage, SystemMessage, AIMessage, ToolMessage]
[docs]
@staticmethod
def get_type_from_role(role: Role) -> type[BaseMessage]:
"""Get the langchain type for a role.
"""
if role not in Message.__ROLES:
raise ValueError(f"Unsupported role: {role}")
return Message.__TYPES[Message.__ROLES.index(role)]
return
[docs]
@staticmethod
def get_role_from_type(message: BaseMessage) -> Role:
"""Get the role from a langchain message type.
"""
if type(message) not in Message.__TYPES:
raise ValueError(f"Unsupported message type: {type(message)}")
return Message.__ROLES[Message.__TYPES.index(type(message))]
def __str__(self):
content_preview = (self.content[:50] + '...') if self.content and len(self.content) > 50 else (self.content or '')
return f"{self.role}: {content_preview}"
[docs]
@staticmethod
def register_checkpoint_message(message: BaseMessage, checkpoint: Checkpoint, agent: Agent) -> Message | None:
"""Convert a LangGraph BaseMessage to a Message model.
Args:
message: BaseMessage instance
checkpoint: Checkpoint instance to associate with
agent: Agent instance (required field)
Returns:
Message instance
Raises:
Message.AlreadyExists: If message with same id already exists
"""
assert isinstance(message, BaseMessage)
role = Message.get_role_from_type(message)
from agency.message_content import serialize_lc_content
content = serialize_lc_content(message.content)
tool_calls = None
if isinstance(message, AIMessage) and hasattr(message, 'tool_calls') and message.tool_calls:
tool_calls = [tc.model_dump() if hasattr(tc, 'model_dump') else dict(tc) for tc in message.tool_calls]
metadata = None
if hasattr(message, 'additional_kwargs') and message.additional_kwargs:
metadata = message.additional_kwargs
user = None
if role == Message.Role.HUMAN:
thread = getattr(checkpoint, "thread", None)
if thread is None and getattr(checkpoint, "run_id", None):
thread = checkpoint.run.thread
if thread is not None:
user = thread.user
try:
return Message.objects.create(
id=message.id,
role=role,
user=user,
agent=agent,
checkpoint=checkpoint,
content=content,
tool_calls=tool_calls,
metadata=metadata,
)
except IntegrityError: # Duplicate primary key
raise Message.AlreadyExists(f"Message with id '{message.id}' already exists")
[docs]
@classmethod
def get_messages_until_checkpoint(cls, thread: 'Thread', checkpoint: 'Checkpoint') -> QuerySet['Message']:
"""Get all messages for a thread up to and including a specific checkpoint.
This is useful when resuming from an interruption to determine which messages
were already persisted before the interruption occurred.
Note: When resuming, a new Run is created, but we want messages from all
previous runs in the same thread. This method queries by thread, not run.
Args:
thread: The Thread to query messages for
checkpoint: The Checkpoint to query up to (messages created at or before
this checkpoint's creation time)
Returns:
QuerySet of Message objects ordered by creation time
"""
return cls.objects.filter(
checkpoint__thread=thread,
checkpoint__id__lte=checkpoint.id,
).order_by('checkpoint__id', 'created_at')
[docs]
def to_message(self) -> BaseMessage:
"""Convert to a LangChain BaseMessage.
Returns:
BaseMessage instance (SystemMessage, HumanMessage, AIMessage, or ToolMessage)
Raises:
ValueError: If role is unsupported
"""
cls = Message.get_type_from_role(self.role)
if not cls:
raise ValueError(f"Unsupported role for message: {self.role}")
from agency.message_content import deserialize_stored_content
raw = deserialize_stored_content(self.content)
if isinstance(raw, str):
content = raw
elif raw is None:
content = ""
else:
content = raw
if isinstance(content, str) and len(content) > Message.MAX_CONTENT:
content = content[: Message.MAX_CONTENT] + "..."
kwargs = {"content": content, "id": self.id}
# Handle tool_calls for AIMessage
if self.role == self.Role.ASSISTANT and self.tool_calls:
kwargs["tool_calls"] = self.tool_calls
# ToolMessage requires tool_call_id (LangChain); prefer metadata, then id.
if self.role == self.Role.TOOL:
meta = self.metadata if isinstance(self.metadata, dict) else {}
tool_call_id = (
meta.get("tool_call_id")
or meta.get("tool_callId")
or self.id
)
kwargs["tool_call_id"] = str(tool_call_id)
if meta.get("name"):
kwargs["name"] = meta["name"]
# Handle additional_metadata if needed
if self.metadata:
kwargs["additional_kwargs"] = self.metadata
return cls(**kwargs)
[docs]
class UserPreferences(Object):
"""Per-user Mission Control prefs: workspace, UI, operator, system voice."""
user = OneToOneField(
User,
on_delete=CASCADE,
related_name="agency_preferences",
primary_key=True,
**T("user"),
)
active_workspace = ForeignKey(
"corpus.Workspace",
on_delete=SET_NULL,
related_name="+",
**NB,
**T("active workspace"),
)
active_agent = ForeignKey(
Agent,
on_delete=SET_NULL,
related_name="+",
**NB,
**T("active agent"),
)
active_thread = ForeignKey(
"Thread",
on_delete=SET_NULL,
related_name="+",
**NB,
**T("active thread"),
)
system_voice = ForeignKey(
Voice,
on_delete=SET_NULL,
related_name="+",
**NB,
**T("system voice"),
)
desktop = JSONField(default=dict, blank=True, **T("desktop"))
language = CharField(max_length=10, blank=True, default="", **T("language"))
timezone = CharField(max_length=64, blank=True, default="", **T("timezone"))
updated_at = DateTimeField(auto_now=True, **T("updated at"))
class Meta:
verbose_name = _("user preferences")
verbose_name_plural = _("user preferences")
def __str__(self) -> str:
return f"prefs:{self.user_id}"
[docs]
def to_api(self) -> dict[str, Any]:
from agency.ui.language import normalize_language
from agency.voices import voice_brief
ws = self.active_workspace_id
agent = self.active_agent
lang = normalize_language(self.language) if self.language else None
tz = (self.timezone or "").strip() or None
system = self.system_voice
thread_id = self.active_thread_id
return {
"workspaceId": str(ws) if ws else None,
"activeAgentId": str(agent.pk) if agent else None,
"activeAgentName": agent.name if agent else None,
"activeThreadId": str(thread_id) if thread_id else None,
"desktop": self.desktop if isinstance(self.desktop, dict) else {},
"language": lang,
"timezone": tz,
"systemVoiceId": str(system.pk) if system else None,
"ttsVoice": system.key if system else None,
"systemVoice": voice_brief(system),
}
[docs]
@classmethod
def for_user(cls, user) -> "UserPreferences":
"""Get or create prefs for an authenticated user."""
from agency.corpus.bootstrap import ensure_default_tenant_membership
prefs, created = cls.objects.select_related(
"active_agent", "system_voice", "active_thread"
).get_or_create(user=user)
if created:
prefs = cls.objects.select_related(
"active_agent", "system_voice", "active_thread"
).get(pk=prefs.pk)
ensure_default_tenant_membership(user)
return prefs
[docs]
class Photo(Object):
"""🖼 User photo library (chat uploads, explorer, generated)."""
[docs]
class Origin(TextChoices):
UPLOAD = "upload", "Upload"
GENERATED = "generated", "Generated"
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = ForeignKey(User, on_delete=CASCADE, related_name="photos", **T("user"))
file = ImageField(upload_to=photo_upload_to, **T("file"))
title = CharField(max_length=255, blank=True, default="", **T("title"))
origin = CharField(
max_length=16,
choices=Origin.choices,
default=Origin.UPLOAD,
db_index=True,
**T("origin"),
)
source = CharField(max_length=64, blank=True, default="", **T("source"))
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
class Meta:
ordering = ["-created_at"]
indexes = [
Index(fields=["user", "origin", "-created_at"], name="agency_photo_user_origin"),
]
verbose_name = _("photo")
verbose_name_plural = _("photos")
def __str__(self) -> str:
return self.title or str(self.id)
[docs]
class UiPanel(Object):
"""Persisted operator UIDocument. Palette = registered schema types."""
[docs]
class Scope(TextChoices):
SESSION = "session", "Session"
WORKSPACE = "workspace", "Workspace"
EPHEMERAL = "ephemeral", "Ephemeral"
id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
owner = ForeignKey(User, on_delete=CASCADE, related_name="ui_panels", **T("owner"))
workspace = ForeignKey(
"corpus.Workspace",
on_delete=SET_NULL,
related_name="ui_panels",
**NB,
**T("workspace"),
)
scope = CharField(
max_length=16,
choices=Scope.choices,
default=Scope.SESSION,
**T("scope"),
)
title = CharField(max_length=255, **T("title"))
channel = CharField(max_length=64, default="agency.panel", **T("channel"))
document = JSONField(default=dict, **T("document"))
created_at = DateTimeField(auto_now_add=True, **T("created at"))
updated_at = DateTimeField(auto_now=True, **T("updated at"))
class Meta:
ordering = ["-updated_at"]
verbose_name = _("UI panel")
verbose_name_plural = _("UI panels")
indexes = [
Index(fields=["owner", "scope", "-updated_at"], name="agency_uipanel_owner"),
]
def __str__(self) -> str:
return self.title or str(self.id)