Runtime Flow

End-to-end trace of a chat request, from the HTTP POST that opens the SSE stream to the batch INSERT that closes telemetry. Concrete example throughout: the user types “Show me RCRA facilities in Louisiana on a map” with a CSV already attached.

This page traces the path that runs today, including where the Uniform Tool Contract changes behavior at the tool boundary. The contract has landed in this worktree and covers most, but not all, tools, see §7 for how to check the current migrated/deferred split rather than trusting a fixed number.

1. Wire entry: POST /api/v1/assistant/chat

The frontend (AssistantPanel) calls chatWithAssistantStream(), which POSTs to /api/v1/assistant/chat with stream: true. The request body is the ChatRequest pydantic model in engine/orchestrator.py:ChatRequest. engine/orchestrator.py has grown substantially since this trace was first written (verify its current size with wc -l rather than trusting a number in prose), treat every citation below as “defined in this function/class,” not as a stable line number.

Field	Purpose
message	the user’s typed text
conversation_id	UUID of the live conversation; ownership-checked
dataset_id	optional explicit dataset selection
model_override	overrides the chat-role model for this turn
file_context	list of context strings from uploaded files (image paths recognised)
data_refs	catalog entry IDs from the Kit/recent picker
inference_params	per-request override of temperature/top_p
focus_context	full-screen artifact viewer context
selection_state	chart/map selections snapshot (CONTEXT_STREAMLINING §3.1)
agentic_mode	gates the planner-first loop
skip_history	sub-agents share the catalog but not the parent’s message tape
advanced.data_isolation	per-request isolation toggle
namespace_id	analyze_data sandbox namespace override

The response is text/event-stream (engine/orchestrator.py:chat, the @router.post("/chat") handler), with Cache-Control: no-cache and X-Accel-Buffering: no to defeat proxy buffering. When body.stream is falsy, chat() instead awaits _chat_no_stream(body, tenant_id) and returns a plain JSON dict.

The non-chat assistant routes (/api/v1/assistant/...) are split between api/downloads.py, api/chart_snapshots.py, api/map_snapshots.py, api/chart_bookmarks.py, and api/chart_url_states.py, all mounted under the same /api/v1/assistant prefix; see app.py where these routers are include_router-ed.

2. Middleware: auth, tenant, gzip

app.py:auth_middleware (an @app.middleware("http") function) is the auth middleware. Public paths bypass JWT decode entirely: the auth-endpoint set (/api/v1/auth/register, /login, /refresh), operational probes (/api/v1/health, /ready, /version), the API docs (/docs, /openapi.json, /redoc), and, newer than this trace’s original write-up, token-authenticated public prefixes where an unguessable token is the auth: /api/v1/ping/ and /api/v1/consent/revoke/ carry the token in the path, while /api/v1/connections/oauth/callback carries an OAuth state nonce in the query string. That callback is unauthenticated by necessity, it is a bare browser redirect from an external IdP (Jira/ServiceNow) that carries no Eunice session, and safe by construction: the state is a 256-bit single-use nonce (shared/oauth.py), deleted on redeem with a ~10-minute TTL, and it authorizes only a scoped credential write bound to the nonce row’s own user (never a resource read, never system_admin). It was the security review’s “undocumented fourth bypass path” (DEV-011), a documentation gap, not a hole. Everything else follows the precedence chain from docs/audits/2026-04-13-pre-aws-audit.md §4:

1. Decode the Authorization: Bearer … JWT (auth_jwt_secret, auth_jwt_algorithm). Expired tokens are accepted only when dev bypass is on.
2. Load User + TenantMembership rows. If X-Tenant-ID was set and the user has no membership in it (and isn’t system_admin), return 403. A token resolving to an app-principal user (User.is_app) is refused outright, no v1 path should ever mint one, but this is the one middleware choke point that would catch a forged or DB-tampered token doing so.
3. Resolve the active tenant: header → JWT tenant_id claim → first membership → settings.default_tenant_id.
4. Resolve the tenant role for that membership and call resolve_capabilities(role, system_role, db).
5. Enter TenantContext(active_tenant) and RequestContextManager(ctx) so downstream code reads the tenant from context vars rather than threading it through every function.

