"""c_work_graph_plan (03-C §3.4): produces a `graph_manifest`, does not execute. U-011 scope: a real `claude --agent -p` call, recorded as a `work_graph_run` row with `status='planning'` (04-P `p_work_graph_execution` start_state). **Empirical finding (2026-07-31)**: `--json-schema` does NOT constrain output when combined with `--agent` -- tried it for real, the response came back as normal prose with an embedded ```json fenced block, `structured_output` was entirely absent from the payload even though the identity itself came through correctly (the response referenced "SOUL 原则 #10" by number, i.e. it was genuinely reasoning as the migrated Planner). So this parses the graph_manifest out of free-form response text instead of trusting `structured_output` -- see `_extract_graph_manifest`. Not documented anywhere; found by actually running it, not assumed. Does NOT execute the resulting plan or produce artifacts -- that's U-012 (`c_work_graph_execute` full)/U-013 (`c_work_node_run`), and nothing today owns `c_artifact_write` either (REQ-02 lists it, no Unit in development-plan.md builds it -- a real gap, flagged rather than silently plugged, since it affects more than just this Unit's scope). REQ-02's happy case (real BRD/PRD/SDD/Benchmark artifacts with lineage) needs all of that; development-plan.md's U-011 Done line claimed it passes here alone, which isn't accurate -- corrected in this session's notes, same pattern as U-009/U-010. """ import json import re import subprocess import uuid GRAPH_MANIFEST_SCHEMA = { "type": "object", "required": ["nodes", "edges"], "properties": { "nodes": { "type": "array", "items": { "type": "object", "required": ["node_id", "artifact_type", "depends_on"], "properties": { "node_id": {"type": "string"}, "artifact_type": { "type": "string", "enum": ["brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"], }, "depends_on": {"type": "array", "items": {"type": "string"}}, # 2026-08-08 (Work Graph self-planning + loop-back) -- both optional, # documentation only (this schema isn't jsonschema-enforced, see # module docstring's --json-schema/--agent note). "judgment": { "type": "object", "properties": { "kind": {"type": "string", "enum": ["none", "mechanical", "llm"]}, "check_ref": {"type": "string"}, "judge_prompt": {"type": "string"}, }, }, "on_fail": { "type": "object", "properties": { "action": {"type": "string", "enum": ["retry_self", "loop_back", "hard_fail"]}, "max_attempts": {"type": "integer"}, }, }, }, }, }, "edges": { "type": "array", "items": { "type": "object", "required": ["from", "to", "edge_type"], "properties": { "from": {"type": "string"}, "to": {"type": "string"}, # "on_failure" (2026-08-08): a node's failure path, expressed the # same way its success path already is -- see _on_failure_target # in work_graph_orchestrate.py. "edge_type": {"type": "string", "enum": ["zero_token", "needs_judgment", "on_failure"]}, }, }, }, }, } # 2026-08-08 (Work Graph self-planning + loop-back, docs/todo/work-graph- # self-planning-and-loop-back-requirements.md, FR-B4 "全部自由规划"): # Planner's doc-writing chain is no longer a fixed architectural constant # (formally reverses docs/done/dynamic-work-graph-design-skill-design.md # §3's decision) -- the LLM decides which node types this goal needs, in # what order, and how each one's judgment/on_fail should work. This # catalog is reference DATA the prompt hands the LLM, not a requirement # that every entry appears -- FR-B2. test_script (disabled 2026-08-07, # docs/todo/validator-test-script-narrative-output-issue.md) is back in # this catalog per that decision's §8 item 7: Part B doesn't wait for the # narrative-output structural fix, that fix is tracked separately # alongside a future Coder-side test_script module. _PRD_JUDGE_PROMPT = ( "Review the following PRD for internally CONTRADICTORY scenario assumptions -- e.g. two " "requirements that cannot both hold at once, or the same scenario described two incompatible " "ways. Do NOT flag mere incompleteness, ambiguity, or style issues, only genuine " "self-contradiction. Reply via the required schema: ok=false only if you found a real " "contradiction, and list each one in issues.\n\nPRD:\n\n{content}" ) _PLANNER_NODE_CATALOG = [ {"artifact_type": "brd", "description": "Business Requirements Document -- why this goal matters, who it's for."}, { "artifact_type": "prd", "description": "Product Requirements Document -- concrete scenarios/behavior.", "suggested_judgment": {"kind": "llm", "judge_prompt": _PRD_JUDGE_PROMPT}, }, {"artifact_type": "sdd", "description": "Solution Design Document -- how it will be implemented."}, {"artifact_type": "benchmark", "description": "Concrete acceptance scenarios with observable pass criteria."}, { "artifact_type": "test_dataset", "description": "Concrete test data (inputs + expected outputs) if this goal has a natural one -- " "many goals (UI wiring, config changes) don't, that's fine, omit this node then.", }, { "artifact_type": "test_script", "description": "A real, runnable script that mechanically checks the Benchmark's scenarios -- " "only include this if a genuine mechanical check is possible for this goal.", "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"}, }, ] # NOTE: no `str.format()` involved in building or filling this template # (unlike _GENERIC_PROMPT_TEMPLATE) -- the embedded catalog JSON and the # instructional JSON examples below (e.g. {"kind": "none"}) are full of # their own literal braces, which .format() would try to parse as format # fields and crash on. plan_work_graph() fills in goal_text with a plain # `.replace("{goal_text}", goal_text)` instead, which treats every other # brace in this string as inert literal text -- no escaping needed at all. _PROMPT_TEMPLATE = ( "Plan (do not execute) the Work Graph for this goal: {goal_text}\n" "Available node types (use any subset, in any order that makes sense -- this goal may not " "need all of them, and 'suggested_judgment' entries are starting points you can reuse or " "override, not requirements):\n" + json.dumps(_PLANNER_NODE_CATALOG) + "\n" "For each node you include, decide: node_id, artifact_type, depends_on (which earlier nodes' " "output it genuinely needs), a judgment config (how to tell if this node's output is " "acceptable -- {\"kind\": \"none\"} if nothing needs checking, {\"kind\": \"mechanical\", " "\"check_ref\": ...} for a named mechanical check, or {\"kind\": \"llm\", \"judge_prompt\": " "\"...{content}...\"} for your own judgment prompt with a {content} placeholder), and an " "on_fail config (what happens if judgment fails: {\"action\": \"retry_self\"}, " "{\"action\": \"loop_back\"}, or {\"action\": \"hard_fail\"}, each optionally with " "\"max_attempts\"). If a node's on_fail.action is 'loop_back', you MUST also add a " "corresponding edge in the top-level 'edges' array with edge_type 'on_failure', from that " "node to the exact earlier node_id it should return to -- the target must be a real upstream " "dependency of this node (something that genuinely produced input this node used), not an " "arbitrary node. This mirrors how the success path already works: every node's 'yes, this " "succeeded, what's next' is expressed as a normal dependency edge, and 'no, this failed, go " "back to X' is expressed the same way, just with edge_type 'on_failure' instead. Every other " "edge needs edge_type: 'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for " "anything requiring the agent to actually reason about content.\n" "End your reply with the graph_manifest as a single fenced ```json code block " "shaped exactly like {\"nodes\": [...], \"edges\": [...]} -- no extra wrapper key." ) # E (2026-08-06): the generic, cross-Org-Agent path -- used for every # org_node OTHER than 'planner'. Planner's own doc-writing chain stays on # _PROMPT_TEMPLATE above unchanged (zero regression risk, see # docs/done/dynamic-work-graph-design-skill-design.md §3): that chain is a # fixed architectural constant (brd->prd->sdd->benchmark->...), not # something that should vary call to call, whereas Coder/Validator's # sub-tasks have no such fixed small vocabulary of node types. Each node # here carries its own self-contained `task_prompt` (see the # design-dynamic-work-graph Skill this prompt points at) instead of an # `artifact_type` that a fixed lookup table like _DOC_PROMPTS could resolve. _GENERIC_PROMPT_TEMPLATE = ( "Plan (do not execute) a Work Graph for this goal: {goal_text}\n" "Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge " "when a node's task_prompt genuinely needs another node's output content -- nodes with " "no real data dependency between them will be scheduled to run CONCURRENTLY, so adding " "an edge just to make the graph look ordered actively slows execution down. If this goal " "is small enough to finish in one call, output a single-node graph_manifest rather than " "forcing a split.\n" "Output a graph_manifest whose nodes each have: node_id (a short readable identifier), " "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt " "for that node's own execution -- assume the agent running it sees nothing except this " "task_prompt and its dependencies' outputs, not this planning conversation). Every edge " "needs edge_type: 'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for " "anything requiring the agent to actually reason about content.\n" "End your reply with the graph_manifest as a single fenced ```json code block " "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key." ) _JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL) class PlanningFailedError(Exception): """PLANNING_FAILED -- 03-C §3.4.""" def _extract_graph_manifest(text): """--json-schema doesn't constrain output when --agent is set (see module docstring) -- pulls the graph_manifest out of free-form response text instead: try the whole text as JSON, then each ```json fenced block; accept either {"nodes":...,"edges":...} directly or wrapped one level deep under a "graph_manifest" key (models sometimes add that wrapper despite being asked not to).""" candidates = [text] + _JSON_FENCE_RE.findall(text) for candidate in candidates: try: parsed = json.loads(candidate) except (json.JSONDecodeError, TypeError): continue if isinstance(parsed, dict): if "nodes" in parsed and "edges" in parsed: return parsed inner = parsed.get("graph_manifest") if isinstance(inner, dict) and "nodes" in inner and "edges" in inner: return inner return None def _transitive_ancestors(graph_manifest, node_id): """node_id itself + every node it (directly or transitively) depends on, via `depends_on`/normal data-dependency edges (`on_failure` edges excluded -- they're control-flow, not data dependency, same reasoning as work_graph_orchestrate.py's own `_transitive_dependents`/ `topological_layers`). Deliberately reimplemented here rather than imported from work_graph_orchestrate.py -- the two modules don't depend on each other today (work_graph_orchestrate.py is a lower-level execution engine, work_graph_plan.py is planning-only) and this is a small enough function that duplicating it doesn't justify introducing that coupling.""" deps = {n.get("node_id"): set(n.get("depends_on") or []) for n in graph_manifest.get("nodes", [])} for edge in graph_manifest.get("edges", []): if edge.get("edge_type") == "on_failure": continue from_id, to_id = edge.get("from"), edge.get("to") 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 _validate_loop_back_targets(manifest): """2026-08-08 (Work Graph self-planning + loop-back): every node declaring `on_fail.action == 'loop_back'` must have exactly one `edge_type: 'on_failure'` edge originating from it (see work_graph_orchestrate.py's `_on_failure_target`), and that edge's target must be a genuine upstream dependency -- looping back to a node that never influenced this one has no coherent meaning. Raises PlanningFailedError (not a generic ValueError) so this surfaces the same way every other planning failure does, before any node ever runs -- catching a malformed loop_back here is far cheaper than discovering it mid-execution.""" node_ids = {n.get("node_id") for n in manifest.get("nodes", [])} for node in manifest.get("nodes", []): if (node.get("on_fail") or {}).get("action") != "loop_back": continue node_id = node.get("node_id") on_failure_edges = [ e for e in manifest.get("edges", []) if e.get("from") == node_id and e.get("edge_type") == "on_failure" ] if len(on_failure_edges) != 1: raise PlanningFailedError( "node {} declares on_fail.action='loop_back' but has {} on_failure edge(s), " "expected exactly 1".format(node_id, len(on_failure_edges)) ) target = on_failure_edges[0].get("to") if target not in node_ids: raise PlanningFailedError( "node {}'s on_failure edge targets {!r}, which does not exist in this graph_manifest".format( node_id, target ) ) if target not in _transitive_ancestors(manifest, node_id): raise PlanningFailedError( "node {}'s on_failure edge targets {!r}, which is not an upstream dependency -- " "loop_back must target a node that genuinely influenced this one".format(node_id, target) ) def _normalize_graph_manifest(manifest): """Real LLM output has been observed (2026-08-01, U-031 integration validation) to use `id` instead of the documented `node_id` key, and to express dependencies purely via `edges` (from/to) without ever populating each node's own `depends_on` array -- even though this schema requires both. Normalizing here, right after parsing and before persisting, means every downstream consumer (work_graph_orchestrate.py's topological sort, v_task_detail's to-do tree, v_graph_visualizer's DAG layering) gets a manifest that actually matches the documented shape, instead of each one needing its own defensive "id or node_id, depends_on or edges" fallback logic repeated independently. Mutates and returns `manifest`. 2026-08-08 (Work Graph self-planning + loop-back, FR-B4 "全部自由规划"): the canonical-doc-chain forced-order enforcement that used to live here (2026-08-03 incident: a real call omitted brd->prd->sdd->benchmark's dependency entirely, scheduling nodes out of order) is REMOVED, not extended -- Planner's node set/order is no longer a fixed architectural constant (formally reverses docs/done/dynamic-work-graph-design-skill- design.md §3). Robustness against that same failure mode now rests entirely on the edge-derived depends_on backfill below (unchanged) plus `_validate_loop_back_targets` (new, below) -- there is no longer a hardcoded canonical chain to fall back on if an LLM response omits dependency info, so a real reproduction of the 2026-08-03 incident shape is a genuine, currently-unmitigated risk this design accepted (see docs/todo/work-graph-self-planning-and-loop-back-requirements.md §7's risk list) -- FR-B3's own acceptance criterion only covers the "wrong key names" failure mode this backfill already handles, not a complete absence of dependency info with no canonical chain to guess from. """ nodes = manifest.get("nodes", []) for node in nodes: if "node_id" not in node and "id" in node: node["node_id"] = node["id"] node.setdefault("depends_on", []) node_ids = {n.get("node_id") for n in nodes} for edge in manifest.get("edges", []): if edge.get("edge_type") == "on_failure": continue # control-flow, not a data dependency -- see _transitive_ancestors from_id, to_id = edge.get("from"), edge.get("to") if from_id in node_ids and to_id in node_ids: for node in nodes: if node.get("node_id") == to_id and from_id not in node["depends_on"]: node["depends_on"].append(from_id) _validate_loop_back_targets(manifest) return manifest def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300): """Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row (status='planning') -- moving it to 'running'/executing it is out of scope. 2026-08-08: `org_node == 'planner'` gets `_PROMPT_TEMPLATE` (now a genuine self-planning prompt over `_PLANNER_NODE_CATALOG`, see that template's own comment for why goal_text is filled in via `.replace()` rather than `.format()`). Every other org_node gets `_GENERIC_PROMPT_TEMPLATE` -- the cross-Org-Agent dynamic planning path described in the design-dynamic-work-graph Skill, whose nodes carry their own task_prompt instead of a fixed artifact_type. Function signature/return shape are unchanged either way.""" if org_node == "planner": prompt = _PROMPT_TEMPLATE.replace("{goal_text}", goal_text) else: prompt = _GENERIC_PROMPT_TEMPLATE.format(goal_text=goal_text) result = subprocess.run( [ "claude", "--agent", org_node, "-p", prompt, "--output-format", "json", "--permission-mode", "auto", ], cwd=target_folder_path, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=timeout, ) if result.returncode != 0: # 2026-07-31 fix (same bug as org_node_trigger.py): include stdout too. raise PlanningFailedError( "plan failed for task {}\nstderr: {}\nstdout (truncated): {}".format( task_id, result.stderr, result.stdout[:4000] ) ) payload = json.loads(result.stdout) if payload.get("is_error"): raise PlanningFailedError("plan failed for task {}: {}".format(task_id, payload)) graph_manifest = _extract_graph_manifest(payload.get("result", "")) if not graph_manifest: raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload)) graph_manifest = _normalize_graph_manifest(graph_manifest) run_id = str(uuid.uuid4()) conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " "VALUES (?, ?, ?, ?, 'planning')", (run_id, task_id, org_node, json.dumps(graph_manifest)), ) conn.commit() return {"run_id": run_id, "graph_manifest": graph_manifest} def generic_node_prompt_builder(node, outputs): """E (2026-08-06): the `build_prompt(node, outputs)` run_graph() needs for a graph_manifest produced via _GENERIC_PROMPT_TEMPLATE -- lives here, not in e2e_pipeline.py, because it's tightly coupled to the `task_prompt` field convention that prompt asks for (keeping both in the same module means a field rename can't silently desync producer and consumer). No factory/closure needed: every node is self-contained per the design-dynamic-work-graph Skill's contract, so `outputs` (already-completed dependency content) isn't needed here.""" return node["task_prompt"]