Agent Pipeline

The agent pipeline manages all background processing in Cortexa through a SQLite-backed job queue and 2 async worker threads with priority scheduling. It handles embedding generation, LLM summarization, the Deep Think thesis pipeline, research synthesis, knowledge graph linking, and audio briefing generation. Lightweight jobs (embed, auto_link) run at higher priority before heavy LLM jobs (reflection, cross-synthesis).

Architecture

                    ┌─────────────────┐
                    │   Tauri Commands │
                    │   HTTP API       │
                    │   MCP Server     │
                    └────────┬────────┘
                             │ enqueue(Job)
                             v
                    ┌─────────────────┐
                    │    Job Queue     │
                    │   (SQLite DB)    │
                    └────────┬────────┘
                             │ claim_next()
               ┌─────────────┴─────────────┐
               v                           v
          ┌─────────┐                ┌─────────┐
          │ Worker 0 │                │ Worker 1 │
          └────┬────┘                └────┬────┘
               │                           │
               v                           v
          handlers::execute(job)
               │
               ├── embed()
               ├── summarize()
               ├── auto_link()
               ├── expand()
               ├── counterpoint()
               ├── synthesize()
               ├── research()
               ├── generate_briefing()
               ├── generate_item_briefing()
               ├── orchestrate_item()
               ├── reflection_pass()
               └── cross_synthesize()

Job Queue

The job queue is implemented in src-tauri/src/agent/queue.rs. It uses a jobs table in the same SQLite database as items, protected by a std::sync::Mutex<Connection>.

Job Types

rust

pub enum Job {
    Embed { item_id: String },
    Summarize { item_id: String },
    AutoLink { item_id: String },
    Expand { item_id: String },
    Counterpoint { item_id: String },
    Synthesize { item_id: String },
    Research { item_id: String },
    GenerateBriefing { briefing_id: String },
    GenerateItemBriefing { item_id: String },
    Orchestrate { item_id: String },
    ReflectionPass,
    CrossSynthesize { peer_item_id: String },
}

Job	Trigger	LLM Required	Description
`Embed`	Every new/updated item	Embedding provider	Generate embedding vector, store in `items.embedding`, insert into HNSW index
`Summarize`	Auto-summarize or manual toggle	Chat provider	Generate one-paragraph summary, store in `items.summary`
`AutoLink`	After embedding completes	None (uses HNSW)	Find top-K nearest neighbors, create/update `links` rows
`Expand`	Deep Think pipeline step 1	Chat provider	Decompose idea into thesis, sub-questions, expansion, thinking lenses
`Counterpoint`	Deep Think pipeline step 2	Chat provider	Critique, find contradictions with prior thoughts, devil's advocate
`Synthesize`	Deep Think pipeline step 3	Chat provider	Produce structured JSON brief with readiness gates
`Research`	Manual research toggle	Chat provider + DuckDuckGo	Pull SERP snippets, synthesize with LLM, store as linked research item
`GenerateBriefing`	Scheduled or manual	Chat provider + TTS	Generate spoken script from N items, optionally produce audio via OpenAI TTS or Piper (local)
`GenerateItemBriefing`	Per-item Brief button or auto-brief	Chat provider + TTS	Generate spoken briefing for a single thought
`Orchestrate`	Deep Think (orchestrator mode)	Chat provider	LLM-powered orchestrator plans and executes registry tools across the specialized agents with parallel scheduling
`ReflectionPass`	Scheduled (configurable interval)	Chat provider	Analyze recent thoughts for themes, contradictions, patterns, gaps. Runs confidence decay. Stores output as `type:reflection` item
`CrossSynthesize`	Hive import of item with thesis	Chat + Embedding	Semantic search for local items with theses on similar topics, LLM synthesis of cross-user perspectives. Stores output as `type:cross-synthesis` item

Job Lifecycle

enqueue() ──> PENDING ──> claim_next() ──> RUNNING
                 ^                            │
                 │                    ┌───────┴───────┐
                 │                    │               │
                 │               success           failure
                 │                    │               │
                 │               complete()       fail()
                 │                    │               │
                 │                    v         attempts < max?
                 │                  DONE         │         │
                 │                              yes        no
                 │                               │         │
                 └──── (backoff) ────────────────┘         v
                                                         DEAD

