"""c_task_execute / c_task_pause (2026-08-03): manual Start/Pause/Stop
controls for v_task_detail, confirmed with the user first:

1. **Soft stop only.** `org_node_trigger.py`/`work_graph_execute.py` both
   invoke the real `claude` CLI via blocking `subprocess.run()` -- no
   process handle is stored anywhere a "stop" action could reach in and
   kill. Pause/Stop let whatever real call is currently in flight finish
   naturally; only the NEXT phase is withheld. A genuine hard-kill would
   need switching those call sites to `Popen` + tracking PIDs somewhere
   reachable by task_id, which is out of scope for this pass.
2. **No new `task.status` enum value.** Paused/stopped tasks reuse
   `status='blocked_pending_human'` with `blocked_reason='user_paused'` /
   `'user_stopped'` (distinguished in the audit trail, not functionally --
   both resume the same way) rather than a 01-M schema change. This reuses
   the already-built, already-tested `manual_resume` mechanism
   (`org_loop_route._manual_resume`) to restore the pre-paused status.

**Why a manual Start button doesn't contradict the "lights-out" design**:
`c_task_create`'s own contract (03-C) already assumed task creation would
auto-trigger scheduling ("若 max_concurrent_goals 已达上限，任务落 queued
但不触发调度" -- implying it DOES trigger scheduling otherwise) via an
always-on Org Graph scheduler that was never actually wired up as a running
process (`src/scheduler.py` builds the mechanism, nothing spins it up
continuously). `start_task_execution_in_background()` below is that same
auto-trigger, just invoked by an explicit user action instead of firing
automatically the instant a task is created.
"""

import threading

from src import e2e_pipeline, org_node_lifecycle, schema, scheduler

_RESUMABLE_STATUSES = ("queued", "planning_brd_prd_sdd", "reviewing_planner_output", "coding", "validating")
_DOC_ARTIFACT_TYPES = ("brd", "prd", "sdd", "benchmark")
_CODE_ARTIFACT_TYPES = ("code_diff", "test_report")


class TaskExecutionError(Exception):
    """The task isn't in a state this module knows how to (re)start from --
    e.g. already running, already done, or blocked for a reason unrelated
    to a user-initiated pause/stop."""


class ExecutionPausedError(Exception):
    """Internal control-flow signal, not a failure: a pause/stop was
    requested while this background thread was between phases. Caught by
    the background runner and treated as a clean stop -- the task is
    ALREADY `blocked_pending_human` by construction (that's literally how
    the pause was requested), nothing more to record."""


def _check_not_paused(conn, task_id):
    status = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
    if status == "blocked_pending_human":
        raise ExecutionPausedError(task_id)


def _latest_artifact_ids(conn, task_id, artifact_types):
    rows = conn.execute(
        "SELECT artifact_type, artifact_id FROM artifact WHERE task_id = ? AND artifact_type IN ({}) "
        "ORDER BY created_at".format(",".join("?" * len(artifact_types))),
        (task_id, *artifact_types),
    ).fetchall()
    return {r[0]: r[1] for r in rows}


def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default):
    """Soft-fail artifact content lookup: returns `default` instead of
    raising when no such artifact exists for this task yet -- e.g. the
    2026-08-06 test_dataset/test_script nodes on a task resumed from before
    that feature existed, or (2026-08-08, Org Graph unification Phase 1,
    the task-execution-resume-assumes-planner-artifacts-exist.md fix) sdd/
    prd/benchmark on a task whose Planner Work Graph never actually
    persisted anything. The old hard-failing sibling this replaced
    (`_artifact_content`, raised TaskExecutionError on a missing row) is
    gone -- every remaining call site in this module wants "have it or
    don't, never crash the resume attempt itself over it"."""
    row = conn.execute(
        "SELECT artifact_id FROM artifact WHERE task_id = ? AND artifact_type = ? ORDER BY created_at DESC LIMIT 1",
        (task_id, artifact_type),
    ).fetchone()
    if row is None:
        return default
    return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0])


