"""Bridges `plan_work_graph()`'s `graph_manifest` to real per-node execution
via `execute_work_graph()` + `run_work_node()` -- the missing integration
between three previously isolated, individually-tested modules (U-011/
U-012/U-013), closed during U-031's regression audit (2026-08-01): each of
the three had its own unit tests (and even real-API validation) but none of
them had ever actually been called by production code, and the one
genuinely-validated closed loop (`e2e_pipeline.py`, U-014) used a much
simpler, separate mechanism (flat `org_node_trigger.trigger_org_node()`
calls) that bypassed all three.

Design: each node's actual work happens via its OWN one-node dynamic
workflow script (`write_workflow_script` + `execute_work_graph`), not one
big multi-node script for the whole graph. This is a deliberate trade-off,
not the "amortize orchestration cost across many nodes in one `claude -p`
call" pattern 03-C §3.5 originally motivated `execute_work_graph` with --
REQ-06's breakpoint-recovery requirement (resume after a process restart
without re-running already-`done` nodes) is only achievable with per-node,
Python-level persisted state (`work_node_run.py`'s whole design), which is
fundamentally incompatible with wrapping the WHOLE graph in one opaque
`claude` subprocess call Python has no visibility into mid-flight.

Validated for real (2026-08-01): a one-node workflow script with a custom
`agentType` + `schema` genuinely loads the target org_node's migrated
persona (confirmed via a real call asking the agent to self-identify) and
returns real `structured_output` -- see 05 §5.4.1.5.

**`verify_fn` (2026-08-02, PH4-closure audit before starting PH5)**: closes
REQ-02's boundary case (PRD internal-contradiction check must intercept
before `verifying`, not silently pass) -- found genuinely missing
(`artifact_write.write_artifact` always marks `validation_status='passed'`
unconditionally, and no Unit ever built a consistency-judgment call).
`work_graph_run.status` was ALSO never once written past its initial
'planning' value anywhere in this codebase before this -- `verify_fn`'s
'reflecting'/'verifying'/'blocked' writes below are the first real use of
that enum. Scoped deliberately narrow: one optional per-node verify
callback, one regeneration attempt on failure (matching 04-P's "回灌受影响
的最小子图" -- for a single content-producing node, the minimal affected
subgraph IS just itself), escalate to `work_graph_run.status='blocked'` +
raise if the regenerated content still fails. Not a general multi-node
re-planning mechanism -- that would be new, unscoped complexity 04-P itself
doesn't describe either.

**DU-01 (2026-08-04, bug A1/A2 root-cause fix)**: `_make_run_fn()`'s `run_fn`
closure is the natural place to call `execution_log.append_log()` -- it's
the one spot that both has `node_exec_id` and wraps the actual blocking
`execute_work_graph()` call, and `work_node_run.run_work_node()` already
guarantees an `execution_log_file` row (status='active') exists before
`run_fn` is ever invoked. Before this, `append_log()` was fully implemented
(U-013/U-027) but never called by any production code path, so
`execution_log_file` rows existed with zero bytes of real content forever
-- the console's "暂无日志内容" was a correct rendering of genuinely empty
data, and downloading it crashed `backend_server._send_file_download()`
with an uncaught `FileNotFoundError` (also fixed here, see that function).
This is one blocking `agent()` call per node (03-C's own dynamic-workflow
design, see this module's own docstring above) with no mid-flight
progress to report, so logging is 4 discrete events around that one call,
not a real character-by-character stream: `node_start` (before the call),
`node_completed` (after a successful return), `node_failed` (any other
exception, re-raised unchanged after logging), `quota_limit` (the
`UsageLimitReachedError` case, logged before translating to
`work_node_run`'s own `_UsageLimitSignal`). A node retried by
`run_work_node`'s own retry loop calls `run_fn` again, appending another
`node_start`/... pair per attempt -- that's an accurate log of what
actually happened (N attempts), not a bug.
"""

import concurrent.futures
import json
import threading
import time
import uuid

from src import execution_log, paths, work_graph_execute, work_node_run


