agent_system.py

import asyncio
import json
import os
import re
import subprocess
import time
import uuid
from collections import defaultdict, deque
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Optional

import httpx

from router import InferenceRouterCircuitBreaker, PredictiveContextFilter, MultiLayerThrottle, DecentralizedThrottle

# ──────────────────────────────────────────────
# Constants
# ──────────────────────────────────────────────
RPM_LIMIT = 40
DAG_MAX_CONCURRENCY = 3
STATE_COMPRESSION_THRESHOLD = 0.75  # 75% token capacity
MAX_TASK_HISTORY = 100
WORKSPACE_DIR = Path("/tmp/altamira-workspace")
WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)

# ──────────────────────────────────────────────
# Enums / Data
# ──────────────────────────────────────────────
class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    REVIEWING = "reviewing"
    APPROVED = "approved"
    REJECTED = "rejected"
    FAILED = "failed"
    COMPLETED = "completed"

class AgentRole(Enum):
    COMMANDER = "commander"
    REVIEWER = "reviewer"
    EXECUTOR = "executor"
    MONITOR = "monitor"

@dataclass
class Task:
    id: str
    dag_id: str
    prompt: str
    role: AgentRole
    status: TaskStatus = TaskStatus.PENDING
    created: float = field(default_factory=time.time)
    updated: float = field(default_factory=time.time)
    output: Optional[str] = None
    error: Optional[str] = None
    patch: Optional[dict] = None
    review_notes: Optional[str] = None
    parent_id: Optional[str] = None
    metadata: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "dag_id": self.dag_id,
            "prompt": self.prompt,
            "role": self.role.value,
            "status": self.status.value,
            "created": self.created,
            "updated": self.updated,
            "output": self.output,
            "error": self.error,
            "patch": self.patch,
            "review_notes": self.review_notes,
            "parent_id": self.parent_id,
            "metadata": self.metadata,
        }

    @staticmethod
    def from_dict(d: dict) -> "Task":
        return Task(
            id=d["id"],
            dag_id=d["dag_id"],
            prompt=d["prompt"],
            role=AgentRole(d["role"]),
            status=TaskStatus(d["status"]),
            created=d.get("created", time.time()),
            updated=d.get("updated", time.time()),
            output=d.get("output"),
            error=d.get("error"),
            patch=d.get("patch"),
            review_notes=d.get("review_notes"),
            parent_id=d.get("parent_id"),
            metadata=d.get("metadata", {}),
        )

# ──────────────────────────────────────────────
# Static DAG Scheduler
# ──────────────────────────────────────────────
class DAGNode:
    def __init__(self, task_id: str, deps: Optional[list[str]] = None, priority: int = 0):
        self.task_id = task_id
        self.deps = deps or []
        self.priority = priority


class DAGScheduler:
    def __init__(self):
        self._nodes: dict[str, DAGNode] = {}
        self._topo_order: list[str] = []
        self._dirty = True

    def add_node(self, task_id: str, deps: Optional[list[str]] = None, priority: int = 0):
        self._nodes[task_id] = DAGNode(task_id, deps, priority)
        self._dirty = True

    def remove_node(self, task_id: str):
        self._nodes.pop(task_id, None)
        self._dirty = True

    def has_cycle(self) -> bool:
        WHITE, GRAY, BLACK = 0, 1, 2
        color = {nid: WHITE for nid in self._nodes}
        def dfs(nid: str) -> bool:
            color[nid] = GRAY
            node = self._nodes.get(nid)
            if node:
                for d in node.deps:
                    if d not in color:
                        continue
                    if color[d] == GRAY:
                        return True
                    if color[d] == WHITE and dfs(d):
                        return True
            color[nid] = BLACK
            return False
        return any(dfs(nid) for nid in self._nodes if color[nid] == WHITE)

    def validate(self) -> list[str]:
        errors = []
        for nid, node in self._nodes.items():
            for d in node.deps:
                if d not in self._nodes:
                    errors.append(f"Node '{nid}' depends on missing node '{d}'")
        if self.has_cycle():
            errors.append("DAG contains a cycle")
        return errors

    def _topological_sort(self) -> list[str]:
        if not self._dirty and self._topo_order:
            return self._topo_order
        visited = set()
        result = []

        def dfs(nid: str):
            if nid in visited:
                return
            visited.add(nid)
            node = self._nodes.get(nid)
            if node:
                for d in node.deps:
                    dfs(d)
                result.append(nid)

        for nid in self._nodes:
            dfs(nid)
        self._topo_order = result
        self._dirty = False
        return result

    def get_ready(self, completed: set[str]) -> list[str]:
        ready = []
        for nid in self._topological_sort():
            if nid in completed:
                continue
            node = self._nodes.get(nid)
            if node and all(d in completed for d in node.deps):
                ready.append(nid)
        ready.sort(key=lambda nid: self._nodes[nid].priority, reverse=True)
        return ready

    def to_dict(self) -> dict:
        return {nid: {"deps": n.deps, "priority": n.priority} for nid, n in self._nodes.items()}


