"""Periodic recovery for tasks abandoned mid-flight (2026-08-06, real
incident: a task sat in status='validating' for 15+ hours with zero
`execution_event` activity and no live process working on it, because
whatever one-off call had driven it that far had already exited and
nothing in production polls for orphaned resumable tasks).

Scope: FR-2 only from `docs/todo/scheduler-production-polling-loop-
requirements.md` (in-flight task reconciliation) -- deliberately NOT FR-1
(new-goal auto-dispatch via `scheduler.run_scheduler_once()`), which is a
separate, larger, deliberately-not-bundled decision (see that doc's
non-goals/risks: "改变系统自主权边界"). This module only ever resumes a
task that already exists and was already in progress; it never creates a
new task or starts work nobody asked for.

A task only becomes a reconciliation candidate if ALL of:
1. `task.status` is in `task_execution._RESUMABLE_STATUSES`, and
2. it has no execution happening in THIS process right now
   (`task_execution.is_in_flight()` -- covers the common case where the
   console's Start button and this loop share the same guard, so an
   actively-running task is never double-dispatched), and
3. its most recent `execution_event` is older than `stale_after_seconds`
   (or it has none at all, in which case `task.created_at` is used) --
   deliberately NOT `task.updated_at`, which does not change while a
   phase is genuinely in progress (`implement_code`/
   `validate_implementation_with_own_graph` never touch it mid-phase --
   only the `task_status_changed` transitions do). Using
   `task.updated_at` was the real blind spot during this incident's own
   diagnosis: it looked "stale" for reasons unrelated to whether the task
   was actually still being worked on.

`_DEFAULT_STALE_AFTER_SECONDS` (1800s = 30 min) is comfortably above the
longest normal single-node blocking call in this codebase (Validator's
own work graph, `timeout=1200s`) so a node that's still genuinely inside
one blocking `claude -p` call is never misdiagnosed as abandoned -- see
the requirements doc's FR-3/non-goal #4 for why this doesn't need to be
more precise than that for v1.
"""

import time
from datetime import datetime, timezone

from src import schema, task_execution

_DEFAULT_STALE_AFTER_SECONDS = 1800


def _to_epoch(sqlite_timestamp):
    # Same parsing as work_node_run._parse_sqlite_timestamp: sqlite's
    # CURRENT_TIMESTAMP / *_at defaults are UTC 'YYYY-MM-DD HH:MM:SS' with
    # no timezone marker.
    return datetime.strptime(sqlite_timestamp, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()


def find_stale_resumable_tasks(conn, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
    """Returns [{"task_id", "target_folder_path", "goal_text"}, ...] for
    every task that's a genuine reconciliation candidate. Read-only --
    does not dispatch anything, kept separate from `reconcile_once()` so
    candidate selection is independently testable without spinning up
    real background threads."""
    cutoff = now() - stale_after_seconds
    placeholders = ",".join("?" * len(task_execution._RESUMABLE_STATUSES))
    rows = conn.execute(
        "SELECT t.task_id, p.target_folder_path, t.goal_text, t.created_at, "
        "(SELECT MAX(e.created_at) FROM execution_event e WHERE e.task_id = t.task_id) AS last_event_at "
        "FROM task t JOIN project p ON p.project_id = t.project_id "
        "WHERE t.status IN ({})".format(placeholders),
        task_execution._RESUMABLE_STATUSES,
    ).fetchall()

    candidates = []
    for task_id, target_folder_path, goal_text, created_at, last_event_at in rows:
        if task_execution.is_in_flight(task_id):
            continue
        last_activity = _to_epoch(last_event_at or created_at)
        if last_activity >= cutoff:
            continue
        candidates.append({"task_id": task_id, "target_folder_path": target_folder_path, "goal_text": goal_text})
    return candidates


def reconcile_once(db_path, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
    """One reconciliation pass: find candidates, dispatch each via
    `task_execution.start_task_execution_in_background()`. Uses its own
    short-lived connection for the scan, closed before dispatching -- the
    background threads each open their own (same pattern as
    `scheduler.py`). Returns the candidate list (dispatched or not) for
    callers/tests to inspect.

    A candidate that fails to dispatch (e.g. `TaskExecutionError` because
    something else claimed it between the scan and here -- a real race,
    not a bug) is skipped, not treated as a fatal error for the whole
    pass: 04-P's "调度器继续执行队列中其他任务" applies here the same way
    it does in `scheduler.run_scheduler_once`."""
    conn = schema.connect(db_path)
    try:
        candidates = find_stale_resumable_tasks(conn, stale_after_seconds, now)
    finally:
        conn.close()
    for c in candidates:
        try:
            task_execution.start_task_execution_in_background(
                db_path, c["task_id"], c["target_folder_path"], c["goal_text"]
            )
        except Exception:
            pass
    return candidates