"""Automatic kanban worker progress capture for operator-facing mission logs.

Records structured tool/status rows into the active run's ``progress_log``,
emits coalesced SSE ``progress`` events, and optionally mirrors to the
activity bus as ``source=kanban-worker``.
"""
from __future__ import annotations

import json
import logging
import threading
import time
from typing import Any

from utils.redaction import cap_str

_log = logging.getLogger(__name__)

_SSE_MIN_INTERVAL_SEC = 0.25  # ≤4/sec/task
_sse_lock = threading.Lock()
_last_sse_at: dict[str, float] = {}
_pending_sse: dict[str, tuple[str, str, int]] = {}
_last_status_note: dict[str, str] = {}
_sse_flush_timers: dict[str, threading.Timer] = {}

_BROWSER_TOOLS = frozenset({
    "read_webpage",
    "search_web",
    "search_web_public",
    "web_search",
    "deep_web_search",
    "analyze_multiple_sources",
})


def _claim_token_for(task_id: str) -> str | None:
    try:
        from utils.kanban_dispatcher import current_claim_token

        bound = current_claim_token()
        if not bound:
            return None
        tid, token = bound
        if tid != task_id or not token:
            return None
        return token
    except Exception:
        return None


def _args_preview(args: dict[str, Any] | None) -> str:
    if not isinstance(args, dict) or not args:
        return ""
    try:
        raw = json.dumps(args, ensure_ascii=False)
    except (TypeError, ValueError):
        raw = str(args)
    return cap_str(raw, 256)


def _extract_url(args: dict[str, Any] | None) -> str:
    if not isinstance(args, dict):
        return ""
    for key in ("url", "query", "search_query", "q"):
        val = args.get(key)
        if isinstance(val, str) and val.strip():
            text = val.strip()
            if key == "url" or text.startswith("http"):
                return cap_str(text, 512)
    return ""


def _extract_paths(args: dict[str, Any] | None) -> list[str]:
    if not isinstance(args, dict):
        return []
    paths: list[str] = []
    for key in ("file_path", "filepath", "path", "output_file_path", "destination"):
        val = args.get(key)
        if isinstance(val, str) and val.strip():
            paths.append(cap_str(val.strip(), 512))
    return paths[:8]


def _emit_progress_sse(task_id: str, note: str, *, run_id: str = "", progress_index: int = 0) -> None:
    note_v = cap_str(note or "", 240)
    now = time.monotonic()
    flush_now = False
    pending: dict[str, tuple[str, str, int]] = {}
    with _sse_lock:
        _pending_sse[task_id] = (note_v, run_id, int(progress_index))
        last = _last_sse_at.get(task_id, 0.0)
        if now - last >= _SSE_MIN_INTERVAL_SEC:
            _last_sse_at[task_id] = now
            pending = dict(_pending_sse)
            _pending_sse.clear()
            flush_now = True
            old_timer = _sse_flush_timers.pop(task_id, None)
            if old_timer is not None:
                old_timer.cancel()
        else:
            _schedule_sse_flush_locked(task_id)
    if flush_now:
        for tid, (n, rid, idx) in pending.items():
            _flush_sse(tid, n, run_id=rid, progress_index=idx)


def _schedule_sse_flush_locked(task_id: str) -> None:
    """Must be called with ``_sse_lock`` held. Flushes coalesced payload after interval."""
    old_timer = _sse_flush_timers.pop(task_id, None)
    if old_timer is not None:
        old_timer.cancel()

    def _fire() -> None:
        with _sse_lock:
            entry = _pending_sse.pop(task_id, None)
            _sse_flush_timers.pop(task_id, None)
            if entry is None:
                return
            _last_sse_at[task_id] = time.monotonic()
        note, run_id, idx = entry
        _flush_sse(task_id, note, run_id=run_id, progress_index=idx)

    timer = threading.Timer(_SSE_MIN_INTERVAL_SEC, _fire)
    timer.daemon = True
    _sse_flush_timers[task_id] = timer
    timer.start()


def _flush_sse(task_id: str, note: str, *, run_id: str = "", progress_index: int = 0) -> None:
    try:
        from utils.kanban_events import notify_kanban_event

        payload = {
            "type": "kanban_event",
            "kanban_event": "progress",
            "task_id": task_id,
            "progress_note": note,
            "progress_index": int(progress_index),
        }
        if run_id:
            payload["run_id"] = run_id
        origin_tid = ""
        try:
            import kanban_storage as ks

            row = ks.get_task(task_id)
            if row:
                origin_tid = (row.get("origin_thread_id") or "").strip()
        except Exception:
            pass
        if origin_tid:
            payload["origin_thread_id"] = origin_tid
        notify_kanban_event(payload, thread_id=origin_tid or None)
    except Exception:
        _log.debug("kanban progress SSE emit failed", exc_info=True)


