--- name: ak-dev-new-queue-transport description: > Step-by-step guide for adding a new queue transport to Agent Kernel's execution pipeline. Use this skill when you need to integrate a new message broker (beyond in_memory, SQS, Kafka, and NATS JetStream) behind the QueueTransport/TransportConsumer interface. Covers the queue-semantics contract every transport must reproduce, factory registration, configuration and extras, the QueueTransportContract test suite (fake and live-broker runs), the transport example, and Helm chart wiring. license: Apache-2.0 metadata: author: yaalalabs category: developer --- # Adding a New Queue Transport This guide walks through adding a new queue transport to the pipeline (`ak-py/src/agentkernel/pipeline/`). Use the shipped implementations as references, in increasing order of complexity: - `transport/in_memory.py`: the semantics in their purest form, no broker - `transport/sqs.py`: a broker with native FIFO groups, visibility timeout, and dedup - `transport/nats.py`: a broker where per-session ordering is built client-side (partitioned subjects, one durable consumer per partition, `max_ack_pending=1`) - `transport/kafka.py` + `transport/bookkeeping.py`: a broker with no per-message acknowledgement model, so receive counts and dedup are rebuilt on a bookkeeping store Read `.agents/skills/ak-dev-architecture` (the pipeline section) first if you have not. ## The Semantics Contract Every transport must reproduce the SQS FIFO semantics the pipeline was extracted from (`docs/specs/495-onprem-kubernetes/research/current-queue-mode.md`): 1. **Per-group FIFO with one in-flight message per group**: `group_id` is the session id; a session's turns never run concurrently or out of order, while distinct sessions run in parallel. 2. **Bounded at-least-once redelivery with an exact `receive_count`**: the `ConsumerLoop` compares it to `max_receive_count` to fire the permanent-failure hook, so it must be exact, not approximate. 3. **Publish-time deduplication** on `dedup_id` within a window (SQS parity: 5 minutes). 4. **Attribute round-tripping**: `QueueMessage.attributes` (request id, user id, status code) must survive the trip byte-identically. 5. **Batch fetch with a bounded wait**, returning fewer than `batch_size` rather than blocking past the wait. 6. **Queue isolation**: INPUT and OUTPUT never see each other's messages. Where the broker genuinely cannot provide a guarantee, the contract suite has an explicit, documented opt-out (see `timeout_redelivery` in `pipeline/testing.py`, which Kafka sets to False because its consumer model has no visibility timeout). Never fake a guarantee; declare its absence and justify it in the subclass. ## Step 1: Implement the Transport Create `ak-py/src/agentkernel/pipeline/transport/.py` implementing both ABCs from `transport/base.py`: - `QueueTransport`: `send(queue, message)` (map `QueueMessage` onto the broker's record: body, attributes as headers/metadata, `group_id` as the ordering key, `dedup_id` as the dedup token), `create_consumer(queue)`, and optionally `check_consumer_capacity(queue, n)` (startup warning when consumer threads exceed what the broker can serve in parallel). - `TransportConsumer`: `fetch(batch_size, wait_seconds)`, `ack`, `nack`, `dead_letter`, `close()`. **One consumer instance is created per consumer thread** (Kafka needs one client object per thread; the design assumes it everywhere), so instance state needs no locking, but anything class-level does. Rules learned from the shipped transports: - Threads, not asyncio: the pipeline's consumers are threads. If the client library is asyncio-only, bridge through one shared event-loop thread (`_NatsLoop` in `nats.py` is the maintainer-recommended pattern; do not spawn a loop per thread). - `receive_count` must be exact. Prefer the broker's own counter (`num_delivered`, `ApproximateReceiveCount`); if none exists, count attempts in a `BookkeepingStore` (`transport/bookkeeping.py`), keyed so a crash-looping poison message cannot reset itself. - Honor `fetch_wait_slice_seconds` semantics: `ConsumerLoop` slices waits to stay responsive to shutdown, so a fetch must tolerate short waits without spinning. - `close()` must actually release broker resources (consumer-group membership, subscriptions, background threads). A leaked consumer keeps CI jobs alive after the tests pass. - Connection/provisioning caches are class-level and keyed by connection target; provide a `reset()` classmethod for test isolation (see `InMemoryTransport.reset`, `NatsTransport.reset`). - Provisioning posture: dev may auto-provision broker objects behind an `auto_provision` flag, but production fails fast with an `AKConfigError` naming the missing object and the declarative alternative (NACK CRs, Strimzi topics). Agent Kernel never silently creates production infrastructure. ## Step 2: Configuration In `ak-py/src/agentkernel/core/config.py`: - Add a `_QueueConfig` model with the broker's connection and tuning fields (mirror `_NatsQueueConfig`; every field needs a real description, since they become user docs). - Add the optional field to `_QueuesConfig` and the type name to its `type` description. - Keep `input`/`output` blocks backend-neutral: `max_receive_count`, `no_of_consumers`, and `batch_size` are shared knobs, never per-backend. If the client library is heavy or compiled, add an extra in `ak-py/pyproject.toml` (`[project.optional-dependencies]`) named after the transport. ## Step 3: Factory Registration `QueueTransportFactory.create()` in `transport/base.py` is an explicit chain: add the branch for your type, guarded by `require_extra("", "execution.queues.type: ")` with the import inside, and add the name to `_BUILTIN_TYPES`. Fail with `AKConfigError` when the config block is missing. Anything not in `_BUILTIN_TYPES` resolves as a dotted path (BYO), so a transport can also live out of tree; built-in status is for transports we test and document. The factory has a second consumer (#503): the **sandbox queue broker** passes its own `_QueuesConfig`-shaped `sandbox.broker.queue` block through the optional `queues_config` parameter on `resolve_type`/`create`/`create_consumer`, so a new transport gets sandbox-broker support for free. Read the block handed to you, never `AKConfig` (the no-argument path keeps reading `execution.queues` and must stay byte-for-byte unchanged; `tests/test_pipeline_factory_seams.py` enforces both properties). ## Step 4: Tests Three layers, all required: 1. **Transport-specific unit tests** (`ak-py/tests/test_pipeline__transport.py`): envelope/header mapping, orderings, error paths, provisioning create-vs-verify, against a fake broker. Build the fake behind the real client's interface so the transport code is exercised unmodified (see the fake JetStream behind the real `_NatsLoop`, and the fake in-memory Kafka cluster). 2. **The contract suite, in-repo**: subclass `QueueTransportContract` (`pipeline/testing.py`) against the fake, implementing `make_transport()`. Tune `ack_wait`/`fetch_wait`/`force_redelivery` per backend; document every capability opt-out. 3. **The contract suite, live** (`ak-py/tests/test_transport_contract_live.py`): add an env-gated subclass pointing at a real broker (`AK_TEST__...` env var, skipped when unset) with per-test unique queues/streams/topics for isolation. The `transport-integration-tests` job in `.github/workflows/test-reusable.yaml` starts the brokers from the transport examples' compose files and runs this file on every PR: add your broker's compose service there. Timing traps that only live brokers catch (both found on real servers, invisible on fakes): - If a fetch holds a pull/poll request open per partition, the per-partition window (`fetch_wait / partitions`) must stay **below** the visibility timeout, or the server redelivers an in-flight message into the still-open request and one fetch returns it twice. - The contract's fixed group ids (`s0`/`s1`/`s2`) must land on distinct partitions under the broker's real partitioner. Partitioners are deterministic: compute the mapping (crc32 for the client-side scheme, murmur2 for Kafka) and choose the partition count accordingly instead of hoping. ## Step 5: Example Add `examples/transport//`: a two-process app (`IOHandler.run()` / `AgentRunner.run()` behind one `app.py`), a `config.yaml` with commented tuning values, a docker compose stack with a healthcheck (the CI job relies on `up -d --wait `), a `_tester.py` harness (bring the stack up, provision what Agent Kernel deliberately does not, inspect queues), and an `app_test.py` covering rest_sync, a multi-turn session, and the retry-to-permanent-failure path. Register it in `.github/test-config.yaml` under the containerized e2e tests. ## Step 6: Deployment and Docs Surfaces - Helm chart (`ak-deployment/ak-k8s/chart/`): a `transport.` values block, its `AK_EXECUTION__QUEUES____*` env injection in `configmap-env.yaml`, a KEDA trigger in `scaledobject.yaml` if a scaler exists, and declarative provisioning CRs if the broker has an operator. - Docs: the transport matrix and a "Running Queue Mode on " section in `docs/docs/advanced/queue-mode-guide.md`; the transports list in `docs/docs/deployment/onprem-kubernetes.md` if the transport is k8s-relevant; the transport roll call on the docs-site features page (`docs/src/pages/features.tsx`: the "Queue broker over SQS, Kafka, or NATS" highlight on the Sandboxed Code Execution card, and any other "SQS, Kafka, or NATS" mention found by grepping `docs/src/pages/*.tsx`). - Landing page inventories (`docs/src/components/*/data.tsx`): a tile in the **Cloud & infrastructure** row of `IntegrationsMarquee/data.tsx` (role `Queue`, `href` to the queue mode guide, logo or `react-icons/si` glyph), and the transport in the **Queue Pipeline** card's `tags` and `description` under the Scale tab in `FeatureExplorer/data.tsx`. Logo sourcing and the build check are in `ak-dev-sync-docs-from-branch`, *Docs-Site Landing and Features Pages*. - Skills: the pipeline section of `.agents/skills/ak-dev-architecture/SKILL.md`, and the user-facing queue/deploy content in `ak-py/src/agentkernel/skills/` where transports are enumerated. ## Definition of Done - `cd ak-py && uv run pytest`: green, including your contract subclass against the fake. - Live contract green against a real broker via the compose stack. - `make lint-check-all`: green. - Example runs end to end locally (its `app_test.py` passes against a live agent). - Factory rejects a missing config block and a missing extra with actionable errors. - Docs and skills surfaces above updated in the same PR.