Populating an agent's context layer from Kafka, end to end

What we mean by the context layer

A model does not know your business. When an agent answers "can we ship this today?" or approves a refund, everything it knows about your stock, your customer and their order comes from data it is handed at that moment: retrieved documents, tool results, rows from a database. We call that the context layer.

For many questions, last night's data is fine. A policy document or a product description does not change by the minute. But some agent actions depend on state that changes all day: available stock, a customer's credit headroom, whether a payment has cleared, whether an order has already shipped. For those actions, stale context does not produce a slower answer. It produces a wrong action, and a wrong action has a direct cost.

So we design the context layer around one question: how old is this value when the agent uses it, and is that old enough to be wrong?

The reference path

The path below is the one we use most often. Each component can be swapped (MySQL for Postgres, Kafka Streams for Flink, Postgres for Redis), but the shape stays the same: capture every change, keep it in order per record, shape it into the state the agent needs, and serve it with its age attached.

Hop 1: capture every change from Postgres

We do not poll tables. Polling misses deletes, misses intermediate states and loads the database. Instead, Debezium reads Postgres's write-ahead log through logical replication: the database is set to wal_level = logical, a publication lists the tables to capture, and a replication slot remembers how far the connector has read. On first start, Debezium takes a consistent snapshot of the tables and then switches to streaming changes.

{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "orders-db.internal",
    "database.port": "5432",
    "database.user": "cdc_reader",
    "database.dbname": "orders",
    "topic.prefix": "orders",
    "table.include.list": "public.orders,public.stock_levels,public.cdc_heartbeat",
    "publication.name": "cdc_orders",
    "publication.autocreate.mode": "filtered",
    "slot.name": "cdc_orders",
    "snapshot.mode": "initial",
    "heartbeat.interval.ms": "10000",
    "heartbeat.action.query": "INSERT INTO cdc_heartbeat (id, ts) VALUES (1, now()) ON CONFLICT (id) DO UPDATE SET ts = now()"
  }
}

Three details in that configuration matter more than they look:

• The replication slot holds on to the write-ahead log until the connector confirms it. If the connector stops for a weekend, the database keeps every change since, and the disk fills. Set max_slot_wal_keep_size (Postgres 13 and later) to cap it, and alert on slot lag in bytes, not just on connector status.
• Heartbeats keep quiet databases moving. If the captured tables are quiet but the rest of the database is busy, the slot cannot advance and the log grows. The heartbeat query writes a tiny row every ten seconds, which advances the slot. As we show below, it also gives a clean way to measure end-to-end freshness.
• Large unchanged columns arrive as a placeholder. Postgres stores big values out of line (TOAST) and does not repeat them in the log when an update leaves them unchanged. Debezium then emits a placeholder instead of the value. Either set REPLICA IDENTITY FULL on those tables, or keep the last known value downstream and fill it in. Otherwise a perfectly good description silently becomes __debezium_unavailable_value in your agent's context.

Hop 2: shape the Kafka topics for state, not just events

An agent needs the latest state of a record, not a log of everything that happened to it. That changes how the topics are set up:

• Key every message by the record's primary key. Kafka only keeps order within a partition, and the key decides the partition, so all changes to SKU 4471 land in order on the same partition.
• Use log compaction (cleanup.policy=compact). Kafka then keeps at least the latest message per key, so a new consumer, or a rebuilt serving store, can replay the topic and recover current state without a fresh database snapshot.
• Treat deletes as first-class messages. Debezium follows a delete with a tombstone (a message with a key and an empty value), which tells compaction to drop the key. Keep delete.retention.ms longer than your slowest consumer could be offline, or that consumer will never learn the record was deleted.
• Decide the partition count up front. Adding partitions later changes which partition a key maps to, which breaks per-key ordering during the change.

Hop 3: turn changes into the state the agent needs

Raw table changes are rarely the answer the agent needs. "Available to promise" is stock on hand minus reserved quantity minus open orders, across two or three tables. Stream processing (Apache Flink in our reference build; Kafka Streams works for smaller cases) joins the changes and computes the derived value as each change arrives.

Two rules keep this stage honest:

