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

import concurrent.futures

from src import e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config

_BUSY_TIMEOUT_MS = 30000
_FIX_LOOP_TIMEOUT = 1200


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."""
    conn = schema.connect(db_path)
    conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
    try:
        result = e2e_pipeline.run_end_to_end(conn, target_folder_path, goal_text)
        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