Org Graph unification Phase 3: frontend Org meta-graph + loop_back replay timeline

FR-5a: a new, standalone four-stage Org meta-graph (renderOrgMetaGraph and friends) rather than extending the existing 3-node renderOrgGraphTopology (left untouched, it's pinned by test_graph_console_smoke.js). Each stage expands in place: planner/coder/validator reuse renderGraphVisualizerPanel unmodified against that stage's latest run (identified via org_nodes_involved, Phase 2c's replacement for the removed single-value org_node column); planner_reviewer has no self-planned work_graph_run, so it shows its planner_review_verdict execution_event history instead (new task_query.py::_planner_review_events_for_task(), mirroring the existing _loop_back_events_for_run() pattern). FR-5b: an attempt_number-ordered, click-to-reveal feedback timeline built from loop_back_events, which was already being fetched but never rendered. New frontend/test_org_meta_graph_smoke.js (35 assertions, same vm.Context sandbox convention as the existing 4 smoke tests). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Org Graph unification Phase 3: frontend Org meta-graph + loop_back replay timeline
FR-5a: a new, standalone four-stage Org meta-graph (renderOrgMetaGraph and friends) rather than extending the existing 3-node renderOrgGraphTopology (left untouched, it's pinned by test_graph_console_smoke.js). Each stage expands in place: planner/coder/validator reuse renderGraphVisualizerPanel unmodified against that stage's latest run (identified via org_nodes_involved, Phase 2c's replacement for the removed single-value org_node column); planner_reviewer has no self-planned work_graph_run, so it shows its planner_review_verdict execution_event history instead (new task_query.py::_planner_review_events_for_task(), mirroring the existing _loop_back_events_for_run() pattern). FR-5b: an attempt_number-ordered, click-to-reveal feedback timeline built from loop_back_events, which was already being fetched but never rendered. New frontend/test_org_meta_graph_smoke.js (35 assertions, same vm.Context sandbox convention as the existing 4 smoke tests). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent f7103fd7
Showing 11 changed files with 725 additions and 1 deletions
docs/todo/org-graph-unification-and-universal-self-planning-requirements.md → docs/done/org-graph-unification-and-universal-self-planning-requirements.md
docs/todo/org-graph-unification-phase2-universal-self-planning-design.md → docs/done/org-graph-unification-phase2-universal-self-planning-design.md
docs/todo/scheduler-new-goal-dispatch-requirements.md → docs/done/scheduler-new-goal-dispatch-requirements.md
docs/todo/work-graph-node-execution-deblackbox-requirements.md → docs/done/work-graph-node-execution-deblackbox-requirements.md
frontend/app.js
frontend/test_org_meta_graph_smoke.js
src/new_goal_dispatch.py
src/task_query.py
src/test_new_goal_dispatch.py
src/test_org_node_trigger_e2e.py
src/test_task_query.py
@@ -1957,6 +1957,141 @@ function renderOrgGraphTopology(activeOrgNode, runStatusByOrgNode, route) {
   return '<div class="graph-topo-row">' + nodesHtml + "</div>" + retryEdgeHtml;
 }
  
+// ---- Org 元图 (FR-5a, Org Graph unification Phase 3, 2026-08-13): a
+// standalone four-stage top-level meta-graph (Planner -> 复核 Planner
+// Reviewer -> Coder -> 终审 Validator), separate from renderOrgGraphTopology
+// above -- that function is still called by mountGraphConsole's own
+// existing "Org Graph" panel and is pinned by test_graph_console_smoke.js's
+// 3-node ORG_NODE_ORDER shape, so it's left untouched rather than stretched
+// to 4 stages. Each stage expands in place into that stage's real Work
+// Graph run, reusing renderGraphVisualizerPanel unmodified -- except
+// planner_reviewer, which never self-plans a Work Graph
+// (org_node_trigger.trigger_org_node() is one flat call, see
+// e2e_pipeline.py::_record_planner_review_event) and instead shows its own
+// verdict trail (task.planner_review_events, task_query.py::
+// _planner_review_events_for_task).
+const ORG_META_STAGE_ORDER = ["planner", "planner_reviewer", "coder", "validator"];
+const ORG_META_STAGE_LABELS = {
+  planner: "Planner", planner_reviewer: "复核（Planner Reviewer）",
+  coder: "Coder", validator: "终审（Validator）",
+};
+
+function orgMetaStageStatus(task, stage) {
+  if (stage === "planner_reviewer") {
+    const events = task.planner_review_events || [];
+    if (events.length === 0) return "not_started";
+    return events[events.length - 1].verdict === "PASS" ? "done" : "reflecting";
+  }
+  const run = getLatestRunForOrgNode(task, stage);
+  return run ? run.status : "not_started";
+}
+
+function renderOrgMetaGraph(task) {
+  const nodesHtml = ORG_META_STAGE_ORDER.map((stage, i) => {
+    const status = orgMetaStageStatus(task, stage);
+    const arrow = i < ORG_META_STAGE_ORDER.length - 1 ? '<div class="graph-edge">→</div>' : "";
+    return (
+      '<div class="graph-topo-node status-' + status + '" data-meta-stage="' + stage + '" style="cursor:pointer;">'
+      + '<div class="role">' + escapeHtml(ORG_META_STAGE_LABELS[stage]) + "</div>"
+      + '<div class="hint">' + escapeHtml(status) + "</div>"
+      + "</div>" + arrow
+    );
+  }).join("");
+  const detailsHtml = ORG_META_STAGE_ORDER.map((stage) =>
+    '<div class="org-meta-stage-detail" id="org-meta-detail-' + stage + '" data-meta-stage="' + stage + '" style="display:none;"></div>'
+  ).join("");
+  return '<div class="graph-topo-row">' + nodesHtml + "</div>" + detailsHtml;
+}
+
+function renderPlannerReviewEventsList(events) {
+  if (!events || events.length === 0) return '<div class="hint">尚无复核记录</div>';
+  return '<ul class="simple-list">' + events.map((e) =>
+    "<li><span>attempt " + e.attempt + " · " + escapeHtml(e.verdict)
+    + (e.used_existing_artifacts ? "（复用已有产出）" : "") + "</span><span>"
+    + escapeHtml(e.created_at || "") + "</span></li>"
+  ).join("") + "</ul>";
+}
+
+// planner/coder/validator branch mirrors renderOrgComboPanelContent's own
+// empty-run/empty-DAG fallback text -- same two states, this is just the
+// meta-graph's own expand target rather than v_task_detail's combo panel.
+function renderOrgMetaStageDetailHtml(task, stage) {
+  if (stage === "planner_reviewer") {
+    return '<div class="panel"><h2>Planner 产出复核记录</h2>' + renderPlannerReviewEventsList(task.planner_review_events) + "</div>";
+  }
+  const run = getLatestRunForOrgNode(task, stage);
+  if (!run) return '<div class="hint">' + escapeHtml(stage) + " 尚未开始（暂无 Work Graph run）</div>";
+  const panelHtml = renderGraphVisualizerPanel(run, { enableNodeClick: false });
+  return panelHtml
+    ? panelHtml + renderLoopBackTimelineHtml(run)
+    : '<div class="hint">最新一次 run_id=' + escapeHtml(run.run_id) + "（status=" + escapeHtml(run.status) + "）· 该 run 尚未生成 Work Graph</div>";
+}
+
+function mountOrgMetaGraph(container, task) {
+  for (const el of container.querySelectorAll(".graph-topo-node[data-meta-stage]")) {
+    el.addEventListener("click", () => {
+      const stage = el.dataset.metaStage;
+      const detail = container.querySelector("#org-meta-detail-" + stage);
+      if (!detail) return;
+      if (detail.style.display !== "none") {
+        detail.style.display = "none";
+        detail.innerHTML = "";
+        return;
+      }
+      detail.innerHTML = renderOrgMetaStageDetailHtml(task, stage);
+      detail.style.display = "block";
+      mountGraphVisualizerTooltips(detail);
+      mountGraphVisualizerEdges(detail);
+      mountLoopBackTimeline(detail);
+    });
+  }
+}
+
+// ---- Loop-back 历史回放时间轴 (FR-5b, Org Graph unification Phase 3,
+// 2026-08-13): task_query.py's _loop_back_events_for_run() already exposes
+// run.loop_back_events (source_node_id/target_node_id/attempt_number/
+// feedback[]/created_at) -- renderGraphVisualizerPanel only ever reduces
+// them to a self-loop badge count / on_failure edge trigger count, never
+// the feedback text itself. This is a separate, additive component (not a
+// rewrite of the DAG renderer) rendered alongside it at each of its call
+// sites, ordered by attempt_number, one entry per real loop_back.
+function renderLoopBackTimelineHtml(run) {
+  const events = run.loop_back_events || [];
+  if (events.length === 0) return "";
+  const itemsHtml = events.map((e, i) =>
+    '<li class="timeline-item" data-attempt-index="' + i + '" style="cursor:pointer;">'
+    + "<span>attempt " + e.attempt_number + " · " + escapeHtml(e.source_node_id) + " → " + escapeHtml(e.target_node_id) + "</span>"
+    + "<span>" + escapeHtml(e.created_at || "") + "</span>"
+    + "</li>"
+  ).join("");
+  const feedbackPanelsHtml = events.map((e, i) =>
+    '<div class="timeline-feedback" data-attempt-index="' + i + '" style="display:none;">'
+    + ((e.feedback || []).length
+      ? "<ul>" + e.feedback.map((f) => "<li>" + escapeHtml(f) + "</li>").join("") + "</ul>"
+      : '<div class="hint">该次 loop_back 未记录 feedback</div>')
+    + "</div>"
+  ).join("");
+  return (
+    '<div class="panel loop-back-timeline"><h2>Loop-back 历史回放（按 attempt_number）</h2>'
+    + '<ul class="simple-list timeline-list">' + itemsHtml + "</ul>"
+    + feedbackPanelsHtml
+    + "</div>"
+  );
+}
+
+function mountLoopBackTimeline(container) {
+  for (const timeline of container.querySelectorAll(".loop-back-timeline")) {
+    for (const item of timeline.querySelectorAll(".timeline-item[data-attempt-index]")) {
+      item.addEventListener("click", () => {
+        const idx = item.dataset.attemptIndex;
+        for (const panel of timeline.querySelectorAll(".timeline-feedback")) {
+          panel.style.display = panel.dataset.attemptIndex === idx ? "block" : "none";
+        }
+      });
+    }
+  }
+}
+
 const GC_DOC_ARTIFACT_TYPES = ["brd", "prd", "sdd", "benchmark"];
 const GC_CODE_ARTIFACT_TYPES = ["code_diff", "test_report"];
  
@@ -2049,6 +2184,10 @@ function renderGraphConsole() {
         <h2>Org Graph</h2>
         <div id="gc_org_topology"><div class="hint">加载中…</div></div>
       </div>
+      <div class="panel">
+        <h2>Org 元图（点击每段展开真实 Work Graph 详情）</h2>
+        <div id="gc_org_meta_graph"><div class="hint">加载中…</div></div>
+      </div>
       <div id="gc_work_graph_panels"><div class="hint">加载中…</div></div>
       <div class="tooltip" id="graph-tooltip"></div>
       <div id="gc_artifacts_overview"><div class="hint">加载中…</div></div>
@@ -2062,6 +2201,7 @@ function mountGraphConsole(queryParams) {
  
   const taskSelect = document.getElementById("gc_task_select");
   const orgTopologyEl = document.getElementById("gc_org_topology");
+  const orgMetaGraphEl = document.getElementById("gc_org_meta_graph");
   const workGraphPanelsEl = document.getElementById("gc_work_graph_panels");
   const artifactsOverviewEl = document.getElementById("gc_artifacts_overview");
  
@@ -2127,9 +2267,11 @@ function mountGraphConsole(queryParams) {
       for (const orgNode of run.org_nodes_involved || []) runStatusByOrgNode[orgNode] = run.status;
     }
     orgTopologyEl.innerHTML = renderOrgGraphTopology(task.assigned_org_node, runStatusByOrgNode, task.route);
+    orgMetaGraphEl.innerHTML = renderOrgMetaGraph(task);
+    mountOrgMetaGraph(orgMetaGraphEl, task);
  
     const panelsHtml = task.work_graph_runs
-      .map((run) => renderGraphVisualizerPanel(run, { enableNodeClick: true }))
+      .map((run) => renderGraphVisualizerPanel(run, { enableNodeClick: true }) + renderLoopBackTimelineHtml(run))
       .join("");
     if (!panelsHtml) {
       workGraphPanelsEl.innerHTML = '<div class="panel"><div class="hint">尚未开始规划</div></div>';
@@ -2139,6 +2281,7 @@ function mountGraphConsole(queryParams) {
     mountGraphVisualizerTooltips(workGraphPanelsEl);
     mountGraphVisualizerEdges(workGraphPanelsEl);
     mountWorkNodeClickExpand(workGraphPanelsEl, { onNodeClick: onWorkNodeClick });
+    mountLoopBackTimeline(workGraphPanelsEl);
   }
  
   async function populateTaskSelect() {
@@ -2146,6 +2289,7 @@ function mountGraphConsole(queryParams) {
     if (tasks.length === 0) {
       taskSelect.innerHTML = '<option value="">暂无任务</option>';
       orgTopologyEl.innerHTML = '<div class="hint">暂无任务 · <a href="#/goals/new">提出第一个目标 →</a></div>';
+      orgMetaGraphEl.innerHTML = "";
       workGraphPanelsEl.innerHTML = "";
       return;
     }
+// Org Graph unification Phase 3 (FR-5a + FR-5b) smoke test. Same no-
+// framework/no-fixture convention as the other frontend/test_*_smoke.js
+// files -- loads app.js in a bare vm.Context and asserts on generated HTML
+// strings for the new pieces:
+//   - FR-5a: ORG_META_STAGE_ORDER / orgMetaStageStatus / renderOrgMetaGraph
+//     (the four-stage top-level meta-graph, Planner -> 复核 -> Coder ->
+//     终审, separate from the existing 3-node renderOrgGraphTopology) and
+//     renderOrgMetaStageDetailHtml's planner_reviewer special-case (a
+//     verdict list, not a faked Work Graph panel).
+//   - FR-5b: renderLoopBackTimelineHtml (the attempt_number-ordered
+//     loop_back replay list, reusing run.loop_back_events which
+//     task_query.py already exposed but nothing rendered before this).
+// Mount-time DOM wiring (mountOrgMetaGraph/mountLoopBackTimeline's click
+// handlers) needs real querySelectorAll traversal this bare sandbox doesn't
+// provide -- same "manual browser verification only" boundary the other
+// smoke tests already draw around mount*/edge-drawing functions; this file
+// only pins the pure render* HTML-string output.
+//
+// Run: node frontend/test_org_meta_graph_smoke.js
+
+"use strict";
+const fs = require("fs");
+const path = require("path");
+const vm = require("vm");
+const assert = require("assert");
+
+function makeFakeDiv() {
+  let text = "";
+  return {
+    set textContent(v) { text = v == null ? "" : String(v); },
+    get textContent() { return text; },
+    get innerHTML() {
+      return text.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;").replace(/"/g, "&quot;");
+    },
+  };
+}
+
+const sandbox = {
+  document: {
+    createElement: () => makeFakeDiv(),
+    getElementById: () => null,
+    querySelectorAll: () => [],
+    addEventListener: () => {},
+  },
+  window: {
+    location: { protocol: "http:", host: "localhost", hash: "" },
+    addEventListener: () => {},
+  },
+  localStorage: { getItem: () => null, setItem: () => {} },
+  console,
+};
+vm.createContext(sandbox);
+
+const src = fs.readFileSync(path.join(__dirname, "app.js"), "utf8");
+// Top-level `const` stays lexically scoped to the script and never reaches
+// the sandbox object (same Annex B quirk the other smoke tests already
+// document for e.g. ARTIFACT_DETAIL_RE) -- re-export what this test needs.
+const augmentedSrc = src + "\nglobalThis.ORG_META_STAGE_ORDER = ORG_META_STAGE_ORDER;\n";
+new vm.Script(augmentedSrc, { filename: "app.js" }).runInContext(sandbox);
+
+let assertionCount = 0;
+function check(condition, message) {
+  assertionCount++;
+  assert.ok(condition, message);
+}
+
+// ---- FR-5a: ORG_META_STAGE_ORDER has all four stages, in order ----
+{
+  check(
+    sandbox.ORG_META_STAGE_ORDER.join(",") === "planner,planner_reviewer,coder,validator",
+    "the meta-graph must cover all four stages in Planner -> 复核 -> Coder -> 终审 order"
+  );
+}
+
+// ---- FR-5a: orgMetaStageStatus ----
+{
+  const emptyTask = { work_graph_runs: [], planner_review_events: [] };
+  check(sandbox.orgMetaStageStatus(emptyTask, "planner") === "not_started", "no run yet -> not_started");
+  check(sandbox.orgMetaStageStatus(emptyTask, "planner_reviewer") === "not_started", "no review events yet -> not_started");
+
+  const taskWithRuns = {
+    work_graph_runs: [
+      { run_id: "run-1", org_nodes_involved: ["coder"], status: "running", started_at: "2026-08-01 00:00:00" },
+    ],
+    planner_review_events: [],
+  };
+  check(sandbox.orgMetaStageStatus(taskWithRuns, "coder") === "running", "status mirrors the latest run's own status");
+
+  const passedReview = { work_graph_runs: [], planner_review_events: [
+    { attempt: 1, verdict: "REJECTED", created_at: "t1" },
+    { attempt: 2, verdict: "PASS", created_at: "t2" },
+  ] };
+  check(sandbox.orgMetaStageStatus(passedReview, "planner_reviewer") === "done", "latest verdict PASS -> done");
+
+  const rejectedReview = { work_graph_runs: [], planner_review_events: [
+    { attempt: 1, verdict: "REJECTED", created_at: "t1" },
+  ] };
+  check(sandbox.orgMetaStageStatus(rejectedReview, "planner_reviewer") === "reflecting", "latest verdict not PASS -> reflecting, not done");
+}
+
+// ---- FR-5a: renderOrgMetaGraph -- four boxes, three arrows, real statuses ----
+{
+  const task = {
+    work_graph_runs: [
+      { run_id: "run-1", org_nodes_involved: ["planner"], status: "done", started_at: "t1" },
+      { run_id: "run-2", org_nodes_involved: ["coder"], status: "running", started_at: "t2" },
+    ],
+    planner_review_events: [{ attempt: 1, verdict: "PASS", created_at: "t1.5" }],
+  };
+  const html = sandbox.renderOrgMetaGraph(task);
+  for (const stage of ["planner", "planner_reviewer", "coder", "validator"]) {
+    check(html.includes('data-meta-stage="' + stage + '"'), "every stage box must carry data-meta-stage=" + stage);
+  }
+  check(html.includes("graph-topo-node status-done"), "planner shows its real run status (done)");
+  check(html.includes("graph-topo-node status-running"), "coder shows its real run status (running)");
+  check(html.includes("graph-topo-node status-not_started"), "validator has no run yet -> not_started");
+  check((html.match(/graph-edge/g) || []).length === 3, "four stages are joined by exactly three arrows");
+  check(html.includes('id="org-meta-detail-planner_reviewer"'), "every stage gets its own (initially hidden) expand target");
+  check(html.includes('style="display:none;"'), "expand targets start collapsed");
+}
+
+// ---- FR-5a: planner_reviewer special-case detail (verdict list, not a
+// faked Work Graph panel) ----
+{
+  const emptyEventsHtml = sandbox.renderOrgMetaStageDetailHtml({ planner_review_events: [] }, "planner_reviewer");
+  check(emptyEventsHtml.includes("尚无复核记录"), "zero review events -> explicit empty state");
+  check(!emptyEventsHtml.includes("work-graph"), "planner_reviewer must never render a fake DAG panel");
+
+  const task = {
+    planner_review_events: [
+      { attempt: 1, verdict: "REJECTED", used_existing_artifacts: false, created_at: "2026-08-12 10:00:00" },
+      { attempt: 2, verdict: "PASS", used_existing_artifacts: true, created_at: "2026-08-12 10:05:00" },
+    ],
+  };
+  const html = sandbox.renderOrgMetaStageDetailHtml(task, "planner_reviewer");
+  check(html.includes("attempt 1"), "each attempt is listed");
+  check(html.includes("REJECTED") && html.includes("PASS"), "each attempt's real verdict is shown");
+  check(html.includes("2026-08-12 10:00:00") && html.includes("2026-08-12 10:05:00"), "each attempt's real timestamp is shown");
+  check(html.includes("复用已有产出"), "used_existing_artifacts=true is surfaced, not silently dropped");
+}
+
+// ---- FR-5a: planner/coder/validator detail reuses renderGraphVisualizerPanel
+// unmodified, with the existing empty-run / empty-DAG fallbacks ----
+{
+  const noRunTask = { work_graph_runs: [] };
+  const noRunHtml = sandbox.renderOrgMetaStageDetailHtml(noRunTask, "coder");
+  check(noRunHtml.includes("尚未开始"), "no run yet for this stage -> explicit empty state");
+
+  const emptyDagTask = {
+    work_graph_runs: [{ run_id: "run-1", org_nodes_involved: ["coder"], status: "failed", graph_manifest: null }],
+  };
+  const emptyDagHtml = sandbox.renderOrgMetaStageDetailHtml(emptyDagTask, "coder");
+  check(emptyDagHtml.includes("run-1") && emptyDagHtml.includes("尚未生成 Work Graph"), "run exists but has no DAG yet -> explicit fallback naming the run");
+
+  const realRunTask = {
+    work_graph_runs: [{
+      run_id: "run-2", org_nodes_involved: ["coder"], status: "running",
+      graph_manifest: { nodes: [{ node_id: "n1", depends_on: [] }] },
+      nodes: [{ node_id: "n1", node_exec_id: "exec-1", subagent_type: "coder", status: "running" }],
+    }],
+  };
+  const realHtml = sandbox.renderOrgMetaStageDetailHtml(realRunTask, "coder");
+  check(realHtml.includes("run_id=run-2"), "a real run renders the actual renderGraphVisualizerPanel DAG output");
+  check(!realHtml.includes("clickable"), "meta-graph expand opts out of click-to-expand-logs (enableNodeClick: false)");
+}
+
+// ---- FR-5b: renderLoopBackTimelineHtml ----
+{
+  const noEventsHtml = sandbox.renderLoopBackTimelineHtml({ loop_back_events: [] });
+  check(noEventsHtml === "", "a run with no loop_back_events renders nothing, not an empty panel");
+  check(sandbox.renderLoopBackTimelineHtml({}) === "", "missing loop_back_events entirely is treated the same as empty");
+
+  const run = {
+    run_id: "run-1",
+    loop_back_events: [
+      { source_node_id: "benchmark", target_node_id: "sdd", attempt_number: 1, feedback: ["missing scenario"], created_at: "t1" },
+      { source_node_id: "benchmark", target_node_id: "sdd", attempt_number: 2, feedback: ["still missing", "and X"], created_at: "t2" },
+    ],
+  };
+  const html = sandbox.renderLoopBackTimelineHtml(run);
+  check(html.includes("attempt 1") && html.includes("attempt 2"), "every attempt gets its own timeline entry, in order");
+  check(html.includes("benchmark") && html.includes("sdd"), "each entry names its real source/target node");
+  check(html.includes('data-attempt-index="0"') && html.includes('data-attempt-index="1"'), "entries are addressable by index for the click-to-reveal wiring");
+  check(html.includes("missing scenario"), "attempt 1's real feedback text is rendered");
+  check(html.includes("still missing") && html.includes("and X"), "attempt 2's real (multi-item) feedback text is rendered");
+  check(html.includes('style="display:none;"'), "feedback panels start collapsed -- a click reveals one at a time");
+
+  const noFeedbackRun = {
+    loop_back_events: [{ source_node_id: "a", target_node_id: "a", attempt_number: 1, feedback: [], created_at: "t1" }],
+  };
+  const noFeedbackHtml = sandbox.renderLoopBackTimelineHtml(noFeedbackRun);
+  check(noFeedbackHtml.includes("未记录 feedback"), "an attempt with empty feedback[] still renders an explicit state, not blank");
+}
+
+console.log("frontend/test_org_meta_graph_smoke.js: all " + assertionCount + " assertions passed");
+"""Periodic auto-dispatch for new Goals that were created but never
+started (2026-08-13, `docs/todo/scheduler-new-goal-dispatch-requirements.md`
+FR-1). `scheduler.run_scheduler_once()` exists and works, but nothing in
+production ever called it: a `task.status='queued'` row (created via the
+API/console, never given to `run_goal_to_completion`) sat there until a
+human clicked the console's Start button.
+
+Scope: deliberately NOT `task_reconciliation.py`'s job -- that module
+resumes tasks that already started and got abandoned mid-flight (status in
+`task_execution._RESUMABLE_STATUSES`, stale by `_DEFAULT_STALE_AFTER_SECONDS`
+= 30 min). This module only ever looks at `status='queued'` (a task that has
+never been dispatched at all) and does not apply a staleness window --
+`docs/done/scheduler-task-reconciliation-design.md` §7 records the decision
+to keep the two loops independent, not merged into one "万能调度循环", and
+that a 'queued' task should NOT wait out a 30-minute staleness window before
+its very first dispatch (the whole point of FR-1 is near-real-time pickup of
+brand-new Goals, not eventual recovery of stuck ones)."""
+
+from src import schema, task_execution
+
+
+def find_queued_tasks(conn):
+    """Returns [{"task_id", "target_folder_path", "goal_text"}, ...] for
+    every `status='queued'` task not already in flight in this process.
+    Read-only -- kept separate from `dispatch_once()` so candidate selection
+    is independently testable without spinning up real background threads
+    (same split as `task_reconciliation.find_stale_resumable_tasks`)."""
+    rows = conn.execute(
+        "SELECT t.task_id, p.target_folder_path, t.goal_text "
+        "FROM task t JOIN project p ON p.project_id = t.project_id "
+        "WHERE t.status = 'queued'"
+    ).fetchall()
+
+    candidates = []
+    for task_id, target_folder_path, goal_text in rows:
+        if task_execution.is_in_flight(task_id):
+            continue
+        candidates.append({"task_id": task_id, "target_folder_path": target_folder_path, "goal_text": goal_text})
+    return candidates
+
+
+def dispatch_once(db_path):
+    """One dispatch pass: find candidates, dispatch each via
+    `task_execution.start_task_execution_in_background()`. Own short-lived
+    connection for the scan, closed before dispatching -- same pattern as
+    `task_reconciliation.reconcile_once`. A candidate that fails to dispatch
+    (e.g. the console's Start button won the race between scan and here) is
+    skipped, not treated as fatal for the whole pass. Returns the candidate
+    list (dispatched or not) for callers/tests to inspect."""
+    conn = schema.connect(db_path)
+    try:
+        candidates = find_queued_tasks(conn)
+    finally:
+        conn.close()
+    for c in candidates:
+        try:
+            task_execution.start_task_execution_in_background(
+                db_path, c["task_id"], c["target_folder_path"], c["goal_text"]
+            )
+        except Exception:
+            pass
+    return candidates
@@ -187,9 +187,32 @@ def get_task(conn, task_id):
             # see docs/done/work-graph-self-planning-and-loop-back-design.md §7).
             "loop_back_events": _loop_back_events_for_run(conn, w[0]),
         })
+    # 2026-08-13 (Org Graph unification Phase 3, FR-5a): planner_reviewer has
+    # no self-planned work_graph_run (org_node_trigger.trigger_org_node() is
+    # one flat call, see e2e_pipeline.py::_record_planner_review_event) --
+    # its only real audit trail is this task-scoped event log, so the Org
+    # meta-graph reads it here instead of faking a run for that stage.
+    task["planner_review_events"] = _planner_review_events_for_task(conn, task_id)
     return task
  
  
+def _planner_review_events_for_task(conn, task_id):
+    rows = conn.execute(
+        "SELECT payload, created_at FROM execution_event "
+        "WHERE task_id = ? AND event_type = 'planner_review_verdict' ORDER BY created_at",
+        (task_id,),
+    ).fetchall()
+    events = []
+    for payload, created_at in rows:
+        data = json.loads(payload)
+        events.append({
+            "attempt": data["attempt"], "verdict": data["verdict"],
+            "used_existing_artifacts": data.get("used_existing_artifacts", False),
+            "created_at": created_at,
+        })
+    return events
+
+
 def _loop_back_events_for_run(conn, run_id):
     rows = conn.execute(
         "SELECT payload, created_at FROM execution_event "
+"""new_goal_dispatch.py tests (2026-08-13, FR-1 --
+docs/todo/scheduler-new-goal-dispatch-requirements.md): same mocked-at-the-
+subprocess-boundary / real-sqlite split as test_task_reconciliation.py.
+
+Run: python3 -m unittest src.test_new_goal_dispatch -v
+"""
+
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from src import new_goal_dispatch, project_scaffold, schema, task_execution
+
+
+class NewGoalDispatchTestBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        project = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
+        self.project_id = project["project_id"]
+
+    def tearDown(self):
+        self.conn.close()
+
+    def _insert_task(self, task_id, status, goal_text="a goal"):
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES (?, ?, ?, 'goal', ?, 'planner', CURRENT_TIMESTAMP)",
+            (task_id, self.project_id, goal_text, status),
+        )
+        self.conn.commit()
+
+
+class FindQueuedTasksTest(NewGoalDispatchTestBase):
+    def test_finds_queued_task(self):
+        self._insert_task("task-1", "queued")
+        result = new_goal_dispatch.find_queued_tasks(self.conn)
+        self.assertEqual(result, [{"task_id": "task-1", "target_folder_path": self.tmp.name, "goal_text": "a goal"}])
+
+    def test_skips_non_queued_statuses(self):
+        for status in ("coding", "validating", "done", "blocked_pending_human", "planning_brd_prd_sdd"):
+            self._insert_task("task-" + status, status)
+        result = new_goal_dispatch.find_queued_tasks(self.conn)
+        self.assertEqual(result, [])
+
+    def test_excludes_in_flight_task(self):
+        self._insert_task("task-1", "queued")
+        with patch.object(task_execution, "is_in_flight", return_value=True):
+            result = new_goal_dispatch.find_queued_tasks(self.conn)
+        self.assertEqual(result, [])
+
+    def test_no_candidates_returns_empty_list(self):
+        result = new_goal_dispatch.find_queued_tasks(self.conn)
+        self.assertEqual(result, [])
+
+
+class DispatchOnceTest(unittest.TestCase):
+    """Real file db (own connection, mirrors ReconcileOnceTest) -- dispatch
+    itself is mocked so this stays at the "did we correctly decide to call
+    it" boundary."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
+        self.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp_db.close()
+        self.db_path = self.tmp_db.name
+        self.addCleanup(lambda: __import__("os").path.exists(self.db_path) and __import__("os").remove(self.db_path))
+
+        conn = schema.connect(self.db_path)
+        schema.create_schema(conn)
+        project = project_scaffold.create_project(conn, self.tmp.name, init_git_if_missing=False)
+        self.project_id = project["project_id"]
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('queued-1', ?, 'goal 1', 'goal', 'queued', 'planner', CURRENT_TIMESTAMP)",
+            (self.project_id,),
+        )
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('coding-1', ?, 'goal 2', 'goal', 'coding', 'coder', CURRENT_TIMESTAMP)",
+            (self.project_id,),
+        )
+        conn.commit()
+        conn.close()
+
+    def test_dispatches_only_queued_candidates(self):
+        with patch.object(task_execution, "start_task_execution_in_background") as mock_start:
+            result = new_goal_dispatch.dispatch_once(self.db_path)
+        self.assertEqual([c["task_id"] for c in result], ["queued-1"])
+        mock_start.assert_called_once_with(self.db_path, "queued-1", self.tmp.name, "goal 1")
+
+    def test_one_dispatch_failure_does_not_block_others(self):
+        conn = schema.connect(self.db_path)
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('queued-2', ?, 'goal 3', 'goal', 'queued', 'planner', CURRENT_TIMESTAMP)",
+            (self.project_id,),
+        )
+        conn.commit()
+        conn.close()
+
+        with patch.object(
+            task_execution, "start_task_execution_in_background",
+            side_effect=[task_execution.TaskExecutionError("already in flight"), None],
+        ) as mock_start:
+            result = new_goal_dispatch.dispatch_once(self.db_path)
+        self.assertEqual(len(result), 2)
+        self.assertEqual(mock_start.call_count, 2)
+
+    def test_no_candidates_returns_gracefully(self):
+        conn = schema.connect(self.db_path)
+        conn.execute("UPDATE task SET status = 'done'")
+        conn.commit()
+        conn.close()
+
+        with patch.object(task_execution, "start_task_execution_in_background") as mock_start:
+            result = new_goal_dispatch.dispatch_once(self.db_path)
+        self.assertEqual(result, [])
+        mock_start.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""trigger_org_node_structured's real-call coverage (2026-08,
+work-graph-node-execution-deblackbox-requirements.md FR-4 -- the migration
+from Dynamic Workflow scripts to flat `claude --agent` calls for the three
+call sites in work_graph_orchestrate.py/e2e_pipeline.py that used to render
+one-off `.js` Workflow scripts purely to get "agent identity + schema-shaped
+output"). Gated behind RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing constraint) --
+costs real API usage, skipped by default, same as src/test_agent_e2e.py.
+
+Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_org_node_trigger_e2e -v
+
+**Real result, recorded 2026-08-13** (one real call, `--agent validator
+--model claude-haiku-4-5`, target org_node's persona = agents/validator/SOUL.md
+via .claude/agents/validator.md): all three FR-4 questions confirmed together
+in a single real subprocess call (spied via `wraps=subprocess.run` so the
+production `trigger_org_node_structured` code path runs completely for real,
+while this test still gets to inspect the raw CompletedProcess underneath it):
+
+1. **Persona loaded correctly** -- the self-identify answer named itself an
+   independent QA/verification agent (PASS/FAIL/INCONCLUSIVE judgment
+   framework, "does not write or fix the Coder's business code"), matching
+   agents/validator/SOUL.md's own red lines, not the generic Claude Code
+   assistant persona. Same technique U-031 used to validate the (now-retired)
+   Dynamic Workflow `agentType` path -- ask the agent to self-identify, check
+   for persona-specific vocabulary, not a canned/generic answer.
+2. **`--model` override took effect for real** -- `claude --agent <org_node>`
+   combined with `--model claude-haiku-4-5` is a genuinely different CLI
+   combination from anything DSL-internal Workflow options ever validated
+   (05 §5.4.1.5's "two different paths" lesson) -- confirmed empirically, not
+   assumed: the raw JSON response's own `modelUsage` key names
+   "claude-haiku-4-5" as the model that actually served the request, not
+   validator's role-declared default (`claude-sonnet-5`, roles/validator/
+   role.json).
+3. **```json fence extraction is stable on real model output** -- the
+   real free-text response (identity narrative + a trailing ```json fenced
+   block, since `--json-schema` is never combined with `--agent`, see 05
+   §5.4.1.1) parsed cleanly via `extract_json_payload`/`trigger_org_node_
+   structured`'s required-keys matching, returning the schema-shaped dict
+   with no manual fallback needed -- the same extraction algorithm
+   `work_graph_plan.py::plan_work_graph()` already validated for real in
+   production (`_extract_graph_manifest`), now proven for `_JSON_FENCE_
+   INSTRUCTION`'s more generic prompt wording too.
+"""
+
+import json
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from src import agent_profile, org_node_trigger, project_scaffold, schema
+from src.test_support import skip_unless_env
+
+_SELF_IDENTIFY_SCHEMA = {
+    "type": "object",
+    "required": ["job", "never_does"],
+    "properties": {
+        "job": {"type": "string", "description": "one short sentence describing your role"},
+        "never_does": {"type": "string", "description": "one short sentence: something you never do"},
+    },
+}
+
+
+@skip_unless_env("RUN_REAL_AGENT_E2E")
+class RealTriggerOrgNodeStructuredTest(unittest.TestCase):
+    def test_identity_model_override_and_json_extraction_all_real(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
+            conn = schema.connect(":memory:")
+            schema.create_schema(conn)
+            project_scaffold.create_project(conn, tmp, init_git_if_missing=False)
+            agent_profile.deploy_agent_definitions(tmp)
+
+            real_run = subprocess.run
+            captured = {}
+
+            def spying_run(*args, **kwargs):
+                result = real_run(*args, **kwargs)
+                captured["result"] = result
+                return result
+
+            # wraps the real subprocess.run rather than replacing it -- this
+            # call goes out to the real `claude` CLI exactly like production
+            # code would, this test just also gets to inspect the raw
+            # CompletedProcess trigger_org_node_structured's own return value
+            # (the extracted payload alone) doesn't expose.
+            with patch("src.org_node_trigger.subprocess.run", side_effect=spying_run):
+                extracted = org_node_trigger.trigger_org_node_structured(
+                    "validator",
+                    "In one short sentence each: what is your job, and what do you never do? "
+                    "Answer only, take no other action.",
+                    tmp,
+                    _SELF_IDENTIFY_SCHEMA,
+                    model="claude-haiku-4-5",
+                    timeout=120,
+                )
+
+        # (3) JSON fence extraction worked on real model output -- a dict
+        # with both required keys, not None/a StructuredOutputExtractionError.
+        self.assertIn("job", extracted)
+        self.assertIn("never_does", extracted)
+
+        # (1) persona genuinely loaded -- validator's own red lines
+        # (agents/validator/SOUL.md), not a generic assistant answer.
+        identity_text = extracted["job"] + " " + extracted["never_does"]
+        self.assertTrue(
+            any(kw in identity_text for kw in (
+                "QA", "验收", "PASS", "FAIL", "INCONCLUSIVE", "independent", "verif", "Verif",
+            )),
+            "validator's answer doesn't look like the migrated QA identity: " + identity_text,
+        )
+
+        # (2) --model override genuinely took effect -- the real response's
+        # own modelUsage key names the override model, not validator's
+        # role-declared default (claude-sonnet-5, roles/validator/role.json).
+        raw_payload = json.loads(captured["result"].stdout)
+        model_usage = raw_payload.get("modelUsage") or {}
+        self.assertIn(
+            "claude-haiku-4-5", model_usage,
+            "--model override not reflected in the real response's modelUsage: " + json.dumps(raw_payload)[:2000],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -90,6 +90,7 @@ class GetTaskTest(TaskQueryTestBase):
         self.assertEqual(result["artifacts"], [])
         self.assertEqual(result["bug_records"], [])
         self.assertEqual(result["work_graph_runs"], [])
+        self.assertEqual(result["planner_review_events"], [])
  
     def test_includes_artifacts(self):
         self._insert_task("task-1")
@@ -239,6 +240,52 @@ class GetTaskTest(TaskQueryTestBase):
         self.assertEqual(node["quota_wait_started_at"], "2026-07-31 09:00:00")
  
  
+    def test_includes_planner_review_events_for_the_task(self):
+        # 2026-08-13 (Org Graph unification Phase 3, FR-5a): planner_reviewer
+        # has no work_graph_run of its own -- its verdicts live in
+        # execution_event (event_type='planner_review_verdict'), see
+        # e2e_pipeline.py::_record_planner_review_event.
+        self._insert_task("task-1")
+        self.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES ('evt-1', 'proj-1', 'task-1', 'planner_review_verdict', ?)",
+            (json.dumps({"attempt": 1, "verdict": "REJECTED", "used_existing_artifacts": False}),),
+        )
+        self.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES ('evt-2', 'proj-1', 'task-1', 'planner_review_verdict', ?)",
+            (json.dumps({"attempt": 2, "verdict": "PASS", "used_existing_artifacts": True}),),
+        )
+        self.conn.commit()
+
+        result = task_query.get_task(self.conn, "task-1")
+        events = result["planner_review_events"]
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0]["attempt"], 1)
+        self.assertEqual(events[0]["verdict"], "REJECTED")
+        self.assertEqual(events[0]["used_existing_artifacts"], False)
+        self.assertEqual(events[1]["attempt"], 2)
+        self.assertEqual(events[1]["verdict"], "PASS")
+        self.assertEqual(events[1]["used_existing_artifacts"], True)
+
+    def test_no_planner_review_events_is_an_empty_list_not_missing_key(self):
+        self._insert_task("task-1")
+        result = task_query.get_task(self.conn, "task-1")
+        self.assertEqual(result["planner_review_events"], [])
+
+    def test_other_task_planner_review_events_excluded(self):
+        self._insert_task("task-1")
+        self._insert_task("task-2")
+        self.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES ('evt-1', 'proj-1', 'task-2', 'planner_review_verdict', ?)",
+            (json.dumps({"attempt": 1, "verdict": "PASS", "used_existing_artifacts": False}),),
+        )
+        self.conn.commit()
+        result = task_query.get_task(self.conn, "task-1")
+        self.assertEqual(result["planner_review_events"], [])
+
+
 class ListTasksSortTest(TaskQueryTestBase):
     def setUp(self):
         super().setUp()