queue.py

# -*- coding: utf-8 -*-
"""TaskQueueManager：長度上限的活躍隊列 + 無限長 pending buffer。

對應架構教學文件 §10.2、interfaces.md §8.2。
線程安全：對外的所有操作都加鎖。
"""
from __future__ import annotations

import threading
from collections import deque
from typing import Any, Deque, Dict, List, Optional


class TaskQueueManager:
    """滑動窗口式任務隊列。

    - `active`：正在或即將執行的子任務，長度上限預設 3。
    - `pending`：等候進入 active 的子任務，無上限。
    - 子任務 dict 必須包含 `subtask_id` 欄位。
    """

    def __init__(self, max_active: int = 3) -> None:
        if max_active < 1:
            raise ValueError("max_active must be >= 1")
        self.max_active = max_active
        self._active: Deque[Dict[str, Any]] = deque()
        self._pending: Deque[Dict[str, Any]] = deque()
        self._lock = threading.RLock()

    # --- 寫操作 -----------------------------------------------------------
    def push(self, subtask: Dict[str, Any]) -> None:
        """新增一個子任務（自動分流到 active 或 pending）。"""
        with self._lock:
            if "subtask_id" not in subtask:
                raise ValueError("subtask must contain 'subtask_id'")
            if len(self._active) < self.max_active:
                self._active.append(subtask)
            else:
                self._pending.append(subtask)

    def push_many(self, subtasks: List[Dict[str, Any]]) -> None:
        for s in subtasks:
            self.push(s)

    def pop_next(self) -> Optional[Dict[str, Any]]:
        """從 active 取出最前面一個子任務，並把 pending 的第一個補上。"""
        with self._lock:
            if not self._active:
                return None
            nxt = self._active.popleft()
            if self._pending:
                self._active.append(self._pending.popleft())
            return nxt

    def peek_active(self) -> Optional[Dict[str, Any]]:
        with self._lock:
            return self._active[0] if self._active else None

    def remove(self, subtask_id: str) -> bool:
        with self._lock:
            for buf in (self._active, self._pending):
                for i, s in enumerate(buf):
                    if s.get("subtask_id") == subtask_id:
                        del buf[i]
                        # 從 pending 補一個進 active
                        while (
                            len(self._active) < self.max_active and self._pending
                        ):
                            self._active.append(self._pending.popleft())
                        return True
            return False

    def reorder(self, new_order: List[str]) -> None:
        """依 subtask_id 列表重新排序 active；不在 new_order 的維持原相對順序。"""
        with self._lock:
            by_id = {s["subtask_id"]: s for s in self._active}
            sorted_subs: List[Dict[str, Any]] = []
            seen = set()
            for sid in new_order:
                if sid in by_id:
                    sorted_subs.append(by_id[sid])
                    seen.add(sid)
            for s in self._active:
                if s["subtask_id"] not in seen:
                    sorted_subs.append(s)
            self._active = deque(sorted_subs)

    def insert(self, idx: int, subtask: Dict[str, Any]) -> None:
        """在 active 指定位置插入；若 active 已滿則把最後一個擠到 pending 開頭。"""
        with self._lock:
            if "subtask_id" not in subtask:
                raise ValueError("subtask must contain 'subtask_id'")
            self._active.insert(idx, subtask)
            while len(self._active) > self.max_active:
                overflow = self._active.pop()
                self._pending.appendleft(overflow)

    def clear(self) -> None:
        with self._lock:
            self._active.clear()
            self._pending.clear()

    # --- 讀操作 -----------------------------------------------------------
    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            return {
                "active": list(self._active),
                "pending": list(self._pending),
                "max_active": self.max_active,
                "active_count": len(self._active),
                "pending_count": len(self._pending),
            }

    def __len__(self) -> int:
        with self._lock:
            return len(self._active) + len(self._pending)

    @property
    def active_count(self) -> int:
        with self._lock:
            return len(self._active)

    @property
    def pending_count(self) -> int:
        with self._lock:
            return len(self._pending)


__all__ = ["TaskQueueManager"]