Dev bypass (CATALYST_AUTH_DEV_BYPASS=true) grants SYSTEM_ADMIN_CAPABILITIES to unauthenticated requests. A startup safety check in app.py refuses to start with bypass enabled on a non-loopback bind. GZipMiddleware compresses responses ≥1 KB (app.add_middleware(GZipMiddleware, minimum_size=1024)), this is load-bearing for the Fabric data endpoints, not the SSE stream (each event is too small to compress meaningfully).

Unhandled OperationalError triggers _handle_db_error (app.py), which disposes the engine, calls init_db(), and returns 503 so the client can retry, It matches both backends’ phrasings: the fallback’s "no such table" and PostgreSQL’s relation "..." does not exist, and it catches ProgrammingError as well as OperationalError, since PostgreSQL raises the former. It previously matched only the fallback’s wording, so the recovery never fired on PostgreSQL; that was corrected in 2026-09.

3. Conversation setup

chat() (engine/orchestrator.py:chat) checks check_conversation_access() and hands off to _chat_stream(), which is the SSE generator wrapped in StreamingResponse. _chat_stream itself is now a thin wrapper (it owns the per-conversation guard used by convened conversations) around _chat_stream_body, which does the actual setup and drives the loop. Setup order:

1. Extract dataset IDs from file_context (regex for ID: <8-hex>) via _extract_dataset_ids_from_context (engine/executor.py).
2. Resolve data_refs through _resolve_data_refs (engine/orchestrator.py): each catalog entry is looked up by ID, a [Dataset: name | rows × cols | ID: …] context string is built, and the ID is added to active_ds_ids.
3. Register dataset activity via register_datasets_for_conversation (engine/state.py) so the dream cycle’s co-occurrence graph sees this turn.
4. Resolve chat config with _resolve_chat_config(model_override) (engine/orchestrator.py), routes through _resolve_chat_plan / shared/model_resolution.py:resolve_model, not a bare DB lookup by model_id; an unknown explicit pick raises MODEL_NOT_FOUND rather than silently substituting another row.
5. Build user content via _build_user_content (engine/orchestrator.py). Image paths in file_context are read from disk, base64-encoded, and emitted as vision blocks; non-image entries fall through as a trailing Attached files: block.
6. Build fallback model chain for rate-limit recovery from get_models_for_role("chat") minus the active model.
7. Instantiate EventRecorder (shared/telemetry.py:EventRecorder), keyed by (conversation_id, message_id, tenant_id). Records a user_message event with the full file_context list so multi-turn vision replies can reconstruct image blocks from history.
8. Run behavioral classifiers on the user message: a correction regex, a meta-question/friction regex, and _label_previous_outcome (engine/orchestrator.py), which classifies the previous turn’s outcome from the current user reply.
9. Construct the DisplayFactory (shared/display.py) with the recorder and the running event loop. The factory owns the SSE queue.
10. Launch _run_agentic_loop as a background task and drain the factory into the SSE response. The try/finally around the drain guarantees factory.close() and telemetry flush even if the client disconnects mid-stream.

4. Prompt build (split static + dynamic)

The system prompt is split into two pieces to preserve provider prompt-cache hits:

• Static prompt (engine/prompt_builder.py:_build_static_system_prompt): identity (CATALYST_SYSTEM_BASE), build_tool_guidance(core_tools), and a one-paragraph note that the model will receive a [CURRENT_STATE] user message and that older tool results may be compacted with call_id references. Byte-stable across the entire agentic loop.
• Dynamic state block (engine/prompt_builder.py:_build_current_state_block): rebuilt at the top of every turn. Sections include [CURRENT_STATE] header, agentic-mode supplement, email availability, MCP connections, [SUGGESTED CONTEXT] from _pre_fetch_context, focus mode, active datasets with full column detail, session state, trained models, and domain context. Injected as the LAST user message, outside the cached system prefix, and never persisted to canonical history.