def _log_event(conn, node_exec_id, event_type, **payload):
    execution_log.append_log(
        conn, node_exec_id, json.dumps({"event": event_type, **payload}) + "\n"
    )


class ConsistencyCheckFailedError(Exception):
    """`verify_fn` rejected a node's output twice in a row (original +
    one regeneration attempt) -- `work_graph_run.status` is already
    'blocked' by the time this is raised."""


class WorkGraphLayerError(Exception):
    """One or more nodes in a concurrently-executed layer failed --
    `errors` is the full [(node_id, exception), ...] list (not just the
    first failure), so a caller can tell whether every failure was the
    same recoverable kind or something more serious."""

    def __init__(self, errors):
        self.errors = errors
        summary = "; ".join("{}: {}".format(node_id, e) for node_id, e in errors)
        super().__init__("{} node(s) failed in one layer: {}".format(len(errors), summary))


class _FetchedRows:
    """A pre-materialized stand-in for a sqlite3.Cursor, holding rows
    already fetched while _LockedConn's lock was held (see _LockedConn's
    own docstring for why this matters -- a real sqlite3.Cursor's
    fetchone()/fetchall() touch the shared connection handle just as much
    as execute() does, so returning a live Cursor and letting the caller
    fetch from it AFTER the lock is released reopens the exact race the
    lock exists to close)."""

    def __init__(self, rows):
        self._rows = rows
        self._index = 0

    def fetchone(self):
        if self._index >= len(self._rows):
            return None
        row = self._rows[self._index]
        self._index += 1
        return row

    def fetchall(self):
        rows = self._rows[self._index:]
        self._index = len(self._rows)
        return rows


class _LockedConn:
    """sqlite3.Connection-compatible thread-safe proxy: execute/executemany
    each take a shared lock for the FULL statement-execution-plus-fetch
    cycle (not just the `execute()` call itself -- a sqlite3.Cursor's
    fetchone()/fetchall() also touch the shared connection handle, so a
    lock that released before the caller's `.fetchone()` runs would still
    let two threads collide on the same connection; this was caught for
    real by RunGraphRealSqliteConcurrencyTest during development, not
    theoretical), commit() takes it for the commit call. SQLite only ever
    allows one writer at a time regardless -- this doesn't create new
    concurrency, it just makes the existing many-scattered-call-sites code
    (`work_node_run.py`/`work_graph_orchestrate.py`/`execution_log.py`) safe
    to call from multiple threads without rewriting every call site, since
    the lock is held only per-statement, not around the (slow) `claude -p`
    call itself. Mirrors `backend_server.py`'s existing shared-connection +
    `threading.Lock()` pattern, generalized to per-statement granularity.
    Attribute access not explicitly wrapped here (cursor()/rollback()/
    close()) passes through unlocked -- nothing on the concurrent path in
    this module uses them."""

    def __init__(self, conn, lock=None):
        self._conn = conn
        self._lock = lock or threading.Lock()

    def execute(self, *args, **kwargs):
        with self._lock:
            return _FetchedRows(self._conn.execute(*args, **kwargs).fetchall())

    def executemany(self, *args, **kwargs):
        with self._lock:
            return _FetchedRows(self._conn.executemany(*args, **kwargs).fetchall())

    def commit(self):
        with self._lock:
            return self._conn.commit()

    def __getattr__(self, name):
        return getattr(self._conn, name)


