"""Bridges `plan_work_graph()`'s `graph_manifest` to real per-node execution
via `org_node_trigger.trigger_org_node_structured()` + `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 blocking
`trigger_org_node_structured()` call, one per node, not one call amortized
across 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 a whole-graph Dynamic Workflow script 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.

**2026-08 (work-graph-node-execution-deblackbox-requirements.md)**: this
module used to render each node's call as a one-off Dynamic Workflow `.js`
script (`work_graph_execute.write_workflow_script` + `.execute_work_graph`,
`claude -p 'Run /<name>...'`) purely to get "agent identity + schema-shaped
output" -- 05 §5.4.1.1's finding that `--json-schema` silently drops
`structured_output` when combined with `--agent` was, at the time (U-031,
2026-08-01), only known to be worked around inside a Workflow script's
`agent(prompt, {agentType, schema})` call. `work_graph_plan.py::
plan_work_graph()` had already been doing the same "agent identity + schema
output" job with a flat `claude --agent` call + ```json-fence extraction the
whole time, in production, without ever needing Workflow -- so the
Workflow indirection was never actually load-bearing here either, just
unnoticed. `org_node_trigger.trigger_org_node_structured()` generalizes
that flat-call mechanism for any org_node/output_schema/model combination;
`work_graph_execute.py`'s `write_workflow_script`/`execute_work_graph` are
kept (not deleted) as a historical mechanism with no new callers -- see
that module's own docstring.

Validated for real (2026-08-01, one-node Dynamic Workflow script;
re-validated 2026-08, flat `trigger_org_node_structured` call -- see
src/test_org_node_trigger_e2e.py): a custom `agentType`/`--agent` +
schema-extraction genuinely loads the target org_node's migrated persona
(confirmed via a real call asking the agent to self-identify) and returns
a real schema-shaped payload -- 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
`trigger_org_node_structured()` 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 (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 os
import threading
import time
import uuid
from collections import Counter

from src import (
    bug_record, execution_log, org_node_trigger, paths, settings_config, skill_query, skill_update,
    work_node_run,
)


def _log_event(conn, node_exec_id, event_type, **payload):
    # ensure_ascii=False (2026-08-13, user-reported): this is the ONLY
    # json.dumps() call in the codebase whose output is displayed as raw
    # TEXT (v_log_viewer reads the .jsonl file's bytes back and shows each
    # line as-is, never JSON.parse()s it back into structured data) --
    # every other json.dumps() call here stores data that's always
    # JSON.parse()d/json.loads()d again before display, so the default
    # \uXXXX escaping round-trips invisibly there. Here it doesn't: real
    # Chinese (or any non-ASCII) content in `payload` showed up as literal
    # \uXXXX escape sequences in the console instead of the actual text.
    execution_log.append_log(
        conn, node_exec_id, json.dumps({"event": event_type, **payload}, ensure_ascii=False) + "\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.

    `node_id` (2026-08-07): the rejected node's id, so a caller salvaging
    OTHER nodes' already-persisted artifacts after a partial run_graph()
    failure (see e2e_pipeline.py's `_write_completed_node_artifacts`) can
    tell this one apart from a node that's `work_node_execution.status
    ='done'` for a genuinely different reason -- 'done' there only means
    "passed schema validation", not "passed verify_fn". The original AND
    its "-reflect" retry both end up schema-valid-and-'done' even when
    THIS error fires (verify_fn rejected both), so status alone can't
    distinguish a verify_fn-rejected node from a genuinely good one."""

    def __init__(self, message, node_id=None):
        super().__init__(message)
        self.node_id = node_id


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 _LoopBackRequested(Exception):
    """Internal signal (2026-08-08, Work Graph self-planning + loop-back):
    a node's on_fail.action='loop_back' fired -- run_graph()'s main loop
    catches this and rewinds to source_node_id's target layer, it's not a
    real failure. Raised as an exception (not returned) because it can
    originate inside a ThreadPoolExecutor worker thread (a multi-node
    layer) -- an exception is the only clean way to carry this out through
    future.result()."""

    def __init__(self, source_node_id, target_node_id, feedback):
        self.source_node_id = source_node_id
        self.target_node_id = target_node_id
        self.feedback = feedback
        super().__init__("node {} requests loop_back to {}".format(source_node_id, target_node_id))


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.

    2026-08-08 (Work Graph self-planning + loop-back): edges with
    `edge_type == "on_failure"` are control-flow (a node's failure path,
    see `_on_failure_target`), not data dependencies -- they're excluded
    here. An `on_failure` edge commonly points from a node back to one of
    its own upstream dependencies, which the OLD unfiltered merge would
    have folded in as an (incorrect, reversed) data dependency, creating a
    spurious cycle that would have made this function reject the graph
    before any node ever ran.
    """
    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", []):
        if edge.get("edge_type") == "on_failure":
            continue
        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 _transitive_dependents(graph_manifest, node_id):
    """node_id itself + every node that (directly or transitively) depends
    on it, via `depends_on`/normal data-dependency edges (`on_failure`
    edges excluded, same reasoning as topological_layers()). Used when a
    loop_back fires: node_id's output is about to be regenerated, so
    everything downstream of it was computed from what's now stale input
    and must be re-run too (2026-08-08, Work Graph self-planning +
    loop-back design §4.2)."""
    dependents = {n["node_id"]: [] for n in graph_manifest["nodes"]}
    deps = {n["node_id"]: set(n.get("depends_on") or []) for n in graph_manifest["nodes"]}
    for edge in graph_manifest.get("edges", []):
        if edge.get("edge_type") == "on_failure":
            continue
        to_id, from_id = edge.get("to"), edge.get("from")
        if to_id in deps and from_id in deps:
            deps[to_id].add(from_id)
    for node_id_, dep_set in deps.items():
        for dep in dep_set:
            dependents.setdefault(dep, []).append(node_id_)

    affected, stack = set(), [node_id]
    while stack:
        current = stack.pop()
        if current in affected:
            continue
        affected.add(current)
        stack.extend(dependents.get(current, []))
    return affected


def _transitive_ancestors(graph_manifest, node_id):
    """Reverse of _transitive_dependents: node_id itself + every node it
    (directly or transitively) depends on. Used to validate that a
    declared on_failure edge's target is a genuine upstream influence, not
    an arbitrary node (work_graph_plan.py's _validate_loop_back_targets)."""
    deps = {n["node_id"]: set(n.get("depends_on") or []) for n in graph_manifest["nodes"]}
    for edge in graph_manifest.get("edges", []):
        if edge.get("edge_type") == "on_failure":
            continue
        to_id, from_id = edge.get("to"), edge.get("from")
        if to_id in deps and from_id in deps:
            deps[to_id].add(from_id)

    ancestors, stack = set(), [node_id]
    while stack:
        current = stack.pop()
        if current in ancestors:
            continue
        ancestors.add(current)
        stack.extend(deps.get(current, []))
    return ancestors


def _on_failure_target(graph_manifest, node_id):
    """The node_id that `node_id`'s on_fail.action='loop_back' should
    return to -- read from the single `{"from": node_id, "to": ...,
    "edge_type": "on_failure"}` edge, not a separate node field (2026-08-08
    design revision: the failure path is expressed the same way the
    success path already is, via `edges`). Callers should have already
    validated (work_graph_plan.py's _validate_loop_back_targets) that
    exactly one such edge exists before relying on this at runtime."""
    for edge in graph_manifest.get("edges", []):
        if edge.get("from") == node_id and edge.get("edge_type") == "on_failure":
            return edge["to"]
    raise ValueError("node {} declares on_fail.action='loop_back' but has no on_failure edge".format(node_id))


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 _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout):
    def run_fn(cwd, model=None):
        _log_event(conn, node_exec_id, "node_start", org_node=org_node, model=model)
        try:
            output = org_node_trigger.trigger_org_node_structured(
                org_node, prompt, cwd, output_schema, model=model, timeout=timeout,
            )
        except org_node_trigger.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


def _coerce_bool(value):
    """Same fix as e2e_pipeline.py's own _coerce_bool (2026-08-02): a real
    call has returned e.g. "false" as a JSON STRING despite "type":
    "boolean" in the schema, and `if "false":` is truthy in Python. Kept as
    its own small copy here rather than importing e2e_pipeline (which
    imports this module -- importing back would be circular) or promoting
    it to a shared utils module for a 3-line helper."""
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() == "true"
    return bool(value)


_JUDGMENT_SCHEMA = {
    "type": "object",
    "required": ["ok", "issues"],
    "properties": {"ok": {"type": "boolean"}, "issues": {"type": "array", "items": {"type": "string"}}},
}


def _llm_judgment_verify_fn(judgment, org_node, target_folder_path, timeout=180):
    """judgment.kind == 'llm': generalizes e2e_pipeline.py's
    _check_prd_consistency -- same self-target worktree isolation (a real
    agent() call with memory:project writes an uncommitted change wherever
    it runs; self-target runs need their own worktree so that doesn't land
    on the main checkout, see that function's own docstring for the
    2026-08-03 incident this guards against), but the judge prompt comes
    from judgment['judge_prompt'] (node-declared, self-contained -- same
    design principle as _GENERIC_PROMPT_TEMPLATE's task_prompt) instead of
    a hardcoded PRD-specific prompt, and the schema is the generic
    {"ok": bool, "issues": [...]} shape instead of PRD's {"consistent": ...}.
    e2e_pipeline.py's own _check_prd_consistency/_prd_consistency_verify_fn
    are untouched and remain the path Validator's fixed manifest uses
    (FR-A8) -- this is a parallel, more general implementation, not a
    replacement. `is_self_target` is checked fresh on each call (not cached
    at closure-creation time) -- matches _check_prd_consistency's own
    behavior, and means a caller building this verify_fn once and reusing
    it always reflects the real current state, not a stale snapshot."""

    def verify_fn(node, output):
        is_self = paths.is_self_target(target_folder_path)
        node_exec_id = "judgment-" + uuid.uuid4().hex[:12]
        worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None
        cwd = worktree_path or target_folder_path
        try:
            prompt = judgment["judge_prompt"].format(content=output["content"])
            result = org_node_trigger.trigger_org_node_structured(org_node, prompt, cwd, _JUDGMENT_SCHEMA, timeout=timeout)
            if worktree_path:
                work_node_run.ensure_worktree_committed(worktree_path, node_exec_id)
                work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift")
                work_node_run.merge_worktree(target_folder_path, node_exec_id)
            ok = _coerce_bool(result["ok"])
            return ok, (result.get("issues") or [])
        finally:
            if worktree_path:
                work_node_run.remove_worktree(target_folder_path, worktree_path)

    return verify_fn


def _resolve_node_verify_fn(node, external_verify_fn, org_node, target_folder_path, mechanical_check_registry):
    """A node's own `judgment` field (if declared) takes over verify_fn
    resolution for that node; a node with no `judgment` (every existing
    graph_manifest today) falls through to whatever the caller passed as
    `verify_fn` -- possibly None, meaning no verification at all -- exactly
    like current behavior (FR-A8)."""
    judgment = node.get("judgment")
    if not judgment or judgment.get("kind", "none") == "none":
        return external_verify_fn
    if judgment["kind"] == "mechanical":
        check_fn = (mechanical_check_registry or {}).get(judgment["check_ref"])
        if check_fn is None:
            raise ValueError("unknown mechanical check_ref: {!r}".format(judgment["check_ref"]))

        def mechanical_verify_fn(node, output, _check_fn=check_fn):
            ok, detail = _check_fn(output["content"])
            return ok, ([detail] if detail else None)

        return mechanical_verify_fn
    if judgment["kind"] == "llm":
        return _llm_judgment_verify_fn(judgment, org_node, target_folder_path)
    raise ValueError("unknown judgment.kind: {!r}".format(judgment["kind"]))


