fix: dead-letter parse and handler failures; add DLQ binding

api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt

@@ -1,4 +1,3 @@
-
 package com.jorisjonkers.personalstack.knowledge.queue
 
 import org.springframework.amqp.core.Binding
@@ -11,9 +10,9 @@ import org.springframework.context.annotation.Configuration
 
 /**
  * RabbitMQ topology owned by knowledge-api's write path. The Python
- * ingest worker (Phase 5) is the eventual consumer; declaring the
- * exchange + queues here means the worker can come up to a pre-bound
- * topology rather than racing to declare it on first boot.
+ * ingest worker is the eventual consumer; declaring the exchange + queues
+ * here means the worker can come up to a pre-bound topology rather than
+ * racing to declare it on first boot.
  *
  * Routing keys (topic exchange):
  *   knowledge.lesson    — capture_lesson tool
@@ -24,6 +23,19 @@ import org.springframework.context.annotation.Configuration
  * `knowledge.*` namespace via wildcard binding. The worker can split
  * into per-type queues later if throughput or backoff semantics
  * diverge — the routing key is already on the message envelope.
+ *
+ * Dead-letter topology:
+ *   knowledge.ingest  →  x-dead-letter-exchange: knowledge.dlx
+ *                         x-dead-letter-routing-key: knowledge.ingest.dlq
+ *   knowledge.dlx (TopicExchange) →  knowledge.ingest.dlq
+ *
+ * The main queue sets `x-dead-letter-routing-key` to `knowledge.ingest.dlq`,
+ * so the DLQ binding uses that exact key. Keeping DLX as a TopicExchange
+ * avoids a PRECONDITION_FAILED error when rolling out onto a broker that
+ * already has `knowledge.dlx` declared as a topic exchange.
+ *
+ * Both parse failures (nacked by the Python worker) and handler errors
+ * route to `knowledge.ingest.dlq` for visibility and manual replay.
  */
 @Configuration
 class IngestQueueConfig {
@@ -59,4 +71,15 @@ class IngestQueueConfig {
         knowledgeIngestQueue: Queue,
         knowledgeExchange: TopicExchange,
     ): Binding = BindingBuilder.bind(knowledgeIngestQueue).to(knowledgeExchange).with("knowledge.*")
+
+    /**
+     * Bind the DLQ to the DLX using the exact routing key the main queue
+     * stamps on dead-lettered messages, so nacked deliveries are stored
+     * rather than dropped by the broker.
+     */
+    @Bean
+    fun knowledgeDlqBinding(
+        knowledgeIngestDlq: Queue,
+        knowledgeDlxExchange: TopicExchange,
+    ): Binding = BindingBuilder.bind(knowledgeIngestDlq).to(knowledgeDlxExchange).with(DLQ)
 }

ingest-worker/README.md

@@ -1,8 +1,16 @@
 # knowledge-ingest-worker
 
-Python consumer that reads `knowledge.ingest`-bound messages from RabbitMQ and (in stacked follow-ups) commits canonical notes to the `knowledge-vault` git repo + populates LightRAG chunks/entities/relations in `knowledge_db`.
+Python consumer that reads `knowledge.ingest`-bound messages from RabbitMQ, commits canonical notes to the `knowledge-vault` git repo, and writes vault pointers back to `kb_notes` in Postgres. The LightRAG chunking + embedding pipeline (Ollama embeddings, pgvector, entity/relation extraction) layers on top of that in stacked follow-ups.
 
-Today the worker ships the skeleton: connect, consume, log each delivery as a structured JSON line, ACK. The git-vault writer lands next; the LightRAG ingest pipeline (Ollama embeddings + pgvector + entity/relation extraction) layers on top of that.
+## Error handling
+
+| Failure mode | Behaviour |
+|---|---|
+| Malformed JSON / schema violation / bad encoding | `basic_nack(requeue=False)` → routed to `knowledge.ingest.dlq` via DLX |
+| Handler error (vault write, DB error) | `basic_nack(requeue=False)` → same DLQ |
+| Clean dispatch | `basic_ack` |
+
+The DLX topology (`knowledge.dlx` exchange + `knowledge.ingest.dlq` queue) is declared by the knowledge-api service on startup. The worker only consumes — it never needs to declare queues.
 
 ## Local development
 