# ──────────────────────────────────────────────
# Round-Robin Key Pool
# ──────────────────────────────────────────────
class KeyPool:
    def __init__(self, keys: Optional[list[str]] = None):
        self._keys: list[str] = keys or []
        self._index = 0
        self._lock = asyncio.Lock()
        self._failures: dict[str, int] = defaultdict(int)
        self._max_failures = 3

    def add_key(self, key: str):
        if key not in self._keys:
            self._keys.append(key)

    def remove_key(self, key: str):
        if key in self._keys:
            self._keys.remove(key)
            self._failures.pop(key, None)

    async def get_key(self) -> Optional[str]:
        async with self._lock:
            if not self._keys:
                return None
            candidates = self._keys[self._index:] + self._keys[:self._index]
            for key in candidates:
                if self._failures.get(key, 0) < self._max_failures:
                    self._index = (self._index + 1) % len(self._keys)
                    return key
            return None

    def record_failure(self, key: str):
        self._failures[key] += 1

    def record_success(self, key: str):
        self._failures[key] = 0

    def is_exhausted(self) -> bool:
        return all(f >= self._max_failures for f in self._failures.values()) if self._keys else True

    def to_dict(self) -> dict:
        return {
            "key_count": len(self._keys),
            "available": sum(1 for k in self._keys if self._failures.get(k, 0) < self._max_failures),
            "failures": dict(self._failures),
        }


# ──────────────────────────────────────────────
# Throttle Controller (40 RPM)
# ──────────────────────────────────────────────
class ThrottleController:
    def __init__(self, rpm_limit: int = RPM_LIMIT):
        self.rpm_limit = rpm_limit
        self._slots: list[float] = []

    def acquire(self) -> float:
        now = time.time()
        cutoff = now - 60
        self._slots = [t for t in self._slots if t > cutoff]
        if len(self._slots) >= self.rpm_limit:
            wait = self._slots[0] + 60 - now
            if wait > 0:
                return wait
        self._slots.append(now)
        return 0.0

    async def wait_if_needed(self):
        wait = self.acquire()
        if wait > 0:
            await asyncio.sleep(wait)

    def usage_pct(self) -> float:
        now = time.time()
        cutoff = now - 60
        active = sum(1 for t in self._slots if t > cutoff)
        return (active / self.rpm_limit) * 100

    def to_dict(self) -> dict:
        return {
            "rpm_limit": self.rpm_limit,
            "current_rpm": len([t for t in self._slots if t > time.time() - 60]),
            "usage_pct": round(self.usage_pct(), 1),
        }


