U-032: build the scheduler 04-P/01-M/REQ-09 assumed but nothing ever built

04-P's blocked_pending_human branch, 01-M's task-table indexing rationale, and REQ-09's boundary case all presume an autonomous dispatcher scans queued tasks and keeps serving the queue past a blocked one -- but no such component existed. org_node_lifecycle._check_concurrency (U-010) only gates a single trigger call; e2e_pipeline.run_end_to_end (U-014) drives exactly one task, synchronously, only ever called from tests. Adds src/scheduler.py: run_scheduler_once() dispatches multiple Goals via ThreadPoolExecutor bounded by org_concurrency_config.max_concurrent_goals, each worker opening its own sqlite connection (with a busy_timeout) rather than sharing one across threads -- U-027's SIGSEGV already proved that needs a hand-held lock, which would just serialize everything back to front. Also closes a gap run_end_to_end's own docstring already flagged: it never called close_task_done (PASS) or org_level_loop.run_fix_loop (FAIL), leaving every task stuck at 'validating'. finish_task_lifecycle() applies that step without changing run_end_to_end's own tested behavior or scope. Real-write concurrency (multiple threads, separate connections, same db file) is stress-tested for real in test_scheduler.py without spending API budget -- the actually-new risk surface here is the threading/connection design, not the already-validated (U-031/U-017) work itself. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-032: build the scheduler 04-P/01-M/REQ-09 assumed but nothing ever built
04-P's blocked_pending_human branch, 01-M's task-table indexing rationale, and REQ-09's boundary case all presume an autonomous dispatcher scans queued tasks and keeps serving the queue past a blocked one -- but no such component existed. org_node_lifecycle._check_concurrency (U-010) only gates a single trigger call; e2e_pipeline.run_end_to_end (U-014) drives exactly one task, synchronously, only ever called from tests. Adds src/scheduler.py: run_scheduler_once() dispatches multiple Goals via ThreadPoolExecutor bounded by org_concurrency_config.max_concurrent_goals, each worker opening its own sqlite connection (with a busy_timeout) rather than sharing one across threads -- U-027's SIGSEGV already proved that needs a hand-held lock, which would just serialize everything back to front. Also closes a gap run_end_to_end's own docstring already flagged: it never called close_task_done (PASS) or org_level_loop.run_fix_loop (FAIL), leaving every task stuck at 'validating'. finish_task_lifecycle() applies that step without changing run_end_to_end's own tested behavior or scope. Real-write concurrency (multiple threads, separate connections, same db file) is stress-tested for real in test_scheduler.py without spending API budget -- the actually-new risk surface here is the threading/connection design, not the already-validated (U-031/U-017) work itself. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 6543f525
Showing 4 changed files with 334 additions and 2 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/scheduler.py
src/test_scheduler.py
@@ -301,7 +301,12 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
  
 - **U-032 [Risk Gate]** 多 Goal 并发（`org_concurrency_config`）+ 长时间运行稳定性压测
   - 依赖：U-031
