G: Work Graph node parallel execution (topological_layers + threaded run_graph)

Implements docs/todo/work-graph-parallel-execution-design.md: run_graph() now schedules graph_manifest nodes layer-by-layer (topological_layers(), Kahn's algorithm grouped by round instead of flattened), running a multi-node layer's nodes concurrently via ThreadPoolExecutor (default 4 workers) with a barrier between layers; a single-node layer stays synchronous, matching every real graph_manifest observed so far. topological_order() is now a thin flattening wrapper, unchanged behavior (regression-tested). _LockedConn/_FetchedRows makes the existing scattered conn.execute() call sites in work_node_run.py/work_graph_orchestrate.py/execution_log.py safe to call from multiple threads without touching those call sites -- note the lock must cover the full execute()-through-fetchone()/fetchall() cycle, not just execute() itself, or two threads can still collide on the shared sqlite3 connection handle. That gap was caught for real (not theoretically) by RunGraphRealSqliteConcurrencyTest during development -- committing the fixed version, not the first draft. create_worktree() (work_node_run.py) gets a module-level lock around the `git worktree add` call for concurrent layers touching the same repo. New tests: layer grouping, same-layer concurrency + inter-layer barrier, one node's failure not cancelling a concurrent sibling (WorkGraphLayerError aggregates), breakpoint-recovery resume of a concurrent layer, a quota-wait timeout coexisting with a sibling's real persisted output, real (non-mocked) sqlite under concurrent writes across several rounds, and real concurrent `git worktree add` calls. Full src (575 tests) + supervisor (125 tests) suites pass with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

G: Work Graph node parallel execution (topological_layers + threaded run_graph)
Implements docs/todo/work-graph-parallel-execution-design.md: run_graph() now schedules graph_manifest nodes layer-by-layer (topological_layers(), Kahn's algorithm grouped by round instead of flattened), running a multi-node layer's nodes concurrently via ThreadPoolExecutor (default 4 workers) with a barrier between layers; a single-node layer stays synchronous, matching every real graph_manifest observed so far. topological_order() is now a thin flattening wrapper, unchanged behavior (regression-tested). _LockedConn/_FetchedRows makes the existing scattered conn.execute() call sites in work_node_run.py/work_graph_orchestrate.py/execution_log.py safe to call from multiple threads without touching those call sites -- note the lock must cover the full execute()-through-fetchone()/fetchall() cycle, not just execute() itself, or two threads can still collide on the shared sqlite3 connection handle. That gap was caught for real (not theoretically) by RunGraphRealSqliteConcurrencyTest during development -- committing the fixed version, not the first draft. create_worktree() (work_node_run.py) gets a module-level lock around the `git worktree add` call for concurrent layers touching the same repo. New tests: layer grouping, same-layer concurrency + inter-layer barrier, one node's failure not cancelling a concurrent sibling (WorkGraphLayerError aggregates), breakpoint-recovery resume of a concurrent layer, a quota-wait timeout coexisting with a sibling's real persisted output, real (non-mocked) sqlite under concurrent writes across several rounds, and real concurrent `git worktree add` calls. Full src (575 tests) + supervisor (125 tests) suites pass with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent b2908009
Showing 4 changed files with 391 additions and 9 deletions
docs/todo/graph-engineering-open-items.md
src/test_work_graph_orchestrate.py
src/test_work_node_run.py
src/work_graph_orchestrate.py
@@ -106,7 +106,7 @@
 ## G. Work Graph 节点并行执行（Python 线程池方案）
-**状态**：✅ 已确认，标记为核心优先级 → 需求文档见 `docs/work-graph-parallel-execution-requirements.md`
+**状态**：✅ 已实现（2026-08-06）→ 需求文档见 `docs/work-graph-parallel-execution-requirements.md`
 **来源**：14 步版对照 §3 开放问题 1
@@ -114,7 +114,7 @@
 **确认结果（2026-08-05）**：确认要做，用户强调"这个很核心"。触发规则明确为：**只要 Work Graph 规划了 fan-out/fan-in 的并行节点，节点触发时就应该并行执行**——这跟需求文档 FR-1/FR-2 已经写的内容一致，不需要改需求文档的功能需求本身，只需要在文档里把这句话作为明确的验收口径固化下来。
-**待办**：✅ 详细设计已产出 → `docs/work-graph-parallel-execution-design.md`。核心设计：新增 `topological_layers()`（层内并发、层间屏障），`_LockedConn` 线程安全代理复用 `backend_server.py` 的锁模式，`create_worktree()` 加防御性锁，`WorkGraphLayerError` 聚合层内多个失败。已知遗留限制（诚实记录，非本轮解决）：单节点配额等待可长达数小时，会拖住整层其他已完成节点的返回——需要更细粒度的可中断调度才能解决，超出本次范围。下一步：实施（见设计文档第 11 节的分步实施建议），或先评审设计。
+**待办**：✅ 已实现（2026-08-06），按设计文档第 11 节的分步实施顺序落地——`src/work_graph_orchestrate.py`：`topological_layers()`（层内并发、层间屏障，`topological_order()` 改为薄封装）、`_LockedConn`/`_FetchedRows`（线程安全 conn 代理，**注意**：实现比设计文档原稿更严格——锁必须覆盖 `execute()` 到 `fetchone()`/`fetchall()` 的整个周期，不能只锁 `execute()` 本身；这个差异是 `RunGraphRealSqliteConcurrencyTest` 真实跑出 `database is locked`/空 `file_path` 类错误后修的，不是理论推导）、`run_graph()` 改层级调度 + `_run_one_node()` + `WorkGraphLayerError`；`src/work_node_run.py`：`create_worktree()` 加 `_worktree_creation_lock`。新增测试：`src/test_work_graph_orchestrate.py`（`TopologicalLayersTest`、`RunGraphConcurrencyTest` 同层并发+层间屏障、`RunGraphLayerErrorTest` 失败不取消同层兄弟节点、`RunGraphConcurrentResumeTest`、`RunGraphQuotaCoexistsWithNormalCompletionTest`、`RunGraphRealSqliteConcurrencyTest` 真实 sqlite 多轮并发）、`src/test_work_node_run.py`（`ConcurrentCreateWorktreeTest` 真实 git worktree 并发创建）。已知遗留限制（设计文档已诚实记录，本轮未解决）：单节点配额等待可长达数小时，会拖住整层其他已完成节点的返回——需要更细粒度的可中断调度才能解决，超出本次范围。`python3 -m unittest discover -s src`（575 tests）+ `discover -s supervisor -p "test_*.py"`（125 tests）全绿，零回归。
 ---
@@ -128,7 +128,7 @@
 | D | ✅ 已确认，需求文档已产出 | `docs/work-graph-time-travel-debugging-requirements.md` |
 | E | ✅ 已确认，需求文档已产出（范围比原开放问题更大） | `docs/dynamic-work-graph-design-skill-requirements.md` |
 | F | ✅ 已确认，需求文档已产出 | `docs/verifier-fan-out-jury-requirements.md` |
-| G | ✅ 已确认，核心优先级，需求文档已产出 | `docs/work-graph-parallel-execution-requirements.md` |
+| G | ✅ 已实现（2026-08-06） | `docs/work-graph-parallel-execution-requirements.md` + `docs/work-graph-parallel-execution-design.md` |
 **当前状态**：A 已落地；B/C/D/E/F/G 六项全部完成需求文档（BRD/PRD 层面），均未进入详细设计（SDD 层面）。
@@ -10,6 +10,8 @@ Run: python3 -m unittest src.test_work_graph_orchestrate -v
 import json
 import os
 import tempfile
+import threading
+import time
 import unittest
 from unittest.mock import patch
@@ -65,6 +67,70 @@ class TopologicalOrderTest(unittest.TestCase):
         self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b"])
+class TopologicalLayersTest(unittest.TestCase):
+    """G: topological_layers() -- the layer-grouped algorithm run_graph()'s
+    concurrent scheduler is built on. topological_order() is now a thin
+    flattening wrapper over this (tested above, unchanged behavior)."""
+
+    def test_diamond_produces_three_layers(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "n1", "depends_on": []},
+                {"node_id": "n2", "depends_on": ["n1"]},
+                {"node_id": "n3", "depends_on": ["n1"]},
+                {"node_id": "n4", "depends_on": ["n2", "n3"]},
+            ],
+            "edges": [],
+        }
+        layers = wgo.topological_layers(manifest)
+        self.assertEqual(
+            [[n["node_id"] for n in layer] for layer in layers],
+            [["n1"], ["n2", "n3"], ["n4"]],
+        )
+
+    def test_linear_chain_is_all_single_node_layers(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "brd", "depends_on": []},
+                {"node_id": "prd", "depends_on": ["brd"]},
+                {"node_id": "sdd", "depends_on": ["prd"]},
+            ],
+            "edges": [],
+        }
+        layers = wgo.topological_layers(manifest)
+        self.assertEqual(
+            [[n["node_id"] for n in layer] for layer in layers],
+            [["brd"], ["prd"], ["sdd"]],
+        )
+
+    def test_independent_nodes_are_a_single_layer(self):
+        manifest = {"nodes": [{"node_id": "b"}, {"node_id": "a"}], "edges": []}
+        layers = wgo.topological_layers(manifest)
+        self.assertEqual(len(layers), 1)
+        self.assertEqual([n["node_id"] for n in layers[0]], ["a", "b"])
+
+    def test_cycle_raises(self):
+        manifest = {
+            "nodes": [{"node_id": "x", "depends_on": ["y"]}, {"node_id": "y", "depends_on": ["x"]}],
+            "edges": [],
+        }
+        with self.assertRaises(ValueError):
+            wgo.topological_layers(manifest)
+
+    def test_topological_order_matches_flattened_layers(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "n1", "depends_on": []},
+                {"node_id": "n2", "depends_on": ["n1"]},
+                {"node_id": "n3", "depends_on": ["n1"]},
+            ],
+            "edges": [],
+        }
+        layers = wgo.topological_layers(manifest)
+        flat = [n["node_id"] for layer in layers for n in layer]
+        self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], flat)
+
+
 class RunGraphTestBase(unittest.TestCase):
     def setUp(self):
         # DU-01: run_fn now really calls execution_log.append_log(), which
