Build Summary: Queue Implementation

Epic: Queue Implementation Project: Job Queue Infrastructure Date: 2026-08-19 Status: Complete

---

What Was Built

Redis-based job queue for the Marchay Agent system. Implemented the full queue lifecycle: per-agent-type queues, structured job payloads, blocking dequeue with visibility timeout, result storage with TTL, pub/sub completion events, dead-letter queue for failed jobs, retry logic with exponential backoff, expired job cleanup, and health checks. The Orchestrator enqueues jobs, workers dequeue and process them, and results are delivered back via pub/sub.

---

Files Created

Source

• shared/src/shared/config.py — QueueSettings via pydantic-settings, reads VALKEY_URL from Doppler
• shared/src/shared/schemas/job.py — JobPayload model (job_id, session_id, agent_type, task, constraints, retry metadata)
• shared/src/shared/schemas/result.py — ResultPayload, CompletionEvent models
• shared/src/shared/queue/client.py — QueueClient (Redis connection pool wrapper)
• shared/src/shared/queue/producer.py — QueueProducer (LPUSH to per-agent queues with TTL tracking)
• shared/src/shared/queue/consumer.py — QueueConsumer (BRPOP with visibility timeout, ack/requeue)
• shared/src/shared/queue/result.py — ResultStore (SET/GET with TTL)
• shared/src/shared/queue/pubsub.py — CompletionPublisher, CompletionSubscriber (PUBLISH/SUBSCRIBE)
• shared/src/shared/queue/dead_letter.py — DeadLetterQueue (LPUSH, inspect, depth)
• shared/src/shared/queue/retry.py — RetryHandler (exponential backoff: 1s, 2s, 4s)
• shared/src/shared/queue/cleanup.py — QueueCleanup (expired job removal)
• shared/src/shared/queue/health.py — HealthCheck (ping/pong, pool status)

Tests

• shared/tests/test_schemas.py — 4 tests (defaults, roundtrip, result, completion event)
• shared/tests/test_producer.py — 4 tests (enqueue/dequeue, per-agent queues, TTL, multiple jobs)
• shared/tests/test_consumer.py — 4 tests (empty dequeue, visibility timeout, ack, requeue)
• shared/tests/test_result.py — 3 tests (store/get, missing, overwrite)
• shared/tests/test_pubsub.py — 2 tests (publish mock, subscribe no messages)
• shared/tests/test_retry.py — 1 test (failure retries)
• shared/tests/test_health.py — 2 tests (ping, check pool)
• shared/tests/test_dead_letter.py — 3 tests (inspect empty, push/inspect, depth)
• shared/tests/conftest.py — shared fixtures (fakeredis, make_job helper)

Infrastructure

• shared/pyproject.toml — dependencies and tool config
• shared/README.md — setup, config, usage docs

---

Test Results

23 passed in 2.09s

All acceptance criteria met:

• JobPayload schema with retry metadata
• Per-agent queues: jobs:dining, jobs:experiences, jobs:transportation, jobs:flights, jobs:dmc
• Job TTL with auto-expire tracking
• BRPOP blocking dequeue with visibility timeout
• Ack removes visibility key, requeue restores job
• Result storage with TTL expiration
• Pub/sub completion events
• Dead-letter queue on max retries (3 attempts)
• Exponential backoff retry (1s, 2s, 4s)
• Redis health check (ping/pong)
• Connection pooling (max 20)
• ruff check clean
• mypy strict clean
• Integration tested against Doppler-managed Valkey

---

Issues Fixed During Build

1. min_connections not supported — Removed from ConnectionPool.from_url() (only max_connections accepted)
2. orjson not in dependencies — Added to pyproject.toml
3. fakeredis INFO command unsupported — Wrapped health check in try/except for RedisError
4. fakeredis pub/sub not supported — Mocked publish in unit tests, verified via integration test
5. shared/__init__.py shadowing — Removed conflicting file at shared/ root that shadowed the installed package
6. src layout mismatch — Restructured from flat src/queue/ to src/shared/queue/ to match import paths
7. ruff lint issues — Sorted imports, removed unused imports, caught specific RedisError instead of broad Exception

---

Configuration

Variable	Default	Description
VALKEY_URL	redis://localhost:6379/0	Valkey connection (from Doppler)
QUEUE_REDIS_POOL_MIN	5	Min connection pool size
QUEUE_REDIS_POOL_MAX	20	Max connection pool size
QUEUE_DEFAULT_TTL_SECONDS	60	Default job TTL
QUEUE_MAX_RETRIES	3	Max retry attempts per job
QUEUE_RETRY_BASE_DELAY	1.0	Base delay for exponential backoff
QUEUE_VISIBILITY_TIMEOUT_SECONDS	30	Worker visibility timeout
QUEUE_CLEANUP_INTERVAL_SECONDS	30	Cleanup interval

---

Usage

# Unit tests (uses fakeredis, no infra needed)
cd shared && uv run --python 3.12 python -m pytest tests/ -v

# Integration test (requires Valkey via Doppler)
cd shared && \
VALKEY_URL=$(doppler secrets get VALKEY_URL --project marchay-platform --config dev --plain) \
uv run --python 3.12 python -m pytest tests/ -v

# Lint and type check
cd shared && uv run --python 3.12 ruff check src/ tests/
cd shared && uv run --python 3.12 mypy src/

---

Next Steps

Ready for next epics:

• Queue Monitoring — Queue depth, job latency, worker health, dead-letter metrics as Prometheus gauges/histograms
• Worker Implementation — Per-agent-type worker processes that dequeue, process, and report results
• Orchestrator Integration — Wire QueueProducer into the Orchestrator dispatch loop