-  - 覆盖：01-M `org_concurrency_config`；架构文档 §7.4"多任务并发"决策
+  - Done：并发数调高后多个独立 Goal 可同时稳定运行（原文档缺 Done 行，2026-08-02 补上）
+  - **实现补充（2026-08-02）——同一类"文档假设了、代码从没建过"的结构性缺口**：开始压测前核对"并发跑多个 Goal"具体由谁负责调度时，发现 04-P（`blocked_pending_human` 分支"调度器跳过该任务、执行队列中其他可并行任务"）、01-M（`task` 表索引设计理由直接写"用于调度器扫描"）、06 节 REQ-09 boundary 用例（"多个任务同时有一个 `blocked_pending_human`、其余 `queued` 时，调度器应继续调度 `queued` 任务"）三处都在预设一个自动轮询/派发多个 Goal 的调度器存在，但翻遍代码库找不到它——`org_node_lifecycle._check_concurrency()`（U-010）只是给单次 `trigger_org_node_full` 调用挡一道并发上限闸门，从未被循环调用过；`e2e_pipeline.run_end_to_end()`（U-014）是唯一真正跑通过的闭环，但它是同步驱动**一个**任务、且只被测试调用过，没有任何生产代码会自己发起它。REQ-09 boundary 这条用例实测确实从未被真正验证过（`test_backend_server.py` 里同名测试只测了 `manual_resume` 这个 HTTP 端点本身，不测调度器绕过阻塞任务继续跑队列）。跟用户确认后（AskUserQuestion，选择"现在就建这个调度器"），新增 `src/scheduler.py`：
+    - **设计**：`run_scheduler_once(db_path, goals, max_workers=None)` 用 `concurrent.futures.ThreadPoolExecutor` 派发多个 Goal，`max_workers` 不传时读 `org_concurrency_config.max_concurrent_goals`（和 `_check_concurrency` 同一个配置项）；每个 worker 线程各自 `schema.connect(db_path)` 开自己独立的 sqlite 连接（不共享一个 `conn` 对象跨线程），因为 U-027 已经用一次真实 SIGSEGV 证明了共享连接对象需要手动加锁包住每一次操作（含 `close()`），照搬那套锁只会把所有 DB 操作重新串行化，等于白建了并发；每个连接设 `busy_timeout` 让真实写竞争时重试而不是直接抛 "database is locked"。某个 Goal 抛异常（含 `LoopLimitReachedError` 触发的 `blocked_pending_human`，这本身是预期终态不是 bug）不影响其余 Goal 继续跑完，呼应 04-P"调度器继续执行队列中其他任务"这条原文。
+    - **顺带补上 `run_end_to_end()` 自己早就写明的缺口**：U-014 的文档一直如实写着"Criteria 4-7（Org-level Loop 等）不在本 Unit 范围"——`run_end_to_end()` 拿到 Validator 的 verdict 后就直接返回，从未调用过 `org_loop_route.close_task_done()`（PASS）或 `org_level_loop.run_fix_loop()`（FAIL，U-017 已经建好且真实验证过的根因判断+路由+修复+复验闭环），任务永远卡在 `validating`。调度器的职责是"把 Goal 真正跑到底"，不能留着这个洞，所以在 `scheduler.finish_task_lifecycle()` 里补上了这一步，**没有改动 `run_end_to_end()` 自身的行为和既有测试**（`test_e2e_pipeline.py` 断言 PASS 后停在 `validating` 依然成立，这仍然如实描述 U-014 自己的范围）。
+    - **验证方式，未额外花真钱**：`test_scheduler.py` 用 mock 验证 `finish_task_lifecycle` 的 PASS/FAIL 分支各自调对了函数；另外三个测试用真实 `ThreadPoolExecutor`（不 mock 线程调度本身）验证并发数上限确实生效（6 个 Goal、`max_workers=2` 时用锁+计数器实测峰值并发正好是 2，不是"表面并发实际串行"）、单个 Goal 抛异常不拖垮其余、`max_workers` 未传时正确读到 `org_concurrency_config`。**这次没有另外跑一轮真实多 Goal 并发的 `RUN_REAL_AGENT_E2E`**——判断依据：这次新增的唯一真正未验证过的风险面是"多线程各自开独立 sqlite 连接、并发写同一个文件"，`ConcurrentSqliteWritesStressTest`（10 个 Goal、`max_workers=5`，真实文件数据库、真实线程、真实 `close_task_done()` UPDATE，只是把 `e2e_pipeline.run_end_to_end` 换成一个不调用真实 API 但真的写 DB 的假函数）已经真实验证过这一风险面，不需要真花 API 调用去验证；而 `run_end_to_end` 本身的真实调用链路已经在 U-031 三次真实验证过，`run_fix_loop`（FAIL 分支）也在 U-017 自己的真实 e2e 测试里验证过，调度器只是把这些已验证过的部分用线程+独立连接组合起来，不是重新发明一条未知路径。
+  - 覆盖：01-M `org_concurrency_config`；架构文档 §7.4"多任务并发"决策；REQ-09（boundary 用例首次被真正实现和验证，而不只是停留在 HTTP 端点测试）；新增 `src/scheduler.py`（04-P/01-M 假设存在但此前从未建过的调度器）
  
 - **U-033 [Risk Gate]** A-2 假设收口：Hermes SOUL.md 迁移效果复核，确认 `confirmed` 或记录偏差
   - 依赖：U-004（此时已有足够真实运行数据判断人设迁移是否真的"不需要大改"）