Enqueue

Jobs are enqueued with a UUID, the job type, a JSON-serialized payload, and status = 'pending':

rust

pub fn enqueue(&self, job: Job) -> Result<String> {
    self.enqueue_at(job, Utc::now(), DEFAULT_MAX_ATTEMPTS)  // max_attempts = 3
}

After insertion, the queue calls notify.notify_one() to wake an idle worker.

Claim

Workers call claim_next() which atomically selects the highest-priority pending job whose next_run_at is in the past and updates it to running. Priority levels: embed=0, auto_link=1, summarize=2, expand/counterpoint/synthesize=3, research/orchestrate=4, briefing=5, reflection/cross_synthesize=6.

sql

SELECT ... FROM jobs WHERE status = 'pending' AND next_run_at <= ?
ORDER BY priority ASC, next_run_at ASC LIMIT 1

UPDATE jobs SET status = 'running', attempts = attempts + 1 WHERE id = ? AND status = 'pending'

The two-step select + conditional update provides atomic claiming without database-level locks. If another worker claims the same job between the SELECT and UPDATE, the UPDATE changes zero rows and the worker retries.

Retry with Exponential Backoff

When a job fails, fail() checks whether the attempt count has reached max_attempts:

Under the limit: Status returns to pending with a delayed next_run_at:
rust
```
fn backoff_secs(attempts: u32) -> u64 {
    let base = 1u64 << attempts.min(10); // 2^attempts
    base.min(60)
}
```
Attempt Backoff
1 2 seconds
2 4 seconds
3 8 seconds
4 16 seconds
5 32 seconds
6+ 60 seconds (cap)
At the limit: Status changes to dead. The worker emits a job-dead event.

Attempt	Backoff
1	2 seconds
2	4 seconds
3	8 seconds
4	16 seconds
5	32 seconds
6+	60 seconds (cap)

Dead Letter Handling

Dead jobs remain in the database for inspection. The UI shows them in the job queue trace panel. Two recovery paths exist:

Manual retry: retry(id) resets a dead job to pending with attempts = 0 and next_run_at = now, making it immediately claimable.
Purge: purge_dead() deletes all dead jobs. purge_done() cleans up completed jobs.

Crash Recovery

On startup, the queue resets any running jobs to pending:

sql

UPDATE jobs SET status = 'pending' WHERE status = 'running'

This handles the case where the app crashed while a job was executing. The job will be re-attempted on next startup.

Worker Pool

Two workers are spawned in src-tauri/src/agent/worker.rs via spawn_workers(handle, 2). Each worker runs an async loop:

rust

async fn worker_loop(handle: AppHandle, notify: Arc<Notify>, worker_id: usize) {
    loop {
        // Pause while vault is locked
        if state.is_locked() { sleep(2s); continue; }

        match queue.claim_next() {
            Ok(Some(rec)) => {
                // Execute the job handler
                handlers::execute(&handle, rec.job).await;
                // Report success or failure
            }
            Ok(None) => {
                // No work -- wait for notify or 2-second timeout
                timeout(2s, notify.notified()).await;
            }
            Err(e) => { sleep(2s); }
        }
    }
}

Key behaviors:

Vault lock: Workers pause while the vault is locked. Jobs accumulate in the queue and resume processing after unlock.
Notify wake: When a job is enqueued, notify.notify_one() wakes an idle worker immediately rather than waiting for the 2-second poll.
Tracing: Every job start, completion, and failure is recorded in the trace system and emitted as Tauri events (job-started, job-completed, job-dead, job-retry-scheduled).

Deep Think Pipeline

The three-stage thesis pipeline is the most complex job chain. Each stage is a separate job, chained automatically:

Stage 1: Expand

The expand() handler:

Reads the item content (decrypts blob, includes text attachments inline, includes image attachments as base64).
Sets thesis status to expanding.
Sends the content to the chat LLM with the EXPAND system prompt, which instructs the LLM to: clarify as a concrete thesis, generate sub-questions, explore implications, and apply thinking lenses (pre-mortem, first principles, inversion, steelman).
Stores the expansion output in theses.expansion.
Enqueues a Counterpoint job.

