# Scalable Agent Worker Pool — Celery + Redis + Claude
import anthropic
from celery import Celery
from redis import Redis
import json, time

# ── Infrastructure Setup ──────────────────────────────────
app     = Celery("agent_pool", broker="redis://localhost:6379/0",
                              backend="redis://localhost:6379/1")
store   = Redis(host="localhost", port=6379, db=2)  # shared state
client  = anthropic.Anthropic()

# ── Agent Worker (runs on N containers) ───────────────────
@app.task(bind=True, max_retries=3, autoretry_for=(Exception,),
          retry_backoff=True)
def run_agent_task(self, task_id: str, task_payload: dict) -> dict:
    """Stateless agent worker — fetches context, runs LLM, stores result."""

    # 1. Load shared context from state store
    ctx_raw  = store.get(f"ctx:{task_payload['session_id']}")
    context  = json.loads(ctx_raw) if ctx_raw else []

    # 2. Build message history + new task
    messages = context + [{
        "role": "user",
        "content": task_payload["instruction"]
    }]

    # 3. Call the LLM (stateless — any worker can run any task)
    response = client.messages.create(
        model     = "claude-opus-4-6",
        max_tokens= 1024,
        system    = task_payload.get("system_prompt", "You are a helpful agent."),
        messages  = messages
    )

    result_text = response.content[0].text

    # 4. Persist result + updated context back to shared store
    updated_ctx = messages + [{"role": "assistant", "content": result_text}]
    store.setex(f"ctx:{task_payload['session_id']}",
                  3600, json.dumps(updated_ctx))  # TTL 1hr

    result = {"task_id": task_id, "output": result_text,
              "tokens": response.usage.output_tokens, "worker": self.request.hostname}

    store.setex(f"result:{task_id}", 7200, json.dumps(result))
    return result


# ── Dispatcher (orchestrator side) ────────────────────────
def dispatch_tasks(tasks: list[dict]) -> list:
    """Fan-out N tasks to the worker pool in parallel."""
    jobs = [
        run_agent_task.apply_async(
            args=[t["id"], t],
            queue=t.get("priority", "default")   # route by priority
        )
        for t in tasks
    ]
    # Fan-in: wait for all results
    return [job.get(timeout=120) for job in jobs]


# ── Example: dispatch 5 parallel tasks ───────────────────
if __name__ == "__main__":
    tasks = [
        {"id": f"task-{i}", "session_id": "sess-001",
         "instruction": f"Analyze document {i} and extract key terms",
         "priority": "high" if i == 0 else "default"}
        for i in range(5)
    ]
    results = dispatch_tasks(tasks)
    for r in results:
        print(f"[{r['worker']}] Task {r['task_id']} → {r['tokens']} tokens")

# Deploy N workers: celery -A agent_pool worker --concurrency=10 -Q high,default
# Scale: docker-compose up --scale worker=50