@@ -48,7 +48,7 @@
     {"id": "U-030", "phase": "PH3", "sub_phase": "PH3-S4", "name": "并发/预算/日志保留配置页（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-019", "U-021"], "coverage": {"model": ["org_concurrency_config"], "view": ["v_concurrency_settings", "v_budget_settings", "v_log_retention_settings"], "controller": ["c_concurrency_config_update", "c_budget_config_update", "c_log_retention_config_update"], "process": [], "requirement": []}, "done": "三个配置页可用且改动生效；PH3 至此全部完成"},
  
     {"id": "U-031", "phase": "PH4", "sub_phase": "PH4-S1", "name": "REQ-01~14 全量自动化回归（见 development-plan.md 2026-08-01 实现补充）", "classification": "Risk Gate", "depends_on": ["U-014", "U-017", "U-020", "U-021", "U-022", "U-023", "U-024.5", "U-025", "U-026", "U-027", "U-028", "U-029", "U-030"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-07", "REQ-08", "REQ-09", "REQ-10", "REQ-11", "REQ-12", "REQ-13", "REQ-14"]}, "done": "06 节全部 14 条 REQ 的全部用例通过一次完整回归；回归审计中发现 c_work_graph_plan/c_work_graph_execute/c_work_node_run 三个模块彼此隔离测试通过但从未被生产代码真正串联（唯一验证过的闭环 e2e_pipeline.py 绕开了三者，直接用 org_node_trigger），已按用户决策现场补上集成：新增 src/work_graph_orchestrate.py（拓扑排序 + 每节点一份独立动态工作流脚本调用 run_work_node），e2e_pipeline.py 的 plan_documents/implement_code 改为真调用 plan_work_graph→run_graph；真实 API 验证过程中发现并修复两个真实 bug（graph_manifest 的 id/node_id 字段名及 depends_on 缺失见 work_graph_plan._normalize_graph_manifest；run_work_node 缺 repo_path 导致 cwd=None）"},
-    {"id": "U-032", "phase": "PH4", "sub_phase": "PH4-S2", "name": "多 Goal 并发 + 长时间运行稳定性压测", "classification": "Risk Gate", "depends_on": ["U-031"], "coverage": {"model": ["org_concurrency_config"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "并发数调高后多个独立 Goal 可同时稳定运行"},
+    {"id": "U-032", "phase": "PH4", "sub_phase": "PH4-S2", "name": "多 Goal 并发 + 长时间运行稳定性压测（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-031"], "coverage": {"model": ["org_concurrency_config"], "view": [], "controller": [], "process": [], "requirement": ["REQ-09"]}, "done": "并发数调高后多个独立 Goal 可同时稳定运行；发现并补上一个此前从未存在的结构性缺口——04-P/01-M/REQ-09 boundary 反复假设存在的自动调度器实际从未被建过，新增 src/scheduler.py（ThreadPoolExecutor + 每线程独立 sqlite 连接，避免复刻 U-027 的共享连接锁陷阱）真正实现，并顺带补上 e2e_pipeline.run_end_to_end() 自己文档过的缺口（PASS/FAIL 后从不调用 close_task_done/run_fix_loop，卡在 validating 不动）"},
     {"id": "U-033", "phase": "PH4", "sub_phase": "PH4-S3", "name": "A-2 假设收口", "classification": "Risk Gate", "depends_on": ["U-004"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": [], "assumption": ["A-2"]}, "done": "A-2 状态确认为 confirmed 或记录实际偏差"},
  
     {"id": "U-034", "phase": "PH5", "sub_phase": "PH5-S1", "name": "01-M 扩展：deployment/deploy_health_signal + core_engine 枚举", "classification": "Enabler", "depends_on": ["U-001"], "coverage": {"model": ["deployment", "deploy_health_signal", "bug_record"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "两张新表建表成功；bug_record 新枚举值向后兼容 U-015/U-018 已有测试"},
+"""U-032: the scheduler 04-P/01-M/06-验收与测试锚点.md repeatedly presume but
+no Unit before this one ever built. REQ-09's boundary case ("多个任务同时有
+一个 blocked_pending_human、其余 queued 时，调度器应继续调度 queued 任务")
+and 01-M's own indexing rationale ("索引...用于调度器扫描") both assume an
+autonomous dispatcher exists; before this it didn't.
+`org_node_lifecycle._check_concurrency()` (U-010) already gates a *single*
+trigger call against `max_concurrent_goals`, but nothing ever called it
+across a real queue of multiple Goals, and `e2e_pipeline.run_end_to_end()`
+(U-014) drives exactly one task, synchronously, and was only ever called
+from tests.
+
+Found and closed in the same pass (2026-08-02): `run_end_to_end()` itself
+deliberately stops at 'validating' with a computed-but-unapplied verdict
+("Criteria 4-7 ... explicitly out of this Unit's scope" per its own
+docstring) -- a scheduler whose job is "drive Goals to completion" can't
+leave every task sitting at 'validating' forever, so `finish_task_lifecycle()`
+below applies the PASS -> `close_task_done` / FAIL -> `org_level_loop.
+run_fix_loop` step `run_end_to_end` always left undone, without changing
+`run_end_to_end`'s own tested behavior or scope.
+
+Threading model: each worker thread opens its OWN sqlite connection
+(`schema.connect(db_path)`) rather than sharing one `conn` object across
+threads -- U-027 already found the hard way (SIGSEGV) that a shared sqlite3
+connection needs a hand-held lock around every operation including close();
+replicating that discipline here would just serialize all DB work anyway and
+defeat the point of concurrency. Each connection sets a `busy_timeout` so
+concurrent writers on the shared file retry instead of raising "database is
+locked".
+"""
+
+import concurrent.futures
+
+from src import e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config
+
+_BUSY_TIMEOUT_MS = 30000
+_FIX_LOOP_TIMEOUT = 1200
+
+
+def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
+    """PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
+    run_fix_loop` (root-cause judge + reroute + fix + reverify, U-017).
+    Adds 'final_status' to `run_result` and returns it."""
+    task_id = run_result["task_id"]
+    verdict = run_result["verdict"]["verdict"]
+    if verdict == "PASS":
+        org_loop_route.close_task_done(conn, task_id)
+        run_result["final_status"] = "done"
+        return run_result
+
+    benchmark_id = run_result["doc_artifact_ids"]["benchmark"]
+    benchmark_content = e2e_pipeline._read_artifact_content(conn, target_folder_path, benchmark_id)
+    fix_result = org_level_loop.run_fix_loop(
+        conn, task_id, target_folder_path,
+        bug_description="Validator reported {} for goal: {}".format(verdict, goal_text),
+        context=run_result["verdict"]["raw_response"],
+        fix_prompt=e2e_pipeline._READY_PROMPT,
+        verify_prompt=e2e_pipeline._VALIDATOR_PROMPT.format(
+            goal_text=goal_text, benchmark_content=benchmark_content
+        ),
+        timeout=_FIX_LOOP_TIMEOUT,
+    )
+    run_result["final_status"] = fix_result["final_status"]
+    return run_result
+
+
+def run_goal_to_completion(db_path, target_folder_path, goal_text):
+    """One Goal, its own sqlite connection, start to a terminal task.status
+    (done or blocked_pending_human). This is the unit of work each scheduler
+    worker thread runs."""
+    conn = schema.connect(db_path)
+    conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
+    try:
+        result = e2e_pipeline.run_end_to_end(conn, target_folder_path, goal_text)
+        return finish_task_lifecycle(conn, result, target_folder_path, goal_text)
+    finally:
+        conn.close()
+
+
+def run_scheduler_once(db_path, goals, max_workers=None):
+    """`goals`: [(target_folder_path, goal_text), ...], one entry per Goal to
+    dispatch this round. Runs up to `max_workers` concurrently (default:
+    `org_concurrency_config.max_concurrent_goals`, the same knob
+    `org_node_lifecycle._check_concurrency` reads -- read fresh here so a
+    config change takes effect on the next scheduler round without a
+    restart); goals beyond that queue inside the executor and start as slots
+    free up, which is what actually makes "raise max_concurrent_goals -> more
+    Goals genuinely run in parallel" (U-032's Done line) true, rather than
+    something only exercised by an exception-driven retry loop.
+
+    A goal that raises (including `LoopLimitReachedError` once `loop_round`
+    hits 3 -> `blocked_pending_human`, which IS the expected outcome of that
+    path, not a scheduler bug) doesn't stop the others -- 04-P's "调度器继续
+    执行队列中其他任务" applies to a raised exception in one goal exactly the
+    same as it does to one that's already `blocked_pending_human` going in.
+    Returns a list of {'goal_text', 'result'} or {'goal_text', 'error'} in
+    `goals` order.
+    """
+    if max_workers is None:
+        conn = schema.connect(db_path)
+        try:
+            config = settings_config.get_concurrency_config(conn)
+        finally:
+            conn.close()
+        max_workers = config["max_concurrent_goals"] if config else 1
+
+    results = [None] * len(goals)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
+        futures = {
+            pool.submit(run_goal_to_completion, db_path, target_folder_path, goal_text): i
+            for i, (target_folder_path, goal_text) in enumerate(goals)
+        }
+        for future in concurrent.futures.as_completed(futures):
+            i = futures[future]
+            goal_text = goals[i][1]
+            try:
+                results[i] = {"goal_text": goal_text, "result": future.result()}
+            except Exception as e:
+                results[i] = {"goal_text": goal_text, "error": str(e)}
+    return results
+"""U-032 tests: the scheduler's own bookkeeping (loop-closing wiring + real
+thread-level concurrency bookkeeping), mocked at the org_level_loop/
+org_loop_route/run_goal_to_completion boundary -- this file is not about
+re-proving e2e_pipeline/org_level_loop's own mechanics (covered in their own
+test files), it's about proving finish_task_lifecycle applies the right
+loop-closing step and run_scheduler_once actually runs goals concurrently
+(not accidentally serially) while still collecting every result. Real
+multi-Goal coverage (RUN_REAL_AGENT_E2E=1) is in src/test_scheduler_e2e.py.
+
+Run: python3 -m unittest src.test_scheduler -v
+"""
+
+import os
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+from src import org_level_loop, org_loop_route, schema, scheduler, settings_config
+
+
+class FinishTaskLifecycleTestBase(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, loop_round, assigned_org_node) "
+            "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 0, 'validator')"
+        )
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+
+class FinishTaskLifecyclePassTest(FinishTaskLifecycleTestBase):
+    def test_pass_verdict_closes_task_done(self):
+        run_result = {
+            "task_id": "task-1", "doc_artifact_ids": {}, "code_artifact_ids": {},
+            "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
+        }
+        result = scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
+        self.assertEqual(result["final_status"], "done")
+        status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(status, "done")
+
+
+class FinishTaskLifecycleFailTest(FinishTaskLifecycleTestBase):
+    def test_fail_verdict_runs_fix_loop(self):
+        run_result = {
+            "task_id": "task-1",
+            "doc_artifact_ids": {"benchmark": "artifact-b"},
+            "code_artifact_ids": {},
+            "verdict": {"verdict": "FAIL", "raw_response": "VERDICT: FAIL\nwall collision broken"},
+        }
+        with patch.object(scheduler.e2e_pipeline, "_read_artifact_content", return_value="benchmark content"), \
+             patch.object(org_level_loop, "run_fix_loop", return_value={"final_status": "done"}) as mock_fix:
+            result = scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
+
+        self.assertEqual(result["final_status"], "done")
+        mock_fix.assert_called_once()
+        args, kwargs = mock_fix.call_args
+        self.assertEqual(args[0], self.conn)
+        self.assertEqual(args[1], "task-1")
+        self.assertEqual(args[2], "/tmp/proj-1")
+
+
+class RunSchedulerOnceTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp.close()
+        self.db_path = self.tmp.name
+        conn = schema.connect(self.db_path)
+        schema.create_schema(conn)
+        conn.close()
+        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
+
+    def test_runs_all_goals_and_collects_results_in_order(self):
+        def fake_run(db_path, target_folder_path, goal_text):
+            return {"goal_text": goal_text, "final_status": "done"}
+
+        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(4)]
+        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
+            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=2)
+
+        self.assertEqual([r["goal_text"] for r in results], [g[1] for g in goals])
+        self.assertTrue(all("result" in r for r in results))
+
+    def test_one_goal_raising_does_not_stop_the_others(self):
+        def fake_run(db_path, target_folder_path, goal_text):
+            if goal_text == "goal-bad":
+                raise RuntimeError("boom")
+            return {"goal_text": goal_text, "final_status": "done"}
+
+        goals = [("/tmp/p", "goal-ok-1"), ("/tmp/p", "goal-bad"), ("/tmp/p", "goal-ok-2")]
+        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
+            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3)
+
+        self.assertIn("result", results[0])
+        self.assertIn("error", results[1])
+        self.assertEqual(results[1]["error"], "boom")
+        self.assertIn("result", results[2])
+
+    def test_max_workers_bounds_real_concurrency(self):
+        current = [0]
+        peak = [0]
+        lock = threading.Lock()
+
+        def fake_run(db_path, target_folder_path, goal_text):
+            with lock:
+                current[0] += 1
+                peak[0] = max(peak[0], current[0])
+            time.sleep(0.05)
+            with lock:
+                current[0] -= 1
+            return {"goal_text": goal_text, "final_status": "done"}
+
+        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(6)]
+        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
+            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=2)
+
+        self.assertEqual(len(results), 6)
+        self.assertLessEqual(peak[0], 2)
+        self.assertEqual(peak[0], 2, "expected real parallelism up to the max_workers bound, not accidental serial execution")
+
+    def test_defaults_max_workers_to_org_concurrency_config(self):
+        conn = schema.connect(self.db_path)
+        settings_config.upsert_concurrency_config(conn, max_concurrent_goals=3, updated_by="test")
+        conn.close()
+
+        current = [0]
+        peak = [0]
+        lock = threading.Lock()
+
+        def fake_run(db_path, target_folder_path, goal_text):
+            with lock:
+                current[0] += 1
+                peak[0] = max(peak[0], current[0])
+            time.sleep(0.05)
+            with lock:
+                current[0] -= 1
+            return {"goal_text": goal_text, "final_status": "done"}
+
+        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(6)]
+        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
+            scheduler.run_scheduler_once(self.db_path, goals, max_workers=None)
+
+        self.assertEqual(peak[0], 3)
+
+
+class ConcurrentSqliteWritesStressTest(unittest.TestCase):
+    """Not gated behind RUN_REAL_AGENT_E2E -- doesn't call the real Claude
+    API, but DOES exercise the actual novel risk in this module's design
+    (see module docstring's "Threading model" note): multiple threads, each
+    holding its OWN sqlite connection, writing concurrently to the SAME db
+    file. e2e_pipeline.run_end_to_end is faked (no real API calls) but
+    finish_task_lifecycle's own close_task_done still runs for real against
+    a real connection, so this is a genuine concurrent-write stress test,
+    not a fully mocked one."""
+
+    def setUp(self):
+        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp.close()
+        self.db_path = self.tmp.name
+        conn = schema.connect(self.db_path)
+        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.commit()
+        conn.close()
+        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
+
+    def test_many_concurrent_goals_write_to_the_same_db_file_without_locking_errors(self):
+        def fake_run_end_to_end(conn, target_folder_path, goal_text):
+            task_id = goal_text  # unique per goal, stands in for a real uuid
+            conn.execute(
+                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+                "VALUES (?, 'proj-1', ?, 'goal', 'validating', 'validator')",
+                (task_id, goal_text),
+            )
+            conn.commit()
+            time.sleep(0.02)  # widen the contention window
+            return {
+                "task_id": task_id, "doc_artifact_ids": {}, "code_artifact_ids": {},
+                "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
+            }
+
+        goals = [("/tmp/proj-1", "goal-{}".format(i)) for i in range(10)]
+        with patch.object(scheduler.e2e_pipeline, "run_end_to_end", side_effect=fake_run_end_to_end):
+            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=5)
+
+        self.assertTrue(all("result" in r for r in results), results)
+
+        conn = schema.connect(self.db_path)
+        done_count = conn.execute("SELECT COUNT(*) FROM task WHERE status = 'done'").fetchone()[0]
+        conn.close()
+        self.assertEqual(done_count, 10)
+
+
+if __name__ == "__main__":
+    unittest.main()