"""Kanban Activity bus helpers and per-tenant create-rate token bucket.

Mirrors :mod:`utils.task_plan_activity` so the kanban subsystem has a single
emit point for activity events and a tiny in-memory rate limiter for
``create_task`` calls per tenant.

Two responsibilities live here intentionally to keep the storage module
free of activity-bus / threading concerns:

1. ``try_acquire_kanban_create_slot(tenant)`` — token-bucket backpressure
   helper used by :func:`kanban_storage.create_task` before inserting.
2. ``emit_kanban_event_activity(...)`` and
   ``emit_kanban_result_parse_failure_activity(...)`` — the only places
   that construct activity-bus payloads, so the schema cannot drift.

Activity payloads are **ciphertext-free** by contract — ids, transitions,
and aggregate counts only.
"""
from __future__ import annotations

import logging
import threading
import time
from collections import defaultdict, deque
from typing import Any, Deque

from utils.redaction import cap_str

_log = logging.getLogger(__name__)

SOURCE_KANBAN = "kanban"
SOURCE_KANBAN_COORDINATOR = "kanban_coordinator"
TAG_KANBAN = "kanban"

# Per-tenant create-rate token bucket.
#
# We track timestamps of the last N creates per tenant in a deque and refuse
# when the count within the trailing 60 s exceeds the configured cap. The
# bucket is rebuilt fresh on process start; this is intentional for v1
# (operators can hot-bounce the runner if they want to clear it).
_RATE_WINDOW_SEC = 60.0
_rate_lock = threading.Lock()
_tenant_history: dict[str, Deque[float]] = defaultdict(deque)


def try_acquire_kanban_create_slot(tenant: str, *, cap_per_minute: int) -> bool:
    """Return True if a kanban_storage.create_task may proceed for *tenant*.

    Caller must pre-validate that ``cap_per_minute > 0``; values <= 0
    disable the limiter (always allow).
    """
    return try_acquire_kanban_create_slots(tenant, count=1, cap_per_minute=cap_per_minute)


def try_acquire_kanban_create_slots(tenant: str, *, count: int, cap_per_minute: int) -> bool:
    """Atomically acquire *count* create slots for *tenant* — all or none.

    June 2026 incident: ``create_batch`` acquired slots one at a time in a
    loop, so a batch rejected mid-loop (or later, by validation) left a
    partial/full burn in the bucket. An agent iterating on a malformed batch
    payload locked itself out of the very call it had just fixed.
    """
    if not isinstance(cap_per_minute, int) or cap_per_minute <= 0:
        return True
    n = max(1, int(count))
    t = (tenant or "default").strip() or "default"
    now = time.monotonic()
    with _rate_lock:
        history = _tenant_history[t]
        cutoff = now - _RATE_WINDOW_SEC
        while history and history[0] < cutoff:
            history.popleft()
        if len(history) + n > cap_per_minute:
            return False
        for _ in range(n):
            history.append(now)
        return True


def reset_kanban_create_rate(*, tenant: str | None = None) -> None:
    """Test/operator helper. Clears the rate limiter for one or all tenants."""
    with _rate_lock:
        if tenant is None:
            _tenant_history.clear()
        else:
            _tenant_history.pop((tenant or "default").strip() or "default", None)


def emit_kanban_event_activity(
    *,
    event_type: str,
    task_id: str,
    from_status: str = "",
    to_status: str = "",
    assignee: str = "",
    tenant: str = "default",
    run_id: str = "",
    attempt: int | None = None,
    summary: str = "",
    origin_thread_id: str = "",
    extra_meta: dict[str, Any] | None = None,
) -> None:
    """Emit an activity-bus event for a kanban lifecycle transition.

    Payload carries only ids and transitions — never ciphertext or
    decrypted user content. ``summary`` is for the activity sidebar text and
    is cap-trimmed; it must not contain task body / metadata.
    """
    try:
        from activity_bus import emit_activity_enriched

        meta: dict[str, Any] = {
            "kanban_event": event_type,
            "task_id": task_id,
            "tenant": tenant,
        }
        if from_status:
            meta["from_status"] = from_status
        if to_status:
            meta["to_status"] = to_status
        if assignee:
            meta["assignee"] = assignee
        if run_id:
            meta["run_id"] = run_id
        if attempt is not None:
            meta["attempt"] = int(attempt)
        if extra_meta:
            for k, v in extra_meta.items():
                if k in meta:
                    continue
                meta[k] = v
        origin_tid = (origin_thread_id or "").strip()
        if origin_tid:
            meta["origin_thread_id"] = origin_tid

        emit_activity_enriched(
            {
                "source": SOURCE_KANBAN,
                "state": to_status or event_type,
                "summary": cap_str(summary or f"kanban {event_type}: {task_id}", 240),
                "thread_id": origin_tid,
                "meta": meta,
            },
            tags=[TAG_KANBAN],
        )
    except Exception:
        _log.debug("emit_kanban_event_activity failed", exc_info=True)