@@ -78,7 +144,10 @@ class RunGraphTestBase(unittest.TestCase):
         os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name
         self.addCleanup(self._restore_graphlemon_home)
-        self.conn = schema.connect(":memory:")
+        # G: check_same_thread=False -- run_graph()'s concurrent layers call
+        # into this conn (via _LockedConn) from ThreadPoolExecutor worker
+        # threads, not just the thread that created the connection.
+        self.conn = schema.connect(":memory:", check_same_thread=False)
         schema.create_schema(self.conn)
         self.conn.execute(
             "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
@@ -404,5 +473,254 @@ class RunFnLogsExecutionEventsTest(RunGraphTestBase):
         self.assertEqual(events[1]["quota_limit_type"], "session")
+class RunGraphConcurrencyTest(RunGraphTestBase):
+    """G, design doc §10 acceptance criteria 1-2: same-layer nodes actually
+    run concurrently (overlapping wall-clock intervals), and the next layer
+    never starts until every node in the previous layer has finished."""
+
+    def test_same_layer_nodes_run_concurrently(self):
+        manifest = {
+            "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}],
+            "edges": [],
+        }
+        events = []
+        events_lock = threading.Lock()
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            start = time.monotonic()
+            time.sleep(0.05)
+            end = time.monotonic()
+            with events_lock:
+                events.append((start, end))
+            return {"content": name}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p-" + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA,
+            )
+
+        self.assertEqual(len(events), 2)
+        (start_a, end_a), (start_b, end_b) = events
+        self.assertLess(start_a, end_b)
+        self.assertLess(start_b, end_a)
+
+    def test_next_layer_waits_for_entire_previous_layer(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "a", "depends_on": []},
+                {"node_id": "b", "depends_on": []},
+                {"node_id": "c", "depends_on": ["a", "b"]},
+            ],
+            "edges": [],
+        }
+        events = []
+        events_lock = threading.Lock()
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            start = time.monotonic()
+            time.sleep(0.03)
+            end = time.monotonic()
+            with events_lock:
+                events.append((start, end))
+            return {"content": "x"}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+            )
+
+        self.assertEqual(len(events), 3)
+        layer1_ends = [end for _, end in events[:2]]
+        layer2_start, _ = events[2]
+        self.assertGreaterEqual(layer2_start, max(layer1_ends))
+
+
+class RunGraphLayerErrorTest(RunGraphTestBase):
+    """G, design doc §10 acceptance criterion (FR-5): one node failing in a
+    concurrent layer must not cancel an already-running sibling, and all
+    failures in the layer are aggregated into one WorkGraphLayerError."""
+
+    def test_one_failing_node_does_not_cancel_sibling_and_aggregates_errors(self):
+        manifest = {
+            "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}],
+            "edges": [],
+        }
+        calls = []
+
+        def fake_run_work_node(conn, node_exec_id, input_payload, input_schema, output_schema, run_fn, **kwargs):
+            calls.append(input_payload["node_id"])
+            if input_payload["node_id"] == "a":
+                raise RuntimeError("boom on a")
+            return {"content": "b done"}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_node_run, "run_work_node", side_effect=fake_run_work_node):
+            with self.assertRaises(wgo.WorkGraphLayerError) as ctx:
+                wgo.run_graph(
+                    self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                    lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                )
+
+        self.assertEqual(sorted(calls), ["a", "b"])  # b was actually attempted, not cancelled
+        error_node_ids = [node_id for node_id, exc in ctx.exception.errors]
+        self.assertEqual(error_node_ids, ["a"])
+
+
+class RunGraphConcurrentResumeTest(RunGraphTestBase):
+    """G, design doc §10 acceptance criterion 3: breakpoint-recovery still
+    works when the interrupted run had a concurrent (2+ node) layer -- a
+    second run_graph() call for the same run_id must not re-run nodes the
+    first call already finished."""
+
+    def test_concurrent_layer_resume_reruns_nothing(self):
+        manifest = {
+            "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}],
+            "edges": [],
+        }
+        calls = []
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            calls.append(name)
+            return {"content": "c-" + name}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+            )
+        self.assertEqual(len(calls), 2)
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute2:
+            mock_execute2.side_effect = AssertionError("should not be called -- both nodes already done")
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+            )
+        self.assertEqual(set(outputs.keys()), {"a", "b"})
+
+
+class RunGraphQuotaCoexistsWithNormalCompletionTest(RunGraphTestBase):
+    """G, design doc §10 acceptance criterion 6: a node stuck on a quota-wait
+    timeout in a concurrent layer must not lose its sibling's real, already-
+    persisted output -- the sibling's row is 'done' with its real payload
+    even though the overall layer raises."""
+
+    def test_quota_timeout_on_one_node_does_not_erase_siblings_persisted_output(self):
+        manifest = {
+            "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}],
+            "edges": [],
+        }
+        prompt_by_script = {}
+        prompt_lock = threading.Lock()
+
+        def fake_write_script(cwd, script_name, script_content):
+            with prompt_lock:
+                prompt_by_script[script_name] = script_content
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            content = prompt_by_script[name]
+            if '"p-a"' in content:
+                raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session")
+            return {"content": "b done"}
+
+        now_calls = {"n": 0}
+        now_lock = threading.Lock()
+
+        def fake_now():
+            # first call records quota_wait_started_at as "0"; second call
+            # (computing elapsed) reports far past the session timeout, so
+            # this times out immediately instead of really sleeping.
+            with now_lock:
+                now_calls["n"] += 1
+                return 0.0 if now_calls["n"] == 1 else 999999.0
+
+        with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            with self.assertRaises(wgo.WorkGraphLayerError) as ctx:
+                wgo.run_graph(
+                    self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                    lambda node, outputs: "p-" + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA,
+                    sleep=lambda s: None, now=fake_now,
+                )
+
+        error_node_ids = [node_id for node_id, exc in ctx.exception.errors]
+        self.assertEqual(error_node_ids, ["a"])
+        self.assertIsInstance(dict(ctx.exception.errors)["a"], work_node_run.UsageLimitWaitTimeoutError)
+
+        b_row = self.conn.execute(
+            "SELECT status, output_payload FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'b'"
+        ).fetchone()
+        self.assertEqual(b_row[0], "done")
+        self.assertEqual(json.loads(b_row[1]), {"content": "b done"})
+
+
+class RunGraphRealSqliteConcurrencyTest(unittest.TestCase):
+    """G, design doc §10 acceptance criterion 4: a real (non-:memory:,
+    non-mocked-conn) sqlite database under genuine concurrent write
+    pressure from a multi-node layer must never raise 'database is
+    locked' -- run several rounds since races like this often only
+    reproduce probabilistically."""
+
+    def setUp(self):
+        self.home_tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.home_tmp.cleanup)
+        self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME")
+        os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name
+        self.addCleanup(self._restore_graphlemon_home)
+        self.db_tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.db_tmp.cleanup)
+
+    def _restore_graphlemon_home(self):
+        if self.old_graphlemon_home is None:
+            os.environ.pop("GRAPHLEMON_HOME", None)
+        else:
+            os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home
+
+    def test_multiple_rounds_of_concurrent_layer_writes_never_raise_locked(self):
+        manifest = {
+            "nodes": [{"node_id": n, "depends_on": []} for n in ("a", "b", "c", "d")],
+            "edges": [],
+        }
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            time.sleep(0.005)
+            return {"content": name}
+
+        for round_n in range(5):
+            db_path = os.path.join(self.db_tmp.name, "test_{}.db".format(round_n))
+            conn = schema.connect(db_path, check_same_thread=False)
+            schema.create_schema(conn)
+            conn.execute(
+                "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+                "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+            )
+            conn.execute(
+                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+                "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')"
+            )
+            conn.execute(
+                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+                "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+            )
+            conn.commit()
+
+            try:
+                with patch.object(work_graph_execute, "write_workflow_script"), \
+                     patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+                    outputs = wgo.run_graph(
+                        conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                        lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                    )
+                self.assertEqual(set(outputs.keys()), {"a", "b", "c", "d"})
+            finally:
+                conn.close()
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -7,6 +7,7 @@ so it&#39;s tested for real here, not mocked.
 Run: python3 -m unittest src.test_work_node_run -v
 """
+import concurrent.futures
 import json
 import os
 import subprocess
@@ -618,5 +619,36 @@ class ResumeWorkGraphRunTest(WorkNodeRunTestBase):
             work_node_run.resume_work_graph_run(self.conn, "run-1")
+class ConcurrentCreateWorktreeTest(unittest.TestCase):
+    """G, design doc §10 acceptance criterion 5: `create_worktree()` calls
+    from multiple threads against the SAME repo at the same time must not
+    race on git's shared `.git/worktrees/` metadata -- real git, real
+    threads, no mocking."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        _init_git_repo(self.tmp.name)
+
+    def test_concurrent_create_worktree_calls_all_succeed_with_unique_paths(self):
+        node_exec_ids = ["node-{}".format(i) for i in range(8)]
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+            paths = list(pool.map(
+                lambda node_exec_id: work_node_run.create_worktree(self.tmp.name, node_exec_id),
+                node_exec_ids,
+            ))
+
+        self.assertEqual(len(paths), len(set(paths)), "every worktree path must be unique")
+        for path in paths:
+            self.assertTrue(os.path.isdir(path))
+
+        listing = subprocess.run(
+            ["git", "worktree", "list", "--porcelain"], cwd=self.tmp.name, check=True, capture_output=True, text=True,
+        ).stdout
+        worktree_lines = [line for line in listing.splitlines() if line.startswith("worktree ")]
+        # +1 for the main checkout itself, which `git worktree list` always includes.
+        self.assertEqual(len(worktree_lines), len(node_exec_ids) + 1)
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -97,10 +97,42 @@ class WorkGraphLayerError(Exception):
         super().__init__("{} node(s) failed in one layer: {}".format(len(errors), summary))
+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/
-    commit each take a shared lock for the duration of the call. SQLite only
-    ever allows one writer at a time regardless -- this doesn't create new
+    """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
@@ -117,11 +149,11 @@ class _LockedConn:
     def execute(self, *args, **kwargs):
         with self._lock:
-            return self._conn.execute(*args, **kwargs)
+            return _FetchedRows(self._conn.execute(*args, **kwargs).fetchall())
     def executemany(self, *args, **kwargs):
         with self._lock:
-            return self._conn.executemany(*args, **kwargs)
+            return _FetchedRows(self._conn.executemany(*args, **kwargs).fetchall())
     def commit(self):
         with self._lock: