From 5d21b17ab1d91dc059657af06cabd3552971a329 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Sun, 2 Aug 2026 01:31:07 +0800 Subject: [PATCH] U-032: build the scheduler 04-P/01-M/REQ-09 assumed but nothing ever built --- docs/development-plan.md | 7 ++++++- docs/sdd/unit_matrix.json | 2 +- src/scheduler.py | 119 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/test_scheduler.py | 208 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 334 insertions(+), 2 deletions(-) create mode 100644 src/scheduler.py create mode 100644 src/test_scheduler.py diff --git a/docs/development-plan.md b/docs/development-plan.md index e7c9e73..dfaa664 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -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（此时已有足够真实运行数据判断人设迁移是否真的"不需要大改"） diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 6828815..109a279 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -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 已有测试"}, diff --git a/src/scheduler.py b/src/scheduler.py new file mode 100644 index 0000000..a291351 --- /dev/null +++ b/src/scheduler.py @@ -0,0 +1,119 @@ +"""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 diff --git a/src/test_scheduler.py b/src/test_scheduler.py new file mode 100644 index 0000000..1da7cb2 --- /dev/null +++ b/src/test_scheduler.py @@ -0,0 +1,208 @@ +"""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() -- libgit2 0.21.4