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

**U-035 self-target audit trail (2026-08-02)**: resolved a real conflict
between supervisor-auto-rollback-design.md §1 (Deploy Supervisor triggers on
"self-target + Validator PASS + merge", no bug_record involved) and 03-C
§3.11/04-P §4.6's literal text (gated on an *existing*
`bug_record.root_cause_category=core_engine`) -- confirmed with the user
that a plain goal-type self-improvement task (task_create.py: `c_task_create`
only ever makes task_type='goal') never gets a bug_record on its happy path
at all, so the literal gate would silently defeat PH5's entire purpose for
that whole task class. Resolution (user-confirmed): trigger on self-target+
PASS alone; `bug_record.record_core_engine_change()` is created as an
audit-trail byproduct of that trigger, not a precondition for it. `deployment`
row creation itself (the actual hand-off) stays in U-037's `c_deploy_
supervisor_stage` as its own Done criterion already specifies -- this module
only detects + records, it doesn't stage a candidate. `p_deploy_supervisor_
blue_green` (04-P §4.6) never touches `task.status` at all -- it's a fully
separate state machine on `deployment`, matching supervisor-auto-rollback-
design.md §2's "no shared failure domain" between Deploy Supervisor and the
Graph Engine -- so `close_task_done()` still runs unconditionally on PASS
(the Coder's work IS genuinely complete and merged), the self-target check
is a purely additive side effect alongside it, not a replacement for it.
"""

import json
import uuid
import concurrent.futures

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

# Rationale: see module docstring's "Threading model" section (lines 21-28) above.
_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, detail=None):
    """`detail` (2026-08-03): the real exception message, when there is one --
    `blocked_reason` itself stays a short routing code (budget_guardrail's
    auto-resume query matches on its exact value, e.g. 'budget_exceeded'), so
    this doesn't touch it. Without `detail` an unroutable failure (anything
    not in `_ROUTABLE_EXCEPTIONS`) landed as a bare 'execution_error' with
    zero diagnostic content anywhere -- found for real via the console's
    manual Start button on a freshly created project (see
    project_scaffold.py's agent-definitions fix, same incident)."""
    task_row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
    payload = {"from_status": task_row[1], "to_status": "blocked_pending_human"}
    if detail:
        payload["detail"] = detail
    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(payload)),
    )
    conn.commit()


def _record_core_engine_audit_if_self_target(conn, task_id, target_folder_path, run_result):
    """U-035 (05 §5.8): additive side effect on a self-target task's PASS
    closure -- see module docstring's "self-target audit trail" note for
    why this doesn't gate close_task_done, just follows it."""
    if not paths.is_self_target(target_folder_path):
        return
    code_diff_id = run_result.get("code_artifact_ids", {}).get("code_diff")
    if not code_diff_id:
        return
    diff_text = e2e_pipeline._read_artifact_content(conn, target_folder_path, code_diff_id)
    changed_files = core_engine_classify.parse_changed_files_from_diff(diff_text)
    if core_engine_classify.classify_core_engine_change(changed_files):
        bug_record.record_core_engine_change(
            conn, task_id,
            "self-target change touched core-engine files: " + ", ".join(sorted(set(changed_files))),
        )


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)
        _record_core_engine_audit_if_self_target(conn, task_id, target_folder_path, run_result)
        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).

    **2026-08-04**: any OTHER exception (e.g. work_graph_execute.
    WorkGraphExecuteError from a real API hiccup -- found for real: a
    "Response stalled mid-stream" error crashed the calling process
    entirely) used to propagate uncaught, same as before this note --
    `run_scheduler_once`'s own try/except only stops it from taking down
    OTHER goals in the same batch, it does NOT apply the task-level
    blocked_pending_human consequence either, so the task itself was left
    stuck in limbo: not running, not blocked, `blocked_reason` still NULL,
    nothing a human could `manual_resume` from the console. This mirrors
    the exact fallback `task_execution._mark_blocked()` already has for its
    own (Start-button) entry point -- both entry points now guarantee a
    task always lands on a real terminal-ish status with a real error
    detail, never a caller-crashes-and-the-task-is-just-stuck outcome.
    """
    # check_same_thread=False (2026-08-06, F): this conn can reach
    # work_graph_orchestrate.run_graph() via finish_task_lifecycle ->
    # org_level_loop.run_fix_loop -> root_cause_trace.
    # judge_root_cause_via_work_graph, whose 3-perspective layer runs on
    # its OWN ThreadPoolExecutor worker threads (G), not this function's
    # thread -- a plain schema.connect() here would raise "SQLite objects
    # created in a thread can only be used in that same thread" the first
    # time a real multi-node concurrent layer actually runs (previously
    # never exercised in production: every graph_manifest before F was
    # either single-node or a strictly serial chain).
    conn = schema.connect(db_path, check_same_thread=False)
    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, detail=str(e))
            return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": blocked_reason}
        except Exception as e:
            _block_task(conn, task_id, "execution_error", detail=str(e))
            return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": "execution_error"}
        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