Incremental Map-Reduce LLM Pipeline with Idempotent State and Prompt-Injection Defense

For the same leadership-development AI platform I built the production infrastructure for (AI Book Companion Platform with Enterprise Security and Observability), I designed and built the insight-synthesis layer — the pipeline that reads each user’s entire conversation history and distills it into a small set of durable, personalized insights that evolve as they keep using the product. The system is an incremental map-reduce over an LLM: a MAP stage extracts a structured signal per conversation, a REDUCE stage folds those signals into insights against a bounded, durable profile of the user’s history. Built on Temporal for reliability, it re-processes only what changed (keeping per-run LLM cost proportional to new activity rather than total history), treats every LLM output as untrusted input, and is idempotent under retries — no false success, no double-counted history, no silently dropped conversations.

The Challenge

The platform runs a Socratic AI companion whose value compounds over time — the more a user reflects, the more the product should understand their evolving practice. Turning that accumulated history into trustworthy, longitudinal insight raised a set of problems that single-shot LLM calls don’t have:

• Longitudinal, not per-session: insights must reflect a user’s whole arc — dozens to hundreds of conversations — not one chat
• Unbounded, ever-growing history: a user’s transcript only grows, so naively re-reading everything on each run is cost-prohibitive at a per-user, minutes-level cadence
• Cost discipline: only conversations that actually changed should ever hit the LLM again
• Correctness under a distributed retry engine: a retried or duplicated run must not double-count history, advance state past unprocessed data, or persist a partial result as “done”
• LLM output is untrusted: extracted content is persisted and fed back into later prompts — out-of-vocabulary or injection-poisoned output must never corrupt durable state
• Prompt injection: conversation text is fully user-controlled and flows directly into extraction prompts
• Never truncate: an arbitrarily long conversation must be fully processed, not clipped to fit a context window

They needed:

• Personalized, evolving insights synthesized from a user’s complete conversation history
• Cost that scales with new activity, not with total history size
• Provable correctness under Temporal’s at-least-once activity execution
• A hardened trust boundary around every LLM call whose output touches persisted state
• Full, lossless processing of long conversations

Solution Architecture

I designed a two-stage map-reduce driven by a completion poller, with a durable per-user profile that lets REDUCE fold new activity forward instead of re-reading history:

MAP — per-conversation structured extraction

The MAP activity extracts one structured ConversationExtract per conversation that needs (re)processing:

• Change detection: a per-conversation source_fingerprint is compared against the persisted extract, so only genuinely changed conversations are re-mapped — every unchanged conversation is skipped before any token is spent
• Bounded, isolated parallelism: candidate conversations are mapped concurrently under an asyncio.Semaphore, with Temporal heartbeats as they complete; a failure on one conversation is isolated and never fails the user’s run
• No-truncation windowing: oversized conversations are split into token-budgeted windows, each extracted separately and then merged — content is never clipped, and the merge deliberately preserves the earliest framing markers as frozen “then” anchors
• Model-facing schema with code-owned provenance: extraction is enforced as a Pydantic model via tool-calling structured output; factual provenance the model can’t know (conversation UUID, timestamps) is injected by code, not produced by the model
• Cold-start draining: when more conversations remain than one batch should map, the user’s row is set to draining and the workflow returns in_progress so the poller re-fires next cycle — bounded batches over poll cycles instead of an unbounded continue_as_new
• Fail-closed coverage gate: if the per-batch error ratio exceeds a configured threshold, REDUCE is skipped entirely and the failed conversations re-map next cycle — the system never synthesizes insights from partial coverage

REDUCE — folding extracts into insights

The REDUCE activity aggregates a user’s extracts into the insight blob, backed by a durable PracticeProfile that keeps the token budget bounded no matter how long the history grows:

• Bootstrap vs. fold-forward: on first run, a schema bump, or an invalidated profile, a chronological hierarchical reduce folds all history into a compacted profile (running outline + principle/dimension tallies + frozen earliest anchors). On every subsequent run it folds forward — advancing a covered_through watermark and folding only the extracts that have aged out of the granular recent window
• Recent-at-full-granularity, history-compacted: the most recent extracts are read verbatim (they drive the current pattern); everything older lives in the bounded profile outline. This is what caps cost as history grows without number
• Idempotency, three ways: a SHA-256 generation_hash over the inputs short-circuits any run whose inputs are unchanged; the fold happens before the watermark is mutated, so a failure can never advance covered_through past an unsummarized batch; and the insight write plus its side effects commit in a single caller-owned transaction
• Two-call gated synthesis: a pattern unit (through-line + unexplored-principle) and a gated “what’s shifted” unit run concurrently with per-call failure isolation; the expensive shift analysis is gated behind minimum-history and minimum-timespan thresholds
• Sharpen, don’t whiplash: prior observations are fed back into the synthesis prompt so insights evolve incrementally rather than lurching between runs, with section-level freshness resolving fresh output against the previous result