Stage 2: Counterpoint

The counterpoint() handler:

Reads the item content and the expansion from Stage 1.
Searches the HNSW index for related prior thoughts.
Sets thesis status to critiquing.
Sends the content + expansion + related thoughts to the LLM with the COUNTERPOINT system prompt, which instructs the LLM to: identify weaknesses, flag contradictions with prior thoughts, cite prior art, and write a devil's advocate argument.
Stores the counterpoints in theses.counterpoints.
Enqueues a Synthesize job.

Stage 3: Synthesize

The synthesize() handler:

Reads the item content, expansion, and counterpoints.
Sets thesis status to synthesizing.
Sends everything to the LLM with the SYNTHESIZE system prompt, which instructs the LLM to produce a strict JSON object matching the ThesisOutput schema.
Parses the JSON response (with fallback handling for fenced code blocks, preamble text, and malformed JSON).
Stores the structured thesis in theses.thesis.
Sets status to done.

Thesis Status Tracking

The ThesisStatus enum tracks pipeline progress:

pending → expanding → critiquing → synthesizing → done
                                                    │
                                      (any failure) → failed

If any stage's job reaches dead status (max retries exhausted), the worker marks the thesis as failed with the error message. The UI shows the pipeline state in real time.

Structured Output

The final ThesisOutput contains:

rust

pub struct ThesisOutput {
    pub core: String,                    // One-sentence thesis
    pub problem_statement: Option<String>,
    pub sub_questions: Vec<String>,
    pub supporting_arguments: Vec<String>,
    pub risks: Vec<String>,
    pub prior_art: Vec<String>,
    pub open_questions: Vec<String>,
    pub next_steps: Vec<String>,
    pub confidence: Option<String>,      // speculation | partially-vetted | well-researched
    pub contradictions: Vec<String>,
    pub lenses: Vec<LensResult>,         // pre-mortem, first-principles, steelman
    pub gates: Vec<Gate>,                // 8 readiness gates with open/closed status
}

The 8 readiness gates are: problem-stated, success-criterion, scope-bounded, approach-selected, open-questions-resolved, prior-art-reviewed, risks-acknowledged, first-milestone-identified.

Research Jobs

The research() handler:

Reads the item content.
Extracts a focused search query via the LLM.
Sends the query to DuckDuckGo and parses SERP HTML for snippets.
Passes the snippets to the LLM with the RESEARCH system prompt for synthesis.
Creates a new item with the research output.
Links the research item to the source item in the links table.

Briefing Jobs

The generate_briefing() handler:

Reads the briefing record to get scope and source item IDs.
Sets status to scripting.
Loads content from the source items.
Sends to the LLM with the BRIEFING system prompt to generate a 2-4 minute spoken script.
Stores the encrypted script.
If TTS is enabled, sets status to voicing and calls the OpenAI TTS API.
Encrypts and stores the audio file.
Sets status to done.

Event Emission

The pipeline emits Tauri events at each stage so the frontend can update in real time:

Event	Payload	When
`job-started`	Job ID	Worker claims a job
`job-completed`	Job ID	Job finishes successfully
`job-dead`	Job ID	Job exhausts all retries
`job-retry-scheduled`	Job ID	Job failed but will retry
`thesis-updated`	Item ID + status	Thesis pipeline state changes
`summary-set`	Item ID + summary text	Summary is stored
`embedding-set`	Item ID	Embedding is stored
`embedding-migration-progress`	Progress info	During re-embed all
`reflection-started`	Reflection ID	Reflection pass begins
`reflection-completed`	New item ID	Reflection pass finishes, item created
`cross-synthesis-completed`	New item ID	Cross-synthesis finishes, item created
`confidence-updated`	Count of updated items	Confidence decay pass completes

Background Reflection

The reflection_pass() handler runs on a configurable schedule (default every 6 hours). It implements cross-item pattern detection — analyzing recent thoughts collectively rather than individually.

Algorithm

