Agent Pipeline Architecture

The agent pipeline is the technical heart of Owlat's evolution from an outbound email platform to a communication intelligence system. Every inbound email flows through a multi-step processing chain that scans, retrieves context, classifies, drafts, and routes to either auto-send or a human review queue.

Step-module architecture

The pipeline is not one big function. Per ADR-0014, each stage is an Agent step module — a small object with two pure functions — and a single walker drives them in sequence. The step modules live under apps/api/convex/agent/steps/<kind>/index.ts; the walker is apps/api/convex/agent/walker.ts.

A step module (apps/api/convex/agent/steps/types.ts) is:

interface AgentStepModule<K, In, Out> {
  readonly kind: K
  readonly llm?: { tier: 'fast' | 'capable' }   // present iff the step calls an LLM
  execute(ctx, input): Promise<AgentStepResult<Out>>  // does the work
  route(output, input, runCtx): AgentRoute            // decides what happens next
}

route() returns one of three shapes:

The walker (runStep) opens an agentActions row, calls module.execute(), applies module.route(), and self-schedules the next step via ctx.scheduler.runAfter(0, …). Any exception inside a step transitions the message to failed with the failing action recorded.

The five steps

The step kinds are exactly: security_scan, context_retrieval, classify, draft, route (AgentStepKind in agent/steps/types.ts). A plan step was designed but dropped pre-prod with ADR-0014 — there is no action-planning stage.

Step 1: Security scan

apps/api/convex/agent/steps/security_scan/index.ts — the entry step, run while processingStatus is received/security_check. It scans the inbound message for prompt injection, instruction smuggling, and spam. The deterministic layer uses detectInjection, detectSmuggling, and calculateSpamScore from patterns.ts; on top of that, a guard-tier LLM injection classifier (getLLMProvider('guard') + runLlmObject) catches novel/obfuscated phrasings the regexes miss. The two verdicts are OR-ed — either flagging with high confidence is treated as an injection — and the LLM call fails open (any model error falls back to the pattern-only verdict so a flaky model never blocks the pipeline).

Routing:

• Injection detected (pattern or guard LLM), confidence ≥ 0.7 → quarantined (held for human release).
• Spam score ≥ 80 → archived (reason spam).
• Clean, agent enabled → transition to classifying and schedule context_retrieval.
• Clean, agent disabled → record the scan and stop (done).

Step 2: Context retrieval

apps/api/convex/agent/steps/context_retrieval/index.ts — runs in_state while processingStatus is classifying. It assembles a briefing string the LLM steps consume from:

• Contact profile — name, language, timezone from contacts
• Recent activity — the last few contactActivities
• Conversation history — earlier messages in the same thread (inboundMessages by thread)
• Knowledge graph — semantically relevant typed entries via internal.knowledge.retrieval.semanticSearch (the [KNOWLEDGE] section, capped by knowledgeEntryLimit)
• Relevant source files — semantically relevant uploaded documents via internal.semanticFileProcessing.semanticSearch (the [RELEVANT FILES] section, capped by fileLimit)
• The current message — from / to / subject / body

The briefing is token-budgeted (≈4 chars/token, 4000-token budget) with three compaction tiers, recorded on inboundMessages.contextTier:

Step 3: Classification

apps/api/convex/agent/steps/classify/index.ts — uses the fast model tier via getLLMProvider('classify') and runLlmObject for type-safe structured output (generateObject under the hood). It produces one classification:

{
  category: 'support' | 'sales' | 'billing' | 'feature_request'
          | 'complaint' | 'spam' | 'internal' | 'other',
  priority: 'urgent' | 'normal' | 'low',
  sentiment: 'positive' | 'neutral' | 'negative',
  intent: 'question' | 'complaint' | 'request' | 'information'
        | 'escalation' | 'acknowledgment',
  confidence: number,  // 0–1
}

Routing: spam → archived; complaint or urgent → straight to draft_ready (skip the drafter, force human review); everything else → transition to drafting and schedule the draft step. The classification is stored on inboundMessages.classification.

Step 4: Draft generation