def emit_kanban_result_parse_failure_activity(
    *,
    task_id: str,
    raw_excerpt: str,
    assignee: str = "",
    tenant: str = "default",
) -> None:
    """Surface malformed worker output (no parseable kanban-result block).

    Operators see this in the activity sidebar so they can debug worker
    prompts or completion contracts. ``raw_excerpt`` is cap-trimmed.
    """
    try:
        from activity_bus import emit_activity_enriched

        emit_activity_enriched(
            {
                "source": SOURCE_KANBAN,
                "state": "result_parse_failure",
                "summary": cap_str(
                    f"kanban worker for {task_id} produced no parseable result block",
                    240,
                ),
                "thread_id": "",
                "meta": {
                    "kanban_event": "result_parse_failure",
                    "task_id": task_id,
                    "assignee": assignee,
                    "tenant": tenant,
                    "raw_excerpt": cap_str(raw_excerpt, 500),
                },
            },
            tags=[TAG_KANBAN],
        )
    except Exception:
        _log.debug("emit_kanban_result_parse_failure_activity failed", exc_info=True)


def emit_council_seat_activity(
    *,
    event_type: str,
    task_id: str,
    seat: str,
    batch_id: str,
    origin_thread_id: str,
    status: str,
    assignee: str = "",
    summary: str = "",
    next_action: str = "",
) -> None:
    """Emit ciphertext-free council seat lifecycle events for the activity feed."""
    try:
        from activity_bus import emit_activity_enriched

        seat_label = (seat or "seat").strip().lower()
        meta: dict[str, Any] = {
            "council_event": event_type,
            "task_id": task_id,
            "council_batch_id": batch_id,
            "council_seat": seat_label,
            "to_status": status,
        }
        if assignee:
            meta["assignee"] = assignee
        origin_tid = (origin_thread_id or "").strip()
        if origin_tid:
            meta["origin_thread_id"] = origin_tid
        next_v = (next_action or "").strip()
        if next_v:
            meta["next_action"] = cap_str(next_v, 240)
        human = summary or f"Council {seat_label} → {status}"
        if next_v and event_type == "council_failed":
            human = cap_str(f"{human} — Next: {next_v}", 240)
        emit_activity_enriched(
            {
                "source": SOURCE_KANBAN,
                "state": event_type,
                "summary": cap_str(human, 240),
                "thread_id": origin_tid,
                "meta": meta,
            },
            tags=[TAG_KANBAN, "council"],
        )
    except Exception:
        _log.debug("emit_council_seat_activity failed", exc_info=True)


def emit_coordinator_board_open_activity(*, thread_id: str, snapshot: str = "") -> None:
    """Activity row when coordinator parent stops with an open board."""
    try:
        from activity_bus import emit_activity_enriched

        origin_tid = (thread_id or "").strip()
        meta: dict[str, Any] = {"snapshot_chars": len(snapshot or "")}
        if origin_tid:
            meta["origin_thread_id"] = origin_tid
        emit_activity_enriched(
            {
                "source": SOURCE_KANBAN_COORDINATOR,
                "state": "coordinator_board_open",
                "summary": cap_str(
                    "Kanban board still running; coordinator waiting for workers.",
                    500,
                ),
                "thread_id": origin_tid,
                "meta": meta,
            },
            tags=[TAG_KANBAN, "coordinator"],
        )
    except Exception:
        _log.debug("emit_coordinator_board_open_activity failed", exc_info=True)