def _validate_mechanical_check_refs(graph_manifest, mechanical_check_registry):
    """Fail fast before running ANY node -- a typo'd check_ref caught only
    when that node is finally reached would waste every token spent on the
    nodes before it."""
    registry = mechanical_check_registry or {}
    for node in graph_manifest["nodes"]:
        judgment = node.get("judgment") or {}
        if judgment.get("kind") == "mechanical" and judgment.get("check_ref") not in registry:
            raise ValueError(
                "node {}'s judgment.check_ref {!r} is not in mechanical_check_registry".format(
                    node["node_id"], judgment.get("check_ref")
                )
            )


_REFLECTION_OUTPUT_SCHEMA = {
    "type": "object", "required": ["target_node_id", "defect_kind", "detail"],
    "properties": {
        "target_node_id": {"type": "string"},
        "defect_kind": {"type": "string", "enum": ["content_error", "skill_defect"]},
        "detail": {"type": "string"},
    },
}

_REFLECTION_FRAMINGS = {
    "content": (
        "CONTENT PERSPECTIVE: for each candidate node below, look at what it was asked to produce "
        "and what it actually produced. Which ONE candidate's output itself is genuinely defective? "
        "Then decide: is this a one-off mistake for this specific run ('content_error' -- the "
        "underlying methodology is fine, this particular attempt just didn't execute it well), or "
        "does it look like a systematic pattern that would recur on other tasks using the same "
        "approach/prompt style ('skill_defect')? A concrete red flag for skill_defect: the node's "
        "content reads as a narrative description of an action taken (e.g. 'I launched/produced/"
        "completed X') instead of the structured content itself -- this is a known recurring failure "
        "mode in nodes that explore with tools before producing a final answer, not a one-off typo."
    ),
    "connectivity": (
        "CONNECTIVITY PERSPECTIVE: if no single candidate's content is obviously wrong on its own, "
        "is the real problem how the candidates connect -- e.g. a node that should have depended on "
        "another candidate's output but didn't, so it worked from stale or missing context? This "
        "kind of problem is almost always 'content_error' (a one-off planning slip for this run), not "
        "a systemic Skill defect -- pick whichever candidate's isolation from its true dependency is "
        "the root issue."
    ),
    "history": (
        "HISTORY PERSPECTIVE: you are given this run's actual loop_back/skill-iteration event history "
        "for each candidate. If a candidate has already been sent back for revision or had its Skill "
        "rewritten before in this same run for a SIMILAR reason, that recurrence itself is strong "
        "evidence of 'skill_defect' -- a Skill that keeps producing the same class of failure even "
        "after a plain retry has methodology problems, not bad luck. If a candidate has no such prior "
        "history, default to 'content_error' unless the content itself (see the CONTENT framing's red "
        "flag) makes skill_defect obvious even on a first occurrence."
    ),
}


def _candidate_context_block(graph_manifest, build_prompt, outputs, candidate_node_ids):
    nodes_by_id = {n["node_id"]: n for n in graph_manifest["nodes"]}
    parts = []
    for node_id in candidate_node_ids:
        node = nodes_by_id[node_id]
        asked = build_prompt(node, outputs)
        produced = (outputs.get(node_id) or {}).get("content", "")
        parts.append(
            "--- candidate node_id={} ---\nWhat it was asked:\n{}\n\nWhat it produced:\n{}".format(
                node_id, asked, produced
            )
        )
    return "\n\n".join(parts)


def _candidate_history_block(conn, run_id, candidate_node_ids):
    events = _loop_back_events(conn, run_id)
    skill_events = _skill_iterated_events(conn, run_id)
    parts = []
    for node_id in candidate_node_ids:
        prior_loop_backs = [e for e in events if e["target_node_id"] == node_id]
        prior_skill_fixes = [e for e in skill_events if e["node_id"] == node_id]
        parts.append(
            "--- candidate node_id={} --- prior loop_back count: {}; prior skill_iterated count: {}".format(
                node_id, len(prior_loop_backs), len(prior_skill_fixes)
            )
        )
    return "\n".join(parts)


def _reflection_prompt(framing_name, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs, conn, run_id):
    framing = _REFLECTION_FRAMINGS[framing_name]
    context = (
        _candidate_history_block(conn, run_id, candidate_node_ids) if framing_name == "history"
        else _candidate_context_block(graph_manifest, build_prompt, outputs, candidate_node_ids)
    )
    return (
        "{framing}\n\nThe node {failed_node_id} failed judgment with this feedback: {feedback}\n\n"
        "Candidates (pick target_node_id from EXACTLY these node_ids, nothing else):\n{candidates}\n\n"
        "Context:\n{context}\n\n"
        "Reply with target_node_id (one of the candidate node_ids above), defect_kind "
        "('content_error' or 'skill_defect'), and detail (one sentence justifying both)."
    ).format(
        framing=framing, failed_node_id=failed_node_id, feedback="; ".join(feedback or []),
        candidates=", ".join(candidate_node_ids), context=context,
    )