@@ -26,3 +34,15 @@ uv run mypy                      # type check
 | `INGEST_PREFETCH`                     | `4`                                      | Bounded by Ollama / LightRAG latency, not AMQP             |
 | `LOG_LEVEL`                           | `INFO`                                   |                                                            |
 | `SERVICE_VERSION`                     | `unknown`                                | Stamped onto each log line (baked into the image at build) |
+| `VAULT_ENABLED`                       | `false`                                  | Set `true` in production; falls back to `LoggingHandler`   |
+| `VAULT_CLONE_URL`                     | `git@github.com:…/knowledge-vault.git`   | SSH URL for the vault repo                                 |
+| `VAULT_CLONE_DIR`                     | `/var/lib/knowledge-vault`               | Persistent volume mount point                              |
+| `VAULT_BRANCH`                        | `main`                                   |                                                            |
+| `VAULT_SSH_KEY_PATH`                  | `/etc/git-secrets/id_ed25519`            | Vault Agent injects deploy key in production               |
+| `VAULT_AUTHOR_NAME`                   | `knowledge-ingest-worker`                |                                                            |
+| `VAULT_AUTHOR_EMAIL`                  | `worker@knowledge.local`                 |                                                            |
+| `KB_PERSIST_ENABLED`                  | `false`                                  | Set `true` to write vault pointers back to `kb_notes`      |
+| `DB_HOST`                             | `postgres.data-system.svc.cluster.local` | k8s service DNS                                            |
+| `DB_PORT`                             | `5432`                                   |                                                            |
+| `DB_NAME`                             | `knowledge_db`                           |                                                            |
+| `DB_USER` / `DB_PASSWORD`             | `kb_user` / `kb_password`                | Vault-projected in production                              |

ingest-worker/src/knowledge_worker/__main__.py

@@ -1,10 +1,14 @@
 """Entry point: ``python -m knowledge_worker`` (or via
 ``opentelemetry-instrument``).
 
-Today the worker only ships the consumer skeleton + the
-`LoggingHandler` — each delivery emits one structured log and is
-ACKed. The real ingest flow (LightRAG, Ollama embeddings,
-knowledge-vault git commits) lands in stacked follow-ups.
+Wires the ``VaultHandler`` (when ``VAULT_ENABLED=true``) or falls
+back to ``LoggingHandler`` for local smoke runs. ``VaultHandler``
+clones the knowledge-vault git repo, writes one markdown file per
+delivery, commits + pushes, and calls back to ``kb_notes`` via
+``PostgresNoteStore`` when ``KB_PERSIST_ENABLED=true``.
+
+Parse failures and handler errors are nacked without requeue so they
+route to ``knowledge.ingest.dlq`` via the DLX declared by knowledge-api.
 """
 
 from __future__ import annotations

ingest-worker/src/knowledge_worker/consumer.py

@@ -1,10 +1,15 @@
 """Blocking pika consumer.
 
-Acks on a clean `handler.handle` return. Reraises everything else
-so pika's default behaviour (`requeue=True`) nacks the delivery and
-returns it to the broker. The knowledge-api side already declares
-the DLX, so terminal failures eventually land on
-`knowledge.ingest.dlq` after the broker's redelivery count trips.
+Acks on a clean ``handler.handle`` return.
+
+Parse failures (malformed JSON, schema violations, bad encoding) are
+immediately dead-lettered: ``basic_nack(requeue=False)`` routes the
+delivery through the DLX that knowledge-api declares on the ingest
+queue so the poison payload lands on ``knowledge.ingest.dlq`` and
+stays recoverable.
+
+Handler failures (vault write errors, DB errors) are also nacked
+without requeue so they flow to the same DLQ rather than vanishing.
 """
 
 from __future__ import annotations
@@ -35,7 +40,7 @@ class _Delivery:
 
 
 class Consumer:
