Job Queue Infrastructure

Redis/Valkey-based message bus between Orchestrator and Research Agents. Chosen for sub-millisecond enqueue/dequeue, built-in primitives (lists, streams, pub/sub, sorted sets), and Valkey as an open-source fork with no licensing concerns.

---

Epic: Queue Implementation ✓

Plan: Build a Redis-based job queue with per-agent-type queues, structured job payloads, result delivery via pub/sub, and dead-letter handling for failed jobs. Use per-agent queues for simpler routing (each worker pool reads its own queue).

Architectural Context: The queue topology uses per-agent queues (jobs:dining, jobs:experiences, etc.) rather than a single stream with consumer groups. This is simpler and sufficient unless rebalancing across agent types is needed. Workers pull via BRPOP, write results to result:{job_id}, and publish completion events via PUBLISH. The Orchestrator subscribes to completion events.

Build Brief · Build Summary

Tasks

• [x] Set up Redis/Valkey instance with persistence (AOF or RDB) ✓
• [x] Define job payload schema matching architecture spec: ✓

  {
    "job_id": "uuid",
    "session_id": "uuid",
    "orchestrator_id": "orch-123",
    "agent_type": "dining",
    "task": "Find 3 dinner reservations in Tokyo for Feb 13-17, party of 2",
    "constraints": {
      "cuisines": ["sushi", "kaiseki"],
      "max_price_per_person": 200,
      "neighborhoods": ["Ginza", "Shibuya"],
      "dietary": ["no_pork"]
    },
    "curated_shortlist": ["sukiyabashi-jiro", "den", "kyubey"],
    "result_format": "RestaurantOption[]",
    "created_at": "2026-01-15T10:30:00Z",
    "ttl_seconds": 60
  }

• [x] Implement job enqueue function with schema validation (reject invalid payloads) ✓
• [x] Build per-agent queues: jobs:dining, jobs:experiences, jobs:transportation, jobs:flights, jobs:dmc ✓
• [x] Implement result storage: SET result:{job_id} {result} EX {ttl_seconds} ✓
• [x] Build pub/sub channel for completion events: PUBLISH agent:{type}:completed {job_id} ✓
• [x] Implement Orchestrator subscriber: SUBSCRIBE agent:*:completed to receive all completion events ✓
• [x] Build dead-letter queue: LPUSH dead:{agent_type} {payload} on max retries exceeded ✓
• [x] Implement job retry logic: max 3 attempts, exponential backoff (1s, 2s, 4s) ✓
• [x] Add job TTL: auto-expire jobs that exceed TTL without completion ✓
• [x] Build queue cleanup: periodic job to remove expired jobs and stale results ✓
• [x] Implement Redis connection pooling: min 5, max 20 connections per service ✓
• [x] Add Redis health check: ping/pong on connection pool ✓

---

Epic: Queue Monitoring

Plan: Expose queue metrics for autoscaling and alerting. Build visibility into queue depth, job latency, worker health, and dead-letter growth. Queue depth per agent type is the primary autoscaling signal.

Architectural Context: The Orchestrator monitors queue depth via LLEN jobs:{agent_type} to determine if workers need scaling. Job latency (enqueue → completion) is the primary SLO. Worker health is tracked via heartbeat or job completion rate. Dead-letter growth indicates systemic issues.

Tasks

• Implement queue depth monitoring: LLEN check per agent type, expose as Prometheus gauge
• Build job latency tracking: record timestamp on enqueue, compute delta on completion, expose as histogram
• Implement worker health monitoring: track active workers per agent type via heartbeat or job completion rate
• Build metrics endpoint: Prometheus-format /metrics endpoint on queue service
• Implement metrics: queue_depth, jobs_enqueued_total, jobs_completed_total, jobs_failed_total, job_latency_seconds, dead_letter_depth
• Add alerting thresholds:
• Queue depth > 100 for > 5 minutes (scale up workers)
• Queue depth = 0 for > 10 minutes (scale down workers)
• Job p95 latency > SLO (investigate worker performance)
• Dead-letter depth > 10 (investigate failures)
• Build Grafana dashboards: queue depth over time, job latency distribution, worker throughput
• Implement alerting via Grafana unified alerting for threshold breaches
• Add runbook links in alert descriptions