def _task_row_fields(task_id: str) -> tuple[str, str, str]:
    """Return ``(assignee, title, origin_thread_id)`` for activity meta."""
    try:
        import kanban_storage as ks

        row = ks.get_task(task_id)
        if not row:
            return "", "", ""
        return (
            (row.get("assignee") or "").strip(),
            (row.get("title") or "").strip(),
            (row.get("origin_thread_id") or "").strip(),
        )
    except Exception:
        return "", "", ""


def _kanban_worker_job_id(task_id: str) -> str:
    tid = (task_id or "").strip()
    return f"kanban:{tid}" if tid else "kanban:unknown"


def emit_kanban_worker_started(task_id: str) -> None:
    """One ``started`` row per claimed task for the subagent popover."""
    assignee, title, origin_tid = _task_row_fields(task_id)
    note = title or f"Kanban worker ({assignee or 'general'})"
    _emit_activity(
        task_id=task_id,
        note=note,
        tool="",
        phase="started",
        origin_thread_id=origin_tid,
        assignee=assignee,
        title=title,
    )


def _clear_task_sse_state(task_id: str) -> None:
    """Evict per-task coalescing state once a worker finishes.

    Without this the module-level maps grow one entry per task forever in a
    long-running backend, and a parked flush timer can outlive its task.
    """
    with _sse_lock:
        _last_sse_at.pop(task_id, None)
        _last_status_note.pop(task_id, None)
        _pending_sse.pop(task_id, None)
        timer = _sse_flush_timers.pop(task_id, None)
    if timer is not None:
        timer.cancel()


def emit_kanban_worker_completed(task_id: str) -> None:
    """Clear the stable kanban-worker activity slot when the worker thread exits."""
    _clear_task_sse_state(task_id)
    assignee, title, origin_tid = _task_row_fields(task_id)
    note = title or f"Worker finished ({assignee or 'general'})"
    try:
        from activity_bus import emit_activity_enriched

        meta: dict[str, Any] = {
            "kanban_task_id": task_id,
            "task_id": task_id,
            "phase": "completed",
        }
        if assignee:
            meta["assignee"] = assignee
            meta["profile"] = assignee
        if title:
            meta["task_title"] = title
        if origin_tid:
            meta["origin_thread_id"] = origin_tid
        emit_activity_enriched(
            {
                "source": "kanban-worker",
                "state": "completed",
                "summary": cap_str(note, 240),
                "job_id": _kanban_worker_job_id(task_id),
                "thread_id": origin_tid,
                "meta": meta,
            },
            tags=["kanban"],
        )
    except Exception:
        _log.debug("kanban-worker completed activity emit failed", exc_info=True)


def _emit_activity(
    *,
    task_id: str,
    call_id: str = "",
    note: str,
    tool: str = "",
    phase: str = "",
    origin_thread_id: str = "",
    assignee: str = "",
    title: str = "",
) -> None:
    try:
        from activity_bus import emit_activity_enriched

        if not assignee and not title:
            fetched_assignee, fetched_title, fetched_origin = _task_row_fields(task_id)
            if not assignee:
                assignee = fetched_assignee
            if not title:
                title = fetched_title
            if not origin_thread_id:
                origin_thread_id = fetched_origin

        meta: dict[str, Any] = {
            "kanban_task_id": task_id,
            "task_id": task_id,
            "tool": tool,
            "phase": phase,
        }
        if assignee:
            meta["assignee"] = assignee
            meta["profile"] = assignee
        if title:
            meta["task_title"] = title
        if origin_thread_id:
            meta["origin_thread_id"] = origin_thread_id
        # Keep one in-flight row per task; tool-done phases update summary in place.
        if phase == "started":
            state = "started"
        elif phase in ("error",):
            state = "progress"
        else:
            state = "progress"
        emit_activity_enriched(
            {
                "source": "kanban-worker",
                "state": state,
                "summary": cap_str(note, 240),
                "job_id": _kanban_worker_job_id(task_id),
                "thread_id": origin_thread_id,
                "meta": meta,
            },
            tags=["kanban"],
        )
    except Exception:
        _log.debug("kanban-worker activity emit failed", exc_info=True)


