"""U-032: the scheduler 04-P/01-M/06-验收与测试锚点.md repeatedly presume but
no Unit before this one ever built. REQ-09's boundary case ("多个任务同时有
一个 blocked_pending_human、其余 queued 时，调度器应继续调度 queued 任务")
and 01-M's own indexing rationale ("索引...用于调度器扫描") both assume an
autonomous dispatcher exists; before this it didn't.
`org_node_lifecycle._check_concurrency()` (U-010) already gates a *single*
trigger call against `max_concurrent_goals`, but nothing ever called it
across a real queue of multiple Goals, and `e2e_pipeline.run_end_to_end()`
(U-014) drives exactly one task, synchronously, and was only ever called
from tests.

Found and closed in the same pass (2026-08-02): `run_end_to_end()` itself
deliberately stops at 'validating' with a computed-but-unapplied verdict
("Criteria 4-7 ... explicitly out of this Unit's scope" per its own
docstring) -- a scheduler whose job is "drive Goals to completion" can't
leave every task sitting at 'validating' forever, so `finish_task_lifecycle()`
below applies the PASS -> `close_task_done` / FAIL -> `org_level_loop.
run_fix_loop` step `run_end_to_end` always left undone, without changing
`run_end_to_end`'s own tested behavior or scope.

Threading model: each worker thread opens its OWN sqlite connection
(`schema.connect(db_path)`) rather than sharing one `conn` object across
threads -- U-027 already found the hard way (SIGSEGV) that a shared sqlite3
connection needs a hand-held lock around every operation including close();
replicating that discipline here would just serialize all DB work anyway and
defeat the point of concurrency. Each connection sets a `busy_timeout` so
concurrent writers on the shared file retry instead of raising "database is
locked".

**Second gap found and closed in the same pass (2026-08-02, PH4-closure
audit before starting PH5)**: `work_node_run.CorruptPersistedStateError`
(REQ-06 failure), `work_graph_execute.ClassifierBlockedError` (REQ-07
failure), and `work_node_run.UsageLimitWaitTimeoutError` (REQ-14 failure)
were each correctly raised and each correctly marked their own
`work_node_execution.status='blocked'` (the last one via U-031's catch-all
in `work_node_run.run_work_node`) -- but nothing ever caught them to apply
the *task*-level consequence 06-验收与测试锚点.md's failure text for all
three literally names: `task.status='blocked_pending_human'` with a specific
`blocked_reason`. `work_node_run.py`'s own docstring already flagged this as
a deliberate layering choice ("callers should route it to
task.status='blocked_pending_human'") -- but no caller ever did. This is
exactly the layer `budget_guardrail.check_budget()` already does the
equivalent translation at for its own case (REQ-10), so `_block_task()`
below follows that same established pattern (UPDATE + `task_status_changed`
execution_event) rather than inventing a new one.
"""

import json
import uuid
import concurrent.futures

from src import (
    e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config,
    task_create, work_graph_execute, work_graph_orchestrate, work_node_run,
)

_BUSY_TIMEOUT_MS = 30000
_FIX_LOOP_TIMEOUT = 1200

_BLOCKED_REASON_FOR_EXCEPTION = (
    (work_node_run.CorruptPersistedStateError, "corrupt_persisted_state"),
    (work_graph_execute.ClassifierBlockedError, "classifier_blocked"),
    (work_node_run.UsageLimitWaitTimeoutError, "usage_limit_wait_timeout"),
    (work_graph_orchestrate.ConsistencyCheckFailedError, "prd_consistency_check_failed"),
)
_ROUTABLE_EXCEPTIONS = tuple(exc for exc, _ in _BLOCKED_REASON_FOR_EXCEPTION)


def _blocked_reason_for(exc):
    for exc_type, reason in _BLOCKED_REASON_FOR_EXCEPTION:
        if isinstance(exc, exc_type):
            return reason
    raise AssertionError("no blocked_reason registered for " + type(exc).__name__)


def _block_task(conn, task_id, blocked_reason):
    task_row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
    conn.execute(
        "UPDATE task SET status = 'blocked_pending_human', blocked_reason = ?, "
        "updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
        (blocked_reason, task_id),
    )
    conn.execute(
        "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
        "VALUES (?, ?, ?, 'task_status_changed', ?)",
        (str(uuid.uuid4()), task_row[0], task_id, json.dumps({"from_status": task_row[1], "to_status": "blocked_pending_human"})),
    )
    conn.commit()


