Add Redis utilities for session management, caching, and task queues

Docs/ARCHITECTURE.md

@@ -0,0 +1,63 @@
+src/
+├── adapters/                    # All source adapters
+│   ├── __init__.py             # Registry + factory
+│   ├── base.py                 # BaseSourceAdapter interface
+│   ├── notion.py               # NotionAdapter
+│   ├── confluence.py           # ConfluenceAdapter
+│   ├── github.py               # GitHubAdapter
+│   ├── slack.py                # SlackAdapter (NEW)
+│   ├── jira.py                 # JiraAdapter
+│   ├── logs.py                 # LogAggregatorAdapter (NEW)
+│   ├── metrics.py              # MetricsAdapter (NEW)
+│   ├── error_traces.py         # ErrorTraceAdapter (NEW)
+│   ├── business_data.py        # BusinessDataAdapter (NEW)
+│   ├── ocr.py                  # OCRAdapter (NEW)
+│   └── pdf.py                  # PDFAdapter
+│
+├── integrations/               # Webhooks, event handlers, polling tasks
+│   ├── __init__.py
+│   ├── github/
+│   │   ├── webhooks.py         # GitHub webhook handler
+│   │   └── tasks.py            # GitHub Celery tasks
+│   ├── slack/
+│   │   ├── webhooks.py         # Slack event handler (NEW)
+│   │   └── tasks.py            # Slack polling tasks (NEW)
+│   ├── jira/
+│   │   ├── webhooks.py         # Jira webhook handler (NEW)
+│   │   └── tasks.py            # Jira polling tasks (NEW)
+│   ├── logs/
+│   │   ├── webhooks.py         # Error log webhook handler (NEW)
+│   │   └── tasks.py            # Log polling tasks (NEW)
+│   ├── metrics/
+│   │   └── tasks.py            # Metrics polling tasks (NEW)
+│   └── business_data/
+│       └── tasks.py            # Business data sync tasks (NEW)
+│
+├── orm/                        # Database ORM connectors
+│   ├── __init__.py
+│   ├── base.py                 # BaseORM interface
+│   ├── postgres.py             # PostgreSQL connector
+│   ├── salesforce.py           # Salesforce REST API
+│   ├── netsuite.py             # NetSuite SuiteTalk API
+│   ├── sap.py                  # SAP OData API
+│   └── generic_rest.py         # Generic REST API wrapper
+│
+├── tasks/                      # Celery task definitions (YOUR DOMAIN)
+│   ├── __init__.py
+│   ├── ingest_tasks.py         # Main ingest coordination
+│   ├── sync_tasks.py           # Polling/incremental sync (notion, confluence, etc)
+│   ├── webhook_tasks.py        # Handle webhook queuing (redis backed)
+│   └── state_management.py     # Redis state tracking
+│
+├── redis/                      # Redis utilities (YOUR DOMAIN)
+│   ├── __init__.py
+│   ├── cache.py                # Caching layer (credentials, last_sync_times)
+│   ├── queues.py               # Task queues (ingest queue, webhook queue)
+│   ├── session_state.py        # Webhook/API session state
+│   └── locks.py                # Distributed locks (prevent parallel syncs)
+│
+└── ingestion/
+    ├── pipeline.py             # Main ingestion pipeline
+    ├── enrichment.py           # Entity/relationship extraction
+    └── orchestrator.py         # Routes docs through adapters
+

src/redis/__init__.py

@@ -0,0 +1,15 @@
+"""Redis utilities for queuing, caching, locks, and session state."""
+
+from .queues import IngestQueue, WebhookQueue, get_queue
+from .cache import RedisCache
+from .locks import DistributedLock
+from .session_state import SessionStateManager
+
+__all__ = [
+    "IngestQueue",
+    "WebhookQueue",
+    "get_queue",
+    "RedisCache",
+    "DistributedLock",
+    "SessionStateManager",
+]

src/redis/cache.py