def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
    """Same return shape as `e2e_pipeline.run_task_lifecycle` -- continues
    from wherever `task.status` indicates real work last completed, instead
    of re-running phases that already produced real artifacts. Pausing only
    ever happens BETWEEN phases (checked right before each org_node
    trigger), so the only statuses a paused task can be resumed from are
    'queued' (nothing ran yet), 'planning_brd_prd_sdd' (Planner done,
    planner_reviewer not yet triggered), 'reviewing_planner_output'
    (planner_reviewer's ready handshake done, but whether the real judgment
    call ever ran is unknown -- see below), 'coding' (Coder done, Validator
    not yet triggered), and 'validating' -- anything else raises
    `TaskExecutionError` rather than guessing at a mid-phase resume this
    module was never designed for.

    **'planning_brd_prd_sdd'/'reviewing_planner_output' (2026-08-08, Org
    Graph unification Phase 1 -- also fixes task-execution-resume-assumes-
    planner-artifacts-exist.md)**: neither status implies sdd/prd/benchmark
    artifacts actually exist -- the Planner's Work Graph can fail on its
    very first node and leave nothing persisted at all (the P0 bug this
    fixes). Resuming from either status now falls straight into
    `e2e_pipeline._review_planner_output_with_retries()`, which does its own
    existence probing (reuse what's there, regenerate via plan_documents()
    if nothing is) instead of this function asserting sdd/prd/benchmark
    exist up front and crashing via the old hard-failing `_artifact_content`
    if they don't. 'planning_brd_prd_sdd' additionally needs the
    planner_reviewer ready handshake first (its own ready handshake hasn't
    happened yet); 'reviewing_planner_output' already has that handshake
    done, so it falls through to the same retry-loop call directly.

    **'validating' (2026-08-06, real stuck-task incident)**: this is NOT a
    "between phases" resume like the others -- `org_node_lifecycle.
    trigger_org_node_full(..., "validator", ...)`'s blocking subprocess call
    already completed successfully (that's the only way `task.status` could
    have become 'validating' in the first place: the UPDATE to 'validating'
    is the last thing that call does). Calling it a second time would fail
    its own `_PRE_STATE["validator"] == "coding"` guard outright, since the
    task is already past that state. The actual gap for a task stuck here is
    narrower and safe to fill directly: `validate_implementation_with_own_
    graph()` -- which creates and runs the Validator's OWN Work Graph --
    never got invoked. Found for real: a task sat in 'validating' for 15+
    hours with zero `execution_event` rows and no `validator`-org_node
    `work_graph_run` ever created, because nothing in production actually
    polls `scheduler.run_scheduler_once()` (a separate, larger, tracked gap
    -- see `docs/done/scheduler-task-reconciliation-requirements.md`, the
    fix that incident drove) and whatever one-off call had driven this task
    that far had already exited by the time Coder finished."""
    status = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
    if status not in _RESUMABLE_STATUSES:
        raise TaskExecutionError("task {} is status={}, not resumable from here".format(task_id, status))

    if status == "queued":
        return e2e_pipeline.run_task_lifecycle(conn, task_id, target_folder_path, goal_text)

    if status == "planning_brd_prd_sdd":
        _check_not_paused(conn, task_id)
        org_node_lifecycle.trigger_org_node_full(
            conn, "planner_reviewer", task_id, target_folder_path, e2e_pipeline._READY_PROMPT
        )
        status = "reviewing_planner_output"

    if status == "reviewing_planner_output":
        _check_not_paused(conn, task_id)
        plan_result = e2e_pipeline._review_planner_output_with_retries(
            conn, task_id, goal_text, target_folder_path
        )
        artifact_ids = plan_result["artifact_ids"]
        sdd_content = e2e_pipeline._read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
        prd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "prd", "N/A: not provided")
        benchmark_content = _artifact_content_or_default(
            conn, target_folder_path, task_id, "benchmark", "N/A: not provided"
        )
        test_dataset_content = _artifact_content_or_default(
            conn, target_folder_path, task_id, "test_dataset", "N/A: not provided"
        )
        test_script_content = _artifact_content_or_default(
            conn, target_folder_path, task_id, "test_script", "N/A: not provided"
        )

        _check_not_paused(conn, task_id)
        org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
        code_artifact_ids = e2e_pipeline.implement_code(
            conn, task_id, sdd_content, goal_text, target_folder_path,
            benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
            test_script_content=test_script_content,
        )
    else:  # status in ("coding", "validating") -- Coder already ran, unchanged from today
        # Soft-fail versions throughout (2026-08-08): even though sdd should
        # never genuinely be missing once a task has reached "coding"/
        # "validating" (Coder can't have started without it -- see
        # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"]), this still uses
        # the soft-fail helper as defense-in-depth against a theoretical
        # edge state, rather than reintroducing the old hard-crash-on-resume
        # bug this whole change exists to fix.
        sdd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "sdd", "N/A: not provided")
        prd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "prd", "N/A: not provided")
        code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES)

    _check_not_paused(conn, task_id)
    if status != "validating":
        # "validating" means this exact call already ran once -- see the
        # docstring's 'validating' section above. Calling it again would
        # hit org_node_lifecycle's own PRE_STATE guard (needs "coding").
        org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
    verdict = e2e_pipeline.validate_implementation_with_own_graph(
        conn, task_id, sdd_content, prd_content, goal_text, target_folder_path,
    )

    return {
        "task_id": task_id, "doc_artifact_ids": _latest_artifact_ids(conn, task_id, _DOC_ARTIFACT_TYPES),
        "code_artifact_ids": code_artifact_ids, "verdict": verdict,
    }