# ──────────────────────────────────────────────
# State Manager (HF Datasets + local fallback)
# ──────────────────────────────────────────────
class StateManager:
    def __init__(self, dataset_repo: str = "", hf_token: str = ""):
        self._dataset_repo = dataset_repo
        self._hf_token = hf_token
        self._local_cache: dict = {}
        self._dirty_keys: set[str] = set()

    async def save(self, key: str, value: Any):
        self._local_cache[key] = value
        self._dirty_keys.add(key)
        if self._dataset_repo and self._hf_token:
            await self._sync_to_dataset(key, value)

    async def load(self, key: str) -> Any:
        if key in self._local_cache:
            return self._local_cache[key]
        if self._dataset_repo and self._hf_token:
            return await self._sync_from_dataset(key)

    async def save_task(self, task: Task):
        tasks = (await self.load("tasks")) or []
        tasks = [t for t in tasks if t.get("id") != task.id]
        tasks.append(task.to_dict())
        if len(tasks) > MAX_TASK_HISTORY:
            tasks = tasks[-MAX_TASK_HISTORY:]
        await self.save("tasks", tasks)

    async def load_tasks(self) -> list[Task]:
        raw = (await self.load("tasks")) or []
        return [Task.from_dict(t) for t in raw]

    async def save_dag(self, dag: DAGScheduler):
        await self.save("dag", dag.to_dict())

    async def _sync_to_dataset(self, key: str, value: Any):
        try:
            from huggingface_hub import HfApi
            api = HfApi()
            local_path = f"/tmp/altamira-state/{key}.json"
            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
            Path(local_path).write_text(json.dumps({key: value}))
            api.upload_file(
                path_or_fileobj=local_path,
                path_in_repo=f"state/{key}.json",
                repo_id=self._dataset_repo,
                repo_type="dataset",
                token=self._hf_token,
            )
        except Exception as e:
            pass

    async def _sync_from_dataset(self, key: str) -> Any:
        from huggingface_hub import hf_hub_download
        try:
            local_path = hf_hub_download(
                repo_id=self._dataset_repo,
                filename=f"state/{key}.json",
                repo_type="dataset",
                token=self._hf_token,
            )
            data = json.loads(Path(local_path).read_text())
            self._local_cache[key] = data.get(key)
            return self._local_cache[key]
        except Exception:
            return None


# ──────────────────────────────────────────────
# Executor Agents — JSON code patching
# ──────────────────────────────────────────────
class ExecutorAgent:
    PATCH_ACTIONS = {"patch", "write", "append", "delete", "rename", "mkdir"}

    async def execute(self, patch: dict, workspace: str = "") -> dict:
        action = patch.get("action", "patch")
        if action not in self.PATCH_ACTIONS:
            return {"status": "error", "error": f"Unknown action: {action}"}
        handler = getattr(self, f"_{action}", None)
        if not handler:
            return {"status": "error", "error": f"No handler for: {action}"}
        try:
            result = await handler(patch, workspace)
            if "error" in result:
                return {"status": "error", "error": result["error"]}
            return {"status": "ok", **result}
        except Exception as e:
            return {"status": "error", "error": str(e)}

    async def _patch(self, patch: dict, workspace: str) -> dict:
        file_path = self._resolve(patch["file"], workspace)
        if not file_path.exists():
            return {"error": f"File not found: {patch['file']}"}
        old = patch.get("old", "")
        new = patch.get("new", "")
        content = file_path.read_text()
        if old:
            if content.count(old) > 1 and not patch.get("all"):
                return {"error": "Multiple matches for old string — set 'all': true or provide more context"}
            if patch.get("all"):
                content = content.replace(old, new)
            else:
                content = content.replace(old, new, 1)
        elif new:
            content = content + "\n" + new
        file_path.write_text(content)
        return {"file": str(file_path), "size": len(content)}

    async def _write(self, patch: dict, workspace: str) -> dict:
        file_path = self._resolve(patch["file"], workspace)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(patch.get("content", ""))
        return {"file": str(file_path), "size": len(patch.get("content", ""))}

    async def _append(self, patch: dict, workspace: str) -> dict:
        file_path = self._resolve(patch["file"], workspace)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        with file_path.open("a") as f:
            f.write(patch.get("content", "") + "\n")
        return {"file": str(file_path)}

    async def _delete(self, patch: dict, workspace: str) -> dict:
        file_path = self._resolve(patch["file"], workspace)
        if file_path.exists():
            file_path.unlink()
            return {"file": str(file_path), "deleted": True}
        return {"error": f"File not found: {patch['file']}"}

    async def _rename(self, patch: dict, workspace: str) -> dict:
        src = self._resolve(patch["source"], workspace)
        dst = self._resolve(patch["dest"], workspace)
        if src.exists():
            dst.parent.mkdir(parents=True, exist_ok=True)
            src.rename(dst)
            return {"from": str(src), "to": str(dst)}
        return {"error": f"Source not found: {patch['source']}"}

    async def _mkdir(self, patch: dict, workspace: str) -> dict:
        path = self._resolve(patch["path"], workspace)
        path.mkdir(parents=True, exist_ok=True)
        return {"path": str(path)}

    def _resolve(self, file_path: str, workspace: str) -> Path:
        p = Path(file_path)
        if p.is_absolute():
            return p
        base = Path(workspace) if workspace else WORKSPACE_DIR
        return base / p