def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
    """PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
    run_fix_loop` (root-cause judge + reroute + fix + reverify, U-017).
    Adds 'final_status' to `run_result` and returns it."""
    task_id = run_result["task_id"]
    verdict = run_result["verdict"]["verdict"]
    if verdict == "PASS":
        org_loop_route.close_task_done(conn, task_id)
        run_result["final_status"] = "done"
        return run_result

    benchmark_id = run_result["doc_artifact_ids"]["benchmark"]
    benchmark_content = e2e_pipeline._read_artifact_content(conn, target_folder_path, benchmark_id)
    fix_result = org_level_loop.run_fix_loop(
        conn, task_id, target_folder_path,
        bug_description="Validator reported {} for goal: {}".format(verdict, goal_text),
        context=run_result["verdict"]["raw_response"],
        fix_prompt=e2e_pipeline._READY_PROMPT,
        verify_prompt=e2e_pipeline._VALIDATOR_PROMPT.format(
            goal_text=goal_text, benchmark_content=benchmark_content
        ),
        timeout=_FIX_LOOP_TIMEOUT,
    )
    run_result["final_status"] = fix_result["final_status"]
    return run_result


def run_goal_to_completion(db_path, target_folder_path, goal_text):
    """One Goal, its own sqlite connection, start to a terminal task.status
    (done or blocked_pending_human). This is the unit of work each scheduler
    worker thread runs.

    Creates the task itself (rather than delegating to e2e_pipeline.
    run_end_to_end) so task_id is in scope even if e2e_pipeline.
    run_task_lifecycle raises partway through -- needed to route
    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
    to task.status='blocked_pending_human' (see module docstring's "Second
    gap" note). Any OTHER exception still propagates uncaught, same as
    before -- run_scheduler_once's own try/except is what keeps one Goal's
    unexpected failure from stopping the rest.
    """
    conn = schema.connect(db_path)
    conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
    try:
        project_id = conn.execute(
            "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
        ).fetchone()[0]
        task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
        try:
            result = e2e_pipeline.run_task_lifecycle(conn, task_id, target_folder_path, goal_text)
        except _ROUTABLE_EXCEPTIONS as e:
            blocked_reason = _blocked_reason_for(e)
            _block_task(conn, task_id, blocked_reason)
            return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": blocked_reason}
        return finish_task_lifecycle(conn, result, target_folder_path, goal_text)
    finally:
        conn.close()


def run_scheduler_once(db_path, goals, max_workers=None):
    """`goals`: [(target_folder_path, goal_text), ...], one entry per Goal to
    dispatch this round. Runs up to `max_workers` concurrently (default:
    `org_concurrency_config.max_concurrent_goals`, the same knob
    `org_node_lifecycle._check_concurrency` reads -- read fresh here so a
    config change takes effect on the next scheduler round without a
    restart); goals beyond that queue inside the executor and start as slots
    free up, which is what actually makes "raise max_concurrent_goals -> more
    Goals genuinely run in parallel" (U-032's Done line) true, rather than
    something only exercised by an exception-driven retry loop.

    A goal that raises (including `LoopLimitReachedError` once `loop_round`
    hits 3 -> `blocked_pending_human`, which IS the expected outcome of that
    path, not a scheduler bug) doesn't stop the others -- 04-P's "调度器继续
    执行队列中其他任务" applies to a raised exception in one goal exactly the
    same as it does to one that's already `blocked_pending_human` going in.
    Returns a list of {'goal_text', 'result'} or {'goal_text', 'error'} in
    `goals` order.
    """
    if max_workers is None:
        conn = schema.connect(db_path)
        try:
            config = settings_config.get_concurrency_config(conn)
        finally:
            conn.close()
        max_workers = config["max_concurrent_goals"] if config else 1

    results = [None] * len(goals)
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(run_goal_to_completion, db_path, target_folder_path, goal_text): i
            for i, (target_folder_path, goal_text) in enumerate(goals)
        }
        for future in concurrent.futures.as_completed(futures):
            i = futures[future]
            goal_text = goals[i][1]
            try:
                results[i] = {"goal_text": goal_text, "result": future.result()}
            except Exception as e:
                results[i] = {"goal_text": goal_text, "error": str(e)}
    return results