The static prompt is built once on asyncio.to_thread at request entry. The dynamic block is rebuilt inside the loop on every iteration (see §6).

Tool filtering happens here too. registry.get_tools_for_llm(tenant_id) returns the resident tier-0 surface (derived by engine/tool_surface_derivation.py, re-derived by the dream cycle’s derive_and_store_tier0; the rest is reachable via search_tools); engine/tool_surface.py:_filter_resident_by_capabilities drops tools a claimed, provider-supported model capability replaces (DB-driven via ToolDefinition.replaced_by_model_capability, not a hardcoded map); a final pass strips agentic-only tools (plan) when not in agentic mode, or restricts to an agentic-mode shortlist when agentic mode is on. See engine/orchestrator.py around where tools is assembled for the current shortlist name and logic.

Pre-fetch behavior (engine/prompt_builder.py:_pre_fetch_context): operational_patterns (top 3) and user_behavior (top 2) corpora are searched via container.vector_searcher() with a keyword fallback. There used to be a third, data_catalog-corpus-backed prefetch gated by PREFETCH_DATA_CATALOG (default off), it was retired because the corpus was a separately-rebuilt, drift-prone artifact that could silently go stale or empty. Its replacement, PREFETCH_DATASET_SUGGESTIONS (engine/prompt_builder.py, default on), ranks the live entry-level CatalogEntryRecord.description_embedding via the same vector-rank + relevance floor as the search_catalog tool, and holds proactive suggestions to a stricter relevance band than the tool itself, an unrequested wrong suggestion in the prompt is more harmful than a weak hit the model went looking for.

5. History load

_load_conversation_history() (engine/history.py) pulls ChatMessage rows, including the structured_content JSON column that carries the canonical tool-call chain. Token budget comes from chat_config["history_token_budget"]; estimation uses ~4 chars/token. If the running estimate exceeds the budget, the loader checks Conversation.summary (with summary_through_id tracking coverage), regenerates the summary via the compaction-role model when stale, and returns [summary] + history_recent_keep raw recent messages. The compaction LLM sees the full conversation text, no truncation, and the call cost lands in usage_log.

Sub-agents (body.skip_history=True) bypass this entirely and start with an empty history (engine/orchestrator.py, gated on body.skip_history).

6. The turn loop

_run_agentic_loop (engine/orchestrator.py) constructs a TurnState and a resolved TurnConfig, then calls engine/runner.py:run_conversation_turns. The runner owns the contract boundary; it builds an EffectAccumulator (engine/effect_accumulator.py) and hands the proxies down into _execute_agentic_turns (engine/executor.py), which is the per-turn body.

Typed turn surface

engine/turn_types.py defines the canonical types:

