Build Brief: Queue Implementation

Epic: Queue Implementation Project: Job Queue Infrastructure Status: Ready for implementation Depends on: None (can start immediately)

---

Context

Build a Redis-based job queue for the Marchay Agent system. This is the message bus between the Orchestrator and Research Agents. The Orchestrator enqueues jobs, workers dequeue and process them, and results are delivered back via pub/sub.

Tech Stack:

• Python 3.12+
• Redis 7 (Valkey fork)
• pytest + pytest-asyncio (testing)
• ruff (linting), mypy (type checking)
• prometheus-client (metrics)

Project Root: shared/ (create this directory in the agent repo)

Directory Layout:

shared/
├── src/
│   ├── __init__.py
│   ├── queue/
│   │   ├── __init__.py
│   │   ├── producer.py          # Job enqueue functions
│   │   ├── consumer.py          # Job dequeue (BRPOP)
│   │   ├── result.py            # Result storage
│   │   ├── pubsub.py            # Completion events
│   │   ├── dead_letter.py       # Dead-letter queue
│   │   ├── retry.py             # Retry logic
│   │   ├── cleanup.py           # Expired job cleanup
│   │   └── health.py            # Redis health check
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── job.py               # Job payload schema
│   │   └── result.py            # Result schema
│   └── config.py                # Settings via pydantic-settings
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_producer.py
│   ├── test_consumer.py
│   ├── test_result.py
│   ├── test_pubsub.py
│   ├── test_retry.py
│   └── test_health.py
├── pyproject.toml
├── Dockerfile
└── README.md

---

Acceptance Criteria

1. Project Setup

• [ ] Initialize shared/ package with pyproject.toml
• [ ] Set up Redis connection with connection pooling (min 5, max 20)
• [ ] Create Redis health check endpoint (ping/pong)
• [ ] Set up pytest, ruff, mypy

2. Job Payload Schema

• [ ] Define job payload schema:

  class JobPayload(BaseModel):
      job_id: str  # UUID
      session_id: str  # UUID
      orchestrator_id: str
      agent_type: str  # dining, experiences, transportation, flights, dmc
      task: str
      constraints: dict[str, Any]
      curated_shortlist: list[str]
      result_format: str
      created_at: datetime
      ttl_seconds: int = 60

3. Producer (Enqueue)

• [ ] Implement job enqueue function with schema validation
• [ ] Build per-agent queues: jobs:dining, jobs:experiences, jobs:transportation, jobs:flights, jobs:dmc
• [ ] Add job TTL: auto-expire jobs that exceed TTL
• [ ] Implement enqueue metrics: jobs_enqueued_total counter

4. Consumer (Dequeue)

• [ ] Implement blocking pop from Redis: BRPOP jobs:{agent_type} 0
• [ ] Build job deserialization and validation
• [ ] Implement job execution timeout handling
• [ ] Add visibility timeout: if worker doesn't ack, job returns to queue

5. Result Storage

• [ ] Implement result storage: SET result:{job_id} {result} EX {ttl_seconds}
• [ ] Build result retrieval: GET result:{job_id}
• [ ] Add result expiration tracking

6. Pub/Sub Completion Events

• [ ] Implement completion event publisher: PUBLISH agent:{type}:completed {job_id}
• [ ] Build Orchestrator subscriber: SUBSCRIBE agent:*:completed
• [ ] Implement completion event schema:

  class CompletionEvent(BaseModel):
      job_id: str
      agent_type: str
      session_id: str
      status: str  # success, failed, timeout
      completed_at: datetime

7. Dead-Letter Queue

• [ ] Build dead-letter queue: LPUSH dead:{agent_type} {payload} on max retries
• [ ] Implement dead-letter inspection: LRANGE dead:{agent_type} 0 -1
• [ ] Add dead-letter metrics: dead_letter_depth gauge

8. Retry Logic

• [ ] Implement max 3 attempts per job
• [ ] Build exponential backoff: 1s, 2s, 4s between retries
• [ ] Track retry count in job metadata
• [ ] Move to dead-letter after max retries exceeded

9. Job TTL and Cleanup

• [ ] Implement job TTL: auto-expire jobs without completion
• [ ] Build periodic cleanup: remove expired jobs and stale results
• [ ] Add cleanup metrics: jobs_expired_total counter

10. Connection Management

• [ ] Implement Redis connection pooling: min 5, max 20
• [ ] Build connection health check: periodic ping/pong
• [ ] Implement reconnection logic with exponential backoff

11. Tests

• [ ] Unit tests for producer (enqueue, validation, TTL)
• [ ] Unit tests for consumer (dequeue, deserialization, timeout)
• [ ] Unit tests for result storage (set, get, expire)
• [ ] Unit tests for pub/sub (publish, subscribe, completion events)
• [ ] Unit tests for retry logic (max attempts, backoff)
• [ ] Unit tests for dead-letter queue
• [ ] Integration tests: full enqueue → dequeue → result flow

---

File Structure

shared/
├── src/
│   ├── __init__.py
│   ├── queue/
│   │   ├── __init__.py
│   │   ├── producer.py
│   │   ├── consumer.py
│   │   ├── result.py
│   │   ├── pubsub.py
│   │   ├── dead_letter.py
│   │   ├── retry.py
│   │   ├── cleanup.py
│   │   └── health.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── job.py
│   │   └── result.py
│   └── config.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_producer.py
│   ├── test_consumer.py
│   ├── test_result.py
│   ├── test_pubsub.py
│   ├── test_retry.py
│   └── test_health.py
├── pyproject.toml
├── Dockerfile
└── README.md

---

Implementation Phases

Phase 1: Project Scaffolding (Day 1)

1. Initialize shared/ package
2. Set up Redis connection with pooling
3. Create health check endpoint
4. Define job and result schemas
5. Write tests for schemas

Phase 2: Producer (Day 1-2)

1. Implement job enqueue function
2. Build per-agent queues
3. Add job TTL
4. Write tests for producer

Phase 3: Consumer (Day 2-3)

1. Implement blocking pop (BRPOP)
2. Build job deserialization
3. Add visibility timeout
4. Write tests for consumer

Phase 4: Result Storage (Day 3)

1. Implement result storage (SET with TTL)
2. Build result retrieval (GET)
3. Write tests for result storage

Phase 5: Pub/Sub (Day 3-4)

1. Implement completion event publisher
2. Build Orchestrator subscriber
3. Write tests for pub/sub

Phase 6: Dead-Letter & Retry (Day 4)

1. Build dead-letter queue
2. Implement retry logic (max 3, exponential backoff)
3. Write tests for retry and dead-letter

Phase 7: Cleanup & Health (Day 4-5)

1. Implement job TTL expiration
2. Build periodic cleanup
3. Add connection health checks
4. Write integration tests

---

Dependencies

Dependency	Version	Purpose
redis	^5.0.0	Redis client
pydantic	^2.7.0	Schema validation
pydantic-settings	^2.3.0	Configuration
pytest	^8.2.0	Testing
pytest-asyncio	^0.23.0	Async test support
ruff	^0.4.0	Linting
mypy	^1.10.0	Type checking
prometheus-client	^0.20.0	Metrics