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