• Carry the source timestamp and position through every join. Debezium includes the commit time (source.ts_ms) and the log position (source.lsn) on every message. When you derive a value from several inputs, keep the oldest commit time among them, because that is how old the derived value really is.
• Prefer safe re-processing over exactly-once claims. Pipelines get replayed: after a failure, a code fix or a backfill. Rather than relying on exactly-once delivery end to end, we make the final write idempotent and version-guarded, so writing the same change twice, or an older change after a newer one, does no harm.

Hop 4: the serving store

The agent should never query Kafka or the source database directly for this kind of state. It reads a small serving store that holds the latest state per key, built for lookups in a few milliseconds. Redis works well; Postgres works too and makes the version guard easy to read:

INSERT INTO stock_context (sku, available_to_promise, source_lsn, source_ts)
VALUES ($1, $2, $3, $4)
ON CONFLICT (sku) DO UPDATE
SET available_to_promise = EXCLUDED.available_to_promise,
    source_lsn           = EXCLUDED.source_lsn,
    source_ts            = EXCLUDED.source_ts
WHERE stock_context.source_lsn < EXCLUDED.source_lsn;  -- older changes are ignored

The WHERE clause is what makes replays safe: a change only lands if it is newer than what is already stored. Deletes need the same care. If you physically delete the row on a tombstone, a late-arriving older update can bring it back to life. We keep a short-lived deleted marker with its log position instead, and let a cleanup job remove markers after the replay window has passed.

If you use Redis, set the eviction policy to noeviction for this data. A cache that silently drops keys under memory pressure turns "we have no record of that order" into an answer the agent will repeat to a customer.

Hop 5: the tool the agent calls

The agent reaches the serving store through a tool, which we expose over the Model Context Protocol (MCP) so it works with any agent framework that supports it. The important part is not the protocol. It is what the tool returns:

{
  "sku": "4471",
  "available_to_promise": 0,
  "as_of": "2026-10-12T11:42:03Z",
  "age_seconds": 3,
  "freshness_budget_seconds": 60,
  "within_budget": true,
  "source": "orders.public.stock_levels"
}

Every value comes with when it was true and whether it is fresh enough for this use. The freshness budget belongs to the business, not to the pipeline: stock for a same-day promise might be allowed a minute, a credit limit five minutes, a product description a day. You set these with the person who owns the outcome.

Then the agent's instructions say what to do when a value is over budget: re-read through a slower path that goes to the source of truth, or tell the user it cannot confirm right now. An agent that says "I can't confirm stock at the moment" costs you a little. An agent that confidently promises stock you do not have costs you a customer.

Measuring freshness at every hop

You cannot manage what you do not measure, and most teams measure the wrong thing: whether the pipeline is running, not how old the data is when it is used. We capture a timestamp at each hop and publish the differences as metrics.

Timestamps from different machines are only as good as their clocks. That is why the heartbeat row is useful: the database writes its own now() into it every ten seconds, the row flows through the same path as the data, and at the serving end you compare the heartbeat value you hold with the database's current time. Both readings come from one clock, so the result is a trustworthy end-to-end age, even when individual hosts drift.

What breaks in production

When a context layer gives wrong answers, the cause is rarely raw speed. In order of how often we see them:

1. A stalled replication slot. The connector is "running" but not advancing, often after a permissions change or a long transaction. Data age grows quietly until someone notices a wrong answer.
2. Unplanned re-snapshots. A connector restarted with lost offsets takes a fresh snapshot, flooding Kafka and pushing real changes behind millions of old rows.
3. Consumer rebalances. Every deploy or scaling event pauses processing while partitions are reassigned. Short pauses are normal; long ones show up directly in t3 − t2.
4. Schema changes. A renamed column upstream breaks the processing job, or worse, does not break it and fills a field with nulls. Use a schema registry with a compatibility rule, and test schema changes against the pipeline before release.
5. Hot keys. One very busy record (a flagship product, a large account) concentrates load on one partition and one processing thread.

Where to start

Do not start by streaming everything. Pick one agent action with a direct cost when it is wrong, such as promising delivery, approving credit or confirming a payment. Map exactly which values that action reads, measure how old they are today when the agent uses them, agree a budget with the business owner, and build only that path. The first path takes the longest; the second reuses the connector, the topics and the measurement.