Load settings, check reflection.enabled.
Query recent N items since last_reflection_at (content + summary + thesis core).
Require at least 3 items (skip if fewer — not enough data for patterns).
Build a combined prompt with all items numbered by ID.
Call LLM with REFLECTION_SYSTEM prompt requesting structured JSON output.
Parse output for: themes, contradictions, patterns, gaps, connections, narrative.
Store the reflection as a new type:reflection item.
Record in the reflections table for tracking.
If confidence is enabled, run recalculate_confidence() — decay all scores and tag stale items.
Update settings.reflection.last_reflection_at.

Scheduling

The reflection schedule is checked in tick_schedules() alongside the daily briefing and weekly review. It fires when Utc::now() - last_reflection_at > interval_hours.

Cross-User Thesis Synthesis

The cross_synthesize() handler is triggered when a Hive peer shares a thought that includes thesis data. It finds related local thoughts and produces a cross-perspective synthesis.

Algorithm

Check reflection.cross_synthesis is enabled.
Load the peer item's content and thesis.
Embed the peer content and search the HNSW index for similar local items.
Find the best local item that also has a thesis (skip if none match above threshold).
Call LLM with CROSS_SYNTHESIS_SYSTEM prompt providing both perspectives.
Parse output for: agreements, disagreements, novel_combinations, synthesis.
Store as a new type:cross-synthesis item with bidirectional links to both sources.

Hook Point

The cross-synthesis job is enqueued in hive_import_feed_item() after the embed job, only when the imported item's ThoughtMeta contains thesis_json.

Memory Confidence & Decay

Confidence scoring adds a numerical quality signal to every thought that decays over time, surfacing stale knowledge for review.

Initial Assignment

On ingest, compute_initial_confidence() assigns a score based on source:

Manual capture: 0.8
MCP / API / CLI: 0.7
Hive shared: 0.6
Telegram: 0.5

Decay Formula

decayed_score = current_score * (decay_rate ^ weeks_since_validated)

With the default rate of 0.95/week:

After 5 weeks: ~77% of original
After 10 weeks: ~60% of original
After 23 weeks: items approach the stale threshold

Stale Tagging

Items below confidence.stale_threshold (default 0.3) are tagged confidence:stale. Items above the threshold have the tag removed. This runs during each reflection pass.

Validation Reset

User interactions that demonstrate engagement with an item reset last_validated_at to now, effectively resetting the decay clock:

Pinning an item
Any future edit/review interaction

Testing

The job queue has unit tests covering:

Enqueue then claim returns the correct job
Fail with retries schedules backoff then reaches dead
Complete marks done and clears last_error
Retry dead job makes it claimable again
Purge dead removes only dead jobs
Backoff grows exponentially with attempt count (capped at 60s)

Agent Pipeline ​

Architecture ​

Job Queue ​

Job Types ​

Job Lifecycle ​

Enqueue ​

Claim ​

Retry with Exponential Backoff ​

Dead Letter Handling ​

Crash Recovery ​

Worker Pool ​

Deep Think Pipeline ​

Stage 1: Expand ​

Stage 2: Counterpoint ​

Stage 3: Synthesize ​

Thesis Status Tracking ​

Structured Output ​

Research Jobs ​

Briefing Jobs ​

Event Emission ​

Background Reflection ​

Algorithm ​

Scheduling ​

Cross-User Thesis Synthesis ​

Algorithm ​

Hook Point ​

Memory Confidence & Decay ​

Initial Assignment ​

Decay Formula ​

Stale Tagging ​

Validation Reset ​

Testing ​

Agent Pipeline

Architecture

Job Queue

Job Types

Job Lifecycle

Enqueue

Claim

Retry with Exponential Backoff

Dead Letter Handling

Crash Recovery

Worker Pool

Deep Think Pipeline

Stage 1: Expand

Stage 2: Counterpoint

Stage 3: Synthesize

Thesis Status Tracking

Structured Output

Research Jobs

Briefing Jobs

Event Emission

Background Reflection

Algorithm

Scheduling

Cross-User Thesis Synthesis

Algorithm

Hook Point

Memory Confidence & Decay

Initial Assignment

Decay Formula

Stale Tagging

Validation Reset

Testing