-    """Wraps a pika `BlockingConnection` + per-delivery dispatch."""
+    """Wraps a pika ``BlockingConnection`` + per-delivery dispatch."""
 
     def __init__(
         self,
@@ -106,18 +111,15 @@ def _on_message(
         try:
             note = self._parse(delivery)
         except (ValidationError, json.JSONDecodeError, UnicodeDecodeError) as exc:
-            # Bad payloads can't get fixed by retrying — ack so the
-            # broker doesn't redeliver the same broken message
-            # forever. A future change can route these to the
-            # dead-letter queue instead by switching to nack +
-            # requeue=false; today the broker's redelivery limit
-            # handles that path well enough.
+            # Poison payloads cannot be fixed by retrying. Dead-letter
+            # them immediately so they land on knowledge.ingest.dlq via
+            # the DLX declared by knowledge-api on this queue.
             self._log.error(
                 "consumer.parse_failed",
                 routing_key=delivery.routing_key,
                 error=str(exc),
             )
-            channel.basic_ack(delivery_tag=method.delivery_tag)
+            channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
             return
 
         try:
@@ -140,6 +142,6 @@ def _parse(delivery: _Delivery) -> CapturedNote:
 
 
 def silence_pika_warning_logs() -> None:
-    """pika emits a noisy WARNING per `basic_qos` round-trip. Demote to INFO."""
+    """pika emits a noisy WARNING per ``basic_qos`` round-trip. Demote to INFO."""
 
     logging.getLogger("pika").setLevel(logging.INFO)

ingest-worker/tests/integration/test_rabbit_consumer.py

@@ -1,12 +1,17 @@
-"""End-to-end smoke test: publish a `CapturedNote` payload onto the
+"""End-to-end smoke test: publish a ``CapturedNote`` payload onto the
 same topic exchange the knowledge-api side declares, then verify the
 worker consumes + dispatches to its handler.
 
 Spins up RabbitMQ via Testcontainers (Docker required). Declares the
-`knowledge` topic exchange + `knowledge.ingest` queue on the test's
+``knowledge`` topic exchange + ``knowledge.ingest`` queue on the test's
 own connection rather than relying on knowledge-api to do it — keeps
 the worker's start-up free of declaration assumptions, which is the
 production posture too.
+
+DLX topology mirrors ``IngestQueueConfig`` on the knowledge-api side:
+  knowledge.ingest  →  x-dead-letter-exchange: knowledge.dlx
+                        x-dead-letter-routing-key: knowledge.ingest.dlq
+  knowledge.dlx     →  knowledge.ingest.dlq (queue, durable)
 """
 
 from __future__ import annotations
@@ -30,6 +35,9 @@
 
 pytestmark = pytest.mark.integration
 
+DLX = "knowledge.dlx"
+DLQ = "knowledge.ingest.dlq"
+
 
 @pytest.fixture(scope="module")
 def rabbit() -> Iterator[dict[str, object]]:
@@ -60,6 +68,7 @@ def _settings(rabbit: dict[str, object]) -> Settings:
 
 
 def _declare_topology(settings: Settings) -> pika.BlockingConnection:
+    """Declare the full DLX topology that knowledge-api owns in production."""
     conn = pika.BlockingConnection(
         pika.ConnectionParameters(
             host=settings.rabbitmq_host,
@@ -71,8 +80,22 @@ def _declare_topology(settings: Settings) -> pika.BlockingConnection:
         )
     )
     ch = conn.channel()
+    # Main exchange
     ch.exchange_declare(exchange="knowledge", exchange_type="topic", durable=True)
-    ch.queue_declare(queue=settings.queue, durable=True)
+    # Dead-letter exchange — topic to match the production IngestQueueConfig declaration
+    ch.exchange_declare(exchange=DLX, exchange_type="topic", durable=True)
+    # DLQ — must be declared before the main queue so the DLX target exists
+    ch.queue_declare(queue=DLQ, durable=True)
+    ch.queue_bind(queue=DLQ, exchange=DLX, routing_key=DLQ)
+    # Main ingest queue wired to the DLX
+    ch.queue_declare(
+        queue=settings.queue,
+        durable=True,
+        arguments={
+            "x-dead-letter-exchange": DLX,
+            "x-dead-letter-routing-key": DLQ,
+        },
+    )
     ch.queue_bind(queue=settings.queue, exchange="knowledge", routing_key="knowledge.*")
     return conn
 
@@ -100,6 +123,22 @@ def _publish(
     conn.close()
 
 
+def _publish_raw(settings: Settings, routing_key: str, body: bytes) -> None:
+    """Publish a raw (potentially malformed) body without JSON encoding."""
+    conn = pika.BlockingConnection(
+        pika.ConnectionParameters(
+            host=settings.rabbitmq_host,
+            port=settings.rabbitmq_port,
+            credentials=pika.PlainCredentials(
+                settings.rabbitmq_user, settings.rabbitmq_password
+            ),
+        )
+    )
+    ch = conn.channel()
+    ch.basic_publish(exchange="knowledge", routing_key=routing_key, body=body)
+    conn.close()
+
+
 def _await_delivery(handler: RecordingHandler, expected: int, timeout_s: float = 10.0) -> None:
     deadline = time.monotonic() + timeout_s
     while time.monotonic() < deadline:
@@ -111,6 +150,32 @@ def _await_delivery(handler: RecordingHandler, expected: int, timeout_s: float =
     )
 
 
+def _await_dlq_message(
+    settings: Settings, timeout_s: float = 10.0
+) -> bytes | None:
+    """Poll the DLQ queue and return the body of the first message found."""
+    conn = pika.BlockingConnection(
+        pika.ConnectionParameters(
+            host=settings.rabbitmq_host,
+            port=settings.rabbitmq_port,
+            credentials=pika.PlainCredentials(
+                settings.rabbitmq_user, settings.rabbitmq_password
+            ),
+        )
+    )
+    ch = conn.channel()
+    deadline = time.monotonic() + timeout_s
+    try:
+        while time.monotonic() < deadline:
+            method, _, body = ch.basic_get(queue=DLQ, auto_ack=True)
+            if method is not None:
+                return body
+            time.sleep(0.05)
+        return None
+    finally:
+        conn.close()
+
+
 def _payload(routing_id: str) -> dict[str, object]:
     return {
         "id": routing_id,
@@ -171,3 +236,26 @@ def test_handles_three_messages_in_order(rabbit: dict[str, object]) -> None:
     finally:
         consumer.stop()
         runner.join(timeout=5)
+
+
+def test_malformed_payload_routes_to_dlq(rabbit: dict[str, object]) -> None:
+    """A poison payload must land on the DLQ, not be silently dropped."""
+    settings = _settings(rabbit)
+    _declare_topology(settings).close()
+
+    handler = RecordingHandler()
+    consumer = Consumer(settings, handler)
+    consumer.start()
+    runner = threading.Thread(target=consumer.run_forever, daemon=True)
+    runner.start()
+
+    try:
+        _publish_raw(settings, "knowledge.lesson", b"{not valid json")
+        dlq_body = _await_dlq_message(settings)
+        assert dlq_body == b"{not valid json", (
+            "malformed payload was not routed to the DLQ"
+        )
+        assert handler.deliveries == [], "handler must not have been called for a poison payload"
+    finally:
+        consumer.stop()
+        runner.join(timeout=5)

ingest-worker/tests/unit/test_consumer.py

@@ -75,19 +75,21 @@ def test_valid_message_is_acked_and_handed_to_handler() -> None:
     assert isinstance(note, CapturedNote)
 
 
-def test_invalid_payload_is_acked_to_avoid_redelivery_loops() -> None:
+def test_invalid_payload_is_nacked_to_dlq() -> None:
+    """Malformed JSON must be dead-lettered, not silently ACK'd."""
     handler = RecordingHandler()
     consumer = Consumer(_settings(), handler)
     channel = _FakeChannel()
 
     consumer._on_message(channel, _FakeMethod(), object(), b"{not valid json")  # type: ignore[arg-type]
 
-    assert channel.acks == [1]
-    assert channel.nacks == []
+    assert channel.acks == []
+    assert channel.nacks == [(1, False)]
     assert handler.deliveries == []
 
 
-def test_payload_missing_required_field_is_acked_and_skipped() -> None:
+def test_payload_missing_required_field_is_nacked_to_dlq() -> None:
+    """Schema violations must be dead-lettered, not silently ACK'd."""
     handler = RecordingHandler()
     consumer = Consumer(_settings(), handler)
     channel = _FakeChannel()
@@ -97,7 +99,8 @@ def test_payload_missing_required_field_is_acked_and_skipped() -> None:
     body = json.dumps(payload).encode()
     consumer._on_message(channel, _FakeMethod(), object(), body)  # type: ignore[arg-type]
 
-    assert channel.acks == [1]
+    assert channel.acks == []
+    assert channel.nacks == [(1, False)]
     assert handler.deliveries == []