AI Security and Trust Boundaries

Because extracted output is persisted and re-fed into later prompts, I treated the LLM as an untrusted component and hardened the boundary on both sides of every call:

• User content is DATA, not instructions: transcript text has its angle brackets escaped before entering the prompt, so a message can’t close the <transcript> wrapper or inject markup the model would treat as instructions — the transcript stays human-readable while being inert
• LLM output is allow-listed before persistence: every principle and behavioral dimension the model emits is checked against the controlled catalog/enum and dropped if it isn’t a member, so a prompt-injection attempt can’t poison durable state with fabricated vocabulary
• Provenance is code-owned: conversation identity and timestamps are injected by code and never taken from model output, so the model cannot misattribute an extract to another conversation or user
• Fail-closed everywhere: partial MAP coverage, a failed profile fold, or a failed core synthesis all refuse to persist a “success” — the run is marked for retry with prior state left intact, rather than shipping a confidently-wrong result

Technical Implementation

• Temporal orchestration: MAP and REDUCE are activities with explicit start-to-close (10 min) and heartbeat (5 min) timeouts and bounded retry policies; the workflow is triggered per-user by a completion poller, and REDUCE re-raises retryable failures so Temporal — not a swallowed exception — owns the retry
• Structured output via tool-calling: every LLM interaction (extraction, profile fold, pattern synthesis, shift synthesis) is a schema-validated tool call into a Pydantic model, eliminating free-text parsing
• Langfuse observability: one trace per conversation in MAP (session-scoped to the conversation, each window a separate generation) and one per-user trace for the whole REDUCE, giving prompt/latency/cost visibility at both granularities
• Pure-function core: partitioning, window/epoch merging, tally accumulation, and the generation-hash preimage are all pure functions, unit-tested independently of the database and the LLM — the deterministic logic is provable without touching a model
• Config-driven budgets: token budgets, concurrency limits, batch sizes, the coverage error-ratio threshold, and the shift-gating thresholds are all externalized settings rather than hard-coded constants

Key Design Decisions

Incremental fold-forward over full re-synthesis:

• Re-reading a user’s entire history on every run is simple but scales cost with total history — untenable at a minutes-level cadence
• The durable profile + watermark makes per-run cost proportional to new conversations, so a long-tenured user costs no more per run than a new one

Watermark advanced only after a successful fold:

• Advancing covered_through before persisting the fold would let a mid-run failure silently drop the unsummarized batch from history forever
• Folding first and mutating second makes the whole REDUCE safely retryable — a post-commit retry sees the advanced watermark and correctly folds nothing

Treat LLM output as untrusted input:

• Most pipelines validate the shape of structured output; here the contents are allow-listed against a controlled vocabulary before they ever reach durable state
• Combined with transcript escaping and code-owned provenance, this closes the prompt-injection path into persisted, re-prompted data

Cold-start drain over continue_as_new:

• Bounded per-cycle batches that lean on the existing poller are simpler to reason about and observe than a self-continuing workflow, with no loss of eventual completeness

Fail-closed synthesis:

• A coverage gate and retryable-error contract guarantee the system would rather produce nothing this cycle than a plausible insight synthesized from partial or corrupted inputs

Results & Impact

• Delivered a longitudinal insight-synthesis layer that distills a user’s full conversation history into three evolving, personalized insights (a through-line pattern, an unexplored principle, and how their framing has shifted over time)
• Architected an incremental map-reduce that keeps per-run LLM cost proportional to changed conversations rather than total history, via per-conversation fingerprint change-detection and a durable fold-forward profile
• Made the pipeline idempotent under Temporal’s at-least-once execution — generation-hash short-circuiting, fold-before-mutate watermarking, and single-transaction side effects eliminate false success and double-counting
• Hardened the LLM trust boundary end-to-end: prompt-injection escaping of user content, controlled-vocabulary allow-listing of persisted model output, and code-owned provenance
• Guaranteed lossless processing of arbitrarily long conversations through token-budgeted windowing that splits, extracts, and merges without truncation
• Instrumented full Langfuse observability at both conversation and user granularity for prompt, latency, and cost tracing
• Isolated all deterministic logic into a pure-function core with dedicated unit tests, keeping correctness verifiable independently of the LLM

Technologies

Python, Temporal, Structured Output (tool-calling), Pydantic, asyncio, Langfuse, PostgreSQL, SQLAlchemy, SHA-256 idempotency, Prompt-Injection Hardening