@@ -0,0 +1,164 @@
+"""Redis caching layer for credentials, sync state, and frequently accessed data."""
+
+import json
+from datetime import datetime, timedelta
+from typing import Any, Optional
+
+import redis
+
+
+class RedisCache:
+    """Simple Redis cache with TTL support."""
+
+    def __init__(self, redis_client: redis.Redis, key_prefix: str = "cache"):
+        self.redis = redis_client
+        self.key_prefix = key_prefix
+
+    def _make_key(self, namespace: str, key: str) -> str:
+        """Build cache key with namespace."""
+        return f"{self.key_prefix}:{namespace}:{key}"
+
+    async def set(
+        self,
+        namespace: str,
+        key: str,
+        value: Any,
+        ttl_seconds: Optional[int] = None,
+    ) -> bool:
+        """
+        Set cache value with optional TTL.
+
+        Args:
+            namespace: Logical grouping (e.g., "credentials", "sync_state")
+            key: Cache key
+            value: Value to cache (auto-serialized to JSON if not string)
+            ttl_seconds: Time-to-live (None = no expiration)
+        """
+        cache_key = self._make_key(namespace, key)
+
+        # Auto-serialize non-string values
+        if isinstance(value, str):
+            cache_value = value
+        else:
+            cache_value = json.dumps(value)
+
+        if ttl_seconds:
+            self.redis.setex(cache_key, ttl_seconds, cache_value)
+        else:
+            self.redis.set(cache_key, cache_value)
+
+        return True
+
+    async def get(
+        self,
+        namespace: str,
+        key: str,
+        deserialize: bool = True,
+    ) -> Optional[Any]:
+        """
+        Get cache value.
+
+        Args:
+            namespace: Logical grouping
+            key: Cache key
+            deserialize: If True, attempt JSON deserialization
+        """
+        cache_key = self._make_key(namespace, key)
+        value = self.redis.get(cache_key)
+
+        if not value:
+            return None
+
+        value_str = value.decode('utf-8')
+
+        if deserialize:
+            try:
+                return json.loads(value_str)
+            except (json.JSONDecodeError, ValueError):
+                return value_str
+
+        return value_str
+
+    async def delete(self, namespace: str, key: str) -> bool:
+        """Delete cache entry."""
+        cache_key = self._make_key(namespace, key)
+        self.redis.delete(cache_key)
+        return True
+
+    async def exists(self, namespace: str, key: str) -> bool:
+        """Check if cache entry exists."""
+        cache_key = self._make_key(namespace, key)
+        return self.redis.exists(cache_key) > 0
+
+    async def clear_namespace(self, namespace: str) -> int:
+        """Clear all entries in a namespace."""
+        pattern = self._make_key(namespace, "*")
+        keys = self.redis.keys(pattern)
+        if keys:
+            return self.redis.delete(*keys)
+        return 0
+
+
+class SyncStateCache(RedisCache):
+    """Specialized cache for tracking last sync times per source."""
+
+    def __init__(self, redis_client: redis.Redis):
+        super().__init__(redis_client, key_prefix="sync_state")
+
+    async def get_last_sync(self, source_type: str, space_id: str) -> Optional[datetime]:
+        """Get last sync timestamp for a source/space."""
+        key = f"{source_type}:{space_id}"
+        timestamp_str = await self.get("last_sync", key, deserialize=False)
+
+        if timestamp_str:
+            return datetime.fromisoformat(timestamp_str)
+        return None
+
+    async def set_last_sync(self, source_type: str, space_id: str):
+        """Update last sync timestamp to now."""
+        key = f"{source_type}:{space_id}"
+        await self.set("last_sync", key, datetime.utcnow().isoformat())
+
+    async def get_last_full_sync(self, source_type: str, space_id: str) -> Optional[datetime]:
+        """Get last full (non-incremental) sync timestamp."""
+        key = f"{source_type}:{space_id}"
+        timestamp_str = await self.get("full_sync", key, deserialize=False)
+
+        if timestamp_str:
+            return datetime.fromisoformat(timestamp_str)
+        return None
+
+    async def set_last_full_sync(self, source_type: str, space_id: str):
+        """Update last full sync timestamp to now."""
+        key = f"{source_type}:{space_id}"
+        await self.set("full_sync", key, datetime.utcnow().isoformat())
+
+
+class CredentialCache(RedisCache):
+    """Specialized cache for storing integration credentials (encrypted in production)."""
+
+    def __init__(self, redis_client: redis.Redis):
+        super().__init__(redis_client, key_prefix="credentials")
+        # In production, use redis-py with encryption or a separate vault service
+        # For now, TTL prevents long-term exposure
+
+    async def get_credentials(self, integration_type: str, org_id: str) -> Optional[dict]:
+        """Get cached credentials for an integration."""
+        key = f"{integration_type}:{org_id}"
+        return await self.get("creds", key, deserialize=True)
+
+    async def set_credentials(
+        self,
+        integration_type: str,
+        org_id: str,
+        credentials: dict,
+        ttl_seconds: int = 3600,  # 1 hour default
+    ):
+        """Cache credentials with short TTL (should fetch fresh when possible)."""
+        key = f"{integration_type}:{org_id}"
+        await self.set("creds", key, credentials, ttl_seconds=ttl_seconds)
+
+    async def clear_credentials(self, integration_type: str, org_id: str):
+        """Revoke cached credentials immediately."""
+        key = f"{integration_type}:{org_id}"
+        await self.delete("creds", key)

