"""Agent runtime for Agency.
Public API (this module — not yet split into subpackages):
- ``ThreadRunner`` — invoke / resume a Thread (creates LangGraph graph + Run)
- ``RunMiddleware`` — run lifecycle, token accounting, Monitor / channel events
- ``CheckpointSaver`` — DB-backed LangGraph checkpointer
- ``TokenUsage`` — token counters attached to a Run
Supporting types: ``State``, ``Configurable``.
Experimental / not product policy: ``agency.agent.policy`` (structured-output
demos). Prefer ``ThreadRunner`` over ad-hoc graph scripts.
Built on LangChain / LangGraph (``create_agent``, optional deep agents).
"""
from __future__ import annotations
import json
import uuid
from typing import TYPE_CHECKING, TypedDict, Any, Iterator, Optional, Sequence, Tuple, List, Callable
if TYPE_CHECKING:
from langgraph.checkpoint import base as chk
from langgraph.checkpoint.serde.base import SerializerProtocol
from langchain.agents.middleware import AgentState
from langchain.agents.middleware.types import ModelCallResult, ToolCallRequest, StateT, ContextT
from django.contrib.auth import get_user_model
from django.db import transaction
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import (
AgentMiddleware,
ModelRequest,
ModelResponse,
SummarizationMiddleware,
HumanInTheLoopMiddleware,
wrap_model_call,
wrap_tool_call,
dynamic_prompt,
)
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
SystemMessage,
ToolMessage,
ToolCall,
UsageMetadata,
)
from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.graph.state import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import Interrupt, Command
from deepagents import create_deep_agent
from agency.models import (Agent, Mission, Thread, Execution,
Run, Tool, Model, Interruption, Checkpoint, CheckpointWrite, Message)
from agency.bus import AgentPresence, Presence
from agency.channel import channel_log
from agency.channel.protocol import Monitor
from agency.utils import app_log, serializable
from agency.llm import build_chat_model, model_audit, ChatModelError
def _is_graph_interrupt(exc: BaseException) -> bool:
"""True when a tool raised LangGraph interrupt (HITL) — not a tool failure."""
name = type(exc).__name__
return name in ("GraphInterrupt", "NodeInterrupt") or "Interrupt" in name
if TYPE_CHECKING:
_ToolCallHandler = Callable[[ToolCallRequest], ToolMessage | Command[Any]]
"""Handler for tool calls."""
_ModelCallHandler = Callable[[ModelRequest], ModelResponse]
"""Handler for model calls."""
[docs]
class State(TypedDict):
question: str
answer: str
docs: list[str]
def _cfg_get(cfg: dict, key: str, default=None):
return (cfg or {}).get("configurable", {}).get(key, default)
def _cfg_set(cfg: dict, **kwargs):
cfg = dict(cfg or {})
configurable = dict(cfg.get("configurable", {}))
configurable.update(kwargs)
cfg["configurable"] = configurable
return cfg
[docs]
class Configurable:
"""A class to wrap the configurable dictionary of the RunnableConfig.
"""
[docs]
def __init__(self, config: "chk.RunnableConfig"):
self.config = config
thread_id = _cfg_get(config, "thread_id")
self.thread_id = str(thread_id) if thread_id else None
checkpoint_id = _cfg_get(config, "checkpoint_id")
self.checkpoint_id:str|None = str(checkpoint_id) if checkpoint_id else None
self.checkpoint_ns:str = _cfg_get(config, "checkpoint_ns", "") or ""
self.run_id = _cfg_get(config, "run_id")
#channel_log(f"Configurable: thread_id={self.thread_id}, checkpoint_id={self.checkpoint_id}, checkpoint_ns={self.checkpoint_ns}, run_id={self.run_id}")
[docs]
class CheckpointSaver(BaseCheckpointSaver):
"""Database-backed checkpoint saver for LangGraph executions.
This class persists LangGraph checkpoints and pending writes in the database
so that graph execution can be paused, resumed, retried, and audited
across process restarts.
It implements LangGraph's checkpointing contract by storing:
- the serialized graph state (checkpoint)
- associated metadata
- parent checkpoint linkage
- pending writes (side effects not yet applied)
- messages from the checkpoint state
This persistence layer is the foundation for:
- human-in-the-loop interruptions
- long-running or asynchronous agent workflows
- deterministic resume via thread_id
- full execution auditability
Conceptually:
- Thread = identity of a long-lived execution context
- Run = a concrete technical execution attempt
- Checkpoint = snapshot of graph state at a point in time
- CheckpointWrite = pending side effects (e.g. interrupts)
The saver is intentionally stateless; all execution state lives in the DB.
"""
[docs]
def __init__(self, *, serde=None):
super().__init__(serde=serde or JsonPlusSerializer())
self.saved_message_ids = []
def _base_message_to_model(self, message: BaseMessage, checkpoint: Checkpoint, agent: Agent) -> Message | None:
"""Convert a LangGraph BaseMessage to a Message model.
Delegates the actual creation logic to ``Message.register_checkpoint_message`` so that
all Message construction rules live in the model instead of the checkpoint
saver.
Args:
message: BaseMessage instance
checkpoint: Checkpoint instance to associate with
agent: Agent instance (required field)
Returns:
Message instance or None if message already exists or cannot be created.
"""
if not isinstance(message, BaseMessage):
return None
if message.id in self.saved_message_ids:
return None
try:
msg_model = Message.register_checkpoint_message(
message=message,
checkpoint=checkpoint,
agent=agent,
)
if msg_model is not None:
self.saved_message_ids.append(message.id)
return msg_model
except Message.AlreadyExists:
# Message already exists, add to cache to avoid retrying
self.saved_message_ids.append(message.id)
return None
except Exception as e:
# Unexpected error occurred inside register_checkpoint_message
channel_log(f"Failed to create message {getattr(message, 'id', None)}: {e}")
self.saved_message_ids.append(message.id)
return None
def _save_messages_from_checkpoint(self, checkpoint: dict, checkpoint_obj: Checkpoint, thread: Thread):
"""Extract and save messages from checkpoint state.
Args:
checkpoint: The checkpoint dict from LangGraph
checkpoint_obj: The saved Checkpoint model instance
thread: The Thread instance
"""
if not thread.agent:
channel_log(f"Cannot save messages: thread {thread.id} has no agent")
return
# Get messages from checkpoint state
# Messages can be in channel_values['messages'] or directly in the state
channel_values = checkpoint.get('channel_values', {})
messages = channel_values.get('messages', [])
# If no messages in channel_values, try the state directly (for AgentState)
if not messages:
messages = checkpoint.get('messages', [])
if not messages:
return
saved_count = 0
for msg_data in messages:
try:
message = None
# Try to deserialize the message
# Messages in checkpoint can be:
# 1. Already BaseMessage instances (unlikely in serialized checkpoint)
# 2. Serialized tuples (type, blob)
# 3. Dict representations
if isinstance(msg_data, tuple) and len(msg_data) == 2:
# Serialized as (type, blob)
msg_type, msg_blob = msg_data
message = self.serde.loads_typed((msg_type, msg_blob))
elif isinstance(msg_data, dict):
# Check if it's a serialized message tuple in dict form
if 'type' in msg_data and 'data' in msg_data:
msg_type = msg_data['type']
msg_blob = msg_data['data']
message = self.serde.loads_typed((msg_type, msg_blob))
elif 'id' in msg_data and 'type' in msg_data:
# Dict representation of a message (from data field)
# Try to reconstruct - this is a fallback
# We need the actual BaseMessage class, skip for now
continue
elif isinstance(msg_data, BaseMessage):
# Already a BaseMessage (shouldn't happen in serialized checkpoint)
message = msg_data
if message:
msg_model = self._base_message_to_model(message, checkpoint_obj, thread.agent)
if msg_model:
saved_count += 1
except Exception as e:
channel_log(f"Failed to save message from checkpoint: {e}", {"message_data": str(msg_data)[:100]})
continue
if saved_count > 0:
channel_log(f"Saved {saved_count} messages from checkpoint {checkpoint_obj.checkpoint_id}")
[docs]
@transaction.atomic
def put(self, config: dict, checkpoint: dict, metadata: dict, new_versions: dict) -> dict:
"""Persist a checkpoint produced by LangGraph.
This method is called by LangGraph whenever the graph reaches a point
where its state should be snapshotted.
Responsibilities:
- Serialize the checkpoint value using JsonPlusSerializer
- Associate the checkpoint with a Thread (via thread_id)
- Optionally associate it with a Run (via run_id, if provided)
- Record metadata and parent checkpoint linkage
- Return an updated config containing the checkpoint identifiers
The returned config may include:
- checkpoint_id
- checkpoint namespace
which LangGraph can use internally for subsequent operations.
"""
checkpoint_id = checkpoint.get("id")
if not checkpoint_id:
raise ValueError("Missing checkpoint_id")
cfg = Configurable(config)
if not cfg.thread_id:
raise ValueError("Missing configurable.thread_id")
thread:Thread = Thread.objects.get(id=cfg.thread_id)
if not thread:
raise ValueError(f"Thread {cfg.thread_id} not found")
checkpoint_ns = cfg.checkpoint_ns
parent = None
parent_cfg = checkpoint.get("parent_config")
if parent_cfg:
parent_id = _cfg_get(parent_cfg, "checkpoint_id")
if parent_id:
parent = Checkpoint.objects.filter(
thread=thread,
namespace=checkpoint_ns,
checkpoint_id=parent_id,
).first()
c_type, c_blob = self.serde.dumps_typed(checkpoint)
ts = checkpoint.get("ts")
dt_ts = parse_datetime(ts) if isinstance(ts, str) else None
# Serialize checkpoint to JSON-serializable dict for data field
# The blob is already JSON-serialized, so decode and parse it
data = serializable(checkpoint)
obj, _ = Checkpoint.objects.update_or_create(
thread=thread,
namespace=checkpoint_ns,
checkpoint_id=checkpoint_id,
defaults=dict(
parent=parent,
run_id=cfg.run_id,
ts=dt_ts,
metadata=metadata or {},
new_versions=new_versions or {},
type=c_type,
blob=c_blob,
data=data,
),
)
channel_log(f"Checkpoint {checkpoint_ns}:{checkpoint_id} "+ "created" if _ else "updated")
# Save messages from checkpoint state
# Save for both new and updated checkpoints to catch any new messages
self._save_messages_from_checkpoint(checkpoint, obj, thread)
return _cfg_set(
config,
thread_id=str(thread.id),
checkpoint_ns=checkpoint_ns,
checkpoint_id=obj.checkpoint_id,
)
[docs]
@transaction.atomic
def put_writes(
self,
config: dict,
writes: Sequence[Tuple[str, Any]],
task_id: str,
task_path: str = "",
) -> None:
"""Persist pending writes generated during graph execution.
Pending writes represent side effects that were produced by a node
but not yet applied to the graph state. Examples include:
- interrupts (__interrupt__)
- tool outputs
- internal control signals
LangGraph uses a two-phase model:
- First, nodes emit writes
- Later, writes are committed and applied
This method stores those writes in the database so they can be:
- replayed on resume
- inspected for auditing/debugging
- used to detect interruptions (e.g. human-in-the-loop)
Each write is stored with:
- thread_id
- checkpoint_id
- channel name
- serialized value
- task metadata (task_id, task_path)
"""
cfg = Configurable(config)
if not (cfg.thread_id and cfg.checkpoint_id):
raise ValueError("put_writes requires thread_id and checkpoint_id")
rows = []
for channel, value in writes:
v_type, v_blob = self.serde.dumps_typed(value)
rows.append(
CheckpointWrite(
thread_id=cfg.thread_id,
checkpoint_id=cfg.checkpoint_id,
checkpoint_ns=cfg.checkpoint_ns,
#run_id=run_id,
task_id=task_id,
task_path=task_path or "",
channel=channel,
value_type=v_type,
value_blob=v_blob,
)
)
channel_log(f"Putting {len(rows)} writes for {cfg.thread_id}:{cfg.checkpoint_id}")
CheckpointWrite.objects.bulk_create(rows)
[docs]
def get_tuple(self, config: dict) -> Optional[CheckpointTuple]:
"""Load a checkpoint tuple for LangGraph execution or resume.
This method reconstructs the full execution context expected by LangGraph
when continuing or resuming a graph. It returns a CheckpointTuple containing:
- the checkpointed graph state
- metadata
- parent checkpoint reference
- pending writes that were not yet applied
The pending writes are critical: they allow LangGraph to correctly
resume interrupted executions (e.g. human-in-the-loop) by replaying
side effects such as '__interrupt__'.
The checkpoint is resolved using:
- thread_id (required)
- optional checkpoint namespace
If multiple checkpoints exist, the most recent one is selected.
This method is the authoritative source of truth for reconstructing
graph execution state from persistent storage.
"""
cfg = Configurable(config)
if not cfg.thread_id:
return None
qs = Checkpoint.objects.filter(
thread_id=cfg.thread_id,
namespace=cfg.checkpoint_ns,
)
if cfg.checkpoint_id:
qs = qs.filter(checkpoint_id=cfg.checkpoint_id)
obj:Checkpoint|None = qs.order_by("-created_at").first()
if not obj:
return None
out_config: "chk.RunnableConfig" = _cfg_set(
config,
thread_id=str(obj.thread.id),
checkpoint_ns=obj.namespace,
checkpoint_id=obj.checkpoint_id,
)
checkpoint: "chk.Checkpoint" = obj.deserialize_blob()
metadata: "chk.CheckpointMetadata" = obj.metadata or {}
parent_config: "chk.RunnableConfig | None" = None
if obj.parent:
parent_config = _cfg_set(
{"configurable": {}},
thread_id=str(obj.thread.id),
checkpoint_ns=obj.namespace,
checkpoint_id=obj.parent.checkpoint_id,
)
pending_writes: list["chk.PendingWrite"] = []
writes = CheckpointWrite.objects.filter(
thread_id=cfg.thread_id,
checkpoint_id=cfg.checkpoint_id
).order_by("created_at")
for w in writes:
val = self.serde.loads_typed((w.value_type, bytes(w.value_blob)))
pending_writes.append((w.task_id, w.channel, val, w.task_path))
channel_log(f"Getting {len(pending_writes)} writes for {cfg.thread_id}:{cfg.checkpoint_id}")
return CheckpointTuple(
out_config,
checkpoint = checkpoint,
metadata = metadata,
parent_config = parent_config,
pending_writes = pending_writes,
)
[docs]
class TokenUsage:
"""💰 Tokens usage helper class.
TODO: Check if we are counting multiple times the same messages.
"""
input: int = 0
output: int = 0
total: int = 0
[docs]
def __init__(self, msg: BaseMessage | dict | None = None):
if isinstance(msg, AIMessage):
usage: UsageMetadata = msg.usage_metadata
self.total = usage.get('total_tokens', 0) or 0
self.input = usage.get('input_tokens', 0) or 0
self.output = usage.get('output_tokens', 0) or 0
elif isinstance(msg, dict):
usage = msg.get('usage_metadata')
if usage:
if isinstance(usage, dict):
self.total = usage.get('total_tokens', 0) or 0
self.input = usage.get('input_tokens', 0) or 0
self.output = usage.get('output_tokens', 0) or 0
def __add__(self, other: TokenUsage) -> TokenUsage:
self.total += other.total
self.input += other.input
self.output += other.output
return self
[docs]
class RunMiddleware(AgentMiddleware):
"""Middleware to wrap agent's tools and model calls.
Manages the run lifecycle and token usage.
Publishes event updates to channels.
"""
[docs]
def __init__(self, run: Run, interrupt_on: set[str] = None):
"""
"""
self.run = run
"""The Run instance."""
self.saved_message_ids: list[str] = []
self.monitor = Monitor(self.run)
self.interrupt_on = interrupt_on or set()
self.tokens = TokenUsage()
self.started = False
self.presence = AgentPresence.for_run(self.run)
"""Agent Bus presence for this run, shared with the ThreadRunner."""
[docs]
def before_agent(self, state: "StateT", runtime: Runtime["ContextT"]) -> dict[str, Any] | None:
"""Execution starts.
"""
# Send standardized agency event
self.run.save_status(Run.Status.RUNNING)
self.monitor.emit(Monitor.Event.RUN_START)
self.presence.emit(Presence.THINKING)
self.started = True
return None
[docs]
def check_new_message(self, message: BaseMessage):
"""Check if a message is new (not yet seen in this run).
This is used to determine if a message should be sent to the user channel.
The actual persistence is handled by CheckpointSaver when checkpoints are saved.
Returns:
True if message is new, False if already seen
"""
if message.id in self.saved_message_ids:
return False
# Mark as seen (will be persisted by CheckpointSaver)
self.saved_message_ids.append(message.id)
return True
[docs]
def wrap_model_call(
self,
request: ModelRequest,
handler: _ModelCallHandler,
) -> "ModelCallResult":
"""Model call wrapper.
"""
channel_log(f"EXE run {self.run.id} calls model {request.model} with {len(request.messages)} messages")
# Send messages to user channel (blue messages)
#self.send.stream(request.messages)
# Publish model request event
model_info = str(request.model) if hasattr(request.model, '__str__') else None
self.monitor.emit(Monitor.Event.MODEL_REQUEST, {
"model": model_info,
"messages_count": len(request.messages),
})
self.presence.emit(Presence.THINKING)
# Messages will be published via post_save signal when saved by CheckpointSaver
try:
response = handler(request)
if isinstance(response, ModelResponse):
result:list[BaseMessage] = response.result
tool_calls_count: int = 0
response_messages:list[BaseMessage] = []
# Handle result as list (common case)
if isinstance(result, list) and len(result) > 0:
# Iterate over all messages and sum tokens from those with usage_metadata
for msg in result:
channel_log(f"*RESULT TYPE {type(msg).__name__}")
usage = TokenUsage(msg)
self.tokens += usage
# Count tool_calls from all messages
if hasattr(msg, 'tool_calls') and msg.tool_calls:
tool_calls_count += len(msg.tool_calls)
elif isinstance(msg, dict) and msg.get('tool_calls'):
tool_calls_count += len(msg['tool_calls'])
# Track new messages (will be published via post_save signal when saved by CheckpointSaver)
if isinstance(msg, (AIMessage, ToolMessage)):
self.check_new_message(msg) # Mark as seen
channel_log(f"EXE run {self.run.id} response tokens {self.tokens.total} tools {tool_calls_count}", {"response": response.result})
# Publish model response event
# Use the last usage for this specific response, but self.tokens has the accumulated total
self.monitor.emit(Monitor.Event.MODEL_RESPONSE, {
"total_tokens": self.tokens.total,
"input_tokens": self.tokens.input,
"output_tokens": self.tokens.output,
"tool_calls_count": tool_calls_count,
"response": response.result,
})
else:
channel_log(f"EXE run {self.run.id} response type {type(response).__name__}", {"response": type(response).__name__})
# Publish model response event (non-standard response type)
self.monitor.emit(Monitor.Event.MODEL_RESPONSE, {
"response_type": type(response).__name__,
})
if self.run.total_tokens != self.tokens.total:
self.run.total_tokens = self.tokens.total
self.run.save(update_fields=['total_tokens'])
return response
except Exception as e:
channel_log("MID MODEL CALL ERROR", {"error": str(e)})
# Publish model error event
self.monitor.emit(Monitor.Event.RUN_ERROR, {
"error": str(e),
"error_type": type(e).__name__,
"context": "model_call",
})
raise
[docs]
def after_agent(self, state: "StateT", runtime: Runtime["ContextT"]) -> dict[str, Any] | None:
"""Run ends.
"""
self.run.save_status(Run.Status.SUCCESS)
channel_log(f"END run {self.run.id} total tokens {self.tokens.total}={self.tokens.input}+{self.tokens.output}")
# Publish run end event
self.monitor.emit(Monitor.Event.RUN_END, {
"total_tokens": self.tokens.total,
"input_tokens": self.tokens.input,
"output_tokens": self.tokens.output,
})
self.presence.emit(Presence.IDLE)
[docs]
class ThreadRunner:
"""
ThreadRunner controls the execution of a LangGraph agent.
Each invocation or resume advances the associated Thread forward.
Creates the executable graph instance (LangGraph agent) with:
- The Agent model and tools
- The CheckpointSaver to persist checkpoints and messages
- The RunMiddleware: Supervises the run lifecycle and token usage
Operational overview:
- Invoke: invokes the graph and checks for interruptions
- Run: Initialize the runner and create a new Run
- Resume: resumes the interrupted execution of the Thread
"""
[docs]
def __init__(self,
user = None,
model: Optional[Model] = None,
thread: Optional[Thread] = None,
deep = False,
):
self.user = user
self.model_override = model
self.thread = thread
self.deep = deep
self.presence: AgentPresence | None = None
"""Presence of the running Agent, owned by the middleware once a graph exists.
Interruption and failure are decided out here, not inside the graph, so
both sides publish through the same object or they would fight over
which state the Agent is in.
"""
[docs]
def resolve_model(self, agent: Agent) -> Model:
"""Model for this run: explicit override, else the agent's configured model."""
if self.model_override is not None:
return self.model_override
if not agent.model_id:
raise ChatModelError(f"Agent {agent.name!r} has no model configured.")
return agent.model
[docs]
def create_graph(self, agent: Agent, run: Run):
"""Compile a LangGraph agent graph.
"""
db_model = self.resolve_model(agent)
chat_model = build_chat_model(db_model)
audit = model_audit(db_model)
run.metadata = {**(run.metadata or {}), "llm": audit}
run.save(update_fields=["metadata"])
middleware = RunMiddleware(run)
self.presence = middleware.presence
kw = {
"model": chat_model,
"tools": agent.get_tools(),
"checkpointer": CheckpointSaver(),
"middleware": [middleware],
}
graph = create_deep_agent(**kw) if (self.deep or bool(agent.deep)) else create_agent(**kw)
return graph
[docs]
def run(self, agent: Agent, input: str, run: Optional[Run] = None, images: Optional[list] = None) -> dict:
"""Execute the mission: Invoke the graph with the given input.
This method performs a single execution attempt ("Run") of the agent
within the associated Thread. It:
1. Creates a Run record to track the technical execution attempt
2. Invokes the LangGraph agent with the current thread configuration
3. Persists checkpoints automatically via the CheckpointSaver
Handles three possible outcomes:
- SUCCESS:
The graph completes normally and returns a final output.
- INTERRUPTION (human-in-the-loop):
A tool calls `interrupt(...)`, causing the graph to pause.
The current graph state is already persisted as a Checkpoint.
An Interruption record is created and execution is suspended
until a human provides input.
- ERROR:
An unexpected exception occurs during execution.
The checkpoint associated with an interruption is resolved by
selecting the most recent checkpoint written for the current Run.
This guarantees that execution can later be resumed from the exact
point where the interruption occurred.
Returns:
- On success: the graph output (dict)
- On interruption: a dict containing the interruption identifier
- On error: raises the exception after persisting Run state
"""
if not self.thread:
self.thread = Thread.objects.create(agent=agent, user=self.user)
run = run or self.thread.create_run()
messages = []
if agent.instructions:
messages.append(SystemMessage(content=agent.instructions))
if images:
from agency.channel.consumer.agent import _build_human_message
messages.append(_build_human_message(input, images))
else:
messages.append(("user", input))
#Aqui no guardamos los mensajes del usuario en la base de datos, preferimos checkpointsaver
return self.invoke({"messages": messages},
config={
"configurable": {
"thread_id": str(self.thread.id),
"run_id": run.id,
"checkpoint_ns": "",
}
},
run=run,
agent=agent)
[docs]
def resume(self, interruption: Interruption) -> dict:
"""Resume a previously interrupted execution.
Resuming continues the same Thread (same thread_id / timeline) from the
latest persisted checkpoint, providing the human answer back to the graph
via Command(resume=...).
This method creates a new Run to track the technical resume attempt. This
keeps auditing clear: the original run may be INTERRUPTED, while the resume
run may succeed, error, or interrupt again.
Args:
interruption: The interruption record containing the persisted checkpoint
reference and the human answer payload.
Returns:
The graph output after resuming, which may complete successfully or
produce new interruptions.
"""
self.thread = interruption.get_thread()
agent:Agent = self.thread.agent
run:Run = self.thread.create_run(status=Run.Status.RESUMING)
try:
execution = self.thread.get_execution()
mission_name = (
execution.mission.name
if execution is not None and execution.mission_id
else "—"
)
except Exception:
mission_name = "—"
channel_log(f'RES Mission "{mission_name}" - {agent.name} {run.status} execution thread')
return self.invoke(Command(resume=interruption.answer),
config={
"configurable": {
"thread_id": str(self.thread.id),
"checkpoint_ns": "",
"run_id": run.id,
}
},
agent=agent,
run=run)
[docs]
def continue_turn(self, message: str, run: Optional[Run] = None, images: Optional[list] = None) -> dict:
"""Advance an existing Thread with a new user message (post-mission chat).
Uses the same ``thread_id`` / checkpointer so LangGraph appends to the
prior timeline. Does **not** re-inject system instructions (already in
checkpoint). Creates a new Run for audit.
"""
if not self.thread:
raise ValueError("continue_turn requires a Thread")
text = (message or "").strip()
if not text and not images:
raise ValueError("continue_turn requires a non-empty message")
if not text:
text = "Describe what you see."
agent: Agent = self.thread.agent
run = run or self.thread.create_run()
try:
execution = self.thread.get_execution()
mission_name = (
execution.mission.name
if execution is not None and execution.mission_id
else "—"
)
except Exception:
mission_name = "—"
channel_log(
f'CONT Mission "{mission_name}" - {agent.name} continues conversation'
)
if images:
from agency.channel.consumer.agent import _build_human_message
user_msg = _build_human_message(text, images)
else:
user_msg = ("user", text)
return self.invoke(
{"messages": [user_msg]},
config={
"configurable": {
"thread_id": str(self.thread.id),
"checkpoint_ns": "",
"run_id": run.id,
}
},
agent=agent,
run=run,
)
[docs]
def invoke(self, input, config: dict, agent: Agent, run: Run) -> dict:
"""Invoke the LangGraph agent and normalize outcomes into Run +
Interruption persistence.
This is the shared execution primitive used by both run() and resume(). It:
- Marks the run as RUNNING
- Calls graph.invoke(...)
- Detects interruptions via the '__interrupt__' key in the output
- Persists Interruption rows linked to the latest checkpoint for this run
- Updates run status to SUCCESS / INTERRUPTED / ERROR
Args:
input: Graph input (e.g. {"messages": [...]}) or a Command for resume.
config: LangGraph config containing thread_id, checkpoint namespace, and run_id.
run: The Run instance tracking this technical attempt.
Returns:
The raw graph output dict. If interrupted, the output includes
'__interrupt__' entries and corresponding Interruption rows are created.
Raises:
Propagates unexpected exceptions after persisting run error details.
"""
graph = self.create_graph(agent, run)
user_id = self.user.pk if self.user is not None else None
if user_id is None and self.thread and self.thread.user_id:
user_id = self.thread.user_id
thread_id = str(self.thread.id) if self.thread else None
execution_id = None
if self.thread and self.thread.execution_id:
execution_id = str(self.thread.execution_id)
from agency.secretary.context import (
bind_secretary_context,
build_secretary_context,
enrich_configurable,
reset_secretary_context,
)
sec_ctx = build_secretary_context(
user_id=user_id,
agent_id=str(agent.pk),
thread_id=thread_id,
execution_id=execution_id,
run_id=run.id,
)
config = enrich_configurable(config, sec_ctx) if sec_ctx else config
token = bind_secretary_context(sec_ctx)
try:
run.save_status(Run.Status.STARTING)
output = graph.invoke(input, config=config)
#for event in graph.stream(input, config=config):
# final = event
# #channel_log(f"*STREAM_EVENT", {"event": event})
#output = final or {}
interrupted = self.check_interrupt(run, output)
if interrupted:
return interrupted
else:
#run.save_status(Run.Status.SUCCESS)
app_log.info(f"Run {run.id} completed successfully")
return output
except Exception as e:
run.status = Run.Status.ERROR
run.error = {"message": str(e), "type": e.__class__.__name__}
run.save()
try:
Monitor(run).emit(Monitor.Event.RUN_ERROR, {
"error": str(e),
"error_type": type(e).__name__,
"context": "invoke",
})
except Exception:
pass
if self.presence:
self.presence.emit(Presence.ERROR)
raise e
finally:
reset_secretary_context(token)
[docs]
def check_interrupt(self, run: Run, output: dict) -> dict | bool:
"""Check if the output contains an interrupt"""
if "__interrupt__" not in output.keys():
return False
run.save_status(Run.Status.INTERRUPTED)
app_log.info(f"Run {run.id} interrupted")
if self.presence:
self.presence.emit(Presence.WAITING_INPUT)
interrupts: List[Interrupt] = output['__interrupt__']
if not interrupts:
Monitor(run).emit(Monitor.Event.RUN_INTERRUPT, {"status": "interrupted"})
return output
checkpoint: Checkpoint = run.checkpoints.order_by("-id").first()
interrupt = interrupts[0]
interrupt_value = interrupt.value if isinstance(interrupt.value, dict) else {}
tool_name = interrupt_value.get('tool_name') or interrupt_value.get('type') or "ask_human"
channel_log(f"INT Checkpoint {checkpoint.id} {run.thread.agent.name}: {repr(interrupt.value)}")
try:
interruption: Interruption = Interruption.objects.get(id=interrupt.id)
channel_log(f"INTERRUPTION {interrupt.id} EXISTS")
except Interruption.DoesNotExist:
interruption = Interruption.objects.create(
id=interrupt.id,
checkpoint=checkpoint,
agent=run.thread.agent,
user=run.thread.user,
question=interrupt.value,
)
Monitor(run).emit(
Monitor.Event.RUN_INTERRUPT,
{
"status": "interrupted",
"interruption_id": str(interrupt.id),
"tool_name": tool_name,
},
)
return output