def _after_write(
    task_id: str,
    *,
    call_id: str,
    note: str,
    tool: str = "",
    phase: str = "",
    ok: bool,
    run_id: str,
    log_len: int,
) -> None:
    if not ok:
        return
    origin_tid = ""
    try:
        import kanban_storage as ks

        row = ks.get_task(task_id)
        if row:
            origin_tid = (row.get("origin_thread_id") or "").strip()
    except Exception:
        pass
    _emit_progress_sse(task_id, note, run_id=run_id, progress_index=log_len)
    _emit_activity(
        task_id=task_id,
        call_id=call_id,
        note=note,
        tool=tool,
        phase=phase,
        origin_thread_id=origin_tid,
    )


def record_tool_start(
    task_id: str,
    call_id: str,
    tool_name: str,
    args: dict[str, Any] | None,
    status_line: str,
) -> None:
    token = _claim_token_for(task_id)
    if not token:
        return
    note = cap_str(status_line or f"{tool_name}: …", 256)
    entry = {
        "phase": "start",
        "call_id": call_id,
        "tool": tool_name,
        "note": note,
        "url": _extract_url(args),
        "paths": _extract_paths(args),
        "args_preview": _args_preview(args),
    }
    try:
        import kanban_storage as ks

        ok, run_id, log_len = ks.upsert_progress_entry(
            task_id, call_id, entry, expected_token=token,
        )
        _after_write(
            task_id,
            call_id=call_id,
            note=note,
            tool=tool_name,
            phase="start",
            ok=ok,
            run_id=run_id,
            log_len=log_len,
        )
    except Exception:
        _log.debug("record_tool_start failed", exc_info=True)


def record_tool_done(
    task_id: str,
    call_id: str,
    tool_name: str,
    result_text: str,
    success: bool,
) -> None:
    token = _claim_token_for(task_id)
    if not token:
        return
    preview = cap_str(str(result_text or ""), 512)
    paths: list[str] = []
    try:
        from agent_worker import _extract_output_path_from_tool_result

        out_path = _extract_output_path_from_tool_result(tool_name, result_text)
        if out_path:
            paths.append(cap_str(out_path, 512))
    except Exception:
        pass
    patch = {
        "phase": "done" if success else "error",
        "tool": tool_name,
        "result_preview": preview,
    }
    if paths:
        patch["paths"] = paths
    try:
        import kanban_storage as ks

        ok, run_id, log_len = ks.upsert_progress_entry(
            task_id, call_id, patch, expected_token=token,
        )
        note = cap_str(f"{tool_name}: {'completed' if success else 'failed'}", 256)
        _after_write(
            task_id,
            call_id=call_id,
            note=note,
            tool=tool_name,
            phase=patch["phase"],
            ok=ok,
            run_id=run_id,
            log_len=log_len,
        )
    except Exception:
        _log.debug("record_tool_done failed", exc_info=True)


def record_tool_policy_block(
    task_id: str,
    call_id: str,
    tool_name: str,
    message: str,
) -> None:
    token = _claim_token_for(task_id)
    if not token:
        return
    note = cap_str(message or f"{tool_name}: blocked", 256)
    patch = {
        "phase": "error",
        "tool": tool_name,
        "note": note,
        "result_preview": note,
    }
    try:
        import kanban_storage as ks

        ok, run_id, log_len = ks.upsert_progress_entry(
            task_id, call_id, patch, expected_token=token,
        )
        _after_write(
            task_id,
            call_id=call_id,
            note=note,
            tool=tool_name,
            phase="error",
            ok=ok,
            run_id=run_id,
            log_len=log_len,
        )
    except Exception:
        _log.debug("record_tool_policy_block failed", exc_info=True)


_LIVE_STATUS_MARKERS = (
    "generating response",
    "worker is thinking",
)


def is_live_worker_status_note(note: str) -> bool:
    lower = (note or "").strip().lower()
    return any(marker in lower for marker in _LIVE_STATUS_MARKERS)


def finalize_progress_log_for_terminal(
    progress_log: list | None,
    *,
    summary: str,
    success: bool,
) -> list:
    """Append a terminal row when a worker ends on a live status tail (no tool_done)."""
    log = list(progress_log) if isinstance(progress_log, list) else []
    terminal_note = cap_str(
        (summary or "").strip() or ("Worker completed" if success else "Worker stopped"),
        256,
    )
    if not log:
        if not terminal_note:
            return log
        call_id = f"finish:{hash((summary, success)) & 0xFFFFFFFF:08x}"
        log.append(
            {
                "call_id": call_id,
                "phase": "done" if success else "error",
                "note": terminal_note,
            }
        )
        return log
    last = log[-1]
    if not isinstance(last, dict):
        return log
    phase = str(last.get("phase") or "").strip().lower()
    note = str(last.get("note") or "")
    if phase not in ("status", "start"):
        return log
    if phase == "status" and not is_live_worker_status_note(note):
        return log
    call_id = f"finish:{hash((summary, success)) & 0xFFFFFFFF:08x}"
    log.append(
        {
            "call_id": call_id,
            "phase": "done" if success else "error",
            "note": terminal_note,
        }
    )
    return log


