Skip to content
← Back to Blog
observabilitytracingpythonreliabilityworkflows

Tracing an Extraction Pipeline Like a Ledger: Trace Nodes, DLQ Boundaries, and Replayable Failures

Daniel Anthony Romitelli Jr. · March 12, 2026

I hit the same failure pattern twice: the extraction result was "wrong," but nothing was actually broken.

The model had taken a different branch.

That's the kind of bug that makes you mistrust everything: your cache, your prompts, even your own memory, because the system looks nondeterministic when it's really just unobserved. So I built a tracer that turns each extraction run into a ledger. Every decision becomes a node. Every retry becomes an edge. And every "we chose variant B" becomes a fact I can point at.

Key insight: extraction is a workflow, so trace it like one

The move that mattered was giving up on "email extraction" as a single operation.

In this codebase it isn't one. Extraction runs as a workflow with multiple steps and multiple tools in play. The platform's own engineering record calls it a LangGraph 3-node workflow (Extract → Research → Validate), with GPT-5 structured extraction, Azure Document Intelligence for email parsing, and SignalR real-time streaming among the pieces.

With that many parts moving at once, logging "start" and "finish" plus a blob of output fails you in three ways:

  1. Branching is invisible. If you have prompt variants, fallbacks, or validation branches, you can't tell which path ran.
  2. Retries look like "it worked eventually." Without a trace, a retry storm is just a slow request.
  3. DLQ boundaries erase causality. Once a job is routed away, you lose the story unless you persist it.

So the tracer is built around trace nodes. A node is one append-only record of what happened at one step, carrying enough data to replay the run.

Distributed tracing works on the same idea, where each span carries the story of a request through a service boundary. The difference is that I'm applying it inside a single pipeline rather than across microservices, which gets me the same causal chain without the operational overhead of a full collector.

The tracer's schema: trace nodes as first-class records

The schema stays small. These nodes are shaped for debugging, not for dashboards.

A trace node has to answer:

  • Where am I in the workflow? (step/node name)
  • What did I feed in? (inputs)
  • What did I get back? (outputs)
  • What did I decide? (branch selection)
  • How expensive was it? (tokens/cost if available)
  • Is this retryable? (attempt count, error)

The codebase explicitly contains an Extraction Workflow Tracer module described as:

"Detailed step-by-step logging for email extraction. Provides complete visibility into the extraction workflow."

That description is the contract: the tracer is not "some logs," it's complete visibility.

A concrete, runnable schema definition

I model a trace node as a dataclass so it's explicit and serializable.

from __future__ import annotations

import json
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Any, Dict, Optional


@dataclass
class TraceNode:
    """A single step in an extraction run, recorded as an append-only event."""

    trace_id: str
    node_name: str
    timestamp: str

    # What we fed into this step
    inputs: Dict[str, Any]

    # What came back out
    outputs: Dict[str, Any]

    # Branching / decisions
    branch_chosen: Optional[str] = None

    # Retry / failure metadata
    attempt: int = 1
    error: Optional[str] = None

    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)