Type	Role
TurnState	mutable working memory: namespace, tool_history, failed_approaches, active_plan, task_stage, policy_violations_this_turn, contextual_signals, last_assistant_response, end_reason, cost_so_far
TurnConfig	resolved per-turn config (model_id, max_output_tokens, inference_params, max_agentic_turns, cost_budget_usd, failed_approach_threshold)
ToolRuntime	outward dependencies a tool needs (llm, tool_registry, catalog, knowledge_service, config, clock, ids)
Effect (sum type)	PersistAssistantMessage, TelemetryEvent, StreamChunks
TurnResult	state, messages_appended, effects, errors, final_text, cards, tool_log, work_log
TurnError	terminal failure with `kind: tool_failure

TurnResult.__post_init__ enforces the invariant that non-empty errors implies state.end_reason == "error". TurnResult.to_legacy_dict() is the bridge for callers that still consume the legacy {"message", "cards", "metadata"} shape: engine/orchestrator.py:_chat_no_stream calls it directly, and tools/delegate.py reaches it indirectly through the chat_no_stream public alias. (Docstrings in engine/runner.py and engine/turn_types.py also name a workflow.py caller, that file does not exist; workflow/ today holds only factory.py and seed_primitives.py, a stale reference worth cleaning up in the source.)

populate_state_from_persistent_store in engine/runner.py:32-99 mirrors the module-level state dicts (dispatcher._conversation_tool_history, session_state._session_states) onto typed fields on the TurnState. This is read-through wiring: the executor body still reads and writes those dicts directly, but downstream code (replay, audit, the upcoming consumer migration) sees a typed view.

Per-turn body

The loop in _execute_agentic_turns (engine/executor.py) runs up to config.max_agentic_turns iterations (default 75, per TurnConfig.max_agentic_turns in engine/turn_types.py):

1. Cancel check, state.is_conversation_cancelled(conversation_id) reads a process-wide flag. The cancel button writes it; the loop honours it before the next LLM call.
2. Build [CURRENT_STATE] on asyncio.to_thread (the dynamic block from §4). Active dataset IDs come from file_context plus the live catalog. Appended as a synthetic user message via _messages_with_current_state.
3. Context guard, estimate prompt tokens (history._estimate_messages_tokens × calibration ratio learned from the previous turn’s real input_tokens). At 80% of (context_window - max_output_tokens), call _compact_large_tool_results to summarize the largest old tool_result payloads. At 95%, break the loop.
4. Stream the LLM call via _stream_llm_response. SDK runs in a worker thread, bridges to the event loop via queue.Queue. text_delta events become display.text(...); tool_use_start becomes a display.thinking(...) line; message_done returns the assembled LLMResponse.
5. Rate-limit fallback, on RateLimitError, walk fallback_configs in order, re-stream with each fallback model, and continue when one succeeds.
6. If stop_reason == "end_turn", clean the response text, run completion guardrails (failed delegate contracts, incomplete plan detection), emit deferred singleton cards (map_display, cited_answer), call _persist_assistant_message, record done, and call _emit_done.
7. If tool_use blocks, append the assistant message to loop_messages, emit a decision telemetry event per tool, then dispatch each tool serially.

Tool dispatch path

For each tool_use block, engine/executor.py calls into helpers in engine/dispatcher.py:

1. Loop guard, this specific guard is inline in engine/executor.py (inside _execute_agentic_turns), not in engine/dispatcher.py: it imports args_hash from the dispatcher, strips a wider set of narration keys than just reasoning (reasoning, explanation, thought, rationale, justification, some tools like analyze_data narrate via explanation, and the model varies that text every call even when the operative code is identical), and hashes what’s left. If the same (tool_name, hash) produced the same logged result signature LOOP_GUARD_THRESHOLD=3 times in a row, the executor refuses the next call outright, ToolResult(error="loop_guard"), not executed, regardless of whether those calls succeeded or failed. This is a distinct, stricter mechanism from the dispatcher’s failed-approach hint below; it catches a model looping on an identical successful call (e.g., repeated display_map), which a failure-only counter would never trip on.
2. Tool display handle, factory.create_tool_handle(name) opens a scoped handle that pushes busy/thinking/card/progress to the thinking tab.
3. Build tool context, a dict carrying tenant_id, user_id, capabilities (as a frozenset; the audit’s “tool can’t see caller caps” bug fixed by propagating them here), auth_dev_bypass, dataset_id, conversation_id, container, display, catalog, progress_callback, _user_message, _recorder, _tool_registry, _budget_ref, _budget_limit, _current_tool_use_id, file_context, focus_context.
4. Execute via AsyncExecutionService, _execute_tool_via_async (engine/dispatcher.py) wraps registry.execute_tool(name, input, context) in a JobSpec so cancellations, timeouts, and queueing are uniform. A watcher task races the tool against is_conversation_cancelled; if the user cancels mid-flight, the watcher returns a ToolResult(content="…cancelled…", error="cancelled") and the worker thread is left to drain naturally (Python can’t force-kill a worker mid-call).
5. Record tool_result, full untruncated content into telemetry. Build a compact tool_log entry (name, sig, params with long strings clipped, outcome, duration_ms) for the work-log surface.
6. Side-effect handling, promote tool_output.datasets into tools.analysis._conversation_namespaces so follow-on analyze_data / display_map can address them by name. Singleton cards (map_display, cited_answer) are deduped by (type, display_id) and merged via _merge_map_card, accumulated for end-of-response emission, not streamed inline. Other cards stream immediately via tool_display.card(...).
7. Exports, _register_tool_exports writes the generated_files row, registers in the conversation catalog, emits an export telemetry event, and fires an export.ready notification.
8. Contextual signals, _capture_contextual_signals updates CatalogEntryRecord.contextual_signals for every dataset the tool referenced.
9. Realtime policy check, _check_policies_realtime runs each active PolicyRule.trigger == tool_name rule against the input, records policy_violation events, reinforces an Observation row, and _format_policy_warnings prefixes the resulting message into tool_output.content so the model actually sees the warning (not just telemetry).
10. Session state update, record_tool_execution and (for maps) record_map_display write to the in-memory session.

The dispatcher separately exposes repeat_failure_hint and record_failed_approach (engine/dispatcher.py). These plug into state.failed_approaches, keyed by config.failed_approach_threshold (default 3), and are the mechanism behind a soft “consider a different approach” hint appended to the result text after repeated failures of the same call, the platform never auto-terminates on this path, the model decides. This is the mechanism the Uniform Contract’s Hint variant composes with; it is separate from the hard loop guard in step 1 above.

7. Uniform Tool Contract: what it changes at the boundary

The Uniform Tool Contract replaces a fractal of per-tool envelopes with one shape. Every migrated tool returns Ok | Hint | Err; one decorator (@uniform_contract) renders that envelope into one templated text block the model sees identically at every boundary. Pre-contract, tools returned a mix of ToolResult (~60 tools), legacy dict (~29), and three async def outliers; the model saw an ad-hoc free-text content field that didn’t generalize between tools. See backend/engine.md and tool-system.md for the contract itself; this section is about how it threads through the runtime flow.

What the runtime path looks like under the contract

The flow above is unchanged at the orchestrator level. The change is at the registry seam (engine/tool_registry.py) and inside the decorator:

1. Spec registration at handler import. Each handler module either source-decorates (@uniform_contract on the function) or bulk-migrates (migrate_tool(handler, name, *, is_tabular, is_cost_rollup=False), which registers a ToolSpec but leaves the handler unchanged). Bulk migration is what makes flipping ~100 handlers safe, changing every source return type at once breaks direct test callers.
2. ToolRegistry.register_system_handler calls _apply_contract(name, handler). If a ToolSpec is registered for name and the handler isn’t already decorated, the registry wraps it. Idempotent; source-decorated handlers pass through unchanged.
3. ToolRegistry.execute_tool(name, inputs, context) strips the reasoning field, populates context["_tool_spec"] and context["_current_tool_name"] so the decorator can read them at call time, invokes the handler, and adapts the returned envelope back to ToolResult via envelope_to_tool_result for the unchanged orchestrator path.
4. The decorator catches HintException BEFORE the generic Exception handler (recoverable corrections become Hint, real bugs become Err), records a contract_envelope telemetry event with the envelope kind + hint_kind/err_kind, persists effects (cards, datasets, exports, cost) for Ok, and validates that tabular tools return a measured schema.

Result variants

Variant	Use	What the model sees
Ok(rows, schema, data, warnings, effects)	success	rendered table preview + advisory warnings + measured schema
Hint(hint_kind, message, alternatives)	recoverable correction (unknown identifier, missing filter, ambiguous match)	“did you mean…” / “add filter:…” / “candidates:…” block; counted as a recoverable correction, not an error
Err(message, err_kind)	actual bug or infra failure (unhandled exception, connection refused, permission denied, OOM)	rendered error block; the registry adapter sets ToolResult.error = message

Ok.schema MUST be measured from real data, declared schemas are the silent-failure mode the contract kills. The validator warns (does not raise) when a tabular tool returns an empty schema. Hint and Err carry NO effects, handlers are responsible for not side-effecting before raising HintException.

Turn-boundary ordering guarantees

The decorator fixes the order of operations the loop body used to interleave:

1. Handler returns envelope (or raises HintException / Exception).
2. Envelope is validated (_validate_envelope).
3. contract_envelope telemetry event is recorded.
4. _persist_effects writes cards, datasets, exports, and (when is_cost_rollup=False) the usage_log row.
5. envelope_to_tool_result adapts to ToolResult for the orchestrator.

is_cost_rollup=True is set exactly on delegate and delegate_batch, the rollup machinery skips the parent usage_log write so leaf rows aren’t double-counted. The model can’t forget; the spec on the registry remembers.

Migration status, as it stands now

The contract has landed and continues to migrate tools incrementally; it is not “incoming” or “post-promotion from a lab branch”, engine/contract.py, engine/contract_decorator.py, engine/contract_resolvers.py, engine/contract_render.py, engine/contract_schema.py, and engine/contract_telemetry.py all live in this worktree today. The migrated/deferred split is not a fixed fraction and shifts every time a tool is decorated or a new tool ships, compute it rather than quoting a number: universe = names in engine.tool_definitions.SYSTEM_TOOLS, migrated = engine.contract._TOOL_SPECS.keys() & universe, deferred = universe - migrated.

By example: display_map, save_map, load_map, spatial_filter, simulate_dispersion, document_conversation, export_excel, generate_document, manipulate_excel, plan, call_llm, web_search, delegate, and delegate_batch, the tools an earlier draft of this doc called “deliberately deferred”, are all migrated as of 2026-09. What’s actually in the deferred set now is mostly tool categories added after the migration baseline was seeded (entity/fact tools, work-item/Jira integration, terminal access, skills authoring, host management, convene) that were never contract-migrated. The CI ratchet (tests/test_uniform_contract.py::test_deferred_set_only_shrinks) is supposed to catch this, a brand-new undecorated tool should fail the test, not silently join the deferred set, but it is currently failing in this worktree, flagging dozens of tools as “new undecorated” that were never added to tests/contract_deferred_baseline.json. That failure is real and worth fixing rather than working around; see the Structural Debt note below.

The legacy ToolResult type in engine/tool_result.py stays in place until the deferred set is empty. Until then, the runtime path is dual: migrated tools speak the envelope at the boundary and the registry adapts back to ToolResult so the rest of the loop is unchanged.

8. Effects, accumulator, and the apply_effect chokepoint

Every side-effect a turn produces is a typed Effect object. The accumulator (engine/effect_accumulator.py) wraps the live EventRecorder and DisplayHandle with recording proxies (_RecordingRecorder, _RecordingDisplay) that:

1. Build a typed TelemetryEvent or SSEChunk.
2. Append it to the shared effects list.
3. Call apply_effect(effect, ctx), engine/effects.py:203-221.

apply_effect is the single chokepoint that writes to the outside world. EffectContext(recorder, display, persister) (engine/effects.py:EffectContext) carries the live services. _SSE_DISPATCH (engine/effects.py) maps each SSE event name (text_delta, card, error, thinking, busy, idle, progress, clear_progress) to the matching display method. _apply_telemetry_event (engine/effects.py) honours the text_delta batching path on the recorder (recorder.record_text_delta); it does not forward TelemetryEvent.blob to the recorder, the real EventRecorder.record(event_type, data) signature takes no blob argument, so blob is captured on the effect purely for replay, and the actual >32 KB payload split into conversation_event_blobs happens later, inside EventRecorder.flush(), based on the serialized payload size. _apply_persist drives the persister callable for PersistAssistantMessage.

The streaming UX is preserved by the call ordering, proxies route through apply_effect synchronously at emission time, so users see deltas as they happen. The contract benefit is the inversion: the loop body still calls recorder.record("event", {...}) inline, but every write to the outside world now flows through one typed object and one routing function. Tests mock apply_effect. Replay drives it with synthetic effects to reconstruct a run. Audit inspects the effects list independently of when/where the writes happened. This is the move from “loop interleaves side-effects with state in five module dicts” to “typed effects ride through one chokepoint.”

PersistAssistantMessage is the end-of-turn write. The runner’s mark_persist records the effect; the legacy inline _persist_assistant_message call in the executor remains the production write path during the transition. When EffectContext.persister is provided (tests, replay), apply_effect drives it directly.

9. SSE emission and frontend rendering

The DisplayFactory (shared/display.py) owns the SSE queue. _chat_stream calls factory.drain(), which yields event: display\ndata: {json}\n\n chunks. The wire payload is DisplayEvent.to_dict() (shared/display.py:222-245) carrying id, type, channel, scope, message_id, verb, priority, data, persistent, ts, and optionally group_id, parent_id, display_id, metadata.

Event types in EventType (shared/display.py:99-134) and their data shapes:

Type	Channel	Data
text_delta	response	{text} accumulated chat-bubble content
text_complete	response	{content} full text at end-of-stream for persistence
reasoning (a.k.a. thinking)	thinking	{text} step entry
tool_call	thinking	{tool, input, tool_use_id}
tool_result	thinking	{tool, output, duration_ms}
busy / idle / status	status	{text} priority-based status bar
progress	status	{text, percent?, current?, total?}, transient, not persisted
card	artifact	{card_type, card_data} artifact panel + inline
cost	system	{usd, tokens_in, tokens_out, model}
error / warning	system	{text, recoverable?}
done	system	{cost, work_log?, max_turns_reached?}

The frontend’s useDisplayEvents.ts hook routes by event type: text_delta accumulates into responseText (via a mutable ref to dodge stale closures), reasoning/tool_call/tool_result append to thinkingEntries with group_id support, card is categorized by ARTIFACT_CARD_TYPES (artifacts go to the side panel; file_download and chart show in both), busy/idle/status mutate statusText with priority arbitration, and done flips isDone. StreamingText renders the response with RAF-batched updates and a CSS mask-image trailing fade.

10. Persistence and telemetry flush

persist_assistant_message (engine/persister.py; callers alias the import as _persist_assistant_message, e.g. in engine/executor.py) is the one function that writes ChatMessage rows. It:

• Extracts everything appended to loop_messages after _agentic_start_idx as the structured_content array (canonical tool-call chain).
• Appends an {role: assistant, content: [{type: text, text: final_text}]} block if the final text isn’t already present.
• Runs _deep_sanitize_for_json to convert numpy scalars, datetimes, NaN/Infinity, and bytes into psycopg-safe JSON primitives (psycopg uses its own JSON serializer, not json.dumps, so a string round-trip doesn’t help here).
• Inserts the ChatMessage row with content (UI display text), structured_content (the chain), cost_usd, cards, and updates Conversation.updated_at.
• Records the final assistant response to session state via record_last_assistant_response so the next turn can offer a “resume context” surface.

Failure is best-effort and logged, persistence must not poison the loop’s exit path.

The EventRecorder (shared/telemetry.py) accumulates every event in memory during streaming; zero DB writes happen on the hot path. text_delta and thinking deltas are buffered (500-char threshold, flushed on next non-text event). At stream end, recorder.flush() performs a single batch INSERT into conversation_events. Payloads ≥BLOB_THRESHOLD (32 KB) are split: the main row stores {"_blob_id": <id>, "summary": "..."} and the full JSON lands in conversation_event_blobs. Telemetry is NEVER truncated.

11. Post-stream extraction

After the SSE stream closes, _chat_stream enqueues knowledge extraction via fabric/knowledge_extraction.enqueue_extraction(conversation_id, tenant_id) (engine/orchestrator.py:662-667). The extraction runs two stages:

• A provisional entry that indexes basic metadata immediately (no LLM).
• A full extraction queued via AsyncExecutionService that uses an LLM to extract operational patterns, tool-usage insights, and contextual signals.

The output feeds the operational_patterns, user_behavior, data_catalog, and tool_definitions system corpora that power the _pre_fetch_context call in §4. This is the closed loop: every conversation that finishes refines the knowledge layer the next request reads from. See backend/knowledge-module.md.

Structural debt worth knowing

A few items are load-bearing for anyone reading this page and then opening the code:

• engine/orchestrator.py is not a thin shell, it has grown, not shrunk. It re-exports underscore-prefixed symbols from dispatcher, executor, persister, state, and prompt_builder for backward compatibility, but it also still owns the live /chat route, ChatRequest, _chat_stream/_chat_no_stream, and, new since the convened-conversations work, prepare_turn / launch_prepared / PreparedTurn / TurnRefusal, which engine/turn_driver.py calls directly to drive a turn for a convened worker without going through HTTP. Check the file’s current size before quoting a number; it has been well over a thousand lines for a while and is not converging on “thin.” New callers should still prefer importing directly from the canonical modules for anything in the per-turn body.
• State still lives in module dicts. _conversation_tool_history (dispatcher), _session_states (session_state), _conversation_namespaces (tools.analysis), _conversation_policy_counts (dispatcher) are the canonical homes today. populate_state_from_persistent_store mirrors them onto TurnState at turn start (read-through), but the executor body still reads and writes the dicts directly. The migration moves consumers field by field; each retired dict reduces the cross-turn coupling surface.
• apply_effect is the chokepoint, but the persister is still called inline. The legacy inline _persist_assistant_message at end-of-turn remains the production write path. mark_persist records the effect for replay; the spec’s pure form (runner calls apply_effects(effects, ctx) once per turn) lands when the proxy flag flips from “record + apply immediately” to “record-only, defer to runner.”
• Uniform Contract migration is real but the ratchet meant to protect it is currently broken. Compute the current migrated/deferred split from engine.contract._TOOL_SPECS vs. engine.tool_definitions.SYSTEM_TOOLS rather than trusting a fraction printed here, it changes with every tool added or migrated. tests/test_uniform_contract.py::test_deferred_set_only_shrinks is failing as of 2026-09, for two causes worth separating. Two tools (convene, present_rows) are registered nowhere and bypass the contract at dispatch. Separately, the test discovers migrated tools through a hand-maintained module list, so it also reports as deferred a number of tools that are in fact migrated, inflating the count and hiding the two real gaps inside it. Both are real gaps to close, not doc inaccuracies to route around. The legacy ToolResult shape stays alive at the registry adapter (envelope_to_tool_result) until the deferred set is empty. The per-tool inventory is tracked separately, but treat any specific counts in them the same way, verify against the registry before repeating one.

Related docs

• backend/engine.md, orchestrator + executor + dispatcher + the contract in detail
• tool-system.md, @uniform_contract, ToolSpec, Ok | Hint | Err, registration
• backend/shared-layer.md, DisplayFactory, EventRecorder, DisplayEvent wire format
• backend/conversations.md, ChatMessage, structured_content, ownership rules
• database-schema.md, conversation_events, conversation_event_blobs, generated_files, usage_log
• glossary.md, DisplayFactory, ToolResult, EventRecorder, TurnState, Effect, Ok | Hint | Err
• tool-system.md, the contract runtime and the typed surface the runner, executor and dispatcher target