apps/api/convex/agent/steps/draft/index.ts — uses the capable model tier via getLLMProvider('draft') and runLlmText. The draft is grounded in agentConfig.toneDescription, agentConfig.signatureTemplate, and the assembled context. As defense-in-depth, the drafter re-scans the assembled context for injection patterns (thread history that wasn't individually scanned upstream); a high-confidence hit fails the step. Output is stored to inboundMessages.draftResponse / draftSubject with a confidenceScore.

Step 5: Routing

apps/api/convex/agent/steps/route/index.ts — the terminal step. It decides auto-approve vs. human review in three tiers, safest-first:

1. Circuit-breaker safety gate. Any open breaker (internal.agentHealth.getCircuitBreakersInternal) forces human_review — a degraded pipeline never auto-sends, regardless of config.
2. Graduated autonomy (when the ai.autonomy flag is on). internal.autonomy.checkPermissionInternal applies the per-category autonomyRules (threshold + daily cap + breaker check). A category with no rule is never auto-approved; a permitted one charges the per-category daily count.
3. Legacy global fallback (autonomy flag off). The single agentConfig singleton — isAutoReplyEnabled, confidenceThreshold, and the daily-cap fields.

On auto_approve it transitions to approved; the lifecycle's schedule_send_approved effect then sends the reply via internal.agent.agentPipeline.sendApprovedReply. Otherwise it transitions to draft_ready (awaiting human review).

Inbound email integration

MTA → Convex flow

Inbound mail for the shared agent inbox arrives over the standard MTA webhook, not a dedicated inbound route:

Inbound SMTP → owlat-mta
  → POST /webhooks/mta  (apps/api/convex/http.ts → handleMtaWebhook)
    → webhook pipeline verifies HMAC, parses the `inbound.received` event
      (apps/api/convex/webhooks/adapters/mta.ts)
    → dispatcher routes inbound.received → internal.inbox.messages.receiveMessage

receiveMessage (apps/api/convex/inbox/messages.ts, an internalMutation) then:

1. Upserts the sender as a contacts row (channel email, source inbound).
2. Resolves or creates a conversationThreads row via the threading cascade.
3. Inserts the inboundMessages row with processingStatus: 'received'.
4. Records an inbound_received contact activity.
5. Schedules internal.agent.walker.start, which kicks the pipeline off at security_scan.

Threading

Threading (owned by the thread module called from receiveMessage) is a three-strategy cascade:

1. In-Reply-To header → find the referenced message's thread.
2. References header → find any referenced message's thread.
3. Fallback: normalized subject (Re:/Fwd: stripped, lowercased) + contactIdentifier.

A matched thread reopens if it was closed; thread messageCount / lastMessageAt are maintained atomically.

Schema additions

The inbox/agent tables are defined in apps/api/convex/schema/inbox.ts.

inboundMessages

Stores every inbound email with its processing state.

conversationThreads

Groups related messages into conversations.

agentActions

One row per pipeline step execution (not per planned action).

agentConfig

The operational-tuning singleton. The master on/off switch is the ai.agent feature flag (isAgentEnabled in agent/agentPipeline.ts calls isFeatureEnabled(ctx, 'ai.agent')), not a column on this table.

Model routing

Different pipeline steps have different requirements. Classification needs speed and structured output; draft generation needs quality and nuance. The provider resolver (apps/api/convex/lib/llmProvider.ts) routes each task to one of two model tiers.

Per-task model selection

getLLMProvider(task) resolves an OpenAI-compatible client and the model id for the task's tier:

// apps/api/convex/lib/llmProvider.ts
export function getLLMProvider(task: LLMTask = 'draft') {
  return getClient()(modelIdForTier(taskTier(task)))
}

Tier resolution lives in taskTier() (apps/api/convex/lib/llmProvider.ts):

Configuration

See ADR-007 for the provider abstraction and ADR-009 for the routing decision. The defaults are gpt-4o-mini (fast) and gpt-4o (capable), per apps/api/convex/lib/llmProvider.ts.

Self-hosters running a single Ollama model can set only LLM_MODEL — both tiers fall back to it. Organizations with GPU budget can split: a small model for classification and a larger model for drafting.

Quarantine

Messages flagged by the security scan are not silently dropped. They are stored with processingStatus: 'quarantined' and a securityFlags object, and surface in a dedicated admin view (apps/web/app/pages/dashboard/inbox/quarantine.vue) where a human can review and release them. Releasing a quarantined message transitions it back to received (the lifecycle's release_quarantine source), which re-runs the pipeline from security_scan.

Verification queue

The verification queue is the human-in-the-loop interface — a view over inboundMessages filtered by processingStatus = 'draft_ready', not a separate system.

UI

• /dashboard/inbox — thread list with filters.
• /dashboard/inbox/[threadId] — full conversation with the agent draft and approve / edit / reject controls.
• /dashboard/inbox/review (review.vue) — focused review queue of items needing attention.
• /dashboard/inbox/quarantine (quarantine.vue) — security-flagged messages.

Actions

• Approve — sends the draft as-is; status → approved, then sent after a confirmed provider dispatch.
• Edit and approve — modify the draft, then send.
• Reject — status → rejected.
• Reassign — route to a different team member.

See the operator-facing Team Inbox and AI Agent & Autonomy guides for the full workflow.

Graduated autonomy

The autonomy and ops layer (tables in schema/inbox.ts) is now wired into the live routing decision, behind the ai.autonomy feature flag.

Message coalescing

Email threads often arrive as bursts — a CC chain with three replies in 30 seconds. Coalescing collapses such a burst into a single pipeline run on the most recent message instead of running it per message.

Multi-intent branching (future)

A future design lets a message with multiple intents (e.g. "check my billing status, and we'd love dark mode") fork into parallel worker branches, each producing its own draft, with the review queue showing both linked to the same inbound message.

Related