def _run_reflection_perspective(framing_name, prompt, target_folder_path, timeout):
    """One perspective's real call, under the Validator identity (Reflection-
    review is diagnostic, not opinion-laden about the failed content itself
    -- same 'independent second opinion' reasoning Fix-Skill's own 'who
    rewrites the Skill' decision uses, see _fix_skill_for_node). Same
    self-target worktree isolation as _llm_judgment_verify_fn (a real
    agent() call with memory:project writes an uncommitted change wherever
    it runs)."""
    is_self = paths.is_self_target(target_folder_path)
    node_exec_id = "reflection-" + framing_name + "-" + uuid.uuid4().hex[:12]
    worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None
    cwd = worktree_path or target_folder_path
    try:
        result = org_node_trigger.trigger_org_node_structured(
            "validator", prompt, cwd, _REFLECTION_OUTPUT_SCHEMA, timeout=timeout
        )
        if worktree_path:
            work_node_run.ensure_worktree_committed(worktree_path, node_exec_id)
            work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift")
            work_node_run.merge_worktree(target_folder_path, node_exec_id)
        return {
            "target_node_id": result["target_node_id"],
            "defect_kind": result["defect_kind"],
            "detail": result.get("detail", ""),
        }
    finally:
        if worktree_path:
            work_node_run.remove_worktree(target_folder_path, worktree_path)


def _tally_reflection(judgments, candidate_node_ids):
    """>=2/3 agreement on target_node_id, same threshold root_cause_trace.py's
    _tally_perspectives uses -- not imported from there (would create a
    circular import, root_cause_trace.py already imports this module) and
    small enough to duplicate, same reasoning work_graph_plan.py's own
    _transitive_ancestors duplication already documents. defect_kind is
    taken from whichever judgment(s) picked the winning target_node_id
    (majority among just those, defaulting to 'content_error' on a tie --
    the conservative choice per §5.7 point 2). <2/3 agreement on
    target_node_id degrades to the nearest direct ancestor + 'content_error'
    rather than escalating to a human -- this is one retry inside a single
    Work Graph run, not the terminal cross-Org-Agent judgment
    root_cause_trace.py's InconclusiveEscalateError guards."""
    target_counts = Counter(j["target_node_id"] for j in judgments)
    winner, count = target_counts.most_common(1)[0]
    if count * 2 < len(judgments) or winner not in candidate_node_ids:
        return {"target_node_id": candidate_node_ids[0], "defect_kind": "content_error", "detail": "no >=2/3 agreement"}
    kind_counts = Counter(j["defect_kind"] for j in judgments if j["target_node_id"] == winner)
    defect_kind = "skill_defect" if kind_counts.get("skill_defect", 0) > kind_counts.get("content_error", 0) else "content_error"
    detail = " | ".join(j["detail"] for j in judgments if j["target_node_id"] == winner)
    return {"target_node_id": winner, "defect_kind": defect_kind, "detail": detail}


