Conversation Message Processing#

Two packages collaborate to turn raw hierarchical message data into AI-model-compatible output:

• conversation-flow — parses a flat message array into structured display representations (contextTree and flatList). Its public API is a single parse() function.
• context-engine — runs a configurable pipeline of processors over a PipelineContext to transform those messages before they are sent to an AI model. Orchestrated by the ContextEngine class.

The two packages serve complementary roles: conversation-flow builds the display model from raw DB messages; context-engine prepares those messages for the model API before each inference call.

conversation-flow: Three-Phase Parse Pipeline#

parse(messages, messageGroups?) is the single entry point 1. It runs three phases:

1. Indexing — buildHelperMaps builds a messageMap and childrenMap for O(1) access patterns.
2. Structuring — buildIdTree converts flat parent-child relationships into a tree of IdNodes.
3. Transformation — a Transformer instance produces two outputs 2:
   • contextTree — semantic tree for navigation (node types: MessageNode, AssistantGroupNode, BranchNode, SignalCallbacksNode, etc.)
   • flatList — flat array optimised for virtual list rendering

A pre-processing step before Phase 1 remaps sub_agent-scoped messages to use their subAgentId as agentId, so downstream collectors never merge messages from different agents into the same group 3.

Post-processing promotes metadata.usage to a top-level usage field on each flat message, normalising token data written by different executor paths 4.

Dual-Form Message Chain Support#

The parser handles two persisted parentage shapes that must produce identical output 5:

Both forms, and any mix within a single turn, collapse to the same assistantGroup node in flatList 6. Key invariants enforced by the reader:

1. A tool message is always inline data of its assistant (both forms).
2. A branch is ≥ 2 non-tool siblings under one parent.

The test suite dualForm.test.ts documents seven fixture classes: old form, new form, mixed forms, parallel tools, regenerate-branch, assistant-anchored regenerated continuation, and async-task summary placement.

MessageCollector#

MessageCollector is the workhorse of the transformation phase. It is constructed with a messageMap, childrenMap, and a BranchResolver. Key methods:

The signal lineage (metadata.signal) is read by getMessageSignal, which deliberately returns undefined when the assistant carries tools — the writer attaches the tag at stream start before knowing whether tools will be used, so the collector defangs that mismatch.

Phase 2 migration note: When a dedicated messages.signal column lands, swap metadata?.signal for (msg as any).signal ?? msg.metadata?.signal in getMessageSignal 7.

BranchResolver#

BranchResolver determines which branch is active when multiple non-tool siblings exist under one parent (e.g. a regenerated reply). It uses a three-tier priority 8:

1. metadata.activeBranchIndex — explicit integer index; if index === children.length, the branch is being optimistically created and undefined is returned.
2. Infer from descendants — the child that already has descendants is the active one.
3. Default — first child.

Two method variants exist for different traversal contexts 9:

• getActiveBranchId(message, idNode) — used during contextTree building (works on IdNode structures).
• getActiveBranchIdFromMetadata(message, childIds, childrenMap) — used during flat-list building.

MessageCollector.resolveActiveContinuationId() calls getActiveBranchIdFromMetadata to pick the correct continuation when >1 non-tool same-agent siblings share a parent 10.

context-engine: ContextEngine Pipeline#

ContextEngine sequences an ordered array of ContextProcessor instances over a mutable PipelineContext 11:

{ messages, metadata, initialState, isAborted } → Processor₁ → Processor₂ → … → PipelineResult

• Early termination: any processor can set context.isAborted = true; subsequent processors are skipped 12.
• Error wrapping: processor failures are caught and re-thrown as PipelineError with the processor name and a 300-char cause summary for dashboard triage 13.
• Stats: PipelineResult includes per-processor processorDurations and totalDuration 14.
• Fluent API: addProcessor, removeProcessor, clear, clone, validate 15.

Key Processors#

All processors extend BaseProcessor (abstract class implementing the ContextProcessor interface) and are exported from packages/context-engine/src/processors/index.ts.

Message Structure#

Tool Handling#

Async Task Support#

Context Management#