宜博 / lemongraph

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

    `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 _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


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"])
            script_name = "graphlemon-judgment-" + uuid.uuid4().hex
            script_content = (
                "export const meta = {{ name: {name}, description: 'graphlemon node judgment' }}\n\n"
                "const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema} }})\n\n"
                "return result\n"
            ).format(
                name=json.dumps(script_name), prompt=json.dumps(prompt),
                org_node_json=json.dumps(org_node), schema=json.dumps(_JUDGMENT_SCHEMA),
            )
            work_graph_execute.write_workflow_script(cwd, script_name, script_content)
            result = work_graph_execute.execute_work_graph(cwd, script_name, {}, _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")
                )
            )


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

    `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)
            if attempt_number > limit:
                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={}".format(
                        sig.source_node_id, sig.target_node_id, 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)."""
    node_id = node["node_id"]
    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, 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) + _loop_history_prompt_suffix(conn, run_id, node_id)
    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,
    )

    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, 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, org_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,
        )
    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