"""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"]}, "depends_on": {"type": "array", "items": {"type": "string"}}, }, }, }, "edges": { "type": "array", "items": { "type": "object", "required": ["from", "to", "edge_type"], "properties": { "from": {"type": "string"}, "to": {"type": "string"}, "edge_type": {"type": "string", "enum": ["zero_token", "needs_judgment"]}, }, }, }, }, } _PROMPT_TEMPLATE = ( "Plan (do not execute) the Work Graph needed to produce BRD, PRD, SDD, and " "Benchmark artifacts for this goal: {goal_text}\n" "Output a graph_manifest with one node per artifact_type (brd, prd, sdd, " "benchmark) and edges expressing their dependency chain (sdd depends on prd " "depends on brd, benchmark depends on sdd). 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 _CANONICAL_DOC_CHAIN = ("brd", "prd", "sdd", "benchmark") 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`. **Canonical doc-chain enforcement (2026-08-03, real self-hosting run)**: a real call was observed to omit the brd->prd->sdd->benchmark dependency entirely -- no `depends_on`, no `edges` connecting them at all, not just the wrong key names the edge-derived backfill above can already recover from. With no dependency info to normalize from, `topological_order()` scheduled `benchmark` before `sdd` even ran (alphabetically first among nodes with an empty in-degree), and `e2e_pipeline.build_prompt()` crashed with `KeyError: 'sdd_content'` trying to format a template that needed sdd's not-yet-produced output. `_PROMPT_TEMPLATE` always asks for exactly this fixed 4-node linear chain -- it's a constant architectural fact about what the Planner's doc-writing graph looks like, not something that genuinely varies call to call, so don't leave it to chance: whenever all four canonical artifact_types are present, deterministically enforce the known linear order regardless of what edges/depends_on the LLM did or didn't provide. Same "机制用代码" principle already applied elsewhere (`work_node_run.ensure_worktree_committed`, etc.). """ 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", []): 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) by_artifact_type = {n.get("artifact_type"): n for n in nodes} if all(t in by_artifact_type for t in _CANONICAL_DOC_CHAIN): for prev_type, cur_type in zip(_CANONICAL_DOC_CHAIN, _CANONICAL_DOC_CHAIN[1:]): prev_id = by_artifact_type[prev_type]["node_id"] cur_node = by_artifact_type[cur_type] if prev_id not in cur_node["depends_on"]: cur_node["depends_on"].append(prev_id) 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.""" prompt = _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}