LangGraph Pipeline

The orchestration engine uses a LangGraph StateGraph to execute agent requests through a deterministic pipeline. Each node performs a specific function and returns a partial state update. Conditional edges route execution based on state values.

Pipeline Overview

START
  |
  v
system_prompt_node      Assemble the agent's system prompt (persona + router manifest)
  |
  v
memory_node             Fetch working + episodic memory; hydrate pending_mutation
  |
  v
resume_guard            Reconcile a HITL-gated mutation against its approval verdict
  |
  +--[approved]---------> tool_node   (replay the frozen call)
  +--[still pending /     output_format --> END
  |   can't evaluate]
  |
  v
initial_llm_call        The single LLM call (may request tool calls)
  |
  +--[has tool calls?]---> tool_node ----+
  |          ^                           |
  |          +------ loop back ----------+
  |   (R36: pull_routers / select_skill run here as ordinary tool calls;
  |    a mutating tool trips the before_tool_call HITL gate ->
  |    awaiting_hitl -> output_format -> END)
  |
  v
approval_node           Check HITL requirements, pause if needed
  |
  +--[awaiting_hitl?]---> output_format --> END
  |
  v
output_format           Finalize status, increment iteration counter
  |
  v
END

State Schema

The entire pipeline shares a single GraphState TypedDict that flows through every node. Each node reads from and writes to this shared state.

class GraphState(TypedDict, total=False):
    # Identity
    execution_id: str
    conversation_id: str
    agent_id: str
    agent_type: str              # "rule_37", "rule_32", "flaring_monitor", etc.
    tenant_id: str
 
    # Execution control
    status: str                  # "pending", "running", "completed", "awaiting_hitl", "failed"
    current_node: str | None
    iteration: int               # Guards against infinite tool-call loops
    error: str | None
 
    # Conversation
    messages: list[dict]         # OpenAI-format messages (role, content, tool_calls)
    system_prompt: str
    model: str                   # code fallback "gpt-4o"; box resolves to
                                 #   anthropic/claude-sonnet-5 (Platform Settings)
 
    # Budget enforcement (two-tier: per-conversation + per-turn)
    tokens_used: int
    cost_usd: float
    max_tokens: int              # cumulative per-conversation ceiling
    max_cost_usd: float          # cumulative per-conversation ceiling
    max_tokens_per_turn: int     # spend since the turn began
    max_cost_per_turn: float     # spend since the turn began
 
    # Tool execution
    pending_tool_calls: list[dict]
 
    # Skill runtime state (R36 — skills load mid-turn via select_skill)
    loaded_skills: list[str]             # skill_keys currently loaded
    loaded_skill_id_map: dict[str, str]  # skill_key -> skill_id (rules engine)
    injected_skill_ids: list[str]
    injected_entity_ids: list[str]
    active_skills: dict[str, int]    # skill_id -> tier level (2 or 3)
    selected_skill_ids: list[str]    # VESTIGIAL (R36) — never populated; retained so
    skill_injection_done: bool       # serialized states stay shape-compatible
 
    # LLM phase (single phase since R41 B0 removed the synthesis node; the field
    # is retained as a constant "initial" for trace/snapshot shape-compat)
    llm_phase: str               # always "initial"
 
    # Memory
    memory_context: dict          # Working memory + episodic search results
 
    # HITL
    hitl_required: bool
    hitl_checkpoint_type: str
    hitl_approval_id: str | None
    hitl_reviewer_id: str | None
    hitl_reviewer_strategy: str   # "named_individual" or "role_based"
 
    # R34 entity resolution / HITL pause-resume
    recent_entity_context: list[str]       # last-10 entity uuids (FIFO), mirrors Redis
    pending_mutation: dict | None          # a gated mutating tool_call awaiting approval
    consecutive_failed_resolves: int       # force-clarify counter (Rule 2 reads it)
 
    # Metadata
    metadata: dict

Node-by-Node Reference

1. system_prompt_node

File: orchestration/nodes.py

Loads the agent’s root skill (from agent_definitions.root_skill_key, R35 P3a) and then assembles the base system message (R35 P3b, design §6.4). The old hardcoded prompt registries are gone: R35 P6 deleted fallback_prompts.py (the per-agent FALLBACK_PROMPTS monolith and the generic DEFAULT_PROMPT) and the demo AGENT_DEFAULT_SKILLS skill map; the live PLATFORM_DEFAULT_PERSONA constant moved into persona.py. An unseeded tenant (no agent_definitions row) now loads no skill and runs skill-less (general mode), rather than falling back to a hardcoded map. Assembly:

1. Persona (voice/identity, §6.3 precedence): explicit agent_definition.persona_key → the root skill’s default persona → PLATFORM_DEFAULT_PERSONA. Resolution keys only on the root skill — dependency-skill personas are inert.
2. USER IDENTITY block: a small === User === block (display name + role from metadata, soft-falling to user_id / "user").
3. Domain facts ride in separately: the loader injects each loaded skill’s context_templates as their own system messages — this node counts them (for the prompt.assemble span), it never re-injects.

The base message (persona + identity) is inserted at index 0 as the stable cache prefix; memory and the skill context_templates are the variable tail. On the Anthropic path the outgoing payload carries a cache_control breakpoint at the end of the prefix (provider-gated — skipped for OpenAI), so a skill load/unload re-pays the tail but never invalidates the prefix.

R36 — router manifest + sealed prefix. For a general-mode run (an agent_definitions row with root_skill_key = NULL) the node also renders an “AVAILABLE SKILL ROUTERS” manifest into the index-0 block: one router_key + recruitment_blurb per active router, fetched from agent-config-service (a derived read). The rendered index-0 block is sealed at conversation birth into conversation_runtime_state (migration 016, trigger-enforced immutable) and reused byte-for-byte on reopen; all later context change is tail-append only, mirrored to the append-only conversation_context_appends log. Per-turn context maintenance (orchestration/context_maintenance.py) applies three append flows: router deltas (routers published/retired since the frozen manifest, tracked by a MAX(skill_routers.updated_at) watermark), lazy skill version upgrades (per-turn for version_check_on_turn = true skills, otherwise at router re-pull) with supersession notices, and deletion tombstones (a soft-deleted skill is unloaded, its tools vanish from the next assembly, and its code-block tools hard-fail at invocation). Agent-mode runs (rule_37, rule_32) load their root skill and render no manifest. This is not a LangGraph checkpointer — messages stay in conversation_messages; only the sealed prefix, watermark, rendered router keys, and the loaded-skills slug→version map persist.

General mode is domain-empty by design (§6.5). A general agent (root_skill_key=NULL) loads no skill at birth → no templates → the platform default persona, which carries no O&G/RRC/Texas content. R35 P6 added GENERAL to the AgentType enum so general mode is reachable via /execute (its agent_definitions row, NULL root, is seeded). Since R36, general mode is no longer stuck at the hard-stop: it recruits skills itself via pull_routers / select_skill (see below). The hard-stop still applies pre-selection — domain tools do not exist in the run until select_skill succeeds.

A caller-supplied state.system_prompt still overrides the assembly (back-compat). Spans emitted: agent.resolve_definition (with the resolved persona) and prompt.assemble (persona_key, persona_source, context_template_count, user_identity_injected, token_estimate).

The node also initializes execution tracking fields:

{
    "llm_phase": "initial",
    "skill_injection_done": False,
    "selected_skill_ids": [],
    "status": "running",
}

2. memory_node

File: orchestration/nodes.py

Makes two HTTP calls to the memory service:

1. Working memory (GET /working-memory/{conversation_id}): Retrieves the conversation’s scratchpad and extracted entities from Redis
2. Episodic memory (POST /episodic/search): Performs semantic search over past conversation summaries using the latest user message as the query (returns top 3 results by default)

The retrieved memory is formatted and injected as a system message after the main system prompt:

=== Memory Context ===

[Working Memory -- Scratchpad]
{scratchpad content}

[Working Memory -- Extracted Entities]
  - Well: 42-329-12345 (Mitchell Ranch 1H)
  - Operator: op-permian-energy (Permian Basin Energy LLC)

[Episodic Memory -- Relevant Past Conversations]
  - (similarity: 0.87) Previously filed Rule 37 exception for Mitchell Ranch 2H...

resume_guard (R34 Phase 5)

File: orchestration/nodes.py

Runs between memory_node and initial_llm_call. If there is no pending_mutation it is a pass-through. Otherwise it reconciles the gated mutation against the approval verdict (get_approval_status → approval-service GET /approvals/{id}):

• approved → reconstructs the assistant tool_calls message from the frozen args + original tool_call_id, sets pending_tool_calls, and routes to tool_node for a single replay.
• rejected → drops pending_mutation, appends a system note (with the reviewer comment), routes to initial_llm_call so the model acknowledges the rejection.
• still pending / can’t evaluate → re-pauses (awaiting_hitl → output_format).

Because pending_mutation is hydrated in memory_node (warm: carried in state; cold: rehydrated from Redis), the guard sees identical state on warm and cold resume — warm/cold equivalence holds by construction.

3. initial_llm_call

File: orchestration/nodes.py (the llm_call function, registered directly as initial_llm_call in orchestration/engine.py)

This is the single LLM node (the synthesis phase was removed in R41 B0). It calls the LLM via LiteLLM with budget enforcement. The node:

1. Calls check_budget() — raises BudgetExceededError if token or cost limits are exceeded (two-tier: per-conversation cumulative and per-turn)
2. Determines the model (request override → tenant Platform Settings → DEFAULT_LLM_MODEL env floor; the code fallback is gpt-4o but the dev box resolves to anthropic/claude-sonnet-5)
3. Assembles the visible tool list (R35 P4): core capabilities + selection tools + the loaded skills’ exposed tools. core_tool_schemas() always contributes the three core capabilities — entity_resolve, context_assemble, and render_chart (R43, presentational); R36 adds the selection tools (pull_routers, select_skill) as their own class for general-mode runs; everything domain-specific arrives from the loaded skills’ llm_visible code blocks (skill_tool_schemas), deduped by name. There is no agent_type-keyed tool list — AGENT_TOOLS / get_tools_for_agent were deleted, and the last exception, the R35-FENCE branch for flaring_monitor / compliance_monitor, was removed when the flaring_watch skill landed (a monitor-typed request now degrades to core tools only, like any agent without a seeded definition). R38 unions the Skill Creator’s authoring tool schemas (orchestration/authoring_tools.py) the same data-driven way — present only when the skill_creator skill is loaded, no agent_type branch. A tools.assemble Langfuse span records the split (core_capability_count, skill_exposed_tool_count, total_tools_visible, skills_loaded).
4. Sends the full message history to the LLM
5. Tracks token usage and cost from the response

If the LLM response includes tool calls, they are stored in pending_tool_calls for the tool node to execute. Otherwise routing proceeds to approval_node.

4. tool_node

File: orchestration/nodes.py

Executes all pending tool calls from the LLM response. For each tool call:

1. Parses the function name and JSON arguments
2. Calls the corresponding tool function via execute_tool()
3. Appends a tool role message with the result

Available tools by provenance (R35 P4 — tools derive from core capabilities + loaded skills, no longer from an agent_type map):

Core capabilities (entity_resolve, context_assemble, render_chart) are always visible regardless of loaded skills. The domain tools arrive from the loaded RRC skills as sandbox code blocks — tool_node routes a call whose name is in the per-conversation skill_code_block_tools manifest to the code-block executor (resolve → validate → jail), whose output carries a {status, message, result} wrapper. spacing_calculation stays reserved in TOOL_REGISTRY (no block may shadow the name) but is never offered — rule_37 reaches spacing via the spacing_assessment block. create_entity (the R34 mutating canary) is no longer universally offered — it rode the deleted AGENT_TOOLS["rule_37"] list; its schema + state-aware dispatch remain, but its HITL-canary role moves to skill rules in R35 P5. The flaring_watch blocks reuse the rrc_rule32 code constants and (for flaring_volume_status) the same resolver, under distinct block keys — same-named blocks on two router-mapped skills would raise CodeBlockToolCollision if both loaded in one conversation. R38: the Skill Creator’s ten authoring tools (R39 P0 added the list_categories read) arrive from the skill_creator skill but execute as native orchestration handlers (orchestration/authoring_tools.py), not code blocks — the skill seeds zero blocks; the handlers call agent-config-service as the conversing user (fail-closed X-User-Id headers, so owner_id lands as the real user).

State-aware tools (entity_resolve, context_assemble, create_entity) dispatch through the STATE_AWARE_TOOL_HANDLERS registry (orchestration/entity_tools.py) rather than execute_tool — they need GraphState to inject context server-side. Each handler returns a declarative StateAwareToolResult (tool output + state/working-memory updates + optional audit trigger) that the loop applies uniformly. R36: the selection tools (pull_routers, select_skill) follow the same contract; a successful select_skill loads the skill mid-turn, and the tool_node node_end SSE event then carries skills_loaded.

After executing all tools, the node clears pending_tool_calls and increments iteration.

before_tool_call HITL gate (R34 Phase 5)

Before dispatching each tool, tool_node fires the before_tool_call rule trigger. A platform rule (platform:hitl_on_mutating_capability) matches any mutating capability (MUTATING_TOOLS maps tool name → capability_key; create_entity was the only invocable one in R34 — R38 adds the Skill Creator’s six authoring write tools, all fail-closed when governance is unevaluable). When the gate sets hitl_required, the node:

1. Creates the approval inline via the approval service (tool_node is upstream of approval_node, so it can’t reach that node in-pass — it reuses the same create_approval_request primitive).
2. Persists pending_mutation to Redis working memory (frozen tool_call_id + args + hitl_approval_id, status="pending") with a generous TTL (PENDING_MUTATION_TTL_SECONDS, default 7 days).
3. Strips the round — truncates messages back to before the assistant tool_calls message, so the turn never ends with a dangling/unmatched tool_call.
4. Sets status="awaiting_hitl" and does not dispatch — the modified _after_tool router carries the pause to output_format → END.

On the resume turn, resume_guard (below) reconstructs the frozen call and routes it back through tool_node, which detects the approved-replay (pending_mutation.status == "approved" && matching tool_call_id), skips the gate, dispatches exactly once, and clears pending_mutation. See HITL Pause-and-Resume.

5. Skill selection — pull_routers / select_skill (R36)

File: orchestration/selection_tools.py (executed inside tool_node)

The dedicated selection nodes are deleted. Skill selection is now two explicit tool calls the LLM makes inside the ordinary tool loop:

1. pull_routers(router_keys[]) — pulls one or more routers from the manifest rendered into the system prompt. Returns each router’s selectable skills with their full author descriptions — a derived read over approved skill_router_map rows × active skills × the latest-approved skill_description_versions row per skill. Cross-topic questions pull multiple routers.
2. select_skill(skill_key) — loads a skill that belongs to a router pulled this turn. On success the existing load_skill() machinery fires unchanged: the skill’s context templates land as system messages (via inject_messages, ordered after the tool-result message so the assistant→tool sequence stays valid) and its llm_visible code-block tools join the next tool assembly.

Enforcement is structural: domain tools do not exist in the run until select_skill succeeds — llm_call only assembles core + selection tools pre-selection. Every validation failure returns a typed error naming the reason plus a hard-stop directive, and every “could not validate” (pool down, read error) fails toward deny, never toward load.

Only general-mode runs get the selection tools and the router manifest. Agent-mode runs (rule_37, rule_32) keep their root skill and do not pull routers. Both handlers follow the R34 P3 state-aware contract (declarative StateAwareToolResult applied by tool_node).

6. approval_node

File: orchestration/nodes.py

Determines whether the current execution requires HITL approval. It reads two sources:

1. The hitl_required flag (with hitl_checkpoint_type) set upstream. For Rule 37 / Rule 32 this is set by a skill’s require_hitl SkillRule firing on after_tool_call in tool_node — the rule matches result.tool_name == '<block_key>' (e.g. rule37_filing_assembly → pre_filing, good_cause_narrative → good_cause_review, rule32_filing_assembly → pre_filing) and its action_parameters.checkpoint_type becomes the label. The before-tool_call mutating-capability gate sets the same flag for create_entity-style mutations.
2. Metadata override: the request metadata can explicitly set hitl_checkpoint_type (a generic escape hatch).

R35 P5 (HITL cutover): the old CHECKPOINT_TRIGGERS / TOOL_HITL_TRIGGERS auto-detect maps — which keyed HITL on injected-skill slug or executed-tool name inside this node — were deleted. Under “behavior comes from skills,” HITL is now driven by the seeded require_hitl SkillRules on rrc_rule37 / rrc_rule32, so the checkpoints live with the skills, not in node code. Each fire emits a hitl.rule_fire Langfuse span carrying the skill + rule provenance. The flaring_watch monitoring skill is read-only and carries no HITL rules by design.

When HITL is required, the node:

1. Builds a state snapshot (messages, skills, entities, token usage, metadata)
2. Creates an approval request via the approval service (POST /approvals)
3. Sets status: "awaiting_hitl" which causes the pipeline to exit straight to output_format (there is no synthesis phase since R41 B0)

7. output_format

File: orchestration/nodes.py

Terminal node that finalizes the execution:

• If status is awaiting_hitl, preserves that status
• Otherwise, sets status: "completed"
• Increments the iteration counter

Routing Logic

Conditional edges control the flow between nodes:

approval_node → output_format is a plain edge — the pause vs. pass-through decision is recorded in status inside the node, not as a branch. The old synthesis_llm_call routes were deleted with the node in R41 B0.

Budget Enforcement

Budget is two-tier (R41 A8): a cumulative per-conversation ceiling (max_tokens / max_cost_usd) and a per-turn ceiling (max_tokens_per_turn / max_cost_per_turn, spend since the current turn began). Every LLM call checks both before execution; tokens_remaining returns the tighter of the two. Per-turn env defaults equal the per-conversation ceiling, so the per-turn tier is dormant until an operator lowers it.

def check_budget(state: GraphState) -> None:
    # Enforces BOTH the per-conversation and per-turn ceilings; raises
    # BudgetExceededError when either token or cost limit is reached.
    ...

The old per-agent budget defaults in agents/*.yaml are historical — those four purpose-built agents were retired (2026-07-07) and the YAML files are not read at runtime. LLM runtime config (default/fallback model, per-execution budgets, daily caps, per-user overrides) is now tenant-configurable via the Platform Settings admin page (/configuration/platform-settings → agent-config /settings). Resolution order: request override → tenant setting → DEFAULT_LLM_MODEL / DEFAULT_MAX_TOKENS_PER_EXECUTION / DEFAULT_MAX_COST_PER_EXECUTION env floor.

The iteration limit (MAX_ITERATIONS) prevents infinite tool-call loops independently of the budget.

API Endpoints

The orchestration engine exposes two execution endpoints:

POST /execute (Synchronous)

Runs the full pipeline and returns the final state. Used for simple request-response interactions.

GET /conversations/{id}/stream (SSE)

Streams node-by-node events as the pipeline executes. Each node completion emits an SSE event:

event: node_end
data: {"event":"node_end","node":"memory_node","data":{"status":"running","tokens_used":0}}

event: node_end
data: {"event":"node_end","node":"initial_llm_call","data":{"status":"running","tokens_used":1234,"message":{"role":"assistant","content":"..."}}}

event: done
data: {"event":"done","data":{"execution_id":"...","status":"completed","tokens_used":1234,"cost_usd":0.05}}

HITL Pause-and-Resume (R34 Phase 5)

A mutating tool call (e.g. create_entity) is paused for human approval before it runs, then replayed on approval. The mechanism turns the dormant before_tool_call trigger into a real governance gate.

Turn 1 (gate)                         Turn 2 (resume, after approval)
-------------                         -------------------------------
initial_llm_call                      memory_node (hydrates pending_mutation)
  emits create_entity tool_call         |
  |                                     v
  v                                   resume_guard
tool_node                               reads verdict via GET /approvals/{id}
  before_tool_call gate FIRES            |
  - create approval (status=pending)     +-- approved  -> reconstruct frozen call
  - persist pending_mutation (Redis)     |               -> tool_node (replay once,
  - strip the round (no dangling tc)     |                  clear pending_mutation)
  - status=awaiting_hitl                 +-- rejected  -> drop + system note
  |                                      |               -> initial_llm_call
  v                                      +-- pending /  -> re-pause
output_format -> END                         can't eval    (output_format -> END)

Source of truth. The durable approval record (approval-service) is authoritative — it survives the Redis TTL. The Redis pending_mutation holds only gate state (frozen tool_call_id, args, hitl_approval_id), never the verdict. get_approval_status (its first consumer is resume_guard) reads the verdict fresh each resume.

By-construction guarantees:

• No dangling tool_call across the turn boundary — the gate strips the round and resume_guard reconstructs the exact call (OpenAI/LiteLLM reject an unmatched assistant tool_calls).
• Replay determinism — the call is replayed from the frozen args + original tool_call_id, independent of any new model output.
• Warm/cold equivalence — pending_mutation is hydrated in memory_node, so an in-memory resume and a cold (server-restarted) resume reconcile identically.
• Graph/index consistency — create_entity routes through KG POST /entities, which mints the derived uuid and write-throughs the name index in the same transaction (R34 0b).

Platform rules (seeded in agent-config-service SEED_PLATFORM_RULES, idempotent — no SQL migration):

The force-clarify counter (consecutive_failed_resolves) is orchestration-computed in entity_tools.run_entity_resolve — incremented on a non-decisive resolve, reset on a decisive auto-select.