def new_trace_node(*, trace_id: str, node_name: str, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> TraceNode:
    return TraceNode(
        trace_id=trace_id,
        node_name=node_name,
        timestamp=datetime.utcnow().isoformat(),
        inputs=inputs,
        outputs=outputs,
    )

What surprised me once I started doing this is how fast "I think the model did X" becomes "node extract.salary chose branch prompt_variant_b on attempt 2." Completely different kind of conversation.

Where the tracing hooks sit in the extraction pipeline

I hook tracing at the boundaries where information is most likely to disappear:

  • right before/after the extraction call
  • right before/after the research step
  • right before/after validation
  • on every retry attempt
  • at the moment a message is routed to a DLQ

This codebase gives strong hints about where those boundaries are:

  • The workflow is managed in the langgraph manager module (called out as the LangGraph workflow file).
  • There's a dedicated extraction workflow tracer module.
  • There's an embedding generator module job that explicitly mentions retry logic and DLQ routing, with content length validation and dimension mismatch detection.

Embeddings and extraction are different pipelines, sure. But the retry/DLQ pattern is the same: if you don't trace at the "route to DLQ" boundary, you lose the evidence you need. (For how Dead-Letter Queues behave and why explicitly recording DLQ routing is important for reconstructing message history, see Azure Service Bus dead-letter documentation.)

A minimal, runnable "hook" pattern

Create a node, emit it, then proceed. That's the whole pattern.

import asyncio
from typing import Any, Dict


class TraceSink:
    """A tiny interface: append-only trace storage."""

    async def emit(self, node: TraceNode) -> None:
        raise NotImplementedError


class InMemoryTraceSink(TraceSink):
    def __init__(self) -> None:
        self.nodes = []

    async def emit(self, node: TraceNode) -> None:
        self.nodes.append(node)


async def traced_step(
    *,
    sink: TraceSink,
    trace_id: str,
    node_name: str,
    inputs: Dict[str, Any],
    run
) -> Dict[str, Any]:
    """Run a step and emit a trace node with inputs/outputs."""

    try:
        outputs = await run(inputs)
        node = new_trace_node(trace_id=trace_id, node_name=node_name, inputs=inputs, outputs=outputs)
        await sink.emit(node)
        return outputs
    except Exception as e:
        node = new_trace_node(trace_id=trace_id, node_name=node_name, inputs=inputs, outputs={"ok": False})
        node.error = str(e)
        await sink.emit(node)
        raise

The point of making it a wrapper is that you can't forget to do it, and the error path comes out just as structured as the success path.

Example: tracing salary extraction with prompt variant A vs. B

Salary is one of the fields recruiting workflows care about most, and one of the easiest to get subtly wrong.

The reason is that it shows up in multiple shapes: ranges, OTE, base-only, "DOE". Emails often bury it in forwarded text or meeting notes. The engineering record also explicitly mentions that the system "handles forwarded emails," which is exactly the kind of input that creates extraction ambiguity.

So here's what I trace for a salary extraction step:

  • the raw email content (or a reference to it)
  • the prompt variant used
  • the model response (structured)
  • post-processing output
  • which branch was chosen

The example below produces two trace nodes, one for variant A and one for variant B, and then records which one was selected.

import asyncio
from typing import Dict, Any


async def fake_model_call(*, prompt_variant: str, email_text: str) -> Dict[str, Any]:
    """Stand-in for the structured extraction call."""

    # This function is intentionally simple: the real system uses structured extraction.
    if prompt_variant == "A":
        return {"salary": "$120k-$150k", "raw": "Base $120k-$150k depending on experience"}
    return {"salary": "$150k OTE", "raw": "$150k OTE (base + bonus)"}


def post_process_salary(model_payload: Dict[str, Any]) -> Dict[str, Any]:
    """Stand-in for post-processing tokens/normalization."""

    return {
        "salary": model_payload.get("salary"),
        "post_processing_tokens": ["salary"],
    }


async def trace_salary_extraction() -> None:
    sink = InMemoryTraceSink()
    trace_id = "trace_salary_001"

    email_text = "Comp: Base $120k-$150k depending on experience."

    # Variant A
    model_a = await traced_step(
        sink=sink,
        trace_id=trace_id,
        node_name="extract.salary.model.variantA",
        inputs={"prompt_variant": "A", "email_text": email_text},
        run=lambda inputs: fake_model_call(prompt_variant=inputs["prompt_variant"], email_text=inputs["email_text"]),
    )
    post_a = await traced_step(
        sink=sink,
        trace_id=trace_id,
        node_name="extract.salary.postprocess.variantA",
        inputs={"model_response": model_a},
        run=lambda inputs: asyncio.sleep(0, result=post_process_salary(inputs["model_response"])),
    )

    # Variant B
    model_b = await traced_step(
        sink=sink,
        trace_id=trace_id,
        node_name="extract.salary.model.variantB",
        inputs={"prompt_variant": "B", "email_text": email_text},
        run=lambda inputs: fake_model_call(prompt_variant=inputs["prompt_variant"], email_text=inputs["email_text"]),
    )
    post_b = await traced_step(
        sink=sink,
        trace_id=trace_id,
        node_name="extract.salary.postprocess.variantB",
        inputs={"model_response": model_b},
        run=lambda inputs: asyncio.sleep(0, result=post_process_salary(inputs["model_response"])),
    )

    # Branch selection (example: choose A if it looks like a range)
    chosen = "A" if isinstance(post_a.get("salary"), str) and "-" in post_a["salary"] else "B"
    node = new_trace_node(
        trace_id=trace_id,
        node_name="extract.salary.branch",
        inputs={"variantA": post_a, "variantB": post_b},
        outputs={"chosen": chosen},
    )
    node.branch_chosen = chosen
    await sink.emit(node)

    for n in sink.nodes:
        print(n.to_json())


if __name__ == "__main__":
    asyncio.run(trace_salary_extraction())

I like this example because it puts on the record the exact thing that used to be invisible: I can see both model responses, I can see the post-processing tokens, and I can see the branch selection recorded as its own node.

How traces are stored and indexed

A trace is only useful if it survives the run.

This platform already has multiple persistence layers in play:

  • Redis cache is referenced directly in multiple places (for example, the SubjectLineBandit uses well_shared.cache.redis_manager.get_cache_manager() and loads variants "from Redis").
  • There's a Service Bus integration in the Azure service manager (the service manager has an _init_service_bus method).
  • The system uses PostgreSQL + pgvector (called out in CLAUDE.md), and other parts of the platform store operational data in tables like ops_error_stream (from the Error Capture Middleware description).

The tracer's job is to write a trace record in a shape those layers can take. The exact storage backend can vary by environment; the important part is the schema stays stable.

Architecture flow (storage + indexing)

The key design choice here is that the tracer emits to an abstraction (TraceSink) so I can send the same nodes to Redis for fast lookups, to ServiceBus for asynchronous processing, and to Postgres for durable querying.

DLQ and retry interactions: making “eventually succeeded” debuggable

Retries are where systems lie.

Not maliciously, just structurally. If you don’t record attempts, the final success overwrites the story. That’s why the Embedding Generator job’s description matters: it explicitly calls out retry logic and DLQ routing plus validation failures like content length validation and dimension mismatch detection.

Those are exactly the kinds of errors that should appear as trace nodes:

  • attempt 1 failed: dimension mismatch
  • attempt 2 failed: content length
  • attempt 3 succeeded
  • routed to DLQ (if it never succeeded)

The tracer schema above includes attempt and error for this reason.

A runnable retry + DLQ trace pattern

import asyncio
from typing import Any, Dict, Callable


class DeadLetterQueue:
    def __init__(self) -> None:
        self.messages = []

    async def send(self, payload: Dict[str, Any]) -> None:
        self.messages.append(payload)


async def run_with_retry_and_dlq(
    *,
    sink: TraceSink,
    dlq: DeadLetterQueue,
    trace_id: str,
    node_name: str,
    inputs: Dict[str, Any],
    max_attempts: int,
    run: Callable[[Dict[str, Any]], Any],
) -> Any:
    for attempt in range(1, max_attempts + 1):
        try:
            out = await run(inputs)
            node = new_trace_node(trace_id=trace_id, node_name=node_name, inputs=inputs, outputs={"ok": True, "result": out})
            node.attempt = attempt
            await sink.emit(node)
            return out
        except Exception as e:
            node = new_trace_node(trace_id=trace_id, node_name=node_name, inputs=inputs, outputs={"ok": False})
            node.attempt = attempt
            node.error = str(e)
            await sink.emit(node)

    dlq_payload = {"trace_id": trace_id, "node_name": node_name, "inputs": inputs}
    await dlq.send(dlq_payload)

    dlq_node = new_trace_node(
        trace_id=trace_id,
        node_name=f"{node_name}.dlq",
        inputs=inputs,
        outputs={"routed": True},
    )
    await sink.emit(dlq_node)

    raise RuntimeError("Exceeded retry attempts; routed to DLQ")

The non-obvious detail is that I trace the DLQ routing as a node too. If you only enqueue a DLQ message, you’ve still lost the narrative unless you can join it back to the original run by trace_id.

Recording the DLQ event as its own trace node is what preserves causality. Without it, the dead-lettered message sits in a separate queue with no backward pointer to the run that produced it, and the trace_id is the only thing that stitches the story back together.

Replayability: feeding a trace back into a local replay rig

Once you have a trace, you can do something that feels like cheating: you can reproduce the run without re-running the whole world.

Replayability is a direct consequence of two choices:

  1. I log inputs/outputs full per-node records, never summaries alone.
  2. I treat nodes as append-only facts.

When an extraction run fails in production, I don’t want to re-fetch attachments, re-run parsing, or depend on external services just to debug a single prompt variant. I want to take the trace and replay it locally.

A runnable replay script

This script replays nodes by name. In a real system, the handlers would call the actual step functions; here it prints what it would do.

import json
from typing import List, Dict, Any, Callable


def replay_trace(nodes_jsonl: str, handlers: Dict[str, Callable[[Dict[str, Any]], Any]]) -> None:
    """Replay trace nodes from JSONL (one JSON per line)."""

    for line in nodes_jsonl.strip().splitlines():
        node = json.loads(line)
        name = node["node_name"]
        inputs = node.get("inputs", {})

        handler = handlers.get(name)
        if handler is None:
            print(f"[replay] no handler for {name}; skipping")
            continue

        print(f"[replay] running {name}")
        result = handler(inputs)
        print(f"[replay] result: {result}")


if __name__ == "__main__":
    sample = """
{"trace_id":"t1","node_name":"extract.salary.branch","timestamp":"2026-03-12T00:00:00","inputs":{"variantA":{"salary":"$120k-$150k"},"variantB":{"salary":"$150k OTE"}},"outputs":{"chosen":"A"},"branch_chosen":"A","attempt":1,"error":null}
""".strip()

    replay_trace(
        sample,
        handlers={
            "extract.salary.branch": lambda inputs: {"chosen": "A" if "-" in inputs["variantA"]["salary"] else "B"}
        },
    )

The first time I replayed a trace like this, the "nondeterministic" bug resolved into a deterministic mismatch between two branches. The tracer paid for itself right there.

Nuances: what a deterministic record changes downstream

I'm careful with the word "deterministic" here.

I'm not claiming the model is deterministic. I'm claiming the record of what happened is deterministic: the trace is an immutable timeline of inputs, outputs, and decisions.

That property matters because three safety mechanisms that are otherwise squishy get something solid to stand on.

Conformal gating

This codebase already has a strong theme of quality gating and budget-aware processing. For example, a proprietary optimization algorithm system describes telemetry events like a proprietary optimization algorithm.quality_score and a proprietary optimization algorithm.processing_cost, and it includes budget warnings at 80% utilization. A deterministic trace lets you associate a gate decision with the exact node outputs that triggered it.

If a gate flips, I can see the evidence.

Regression detection

Change a prompt and the regression often arrives as "the output looks different." Without traces, you can't tell whether the input changed, the prompt changed, or the branch changed.

With traces you can diff node by node. Variant A response changed. Post-processing tokens changed. Validation branch changed.

Safer cache invalidation

This platform uses caching in multiple places (Redis is a recurring component, and the Embedding Agent explicitly implements Redis caching with a 24hr TTL). Invalidation gets safer when you can tie a cached artifact to the trace node that produced it.

If a prompt variant changes, I can invalidate only the nodes that depended on that variant, instead of flushing everything and hoping. Concretely: each cache key includes the node name and a hash of the inputs that produced it. When variant B replaces variant A, I walk the trace graph to find every downstream node whose inputs included variant A's output (post-processing, validation, branch selection) and expire exactly those keys. Everything else survives. That turns cache invalidation from a prayer ("flush it all and re-warm") into a targeted operation with a clear blast radius, which matters a lot when you're running extraction at scale and a full cache miss means re-calling the model for every in-flight job.

Closing

Once extraction runs became a ledger of trace nodes, with retries and DLQ boundaries and branch decisions all written down, the system stopped feeling like a black box and started feeling like a machine I could actually tune. Better logs weren't the real win. The win was watching "I think it chose the other prompt" turn into a node I could point at and replay until it behaved.