src/redis/locks.py

@@ -0,0 +1,141 @@
+"""Distributed locks to prevent concurrent syncing of same source."""
+
+import uuid
+from datetime import datetime, timedelta
+from typing import Optional
+
+import redis
+
+
+class DistributedLock:
+    """
+    Redis-backed distributed lock for preventing concurrent operations on same resource.
+
+    Prevents multiple Celery workers from syncing the same source simultaneously.
+    """
+
+    def __init__(self, redis_client: redis.Redis, key_prefix: str = "locks"):
+        self.redis = redis_client
+        self.key_prefix = key_prefix
+        self.lock_id = str(uuid.uuid4())
+
+    def _make_lock_key(self, resource: str) -> str:
+        """Build lock key."""
+        return f"{self.key_prefix}:{resource}"
+
+    async def acquire(
+        self,
+        resource: str,
+        timeout_seconds: int = 3600,
+    ) -> bool:
+        """
+        Try to acquire a lock on a resource.
+
+        Args:
+            resource: Unique identifier (e.g., "notion:workspace-123")
+            timeout_seconds: How long lock lasts before auto-releasing
+
+        Returns:
+            True if lock acquired, False if resource already locked
+        """
+        lock_key = self._make_lock_key(resource)
+
+        # SET NX: only set if not exists
+        # SET with EX: auto-expire after timeout
+        result = self.redis.set(
+            lock_key,
+            self.lock_id,
+            nx=True,
+            ex=timeout_seconds,
+        )
+
+        return result is not None
+
+    async def release(self, resource: str) -> bool:
+        """
+        Release a lock (only if we own it).
+
+        Uses Lua script for atomic check-and-delete.
+        """
+        lock_key = self._make_lock_key(resource)
+
+        # Lua script: delete only if value matches our lock_id (prevents freeing other locks)
+        lua_script = """
+        if redis.call("get", KEYS[1]) == ARGV[1] then
+            return redis.call("del", KEYS[1])
+        else
+            return 0
+        end
+        """
+
+        script = self.redis.register_script(lua_script)
+        result = script(keys=[lock_key], args=[self.lock_id])
+
+        return result > 0
+
+    async def is_locked(self, resource: str) -> bool:
+        """Check if resource is currently locked."""
+        lock_key = self._make_lock_key(resource)
+        return self.redis.exists(lock_key) > 0
+
+    async def wait_and_acquire(
+        self,
+        resource: str,
+        timeout_seconds: int = 3600,
+        max_wait_seconds: int = 300,
+    ) -> bool:
+        """
+        Poll and wait until lock is available, then acquire it.
+
+        Useful for ensuring sequential processing of the same resource.
+        """
+        wait_start = datetime.utcnow()
+        wait_deadline = wait_start + timedelta(seconds=max_wait_seconds)
+
+        while datetime.utcnow() < wait_deadline:
+            if await self.acquire(resource, timeout_seconds):
+                return True
+
+            # Back off: sleep 100ms before retrying
+            await asyncio.sleep(0.1)
+
+        return False
+
+
+class LockPool:
+    """Manage multiple locks efficiently."""
+
+    def __init__(self, redis_client: redis.Redis):
+        self.redis = redis_client
+        self.locks = {}  # resource -> DistributedLock instance
+
+    def get_lock(self, resource: str) -> DistributedLock:
+        """Get or create a lock for a resource."""
+        if resource not in self.locks:
+            self.locks[resource] = DistributedLock(self.redis)
+
+        return self.locks[resource]
+
+    async def acquire_many(
+        self,
+        resources: list[str],
+        timeout_seconds: int = 3600,
+    ) -> dict[str, bool]:
+        """Try to acquire locks on multiple resources."""
+        results = {}
+
+        for resource in resources:
+            lock = self.get_lock(resource)
+            results[resource] = await lock.acquire(resource, timeout_seconds)
+
+        return results
+
+    async def release_all(self):
+        """Release all locks held by this pool."""
+        for resource, lock in self.locks.items():
+            await lock.release(resource)
+
+        self.locks.clear()
+
+
+import asyncio