# ──────────────────────────────────────────────
# Monitoring Framework
# ──────────────────────────────────────────────
@dataclass
class MetricPoint:
    timestamp: float = field(default_factory=time.time)
    value: float = 0.0
    label: str = ""


class Monitor:
    def __init__(self, max_history: int = 1000):
        self._metrics: dict[str, deque] = defaultdict(lambda: deque(maxlen=max_history))
        self._alerts: list[dict] = []
        self._alert_rules: list[dict] = []

    def record(self, metric: str, value: float, label: str = ""):
        self._metrics[metric].append(MetricPoint(value=value, label=label))

    def add_alert_rule(self, name: str, metric: str, threshold: float, direction: str = "gt"):
        self._alert_rules.append({
            "name": name, "metric": metric,
            "threshold": threshold, "direction": direction,
        })

    def check_alerts(self) -> list[dict]:
        triggered = []
        for rule in self._alert_rules:
            points = self._metrics.get(rule["metric"])
            if not points:
                continue
            latest = points[-1].value
            if rule["direction"] == "gt" and latest > rule["threshold"]:
                triggered.append(self._fire_alert(rule["name"], latest, rule["threshold"]))
            elif rule["direction"] == "lt" and latest < rule["threshold"]:
                triggered.append(self._fire_alert(rule["name"], latest, rule["threshold"]))
        self._alerts.extend(triggered)
        return triggered

    def _fire_alert(self, name: str, value: float, threshold: float) -> dict:
        return {
            "alert": name,
            "value": value,
            "threshold": threshold,
            "timestamp": time.time(),
            "severity": "critical" if abs(value - threshold) / threshold > 0.2 else "warning",
        }

    def summary(self) -> dict:
        return {
            "metrics": {k: [{"value": p.value, "label": p.label, "t": p.timestamp} for p in list(v)[-20:]]
                        for k, v in self._metrics.items()},
            "alerts": self._alerts[-20:],
            "alert_rules": self._alert_rules,
        }