def topological_layers(graph_manifest):
    """Kahn's algorithm, layered: returns List[List[node]], where each inner
    list is one round's batch of in-degree-zero nodes -- nodes within a
    layer are, by construction, mutually independent (neither depends on
    the other), so they're safe to run concurrently. Layers are ordered
    (layer N+1 only becomes ready once layer N is fully accounted for);
    each layer's nodes are sorted by node_id for deterministic test
    assertions. Dependencies are read from BOTH `node['depends_on']` and
    `graph_manifest['edges']` and merged -- `plan_work_graph()`'s own
    output isn't schema-validated (it can't be, `--json-schema` doesn't
    work with `--agent`, see work_graph_plan.py), so a real LLM response
    might populate one and not the other; trusting only one would silently
    mis-order nodes rather than raise. Raises ValueError on a cycle or a
    dependency referencing an unknown node_id.
    """
    nodes = {n["node_id"]: n for n in graph_manifest["nodes"]}
    deps = {node_id: set(n.get("depends_on") or []) for node_id, n in nodes.items()}
    for edge in graph_manifest.get("edges", []):
        to_id, from_id = edge.get("to"), edge.get("from")
        if to_id in deps and from_id in nodes:
            deps[to_id].add(from_id)

    in_degree = {node_id: len(d) for node_id, d in deps.items()}
    dependents = {node_id: [] for node_id in nodes}
    for node_id, d in deps.items():
        for dep in d:
            if dep in dependents:
                dependents[dep].append(node_id)

    layers = []
    current = sorted(node_id for node_id, degree in in_degree.items() if degree == 0)
    seen_count = 0
    while current:
        layers.append([nodes[node_id] for node_id in current])
        seen_count += len(current)
        next_layer = set()
        for node_id in current:
            for dependent in dependents[node_id]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    next_layer.add(dependent)
        current = sorted(next_layer)
    if seen_count != len(nodes):
        raise ValueError(
            "graph_manifest has a dependency cycle or a depends_on/edge references an unknown node_id"
        )
    return layers


def topological_order(graph_manifest):
    """Flat view of topological_layers() -- kept for callers (e.g.
    e2e_pipeline.py) that only need a total order, not the layer grouping."""
    return [node for layer in topological_layers(graph_manifest) for node in layer]


def ensure_placeholder_skill(conn, skill_id, owner_org_node):
    """No Skill content has actually been migrated from the Hermes profiles
    yet (skill_update.py's own documented gap) -- work_node_execution.
    skill_ref is a required FK, so a real row has to exist regardless.
    Idempotent."""
    exists = conn.execute("SELECT 1 FROM skill WHERE skill_id = ?", (skill_id,)).fetchone()
    if exists is None:
        conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES (?, ?, ?, 'unknown', 'new', 'active')",
            (skill_id, owner_org_node, "skills/" + skill_id + "/"),
        )
        conn.commit()


def _ensure_node_row(conn, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, input_payload):
    existing = conn.execute(
        "SELECT node_exec_id FROM work_node_execution WHERE run_id = ? AND node_id = ?", (run_id, node_id)
    ).fetchone()
    if existing:
        return existing[0]
    node_exec_id = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
        "input_schema_ref, output_schema_ref, input_payload, status) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')",
        (node_exec_id, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, json.dumps(input_payload)),
    )
    conn.commit()
    return node_exec_id


def _render_node_script(script_name, org_node, prompt, output_schema, model):
    model_opt = ", model: " + json.dumps(model) if model else ""
    return (
        "export const meta = {{ name: {name}, description: 'graphlemon Work Node ({org_node})' }}\n\n"
        "const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema}{model_opt} }})\n\n"
        "return result\n"
    ).format(
        name=json.dumps(script_name), org_node=org_node, org_node_json=json.dumps(org_node),
        prompt=json.dumps(prompt), schema=json.dumps(output_schema), model_opt=model_opt,
    )


def _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout):
    script_name = "graphlemon-node-" + node_exec_id.replace("-", "")

    def run_fn(cwd, model=None):
        _log_event(conn, node_exec_id, "node_start", org_node=org_node, model=model)
        script_content = _render_node_script(script_name, org_node, prompt, output_schema, model)
        work_graph_execute.write_workflow_script(cwd, script_name, script_content)
        try:
            output = work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout)
        except work_graph_execute.UsageLimitReachedError as e:
            _log_event(conn, node_exec_id, "quota_limit", quota_limit_type=e.quota_limit_type)
            work_node_run.raise_usage_limit_signal(e.quota_limit_type)
        except Exception as e:
            # ClassifierBlockedError / generic WorkGraphExecuteError -- logged
            # then propagated as-is; run_work_node's own `except Exception`
            # clause marks the node 'blocked' and re-raises, no special
            # handling needed here beyond recording what actually happened.
            _log_event(conn, node_exec_id, "node_failed", error=str(e))
            raise
        else:
            _log_event(conn, node_exec_id, "node_completed", output=output)
            return output

    return run_fn