src/redis/queues.py

@@ -0,0 +1,234 @@
+"""Redis-backed task queues for document ingestion and webhook events."""
+
+import json
+from datetime import datetime
+from typing import Optional
+from enum import Enum
+
+import redis
+from pydantic import BaseModel
+
+
+class Priority(str, Enum):
+    """Task priority levels."""
+    CRITICAL = "critical"  # Real-time (errors, alerts)
+    HIGH = "high"          # Fast track (key decisions, breaking changes)
+    NORMAL = "normal"      # Standard (routine data, logs)
+    LOW = "low"            # Background (metrics, non-urgent)
+
+
+class QueuedTask(BaseModel):
+    """Task in a queue."""
+    id: str
+    source_type: str
+    payload: dict
+    rbac_tags: dict
+    priority: Priority
+    created_at: datetime
+    attempt: int = 0
+    max_retries: int = 3
+
+
+class IngestQueue:
+    """Queue for documents ready for ingestion pipeline."""
+
+    def __init__(self, redis_client: redis.Redis, key_prefix: str = "ingest"):
+        self.redis = redis_client
+        self.key_prefix = key_prefix
+
+    async def add(
+        self,
+        source_type: str,
+        payload: dict,
+        rbac_tags: dict,
+        priority: Priority = Priority.NORMAL,
+    ) -> str:
+        """
+        Add document to ingest queue.
+
+        Uses sorted set with priority as score for ordering:
+        - CRITICAL (0): real-time processing
+        - HIGH (1): priority batch
+        - NORMAL (2): standard batch
+        - LOW (3): low-priority batch
+        """
+        task_id = f"{source_type}:{payload.get('uri', 'unknown')}:{datetime.utcnow().timestamp()}"
+
+        priority_scores = {
+            Priority.CRITICAL: 0,
+            Priority.HIGH: 1,
+            Priority.NORMAL: 2,
+            Priority.LOW: 3,
+        }
+
+        task = {
+            "id": task_id,
+            "source_type": source_type,
+            "payload": json.dumps(payload),
+            "rbac_tags": json.dumps(rbac_tags),
+            "priority": priority.value,
+            "created_at": datetime.utcnow().isoformat(),
+            "attempt": 0,
+        }
+
+        # Add to sorted set: score = priority (lower = higher priority)
+        self.redis.zadd(
+            f"{self.key_prefix}:queue",
+            {json.dumps(task): priority_scores[priority]}
+        )
+
+        return task_id
+
+    async def pop(self, batch_size: int = 10) -> list[QueuedTask]:
+        """
+        Pop highest-priority tasks from queue.
+        Returns up to batch_size tasks, prioritizing CRITICAL > HIGH > NORMAL > LOW.
+        """
+        key = f"{self.key_prefix}:queue"
+        tasks = []
+
+        # Pop from lowest score (highest priority)
+        for _ in range(batch_size):
+            result = self.redis.zrange(key, 0, 0)  # Get first item
+            if not result:
+                break
+
+            task_json = result[0].decode('utf-8')
+            task_dict = json.loads(task_json)
+
+            # Remove from queue
+            self.redis.zrem(key, task_json)
+
+            task = QueuedTask(
+                id=task_dict["id"],
+                source_type=task_dict["source_type"],
+                payload=json.loads(task_dict["payload"]),
+                rbac_tags=json.loads(task_dict["rbac_tags"]),
+                priority=Priority(task_dict["priority"]),
+                created_at=datetime.fromisoformat(task_dict["created_at"]),
+                attempt=task_dict["attempt"],
+            )
+            tasks.append(task)
+
+        return tasks
+
+    async def requeue_on_failure(self, task: QueuedTask) -> bool:
+        """
+        Re-queue task if it hasn't exceeded max retries.
+        Returns True if re-queued, False if max retries exceeded.
+        """
+        if task.attempt >= task.max_retries:
+            # Send to deadletter queue
+            await self.send_to_deadletter(task, "max_retries_exceeded")
+            return False
+
+        task.attempt += 1
+        await self.add(
+            task.source_type,
+            task.payload,
+            task.rbac_tags,
+            task.priority,
+        )
+        return True
+
+    async def send_to_deadletter(self, task: QueuedTask, reason: str):
+        """Send task to deadletter queue for manual inspection."""
+        dlq_key = f"{self.key_prefix}:deadletter"
+
+        dlq_item = {
+            "task": json.dumps(task.dict()),
+            "reason": reason,
+            "failed_at": datetime.utcnow().isoformat(),
+        }
+
+        self.redis.lpush(dlq_key, json.dumps(dlq_item))
+        # Keep deadletter for 30 days
+        self.redis.expire(dlq_key, 86400 * 30)
+
+    async def get_stats(self) -> dict:
+        """Get queue statistics."""
+        key = f"{self.key_prefix}:queue"
+        dlq_key = f"{self.key_prefix}:deadletter"
+
+        queue_length = self.redis.zcard(key)
+        dlq_length = self.redis.llen(dlq_key)
+
+        return {
+            "queue_length": queue_length,
+            "deadletter_length": dlq_length,
+            "total": queue_length + dlq_length,
+        }
+
+
+class WebhookQueue:
+    """Queue for raw webhook events (before processing)."""
+
+    def __init__(self, redis_client: redis.Redis, key_prefix: str = "webhooks"):
+        self.redis = redis_client
+        self.key_prefix = key_prefix
+
+    async def add(
+        self,
+        event_type: str,
+        payload: dict,
+        priority: Priority = Priority.NORMAL,
+    ) -> str:
+        """Queue a raw webhook event for async processing."""
+        event_id = f"{event_type}:{datetime.utcnow().timestamp()}"
+
+        priority_scores = {
+            Priority.CRITICAL: 0,
+            Priority.HIGH: 1,
+            Priority.NORMAL: 2,
+            Priority.LOW: 3,
+        }
+
+        event = {
+            "id": event_id,
+            "event_type": event_type,
+            "payload": json.dumps(payload),
+            "priority": priority.value,
+            "received_at": datetime.utcnow().isoformat(),
+        }
+
+        # Add to sorted set by priority
+        self.redis.zadd(
+            f"{self.key_prefix}:pending",
+            {json.dumps(event): priority_scores[priority]}
+        )
+
+        return event_id
+
+    async def pop(self, batch_size: int = 5) -> list[dict]:
+        """Pop highest-priority webhook events."""
+        key = f"{self.key_prefix}:pending"
+        events = []
+
+        for _ in range(batch_size):
+            result = self.redis.zrange(key, 0, 0)
+            if not result:
+                break
+
+            event_json = result[0].decode('utf-8')
+            event = json.loads(event_json)
+
+            # Remove from queue
+            self.redis.zrem(key, event_json)
+
+            events.append(event)
+
+        return events
+
+
+def get_queue(queue_type: str, redis_client: redis.Redis) -> Optional[object]:
+    """Factory function to get queue instance."""
+    queues = {
+        "ingest": IngestQueue,
+        "webhook": WebhookQueue,
+    }
+
+    queue_class = queues.get(queue_type)
+    if not queue_class:
+        return None
+
+    return queue_class(redis_client)