def record_worker_status(task_id: str, message: str) -> None:
    msg = cap_str((message or "").strip(), 256)
    if not msg:
        return
    with _sse_lock:
        if _last_status_note.get(task_id) == msg:
            return
    token = _claim_token_for(task_id)
    if not token:
        return
    call_id = f"status:{hash(msg) & 0xFFFFFFFF:08x}"
    entry = {"phase": "status", "call_id": call_id, "note": msg}
    try:
        import kanban_storage as ks

        ok, run_id, log_len = ks.upsert_progress_entry(
            task_id, call_id, entry, expected_token=token,
        )
        if ok:
            with _sse_lock:
                _last_status_note[task_id] = msg
        _after_write(
            task_id,
            call_id=call_id,
            note=msg,
            phase="status",
            ok=ok,
            run_id=run_id,
            log_len=log_len,
        )
    except Exception:
        _log.debug("record_worker_status failed", exc_info=True)


def _cap_thumb_b64(raw: str) -> str:
    text = (raw or "").strip()
    if not text:
        return ""
    if len(text) <= 8192:
        return text
    return text[:8192]


def record_live_preview(task_id: str, evt: dict[str, Any] | None) -> None:
    if not isinstance(evt, dict):
        return
    b64 = evt.get("preview_image_base64") or evt.get("preview_image")
    if not isinstance(b64, str) or not b64.strip():
        return
    thumb = _cap_thumb_b64(b64)
    token = _claim_token_for(task_id)
    if not token:
        return
    tool_name = str(evt.get("tool_name") or "").strip().lower()
    call_id = ""
    try:
        import kanban_storage as ks

        runs = ks.list_runs(task_id, limit=10)
        if not runs:
            return
        run_row = next(
            (r for r in runs if isinstance(r, dict) and r.get("outcome") == ks.OUTCOME_ACTIVE),
            runs[0] if isinstance(runs[0], dict) else None,
        )
        if not run_row:
            return
        md = run_row.get("metadata") if isinstance(run_row, dict) else {}
        log_arr = md.get("progress_log") if isinstance(md, dict) else None
        if not isinstance(log_arr, list):
            return
        target_idx = None
        for idx in range(len(log_arr) - 1, -1, -1):
            row = log_arr[idx]
            if not isinstance(row, dict):
                continue
            if row.get("phase") == "start" and (
                not tool_name
                or str(row.get("tool") or "").lower() == tool_name
                or str(row.get("tool") or "").lower() in _BROWSER_TOOLS
            ):
                target_idx = idx
                call_id = str(row.get("call_id") or "")
                break
        if target_idx is None and log_arr:
            row = log_arr[-1]
            if isinstance(row, dict):
                call_id = str(row.get("call_id") or "")
        if not call_id:
            call_id = f"preview:{int(time.time() * 1000)}"
        ok, run_id, log_len = ks.upsert_progress_entry(
            task_id,
            call_id,
            {"preview_thumb": thumb},
            expected_token=token,
        )
        if ok:
            _emit_progress_sse(task_id, "Preview updated", run_id=run_id, progress_index=log_len)
    except Exception:
        _log.debug("record_live_preview failed", exc_info=True)


def flush_pending_sse(task_id: str | None = None) -> None:
    """Flush coalesced SSE payloads (all tasks, or one task when *task_id* set)."""
    with _sse_lock:
        if task_id:
            timer = _sse_flush_timers.pop(task_id, None)
            if timer is not None:
                timer.cancel()
            pending = {task_id: _pending_sse.pop(task_id)} if task_id in _pending_sse else {}
        else:
            for timer in _sse_flush_timers.values():
                timer.cancel()
            _sse_flush_timers.clear()
            pending = dict(_pending_sse)
            _pending_sse.clear()
        now = time.monotonic()
        for tid in pending:
            _last_sse_at[tid] = now
    for tid, entry in pending.items():
        if entry is None:
            continue
        note, run_id, idx = entry
        _flush_sse(tid, note, run_id=run_id, progress_index=idx)