_DEFAULT_MAX_WORKERS = 4  # see this module's own docstring section on G for rationale


def run_graph(
    conn, run_id, org_node, skill_id, target_folder_path, graph_manifest,
    build_prompt, input_schema, output_schema, timeout=600, needs_worktree=False,
    verify_fn=None, sleep=time.sleep, now=time.time, max_workers=_DEFAULT_MAX_WORKERS,
):
    """Runs every node in `graph_manifest` through `run_work_node()`, each
    backed by a real one-node dynamic-workflow `agent()` call using
    `org_node`'s migrated persona. Nodes are grouped into topological
    layers (`topological_layers()`) and run layer-by-layer with a barrier
    in between; within a layer -- nodes that are mutually independent by
    construction -- a layer with 2+ nodes runs them concurrently via a
    `ThreadPoolExecutor` (bounded by `max_workers`), a layer with exactly 1
    node runs it synchronously (no thread-pool overhead for the common
    case: every real graph_manifest observed so far is a single-node-per-
    layer chain).

    `build_prompt(node, outputs)` is called right before running that
    node -- `outputs` contains every already-completed EARLIER layer's
    output_payload (a layer never starts until every node in the previous
    layer has finished, so this is always safe to read without locking),
    so prompts needing upstream content (e.g. "here is the approved
    BRD...") can be built just-in-time instead of requiring the whole
    graph's prompts precomputed before anything runs. `build_prompt` never
    sees a same-layer sibling's output -- by definition, same-layer nodes
    don't depend on each other. This also makes resume-after-restart work
    correctly: `run_work_node` short-circuits already-`done` nodes and
    returns their persisted output, which lands in `outputs` the same as a
    freshly-run one would.

    `needs_worktree=True` (03-C §3.6: "并行写代码的节点必须先申请
    git_worktree_ref") isolates each node's actual file changes in its own
    git worktree, merged back only once the node succeeds -- callers
    writing code (Coder) should pass this; callers only producing document
    content (Planner) don't need it.

    **U-035 (05 §5.8) self-target override**: when `target_folder_path` IS
    graphlemon's own checkout (`paths.is_self_target`), `needs_worktree` is
    forced True regardless of what the caller passed -- the currently-running
    backend process is reading this exact code from disk, so an in-place edit
    to the main checkout (even a serial, non-parallel one) risks corrupting
    state out from under itself mid-execution. This is unconditional (not
    "only when parallel", 03-C §3.6's general rule) precisely because
    self-target has no safe non-worktree case at all.

    `verify_fn(node, output) -> (ok: bool, feedback)`, if given, is called
    right after each node completes, BEFORE moving on to nodes that depend
    on it -- lets a caller intercept a bad output before it propagates
    downstream (e.g. REQ-02's PRD-internal-contradiction check), not just
    after the fact. On `ok=False`: `work_graph_run.status` -> 'reflecting',
    the SAME node is regenerated once with `feedback` appended to its
    prompt, and re-verified. If it fails again: `work_graph_run.status`
    -> 'blocked' and `ConsistencyCheckFailedError` is raised. If it now
    passes: `work_graph_run.status` -> 'verifying' and the regenerated
    output replaces the original in `outputs`.

    **G (2026-08-06, Work Graph parallel execution)**: a multi-node layer's
    failures are collected from every node in that layer (a sibling's
    failure never cancels an already-running sibling -- they're
    independent work, no reason to throw away completed work), then raised
    together as `WorkGraphLayerError` once the whole layer has settled.
    One real limitation, accepted deliberately (see
    `docs/done/work-graph-parallel-execution-design.md` §6): a node
    waiting on a quota-backoff sleep inside `run_work_node` blocks the
    thread pool worker running it, which in turn blocks the layer barrier
    even after every sibling has finished -- Python threads have no clean
    preemptible cancel, and building one is out of scope for this pass.

    Returns {node_id: output_payload}.
    """
    ensure_placeholder_skill(conn, skill_id, org_node)
    effective_needs_worktree = needs_worktree or paths.is_self_target(target_folder_path)
    layers = topological_layers(graph_manifest)
    outputs = {}
    locked_conn = _LockedConn(conn)

    for layer in layers:
        if len(layer) == 1:
            node = layer[0]
            outputs[node["node_id"]] = _run_one_node(
                conn, run_id, org_node, skill_id, target_folder_path, node, outputs,
                build_prompt, input_schema, output_schema, timeout,
                effective_needs_worktree, verify_fn, sleep, now,
            )
            continue

        worker_count = min(max_workers, len(layer))
        errors = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=worker_count) as pool:
            futures = {
                pool.submit(
                    _run_one_node, locked_conn, run_id, org_node, skill_id, target_folder_path,
                    node, outputs, build_prompt, input_schema, output_schema, timeout,
                    effective_needs_worktree, verify_fn, sleep, now,
                ): node
                for node in layer
            }
            done, _ = concurrent.futures.wait(futures, return_when=concurrent.futures.ALL_COMPLETED)
            for future in done:
                node = futures[future]
                try:
                    outputs[node["node_id"]] = future.result()
                except Exception as e:
                    errors.append((node["node_id"], e))

        if errors:
            raise WorkGraphLayerError(errors)

    return outputs