# ──────────────────────────────────────────────
# Resource Manager (CPU/Memory aware concurrency)
# ──────────────────────────────────────────────
class ResourceManager:
    def __init__(self, max_concurrency: int = DAG_MAX_CONCURRENCY,
                 cpu_threshold: float = 80.0, mem_threshold: float = 80.0):
        self._max_concurrency = max_concurrency
        self._cpu_threshold = cpu_threshold
        self._mem_threshold = mem_threshold

    def cpu_usage(self) -> float:
        try:
            with open("/proc/stat") as f:
                line = f.readline()
            parts = line.split()
            if len(parts) >= 5:
                user = int(parts[1])
                nice = int(parts[2])
                system = int(parts[3])
                idle = int(parts[4])
                total = user + nice + system + idle
                if total > 0:
                    return 100.0 * (total - idle) / total
        except Exception:
            pass
        return 0.0

    def mem_usage(self) -> float:
        try:
            with open("/proc/meminfo") as f:
                lines = f.readlines()
            total = 0
            available = 0
            for line in lines:
                if line.startswith("MemTotal:"):
                    total = int(line.split()[1])
                elif line.startswith("MemAvailable:"):
                    available = int(line.split()[1])
            if total > 0:
                return 100.0 * (total - available) / total
        except Exception:
            pass
        return 0.0

    def effective_concurrency(self) -> int:
        cpu = self.cpu_usage()
        mem = self.mem_usage()
        if cpu > self._cpu_threshold or mem > self._mem_threshold:
            return max(1, self._max_concurrency // 2)
        return self._max_concurrency

    def should_skip_review(self) -> bool:
        cpu = self.cpu_usage()
        mem = self.mem_usage()
        return cpu > self._cpu_threshold or mem > self._mem_threshold

    def summary(self) -> dict:
        return {
            "cpu_pct": round(self.cpu_usage(), 1),
            "mem_pct": round(self.mem_usage(), 1),
            "effective_concurrency": self.effective_concurrency(),
            "skip_review": self.should_skip_review(),
        }


# ──────────────────────────────────────────────
# Commander / Reviewer Subsystem
# ──────────────────────────────────────────────
COMMANDER_TOOLS = """Available actions:
- patch: replace old string with new string in a file
- write: create or overwrite a file with content
- append: append content to an existing file
- delete: remove a file
- rename: move/rename a file
- mkdir: create a directory
- shell: execute a shell command (use sparingly)

Each task must include a "patch" dict with the action and relevant parameters.
Dependencies ("deps") declare ordering: a task only runs after all its deps complete.
"""

COMMANDER_PROMPT = f"""You are the Commander — an autonomous software development coordinator.
Your job is to:
1. Analyze the user's request and break it into concrete code tasks
2. For each task, produce a JSON patch command using the available actions
3. Delegate review to the Reviewer before applying changes
4. Ensure no task proceeds without approval
5. Self-throttle: you have a 40 RPM limit. Group related changes to minimize API calls.
   If you get rate-limited, back off and retry locally before escalating.

{COMMANDER_TOOLS}
Output a JSON array of tasks, each with:
- "id": unique task name
- "prompt": what to do
- "patch": {{"action": "patch|write|append|delete|rename|mkdir", "file": "...", "old": "...", "new": "..."}}
- "deps": list of task IDs this depends on (empty list if none)
- "priority": integer, higher = more urgent (default 0)
"""

REVIEWER_PROMPT = """You are the Reviewer — a strict code quality gate.
Your job is to:
1. Review the patch or shell command proposed by the Commander
2. Check for syntax errors, security vulnerabilities, and correctness
3. Return APPROVED with notes, or REJECTED with specific issues to fix
4. Never approve code that introduces security vulnerabilities
5. Flag any use of eval(), exec(), raw subprocess without sanitization

Respond with JSON: {"decision": "APPROVED"|"REJECTED", "notes": "..."}
"""


class Commander:
    def __init__(self, infer_breaker: InferenceRouterCircuitBreaker,
                 key_pool: KeyPool, throttle: ThrottleController,
                 monitor: Monitor):
        self._breaker = infer_breaker
        self._key_pool = key_pool
        self._throttle = throttle
        self._monitor = monitor
        self._context_filter = PredictiveContextFilter()

    async def plan(self, user_request: str) -> list[dict]:
        await self._throttle.wait_if_needed()
        key = await self._key_pool.get_key()
        prompt = f"{COMMANDER_PROMPT}\n\nUser request: {user_request}\n\nOutput JSON array of tasks:"
        compressed = self._context_filter.compact(prompt, target_ratio=STATE_COMPRESSION_THRESHOLD)
        start = time.time()
        result = await self._breaker.infer(compressed, nvidia_key=key or "")
        elapsed = time.time() - start
        self._monitor.record("commander_latency", elapsed)
        self._monitor.record("commander_output_len", len(result))
        self._monitor.record("commander_compressed", len(compressed))
        tasks = self._parse_tasks(result)
        return tasks

    def _parse_tasks(self, text: str) -> list[dict]:
        json_match = re.search(r"\[.*?\]", text, re.DOTALL)
        if json_match:
            try:
                return json.loads(json_match.group())
            except json.JSONDecodeError:
                pass
        return [{"id": "task-1", "prompt": text[:500], "patch": {}, "deps": []}]


class Reviewer:
    def __init__(self, infer_breaker: InferenceRouterCircuitBreaker,
                 key_pool: KeyPool, throttle: ThrottleController,
                 monitor: Monitor):
        self._breaker = infer_breaker
        self._key_pool = key_pool
        self._throttle = throttle
        self._monitor = monitor
        self._context_filter = PredictiveContextFilter()

    async def review(self, task: Task) -> tuple[bool, str]:
        await self._throttle.wait_if_needed()
        key = await self._key_pool.get_key()
        review_prompt = (
            f"{REVIEWER_PROMPT}\n\n"
            f"Task: {task.prompt}\n"
            f"Patch: {json.dumps(task.patch, indent=2)}\n\n"
            f"Review the above. Respond with JSON decision."
        )
        compressed = self._context_filter.compact(review_prompt, target_ratio=STATE_COMPRESSION_THRESHOLD)
        start = time.time()
        result = await self._breaker.infer(compressed, nvidia_key=key or "")
        elapsed = time.time() - start
        self._monitor.record("reviewer_latency", elapsed)
        self._monitor.record("reviewer_compressed", len(compressed))
        decision, notes = self._parse_decision(result)
        return decision, notes

    def _parse_decision(self, text: str) -> tuple[bool, str]:
        json_match = re.search(r"\{.*?\}", text, re.DOTALL)
        if json_match:
            try:
                data = json.loads(json_match.group())
                approved = data.get("decision", "").upper() == "APPROVED"
                notes = data.get("notes", "")
                return approved, notes
            except json.JSONDecodeError:
                pass
        return "approv" in text.lower(), text[:200]


# ──────────────────────────────────────────────
# Main Agent Subsystem
# ──────────────────────────────────────────────
class AgentSubsystem:
    def __init__(self, infer_breaker: InferenceRouterCircuitBreaker,
                 dataset_repo: str = "", hf_token: str = ""):
        self._dag = DAGScheduler()
        self._key_pool = KeyPool()
        self._throttle = ThrottleController()
        self._multi_layer = MultiLayerThrottle()
        self._monitor = Monitor()
        self._resources = ResourceManager()
        self._state = StateManager(dataset_repo, hf_token)
        self._executor = ExecutorAgent()
        self._commander = Commander(infer_breaker, self._key_pool, self._throttle, self._monitor)
        self._reviewer = Reviewer(infer_breaker, self._key_pool, self._throttle, self._monitor)
        self._tasks: dict[str, Task] = {}
        self._completed: set[str] = set()
        self._active_tasks: dict[str, asyncio.Task] = {}
        self._lock = asyncio.Lock()
        self._running = False

        self._monitor.add_alert_rule("high_rpm", "throttle_usage", 80, "gt")
        self._monitor.add_alert_rule("high_latency", "commander_latency", 30.0, "gt")
        self._monitor.add_alert_rule("key_exhaustion", "key_exhausted", 0.5, "lt")
        self._monitor.add_alert_rule("high_cpu", "cpu_usage", 80, "gt")
        self._monitor.add_alert_rule("high_mem", "mem_usage", 80, "gt")

    def add_api_key(self, key: str):
        self._key_pool.add_key(key)

    async def submit_request(self, user_request: str) -> str:
        tasks_data = await self._commander.plan(user_request)
        dag_id = f"dag-{uuid.uuid4().hex[:8]}"
        parent_task = Task(
            id=f"{dag_id}-root",
            dag_id=dag_id,
            prompt=user_request,
            role=AgentRole.COMMANDER,
        )
        async with self._lock:
            self._tasks[parent_task.id] = parent_task
            self._dag.add_node(parent_task.id)
            for td in tasks_data:
                tid = td.get("id", f"task-{uuid.uuid4().hex[:6]}")
                deps = [parent_task.id] + td.get("deps", [])
                task = Task(
                    id=tid,
                    dag_id=dag_id,
                    prompt=td.get("prompt", ""),
                    role=AgentRole.EXECUTOR,
                    patch=td.get("patch", {}),
                    parent_id=parent_task.id,
                    metadata=td,
                )
                self._tasks[tid] = task
                self._dag.add_node(tid, deps)
            await self._state.save_dag(self._dag)
            for t in self._tasks.values():
                await self._state.save_task(t)
        return dag_id

    async def run_cycle(self):
        if self._running:
            return
        self._running = True
        try:
            max_active = self._resources.effective_concurrency()
            ready = self._dag.get_ready(self._completed)
            for tid in ready:
                if len(self._active_tasks) >= max_active:
                    break
                if tid in self._active_tasks or tid in self._completed:
                    continue
                task = self._tasks.get(tid)
                if not task:
                    continue
                self._active_tasks[tid] = asyncio.create_task(self._process_task(tid))
            self._monitor.record("active_tasks", len(self._active_tasks))
            self._monitor.record("completed_tasks", len(self._completed))
            self._monitor.record("throttle_usage", self._throttle.usage_pct())
            self._monitor.record("key_exhausted", 1.0 if self._key_pool.is_exhausted() else 0.0)
            self._monitor.record("cpu_usage", self._resources.cpu_usage())
            self._monitor.record("mem_usage", self._resources.mem_usage())
            self._monitor.check_alerts()
        finally:
            self._running = False

    async def _process_task(self, tid: str):
        task = self._tasks[tid]
        task.status = TaskStatus.RUNNING
        task.updated = time.time()
        await self._state.save_task(task)
        try:
            if task.role == AgentRole.EXECUTOR and task.patch:
                skip_review = self._resources.should_skip_review()
                if skip_review:
                    task.review_notes = "Review skipped — resource constraints"
                    approved = True
                else:
                    task.status = TaskStatus.REVIEWING
                    await self._state.save_task(task)
                    approved, notes = await self._reviewer.review(task)
                    task.review_notes = notes
                if approved:
                    result = await self._executor.execute(task.patch, str(WORKSPACE_DIR))
                    if result.get("status") == "ok":
                        task.status = TaskStatus.COMPLETED
                        task.output = json.dumps(result)
                    else:
                        task.status = TaskStatus.FAILED
                        task.error = result.get("error", "execution failed")
                else:
                    task.status = TaskStatus.REJECTED
                    task.error = task.review_notes or ""
            else:
                task.status = TaskStatus.COMPLETED
                task.output = f"Processed: {task.prompt[:100]}"
        except Exception as e:
            task.status = TaskStatus.FAILED
            task.error = str(e)
        task.updated = time.time()
        async with self._lock:
            self._completed.add(tid)
            del self._active_tasks[tid]
        await self._state.save_task(task)

    def get_task(self, tid: str) -> Optional[Task]:
        return self._tasks.get(tid)

    def get_tasks(self, dag_id: Optional[str] = None) -> list[Task]:
        if dag_id:
            return [t for t in self._tasks.values() if t.dag_id == dag_id]
        return list(self._tasks.values())

    def get_status(self) -> dict:
        return {
            "tasks": {s.value: sum(1 for t in self._tasks.values() if t.status == s) for s in TaskStatus},
            "completed_ids": list(self._completed),
            "active_ids": list(self._active_tasks.keys()),
            "throttle": self._throttle.to_dict(),
            "multi_layer": self._multi_layer.to_dict(),
            "key_pool": self._key_pool.to_dict(),
            "dag": self._dag.to_dict(),
            "monitor": self._monitor.summary(),
            "resources": self._resources.summary(),
        }