src/redis/session_state.py

@@ -0,0 +1,222 @@
+"""Session state management for API requests, webhook processing, and user interactions."""
+
+import json
+from datetime import datetime, timedelta
+from typing import Any, Optional
+from uuid import uuid4
+
+import redis
+
+
+class SessionStateManager:
+    """Manage session state in Redis (API sessions, webhook processing state, etc.)."""
+
+    def __init__(self, redis_client: redis.Redis, key_prefix: str = "session"):
+        self.redis = redis_client
+        self.key_prefix = key_prefix
+
+    def _make_session_key(self, session_id: str) -> str:
+        """Build session key."""
+        return f"{self.key_prefix}:{session_id}"
+
+    async def create_session(
+        self,
+        user_id: str,
+        org_id: str,
+        ttl_seconds: int = 3600,
+    ) -> str:
+        """Create a new session."""
+        session_id = str(uuid4())
+        session_key = self._make_session_key(session_id)
+
+        session_data = {
+            "session_id": session_id,
+            "user_id": user_id,
+            "org_id": org_id,
+            "created_at": datetime.utcnow().isoformat(),
+            "last_activity": datetime.utcnow().isoformat(),
+        }
+
+        self.redis.setex(
+            session_key,
+            ttl_seconds,
+            json.dumps(session_data),
+        )
+
+        return session_id
+
+    async def get_session(self, session_id: str) -> Optional[dict]:
+        """Retrieve session data."""
+        session_key = self._make_session_key(session_id)
+        data = self.redis.get(session_key)
+
+        if not data:
+            return None
+
+        return json.loads(data.decode('utf-8'))
+
+    async def update_session(self, session_id: str, updates: dict, ttl_seconds: int = 3600):
+        """Update session data."""
+        session_key = self._make_session_key(session_id)
+        session_data = await self.get_session(session_id)
+
+        if not session_data:
+            return False
+
+        session_data.update(updates)
+        session_data["last_activity"] = datetime.utcnow().isoformat()
+
+        self.redis.setex(
+            session_key,
+            ttl_seconds,
+            json.dumps(session_data),
+        )
+
+        return True
+
+    async def delete_session(self, session_id: str):
+        """Invalidate a session."""
+        session_key = self._make_session_key(session_id)
+        self.redis.delete(session_key)
+
+    async def session_exists(self, session_id: str) -> bool:
+        """Check if session is still valid."""
+        session_key = self._make_session_key(session_id)
+        return self.redis.exists(session_key) > 0
+
+
+class WebhookProcessingState:
+    """Track webhook processing state (for idempotency and retry logic)."""
+
+    def __init__(self, redis_client: redis.Redis):
+        self.redis = redis_client
+        self.key_prefix = "webhook_state"
+
+    def _make_key(self, webhook_id: str) -> str:
+        """Build key for webhook state."""
+        return f"{self.key_prefix}:{webhook_id}"
+
+    async def mark_processing(self, webhook_id: str, ttl_seconds: int = 86400):
+        """Mark webhook as being processed."""
+        key = self._make_key(webhook_id)
+        state = {
+            "webhook_id": webhook_id,
+            "status": "processing",
+            "started_at": datetime.utcnow().isoformat(),
+        }
+
+        self.redis.setex(key, ttl_seconds, json.dumps(state))
+
+    async def mark_completed(self, webhook_id: str, result: dict):
+        """Mark webhook as completed."""
+        key = self._make_key(webhook_id)
+        state = {
+            "webhook_id": webhook_id,
+            "status": "completed",
+            "completed_at": datetime.utcnow().isoformat(),
+            "result": result,
+        }
+
+        # Keep for 7 days after completion
+        self.redis.setex(key, 86400 * 7, json.dumps(state))
+
+    async def mark_failed(self, webhook_id: str, error: str, retry_at: Optional[datetime] = None):
+        """Mark webhook as failed."""
+        key = self._make_key(webhook_id)
+        state = {
+            "webhook_id": webhook_id,
+            "status": "failed",
+            "failed_at": datetime.utcnow().isoformat(),
+            "error": error,
+            "retry_at": retry_at.isoformat() if retry_at else None,
+        }
+
+        self.redis.setex(key, 86400 * 7, json.dumps(state))
+
+    async def get_state(self, webhook_id: str) -> Optional[dict]:
+        """Get current state of a webhook."""
+        key = self._make_key(webhook_id)
+        data = self.redis.get(key)
+
+        if not data:
+            return None
+
+        return json.loads(data.decode('utf-8'))
+
+    async def is_processing(self, webhook_id: str) -> bool:
+        """Check if webhook is currently being processed."""
+        state = await self.get_state(webhook_id)
+        return state and state.get("status") == "processing"
+
+    async def is_completed(self, webhook_id: str) -> bool:
+        """Check if webhook has been completed (idempotency check)."""
+        state = await self.get_state(webhook_id)
+        return state and state.get("status") == "completed"
+
+
+class AgentJobTracker:
+    """Track agent job execution state (for orchestrator coordination)."""
+
+    def __init__(self, redis_client: redis.Redis):
+        self.redis = redis_client
+        self.key_prefix = "agent_jobs"
+
+    def _make_job_key(self, job_id: str) -> str:
+        """Build key for agent job."""
+        return f"{self.key_prefix}:{job_id}"
+
+    async def create_job(
+        self,
+        job_id: str,
+        agent_name: str,
+        input_data: dict,
+        ttl_seconds: int = 3600,
+    ):
+        """Create a new agent job."""
+        key = self._make_job_key(job_id)
+        job = {
+            "job_id": job_id,
+            "agent_name": agent_name,
+            "status": "queued",
+            "input": json.dumps(input_data),
+            "created_at": datetime.utcnow().isoformat(),
+            "output": None,
+        }
+
+        self.redis.setex(key, ttl_seconds, json.dumps(job))
+
+    async def mark_job_running(self, job_id: str):
+        """Mark job as running."""
+        key = self._make_job_key(job_id)
+        job = json.loads(self.redis.get(key).decode('utf-8'))
+
+        job["status"] = "running"
+        job["started_at"] = datetime.utcnow().isoformat()
+
+        self.redis.set(key, json.dumps(job))
+
+    async def mark_job_completed(self, job_id: str, output: dict):
+        """Mark job as completed with output."""
+        key = self._make_job_key(job_id)
+        job = json.loads(self.redis.get(key).decode('utf-8'))
+
+        job["status"] = "completed"
+        job["output"] = json.dumps(output)
+        job["completed_at"] = datetime.utcnow().isoformat()
+
+        self.redis.set(key, json.dumps(job))
+
+    async def get_job(self, job_id: str) -> Optional[dict]:
+        """Get job status and output."""
+        key = self._make_job_key(job_id)
+        data = self.redis.get(key)
+
+        if not data:
+            return None
+
+        return json.loads(data.decode('utf-8'))
+
+    async def is_job_complete(self, job_id: str) -> bool:
+        """Check if job is completed."""
+        job = await self.get_job(job_id)
+        return job and job.get("status") == "completed"