"""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", "coding")
_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(conn, target_folder_path, task_id, artifact_type):
    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:
        raise TaskExecutionError("task {} has no {} artifact to resume from".format(task_id, artifact_type))
    return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0])


def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default):
    """Like _artifact_content, but for the 2026-08-06 test_dataset/
    test_script nodes -- a task resumed from before this feature existed
    (or any other reason those two specific artifacts are missing) should
    still be resumable, just without them, not hard-fail."""
    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, Coder
    not yet triggered), and 'coding' (Coder done, Validator not yet
    triggered) -- anything else raises `TaskExecutionError` rather than
    guessing at a mid-phase resume this module was never designed for."""
    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)

    doc_artifact_ids = _latest_artifact_ids(conn, task_id, _DOC_ARTIFACT_TYPES)
    sdd_content = _artifact_content(conn, target_folder_path, task_id, "sdd")
    prd_content = _artifact_content(conn, target_folder_path, task_id, "prd")
    benchmark_content = _artifact_content(conn, target_folder_path, task_id, "benchmark")
    # 2026-08-06 nodes -- see _artifact_content_or_default's own docstring
    # for why these two specifically fall back instead of hard-failing.
    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"
    )

    if status == "planning_brd_prd_sdd":
        _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 == "coding" -- Coder already ran
        code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES)

    _check_not_paused(conn, task_id)
    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": doc_artifact_ids,
        "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))


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."""
    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()

    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)."""
    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", "coding"):
        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}