poimen/tasks/T10.7-kafka-topics-and-inbox-relay.md

T10.7 — Kafka topics and inbox relay

Field	Value
Phase	P10 — Orchestration
Size	L — over 3 days
Status	Not started
Flags	—
Spec	inlined below
Blocks	T10.8

Goal

poimen.<tenant>.events (outbound, existing outbox unchanged) and poimen.<tenant>.dispatch (inbound, new) — an inbox relay symmetric to T7.4's outbox, turning dispatch messages into spawn calls through the same admission path, never a raw kernel write.

Facts (inlined — no spec read needed)

• Outbound is unchanged: T7.4's relay ships goal-tagged WorkEvents exactly as it ships every other event. No new producer logic on this side.
• Inbound is genuinely new: the orchestrator (T10.8) publishes to .dispatch after a Reconciler decision; this task builds only the relay that drains it, not the orchestrator itself.
• The inbox relay's failure-domain rule mirrors T7.4's exactly: own process/task, own retry, own supervision, not co-located with the executor's lifecycle.
• Every dispatch message becomes a call to the same POST /v1/runs (T10.4) the HTTP API exposes — the relay is an HTTP client to the local API surface, not a kernel-internal spawn path. This is what makes the "no bypass" guarantee mechanical, not a code-review convention.
• Redelivery on .dispatch is handled entirely by T10.3's dispatch_key admission dedup — the relay passes dispatch_key through unchanged, no separate idempotency layer.
• Partition key on both topics: (TenantId, RunId) per T7.5, consistent with the outbox.

Steps

1. Declare topics poimen.<tenant>.events, poimen.<tenant>.dispatch — partition key (TenantId, RunId) on both (T7.5).
2. No change to T7.4's relay beyond including goal-tagged records (already covered since GoalId is a field on WorkEvent, T10.1).
3. Build the inbox relay: consume .dispatch, for each message call POST /v1/runs with {workflow_ref, input, goal_id, dispatch_key} verbatim.
4. Own retry/backoff on relay-to-API-call failure; never crash the relay on a single bad message — dead-letter it, keep draining.
5. Own failure domain: separate process/task, no shared lifecycle with the executor or the HTTP API service.
6. Emit inbox lag as a metric, alongside T7.4's existing outbox-lag metric.
7. Test with the broker down for a stretch, and with forced redelivery of the same dispatch message.

Acceptance

• A message published to .dispatch results in exactly one run, even under forced redelivery.
• Broker down for the inbox side does not affect runs already in flight; queued dispatch messages are processed once the broker returns.

Verify

Harness: same broker-down/broker-up harness as T7.4, plus a second consumer/producer pair for the .dispatch direction.

Integration test — tests/it_inbox_relay.rs:

1. Publish one message to .dispatch; assert exactly one run created via the same POST /v1/runs path — instrument the API, assert the call, not a direct kernel spawn.
2. Force redelivery of the same message; assert still exactly one run (T10.3's admission dedup catches it) — no relay-side special-casing needed.
3. Take the broker down; publish a message; bring the broker up; assert the run is eventually created, no message lost.
4. Publish a malformed message; assert it's dead-lettered, relay keeps draining subsequent valid messages.
5. Assert the relay never calls kernel spawn directly — instrument the kernel's internal admission entrypoint, assert all calls originate from the HTTP layer, none from relay threads.
6. Assert inbox lag is emitted as a metric.

Command: cargo test -p distribution inbox_relay -- --test-threads=1

False pass:

• Step 2 without step 1's exact-call-path instrumentation: a relay that spawns directly and also happens to dedupe correctly would still pass a coarser "one run" check while violating the no-bypass rule.
• Skipping the malformed-message case: one bad message halting the whole relay is a silent single point of failure discovered only in production.

Traps

• Relay calling an in-process spawn function "for performance" instead of going through POST /v1/runs — same trap T7.4 names for the outbound side, mirrored here.
• Building custom dedup logic in the relay instead of trusting T10.3's admission-layer dedup. Two idempotency layers can disagree under partial failure.

---

Background (not required to do this task): rust-agentic-sys.md §8.3, §9.2, §9.3, §13.3 · T7.4-outbox-relay.md · T7.5-partition-keys-on-adapters.md · T10.3-idempotent-run-admission.md