def _mark_blocked(conn, task_id, exc):
    reason = scheduler._blocked_reason_for(exc) if isinstance(exc, scheduler._ROUTABLE_EXCEPTIONS) else "execution_error"
    scheduler._block_task(conn, task_id, reason, detail=str(exc))


# 2026-08-06 (task_reconciliation): in-process guard against dispatching the
# SAME task_id twice concurrently -- e.g. the console's Start button firing
# while a reconciliation pass (task_reconciliation.py) has already picked up
# the same stale task, or two rapid double-clicks on Start. Deliberately
# in-memory only (not a DB column/lock): this only needs to protect against
# two threads inside THIS process racing each other; it does not (and isn't
# meant to) survive a process restart -- a restart's whole point is that
# nothing in this process is running anymore, so an empty set on startup is
# the correct state, not a bug.
_in_flight_lock = threading.Lock()
_in_flight_task_ids = set()


def is_in_flight(task_id):
    with _in_flight_lock:
        return task_id in _in_flight_task_ids


def _try_claim(task_id):
    with _in_flight_lock:
        if task_id in _in_flight_task_ids:
            return False
        _in_flight_task_ids.add(task_id)
        return True


def _release(task_id):
    with _in_flight_lock:
        _in_flight_task_ids.discard(task_id)


def start_task_execution_in_background(db_path, task_id, target_folder_path, goal_text):
    """Spawns a daemon thread that runs (or resumes) this task's real
    execution to completion. Own sqlite connection per thread -- same
    reasoning as `scheduler.run_goal_to_completion`, a shared connection
    across threads needs the same `conn_lock` discipline U-027 already
    proved is easy to get wrong. On PASS/FAIL, applies the same
    `scheduler.finish_task_lifecycle` closing step (`close_task_done` /
    `org_level_loop.run_fix_loop`) a scheduled task gets, so a manually
    started one reaches the same real terminal state. Returns the Thread
    (already started) so callers/tests can `.join()` it if needed.

    Raises `TaskExecutionError` immediately (no thread spawned) if this
    task_id already has an execution in flight in this process -- see
    `_in_flight_task_ids` above."""
    if not _try_claim(task_id):
        raise TaskExecutionError("task {} already has an execution in flight".format(task_id))

    def run():
        # check_same_thread=False (2026-08-06, F): same reasoning as
        # scheduler.run_goal_to_completion's own conn -- this thread's conn
        # can reach work_graph_orchestrate.run_graph()'s multi-node
        # concurrent layers (G) via finish_task_lifecycle -> run_fix_loop ->
        # judge_root_cause_via_work_graph, whose worker threads aren't this
        # one.
        conn = schema.connect(db_path, check_same_thread=False)
        try:
            result = run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text)
            scheduler.finish_task_lifecycle(conn, result, target_folder_path, goal_text)
        except ExecutionPausedError:
            pass
        except Exception as e:
            _mark_blocked(conn, task_id, e)
        finally:
            conn.close()
            _release(task_id)

    thread = threading.Thread(target=run, daemon=True)
    thread.start()
    return thread


def request_pause(conn, task_id, reason):
    """`reason`: 'user_paused' or 'user_stopped' -- see module docstring
    for why these are functionally identical (both a soft stop reusing
    blocked_pending_human), kept as distinct strings only so the audit
    trail (execution_event / blocked_reason) preserves which button the
    user actually clicked. Only valid while a background execution could
    plausibly be in flight; raises TaskExecutionError otherwise (nothing
    running to pause, or already terminal).

    'validating' included (2026-08-06, same incident as
    `run_or_resume_task_lifecycle`'s 'validating' section): a resumed
    validation run stays in `task.status='validating'` for its whole
    duration (`validate_implementation_with_own_graph` never updates
    `task.status` itself, same as `implement_code` doesn't during
    'coding') -- excluding it here would leave that in-flight run with no
    way to request a pause/stop, unlike the other resumable phases.

    'reviewing_planner_output' included (2026-08-08, Org Graph unification
    Phase 1, same reasoning): `_review_planner_output_with_retries()` can
    run for a while (up to 3 full replan+review attempts) without ever
    updating `task.status` itself -- same "stays under one status name for
    its whole duration" shape as 'coding'/'validating', so it needs the same
    pause/stop reachability."""
    row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
    if row is None:
        raise TaskExecutionError("no such task: " + task_id)
    project_id, status = row
    if status not in ("planning_brd_prd_sdd", "reviewing_planner_output", "coding", "validating"):
        raise TaskExecutionError(
            "task {} is status={}, nothing in-flight to pause/stop".format(task_id, status)
        )
    scheduler._block_task(conn, task_id, reason)
    return {"task_id": task_id, "status": "blocked_pending_human", "blocked_reason": reason}