def _judge_root_cause_node(
    conn, run_id, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs,
    target_folder_path, timeout=180,
):
    """Reflection-review (org-graph-unification-and-universal-self-planning-
    requirements.md §3.1 step 2): three independent real perspectives --
    content, connectivity, history -- each picks a target_node_id from
    candidate_node_ids and a defect_kind, run concurrently (own real claude
    call, own worktree if self-target). Majority vote via _tally_reflection.
    candidate_node_ids must be non-empty (the caller, _run_one_node's
    dynamic_loop_back branch, only calls this when the failing node has at
    least one real ancestor -- a terminal judgment node with zero ancestors
    has nothing to loop back to and shouldn't declare dynamic_loop_back)."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        futures = [
            pool.submit(
                _run_reflection_perspective, name,
                _reflection_prompt(name, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs, conn, run_id),
                target_folder_path, timeout,
            )
            for name in _REFLECTION_FRAMINGS
        ]
        judgments = [f.result() for f in futures]
    return _tally_reflection(judgments, candidate_node_ids)


_SKILL_REWRITE_SCHEMA = {
    "type": "object", "required": ["new_content", "change_summary"],
    "properties": {"new_content": {"type": "string"}, "change_summary": {"type": "string"}},
}

_SKILL_REWRITE_PROMPT = (
    "You are revising a Skill file (the reusable methodology/prompt guidance a graphlemon Org Agent "
    "follows) because it produced a real, repeated failure. Your job is surgical: fix the SPECIFIC "
    "gap that caused this failure, without breaking the Skill's existing correct guidance for other "
    "cases it already handles well.\n\n"
    "Goal this task is working on: {goal_text}\n\n"
    "Skill being revised (skill_id={skill_id}):\n{current_content}\n\n"
    "What the node using this Skill was asked to do:\n{what_was_asked}\n\n"
    "What it actually produced (the defective output):\n{what_it_produced}\n\n"
    "Why this was judged a failure: {failure_feedback}\n\n"
    "This node's prior rejection history in this same run (if any): {loop_history}\n\n"
    "Reply with new_content (the COMPLETE revised Skill file content -- not a diff, not a partial "
    "excerpt) and change_summary (one sentence describing what changed and why)."
)


def _read_skill_content(skill_id):
    """Skill library content lives in graphlemon's own repo checkout
    (skill_query._REPO_ROOT), never in target_folder_path -- same
    established boundary c_skill_diff/c_skill_rollback's HTTP routes
    already rely on (05 §5.4.1.4), reused here rather than re-derived."""
    path = os.path.join(skill_query._REPO_ROOT, "skills", skill_id, "SKILL.md")
    with open(path, encoding="utf-8") as f:
        return f.read()


def _log_skill_iterated_event(conn, run_id, node_id, skill_id, iteration_id):
    """Persists a 'skill_iterated' execution_event -- this CHECK-enum value
    has existed since the original U-001 schema but was never actually
    written by any code path before Fix-Skill (skill_update.update_skill()
    itself doesn't log one; it's this caller's job, same division of
    responsibility loop_back_triggered already has between run_graph() and
    _record_loop_back_event). No schema migration needed."""
    row = conn.execute(
        "SELECT wgr.task_id, t.project_id FROM work_graph_run wgr "
        "JOIN task t ON wgr.task_id = t.task_id WHERE wgr.run_id = ?", (run_id,)
    ).fetchone()
    task_id, project_id = row
    conn.execute(
        "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) "
        "VALUES (?, ?, ?, ?, 'skill_iterated', ?)",
        (str(uuid.uuid4()), project_id, task_id, run_id, json.dumps({
            "node_id": node_id, "skill_id": skill_id, "iteration_id": iteration_id,
        })),
    )
    conn.commit()


def _skill_iterated_events(conn, run_id):
    rows = conn.execute(
        "SELECT payload FROM execution_event WHERE run_id = ? AND event_type = 'skill_iterated' "
        "ORDER BY created_at", (run_id,)
    ).fetchall()
    return [json.loads(p) for (p,) in rows]


_MAX_SKILL_FIXES_PER_RUN_PER_SKILL = 2  # §5.7 point 5: cap real Git rewrites of the same Skill within one run


def _fix_skill_for_node(conn, run_id, target_node_id, graph_manifest, build_prompt, outputs, feedback, target_folder_path, timeout=300):
    """Fix-Skill (org-graph-unification-and-universal-self-planning-
    requirements.md §3.1 step 3): a real, Git-versioned Skill rewrite, not a
    per-run prompt patch -- calls the existing but (until now) never
    actually invoked by production code skill_update.py::update_skill()
    (see design doc §5's 'existing fact B': org_level_loop.py's
    _ROUTE_FOR_CATEGORY has no 'skill_defect' key, so this capability has
    sat unused since it was built). Always runs under the Validator
    identity (user-confirmed 2026-08-12 via AskUserQuestion): letting a
    node's own org_node rewrite its own Skill risks reproducing the exact
    blind spot that caused the failure.

    §5.7 point 5 rate limit: if this (run_id, skill_id) pair has already
    been rewritten _MAX_SKILL_FIXES_PER_RUN_PER_SKILL times this run, skip
    the rewrite silently (the caller's dynamic_loop_back still proceeds as
    a plain re-run of target_node_id -- not touching the Skill again is a
    safe degradation, not a failure)."""
    target_node = next(n for n in graph_manifest["nodes"] if n["node_id"] == target_node_id)
    skill_id = target_node.get("skill_ref")
    if not skill_id:
        return None  # nothing declared for this node to attribute the fix to -- see caller's docstring

    prior_fixes = sum(1 for e in _skill_iterated_events(conn, run_id) if e["skill_id"] == skill_id)
    if prior_fixes >= _MAX_SKILL_FIXES_PER_RUN_PER_SKILL:
        return None

    row = conn.execute("SELECT task_id FROM work_graph_run WHERE run_id = ?", (run_id,)).fetchone()
    task_id = row[0]
    goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
    current_content = _read_skill_content(skill_id)

    prompt = _SKILL_REWRITE_PROMPT.format(
        goal_text=goal_text, skill_id=skill_id, current_content=current_content,
        what_was_asked=build_prompt(target_node, outputs),
        what_it_produced=(outputs.get(target_node_id) or {}).get("content", ""),
        failure_feedback="; ".join(feedback or []),
        loop_history=_loop_history_prompt_suffix(conn, run_id, target_node_id) or "(none)",
    )

    is_self = paths.is_self_target(target_folder_path)
    node_exec_id = "fixskill-" + uuid.uuid4().hex[:12]
    worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None
    cwd = worktree_path or target_folder_path
    try:
        rewrite_result = org_node_trigger.trigger_org_node_structured(
            "validator", prompt, cwd, _SKILL_REWRITE_SCHEMA, timeout=timeout
        )
        if worktree_path:
            work_node_run.ensure_worktree_committed(worktree_path, node_exec_id)
            work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift")
            work_node_run.merge_worktree(target_folder_path, node_exec_id)
    finally:
        if worktree_path:
            work_node_run.remove_worktree(target_folder_path, worktree_path)

    # skill_query._REPO_ROOT (graphlemon's own checkout), NOT target_folder_path
    # -- the Skill library is graphlemon's own platform asset (05 §5.4.1.4
    # precedent), independent of which target project this task is working on.
    bug_result = bug_record.create_bug_record(
        conn, task_id=task_id,
        description="Skill {} produced a repeated defective output at node {}".format(skill_id, target_node_id),
        root_cause_category="skill_defect", root_cause_detail=rewrite_result["change_summary"],
        suggested_route="full_replan",
    )
    iteration = skill_update.update_skill(
        conn, skill_id, bug_result["bug_id"], rewrite_result["new_content"],
        rewrite_result["change_summary"], repo_path=skill_query._REPO_ROOT,
    )
    _log_skill_iterated_event(conn, run_id, target_node_id, skill_id, iteration["iteration_id"])
    return iteration


_DEFAULT_LOOP_LIMIT = 3  # only reached if even the "default" config_id row is unseeded


def _resolve_loop_limit(conn, org_node, graph_manifest, source_node_id):
    """How many times `source_node_id` (the node whose judgment failed and
    declared on_fail.action in {'retry_self', 'loop_back'}) may trigger a
    loop before run_graph() gives up. Lookup order (2026-08-08 design §6):
    the failing node's own on_fail.max_attempts, then a settings_config.py
    row keyed by "{org_node}:{node_type}", then "{org_node}:*", then
    "default", then a hardcoded fallback."""
    source_node = next(n for n in graph_manifest["nodes"] if n["node_id"] == source_node_id)
    node_level = (source_node.get("on_fail") or {}).get("max_attempts")
    if node_level is not None:
        return node_level
    node_type = source_node.get("artifact_type", source_node_id)
    for config_id in ("{}:{}".format(org_node, node_type), "{}:*".format(org_node), "default"):
        config = settings_config.get_loop_limit_config(conn, config_id)
        if config:
            return config["max_loop_attempts"]
    return _DEFAULT_LOOP_LIMIT


def _resolve_total_loop_limit(conn, org_node, source_node_id):
    """Phase 2 §5.7 point 3: total dynamic_loop_back attempts for
    source_node_id, regardless of which target each attempt picked --
    guards against dynamic_loop_back dodging _resolve_loop_limit's
    per-(target,source) cap by picking a different target each round.
    Lookup: a settings_config.py row keyed "{org_node}:{source_node_id}:total"
    (config_id has no format constraint, this key shape needs no schema
    change), then "default" (shared with _resolve_loop_limit's own default
    row -- deliberately the SAME fallback chain/value, so a source_node_id
    that never uses dynamic_loop_back sees an unrestrictive default here)."""
    for config_id in ("{}:{}:total".format(org_node, source_node_id), "default"):
        config = settings_config.get_loop_limit_config(conn, config_id)
        if config:
            return config["max_loop_attempts"]
    return _DEFAULT_LOOP_LIMIT


def _record_loop_back_event(conn, run_id, sig, attempt_number):
    """Persists one 'loop_back_triggered' execution_event row -- this IS
    the loop history (2026-08-08 design §7): no separate table, so a
    node's full "sent back N times, here's why each time" story is
    reconstructed purely by querying execution_event."""
    row = conn.execute(
        "SELECT wgr.task_id, t.project_id FROM work_graph_run wgr "
        "JOIN task t ON wgr.task_id = t.task_id WHERE wgr.run_id = ?", (run_id,)
    ).fetchone()
    task_id, project_id = row
    conn.execute(
        "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) "
        "VALUES (?, ?, ?, ?, 'loop_back_triggered', ?)",
        (str(uuid.uuid4()), project_id, task_id, run_id, json.dumps({
            "source_node_id": sig.source_node_id, "target_node_id": sig.target_node_id,
            "attempt_number": attempt_number, "feedback": sig.feedback,
        })),
    )
    conn.commit()


def _loop_back_events(conn, run_id):
    rows = conn.execute(
        "SELECT payload FROM execution_event WHERE run_id = ? AND event_type = 'loop_back_triggered' "
        "ORDER BY created_at", (run_id,)
    ).fetchall()
    return [json.loads(p) for (p,) in rows]


def _loop_attempt_count_from_events(conn, run_id, target_node_id, source_node_id):
    """Real count of prior loop_back triggers for this exact (target,
    source) pair, read from execution_event -- not an in-memory counter,
    so a process restart mid-loop resumes with the true count (REQ-06)."""
    return sum(
        1 for e in _loop_back_events(conn, run_id)
        if e["target_node_id"] == target_node_id and e["source_node_id"] == source_node_id
    )


def _loop_attempt_count_from_events_total(conn, run_id, source_node_id):
    """Real count of every prior loop_back triggered by source_node_id,
    regardless of target -- see _resolve_total_loop_limit's own docstring
    for why this differs from _loop_attempt_count_from_events."""
    return sum(1 for e in _loop_back_events(conn, run_id) if e["source_node_id"] == source_node_id)


def _loop_history_prompt_suffix(conn, run_id, node_id):
    """When node_id is about to be (re-)executed, append the full history
    of every prior time it was sent back for revision (not just the most
    recent) -- empty string if node_id was never a loop_back target."""
    history = [e for e in _loop_back_events(conn, run_id) if e["target_node_id"] == node_id]
    if not history:
        return ""
    lines = [
        "Attempt {}: rejected because: {}".format(h["attempt_number"], "; ".join(h["feedback"] or []))
        for h in history
    ]
    return "\n\nThis node was sent back for revision before, with this history:\n" + "\n".join(lines)


def _next_exec_node_id(conn, run_id, node_id, graph_manifest):
    """The exec node_id to use for node_id's NEXT real execution. A node
    that has never had its cached output invalidated by a loop_back (the
    overwhelming common case, and every node in a graph_manifest with no
    judgment/on_fail at all) always reuses the plain node_id -- preserving
    run_work_node's normal breakpoint-resume short-circuit (a 'done' row
    for node_id returns its persisted output instead of re-running)
    completely unchanged, including across a process restart on a run
    that never looped at all.

    2026-08-08 (found via a real test failure, not just theory): counting
    only "was node_id EVER the direct loop_back target" undercounts --
    every node in `_transitive_dependents(target)` gets its `outputs`
    entry popped and genuinely re-executed too (run_graph()'s own
    control flow), not just the target itself. A downstream node that's
    never itself a `target_node_id` but IS swept up by some earlier
    node's loop_back still needs its own fresh '-attemptN' row each time,
    or it silently short-circuits to its own stale pre-invalidation
    output on the second pass. So this counts every loop_back event
    whose target's `_transitive_dependents` set includes node_id, not
    just events where node_id itself is the literal target."""
    prior_invalidations = sum(
        1 for e in _loop_back_events(conn, run_id)
        if node_id in _transitive_dependents(graph_manifest, e["target_node_id"])
    )
    if prior_invalidations == 0:
        return node_id
    return "{}-attempt{}".format(node_id, prior_invalidations + 1)


_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,
    mechanical_check_registry=None,
):
    """Runs every node in `graph_manifest` through `run_work_node()`, each
    backed by a real `trigger_org_node_structured()` 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.

    `mechanical_check_registry` (2026-08-08, Work Graph self-planning +
    loop-back): {check_ref: callable(content) -> (ok, detail)}, consulted
    by any node declaring `judgment: {"kind": "mechanical", "check_ref":
    ...}` -- see `_resolve_node_verify_fn`. A node with no `judgment` field
    (every graph_manifest that predates this) ignores this entirely and
    keeps using whatever `verify_fn` the caller passed, unchanged (FR-A8).

    **loop_back (2026-08-08, Work Graph self-planning + loop-back)**: a
    node declaring its own `judgment` can also declare `on_fail.action`:
    `hard_fail` (fail the whole run immediately, no retry), `retry_self`
    (loop back to itself), or `loop_back` (jump to an earlier node named
    by that node's own `edge_type: "on_failure"` edge -- see
    `_on_failure_target`). `retry_self`/`loop_back` share one mechanism:
    the target node (and everything transitively depending on it, since
    their input is now stale) is re-executed with the full accumulated
    history of every prior rejection appended to its prompt
    (`_loop_history_prompt_suffix`), bounded per (target, source) pair by
    `_resolve_loop_limit` -- exceeding it raises `ConsistencyCheckFailedError`
    same as the legacy single-retry path. A node with NO `judgment` field
    (every graph_manifest that predates this) is entirely unaffected --
    it keeps going through `_verify_and_maybe_regenerate` exactly as
    before, regardless of any of this (FR-A8).

    Returns {node_id: output_payload}.
    """
    _validate_mechanical_check_refs(graph_manifest, mechanical_check_registry)
    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)
    node_layer_index = {node["node_id"]: i for i, layer in enumerate(layers) for node in layer}
    outputs = {}
    locked_conn = _LockedConn(conn)

    try:
        return _run_layers(
            conn, run_id, org_node, skill_id, target_folder_path, layers, node_layer_index, outputs,
            graph_manifest, build_prompt, input_schema, output_schema, timeout, effective_needs_worktree,
            verify_fn, sleep, now, max_workers, mechanical_check_registry, locked_conn,
        )
    except Exception:
        # 2026-08-08 (found via a real self-planned run): only three specific,
        # anticipated failure paths (loop_back exceeding its limit,
        # on_fail.action='hard_fail', the legacy _verify_and_maybe_regenerate
        # double-failure) ever mark work_graph_run terminal. Any OTHER
        # exception -- e.g. a bare KeyError from a caller's build_prompt(),
        # which is exactly what a real run hit -- used to propagate straight
        # out of run_graph() leaving the row stuck at its non-terminal status
        # forever, permanently blocking every future attempt for this
        # (task_id, org_node) via ux_work_graph_run_active's UNIQUE index
        # (schema.py) -- the retry itself would fail immediately on the
        # INSERT, with no clean error explaining why. This is a catch-all
        # safety net, not a substitute for the three specific handlers above
        # (which run first and set more specific end-states along the way);
        # the WHERE clause is a no-op if one of them (or a nested call) had
        # already marked this run terminal.
        conn.execute(
            "UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ? "
            "AND status IN ('planning', 'running', 'merging', 'verifying', 'reflecting')",
            (run_id,),
        )
        conn.commit()
        raise


def _run_layers(
    conn, run_id, org_node, skill_id, target_folder_path, layers, node_layer_index, outputs,
    graph_manifest, build_prompt, input_schema, output_schema, timeout, effective_needs_worktree,
    verify_fn, sleep, now, max_workers, mechanical_check_registry, locked_conn,
):
    """run_graph()'s actual node-by-node/layer-by-layer execution loop --
    split out so run_graph() itself can wrap the whole thing in one
    try/except (see its own comment) without deeply nesting this loop."""
    layer_index = 0
    while layer_index < len(layers):
        pending = [node for node in layers[layer_index] if node["node_id"] not in outputs]
        if not pending:
            layer_index += 1
            continue

        loop_back_signal = None
        if len(pending) == 1:
            node = pending[0]
            try:
                outputs[node["node_id"]] = _run_one_node(
                    conn, run_id, org_node, skill_id, target_folder_path, node, outputs, graph_manifest,
                    build_prompt, input_schema, output_schema, timeout,
                    effective_needs_worktree, verify_fn, sleep, now, mechanical_check_registry,
                )
            except _LoopBackRequested as sig:
                loop_back_signal = sig
        else:
            worker_count = min(max_workers, len(pending))
            errors = []
            loop_back_signals = []
            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, graph_manifest, build_prompt, input_schema, output_schema, timeout,
                        effective_needs_worktree, verify_fn, sleep, now, mechanical_check_registry,
                    ): node
                    for node in pending
                }
                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 _LoopBackRequested as sig:
                        loop_back_signals.append(sig)
                    except Exception as e:
                        errors.append((node["node_id"], e))

            if loop_back_signals:
                # A layer can have at most one *winning* rewind per pass --
                # pick deterministically (lowest source_node_id). Any plain
                # `errors` from OTHER siblings in this same pass are
                # dropped, not raised: they'll surface again naturally on
                # the next pass if still genuinely wrong after the rewind
                # (same "accept a known, documented limitation rather than
                # build a bigger coordinator" philosophy as the quota-wait
                # limitation already documented above for G).
                loop_back_signal = sorted(loop_back_signals, key=lambda s: s.source_node_id)[0]
            elif errors:
                raise WorkGraphLayerError(errors)

        if loop_back_signal is not None:
            sig = loop_back_signal
            attempt_number = _loop_attempt_count_from_events(
                conn, run_id, sig.target_node_id, sig.source_node_id
            ) + 1
            limit = _resolve_loop_limit(conn, org_node, graph_manifest, sig.source_node_id)
            # Phase 2 §5.7 point 3: attempt_number/limit above count only
            # THIS (target, source) pair -- dynamic_loop_back can pick a
            # DIFFERENT target each round, which would dodge that per-pair
            # limit indefinitely. total_attempts/total_limit count every
            # loop_back triggered by this source_node_id regardless of
            # target. For retry_self/static loop_back (always the same
            # target every time), total_attempts == attempt_number by
            # construction, so this is a no-op there -- only dynamic_loop_back
            # can actually make them diverge.
            total_attempts = _loop_attempt_count_from_events_total(conn, run_id, sig.source_node_id) + 1
            total_limit = _resolve_total_loop_limit(conn, org_node, sig.source_node_id)
            if attempt_number > limit or total_attempts > total_limit:
                # 2026-08-13 (real incident, task b243dfe8, diagnosed live
                # with the user): this branch used to jump straight to
                # raise, skipping _record_loop_back_event() below entirely
                # -- sig.feedback (the judge's REAL reason THIS FINAL
                # attempt was rejected, the one that actually exhausted the
                # budget) was silently discarded. Nothing in the DB could
                # answer "what was specifically wrong with the last try",
                # only "it happened N times" (this exception's own summary
                # message). The attempt that exhausted the budget is
                # recorded the same way every other rejected attempt already
                # is, before giving up -- so a human (or a future arbiter)
                # diagnosing the block can see the real judgment, not just a
                # count.
                _record_loop_back_event(conn, run_id, sig, attempt_number)
                conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
                conn.commit()
                raise ConsistencyCheckFailedError(
                    "loop_back from {} to {} exceeded max_attempts={} (per-target={}/{}, total={}/{})".format(
                        sig.source_node_id, sig.target_node_id, max(limit, total_limit),
                        attempt_number, limit, total_attempts, total_limit,
                    ),
                    node_id=sig.source_node_id,
                )
            _record_loop_back_event(conn, run_id, sig, attempt_number)
            for affected_id in _transitive_dependents(graph_manifest, sig.target_node_id):
                outputs.pop(affected_id, None)
            layer_index = node_layer_index[sig.target_node_id]
            continue

        layer_index += 1

    return outputs


def _run_one_node(
    conn, run_id, org_node, skill_id, target_folder_path, node, outputs, graph_manifest,
    build_prompt, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
    mechanical_check_registry=None,
):
    """One node's full lifecycle (row creation, prompt, run, optional
    verify+regenerate/loop_back) -- 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.

    2026-08-08: uses `_next_exec_node_id` rather than always the plain
    node_id -- see that function's own docstring for why (a loop_back
    target must never collide with, and short-circuit against, its own
    stale pre-loop_back 'done' row; every other node keeps resuming
    exactly like before).

    2026-08-12 (Phase 2 §4, node-level subagent/skill binding): a node can
    declare its own `subagent_type`/`skill_ref`, overriding the run-level
    `org_node`/`skill_id` defaults this whole function used to use
    unconditionally for every node. A node with neither field (every
    existing graph_manifest -- Planner's fixed chain, Coder's synthetic
    single-node manifest) behaves identically to before: `node_org_node`/
    `node_skill_id` just equal the run-level defaults."""
    node_id = node["node_id"]
    node_org_node = node.get("subagent_type", org_node)
    node_skill_id = node.get("skill_ref", skill_id)
    # run_graph() only ensures the RUN-LEVEL default skill_id exists as a
    # placeholder row (work_node_execution.skill_ref's FK target) -- a
    # node declaring its own skill_ref needs the same guarantee, or the
    # INSERT below fails FOREIGN KEY constraint the first time that
    # specific skill_id is ever seen. Idempotent, cheap to call per node.
    if node_skill_id != skill_id:
        ensure_placeholder_skill(conn, node_skill_id, node_org_node)
    exec_node_id_str = _next_exec_node_id(conn, run_id, node_id, graph_manifest)

    node_exec_id = _ensure_node_row(
        conn, run_id, exec_node_id_str, node_org_node, node_skill_id,
        "work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
        {"node_id": node_id},
    )
    prompt = build_prompt(node, outputs) + _loop_history_prompt_suffix(conn, run_id, node_id)
    run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, node_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,
    )

    judgment = node.get("judgment")
    has_own_judgment = bool(judgment) and judgment.get("kind", "none") != "none"
    resolved_verify_fn = _resolve_node_verify_fn(
        node, verify_fn, node_org_node, target_folder_path, mechanical_check_registry
    )
    if resolved_verify_fn is None:
        return output

    if not has_own_judgment:
        # Legacy path (FR-A8): this node has no judgment of its own --
        # whatever external verify_fn the caller passed behaves exactly as
        # before (_verify_and_maybe_regenerate does its OWN verify_fn
        # call internally -- checking here too would call it twice for
        # every node, once redundantly), regardless of any on_fail field
        # (legacy nodes don't have one).
        return _verify_and_maybe_regenerate(
            conn, run_id, node_org_node, node_skill_id, target_folder_path, node, node_id, prompt,
            output, input_schema, output_schema, timeout, needs_worktree, resolved_verify_fn, sleep, now,
        )

    ok, feedback = resolved_verify_fn(node, output)
    if ok:
        return output

    action = (node.get("on_fail") or {}).get("action", "retry_self")
    if action == "hard_fail":
        conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
        conn.commit()
        raise ConsistencyCheckFailedError(
            "node {} failed judgment with on_fail.action='hard_fail': {}".format(
                node_id, "; ".join(feedback or [])
            ),
            node_id=node_id,
        )
    if action == "dynamic_loop_back":
        # Phase 2 §5: Reflection-review -> [Fix-Skill] -> Re-loop. See
        # _judge_root_cause_node/_fix_skill_for_node docstrings for the
        # full four-step methodology (org-graph-unification-and-universal-
        # self-planning-requirements.md §3.1).
        ancestors = sorted(_transitive_ancestors(graph_manifest, node_id) - {node_id})
        diagnosis = _judge_root_cause_node(
            conn, run_id, node_id, feedback, ancestors, graph_manifest, build_prompt, outputs,
            target_folder_path,
        )
        target = diagnosis["target_node_id"]
        if target not in ancestors:
            raise ValueError(
                "dynamic_loop_back judge picked {!r}, not a real ancestor of {}".format(target, node_id)
            )
        if diagnosis["defect_kind"] == "skill_defect":
            _fix_skill_for_node(
                conn, run_id, target, graph_manifest, build_prompt, outputs, feedback, target_folder_path,
            )
        raise _LoopBackRequested(source_node_id=node_id, target_node_id=target, feedback=feedback)
    target = node_id if action == "retry_self" else _on_failure_target(graph_manifest, node_id)
    raise _LoopBackRequested(source_node_id=node_id, target_node_id=target, feedback=feedback)


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),
            node_id=node_id,
        )

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