def _run_one_node(
    conn, run_id, org_node, skill_id, target_folder_path, node, outputs,
    build_prompt, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
):
    """One node's full lifecycle (row creation, prompt, run, optional
    verify+regenerate) -- shared by both the synchronous single-node-layer
    path and the ThreadPoolExecutor multi-node-layer path in run_graph(),
    so there's exactly one implementation of "run a node" to maintain."""
    node_id = node["node_id"]
    node_exec_id = _ensure_node_row(
        conn, run_id, node_id, org_node, skill_id,
        "work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
        {"node_id": node_id},
    )
    prompt = build_prompt(node, outputs)
    run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout)
    output = work_node_run.run_work_node(
        conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn,
        needs_worktree=needs_worktree, repo_path=target_folder_path, sleep=sleep, now=now,
    )
    if verify_fn is not None:
        output = _verify_and_maybe_regenerate(
            conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
            output, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
        )
    return output


def _verify_and_maybe_regenerate(
    conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
    output, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
):
    ok, feedback = verify_fn(node, output)
    if ok:
        return output

    conn.execute("UPDATE work_graph_run SET status = 'reflecting' WHERE run_id = ?", (run_id,))
    conn.commit()

    reflect_node_id = node_id + "-reflect"
    reflect_node_exec_id = _ensure_node_row(
        conn, run_id, reflect_node_id, org_node, skill_id,
        "work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
        {"node_id": node_id, "feedback": feedback},
    )
    reflect_prompt = (
        prompt + "\n\nA consistency reviewer flagged issues with your previous draft: "
        + "; ".join(feedback or []) + "\nProduce a corrected version resolving these issues."
    )
    reflect_run_fn = _make_run_fn(
        conn, target_folder_path, reflect_node_exec_id, org_node, reflect_prompt, output_schema, timeout
    )
    regenerated = work_node_run.run_work_node(
        conn, reflect_node_exec_id, {"node_id": node_id}, input_schema, output_schema, reflect_run_fn,
        needs_worktree=needs_worktree, repo_path=target_folder_path,
        sleep=sleep, now=now,
    )

    ok2, feedback2 = verify_fn(node, regenerated)
    if not ok2:
        conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
        conn.commit()
        raise ConsistencyCheckFailedError(
            "node {} still failed verification after one regeneration attempt: {}".format(node_id, feedback2)
        )

    conn.execute("UPDATE work_graph_run SET status = 'verifying' WHERE run_id = ?", (run_id,))
    conn.commit()
    return regenerated