"""会话内 goal / 自动化标记工具单测：LLM 自己判断意图那条路。 不跑真实模型、不联网： - 解析层是纯函数（goal-convert、automation-create、 first_run_at 的 ISO / 无时区 / 毫秒解析与非法跳过）； - automations 层用内存 SQLite，测 first_run_at 到期与 run_once 跑过后不再到期（Python 口径与 SQL due/claim 口径一致）； - worker 投递用 init_db 建的临时库 + 最小任务行，直接调 _deliver_marker_tools，不走 run_one 全量流程。 注意：本文件不测“LLM 真的会写块”——那是模型行为， 且本批刻意不向 LLM 自动注入工具指南（避免 prompt 臃肿）；这里测的是块写出来后系统一定按约定执行，且 非法块一定不静默变调度。工具格式见 docs/marker-tools.md， 需要时由用户在会话里手动告诉 agent。 """ import os import sqlite3 import sys import tempfile import time import unittest from pathlib import Path try: from cloud.automations import ( claim_due_automation, create_automation, due_automations, get_automation, update_automation, ) from cloud.marker_tools import ( extract_automation_creates, extract_goal_converts, parse_first_run_at, ) from cloud.server import init_db from cloud.worker import Worker except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.automations import ( claim_due_automation, create_automation, due_automations, get_automation, update_automation, ) from cloud.marker_tools import ( extract_automation_creates, extract_goal_converts, parse_first_run_at, ) from cloud.server import init_db from cloud.worker import Worker FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} class ParseTestCase(unittest.TestCase): def test_goal_convert_basic_and_alias(self): text = ( "```goal-convert\n{\"goal\": \"收 xxx\"}\n```\n" "```goal_convert\n{\"goal\": \"收 yyy\"}\n```\n") converts = extract_goal_converts(text) # 每轮最多 1 块，第二个被忽略 self.assertEqual([c["goal"] for c in converts], ["收 xxx"]) def test_goal_convert_invalid_skipped(self): text = ( "```goal-convert\nnot json\n```\n" "```goal-convert\n{\"goal\": \" \"}\n```\n" "```goal-convert\n{\"goal\": \"ok\"}\n```\n") self.assertEqual( [c["goal"] for c in extract_goal_converts(text)], ["ok"]) def test_first_run_at_formats(self): iso = parse_first_run_at("2026-10-30T09:00:00+08:00") naive = parse_first_run_at("2026-10-30 09:00") self.assertEqual(iso, naive) self.assertEqual(parse_first_run_at(iso), iso) self.assertEqual(parse_first_run_at(int(iso * 1000)), iso) self.assertIsNone(parse_first_run_at("not a time")) self.assertIsNone(parse_first_run_at("2026-13-40 09:00")) self.assertIsNone(parse_first_run_at(True)) self.assertIsNone(parse_first_run_at(-1)) def test_automation_create_basic(self): text = ( "```automation-create\n" "{\"prompt\": \"看 yyy\", \"interval_seconds\": 86400," " \"first_run_at\": \"2026-10-30T09:00:00+08:00\"," " \"once\": true, \"bind\": \"heartbeat\"}\n```\n") blocks = extract_automation_creates(text) self.assertEqual(len(blocks), 1) block = blocks[0] self.assertEqual(block["prompt"], "看 yyy") self.assertEqual(block["interval_seconds"], 86400) self.assertEqual( block["first_run_at"], parse_first_run_at("2026-10-30T09:00:00+08:00")) self.assertTrue(block["once"]) self.assertEqual(block["bind"], "heartbeat") self.assertIsNone(block["name"]) def test_automation_create_invalid_interval_and_bind_skipped(self): text = ( "```automation-create\n{\"prompt\": \"x\"," " \"interval_seconds\": 30}\n```\n" "```automation-create\n{\"prompt\": \"y\"," " \"bind\": \"weird\"}\n```\n" "```automation-create\n{\"prompt\": \"good\"}\n```\n") blocks = extract_automation_creates(text) self.assertEqual([b["prompt"] for b in blocks], ["good"]) self.assertEqual(blocks[0]["interval_seconds"], 86400) class AutomationsTestCase(unittest.TestCase): def setUp(self): self.conn = sqlite3.connect(":memory:") self.conn.row_factory = sqlite3.Row def tearDown(self): self.conn.close() def test_first_run_at_not_due_early_then_due(self): now = 1700000000.0 auto = create_automation( self.conn, "p", "future", "看 x", "u", now=now, first_run_at=now + 3600) self.assertFalse(auto["due"]) self.assertEqual(auto["next_run_at"], now + 3600) self.assertEqual(due_automations(self.conn, "p", now=now), []) self.assertFalse( claim_due_automation(self.conn, auto["id"], now=now)) self.assertEqual( [a["name"] for a in due_automations( self.conn, "p", now=now + 3600)], ["future"]) self.assertTrue( claim_due_automation(self.conn, auto["id"], now=now + 3600)) def test_run_once_never_due_after_first_run(self): now = 1700000000.0 auto = create_automation( self.conn, "p", "once", "看 y", "u", now=now, first_run_at=now + 10, run_once=True) self.assertTrue( claim_due_automation(self.conn, auto["id"], now=now + 20)) after = get_automation(self.conn, auto["id"]) self.assertFalse(after["due"]) self.assertIsNone(after["next_run_at"]) self.assertEqual( due_automations(self.conn, "p", now=now + 90000), []) def test_update_first_run_at_and_run_once(self): now = 1700000000.0 auto = create_automation( self.conn, "p", "upd", "看 z", "u", now=now) ok, updated = update_automation( self.conn, auto["id"], {"first_run_at": now + 7200, "run_once": True}, now=now) self.assertTrue(ok) self.assertEqual(updated["first_run_at"], now + 7200) self.assertTrue(updated["run_once"]) self.assertFalse(updated["due"]) class WorkerDeliverTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self._insert("PM", status="running", created_by="tester") self._member("tester", "maintainer") def tearDown(self): self.tmp.cleanup() def _insert(self, task_id, status="queued", created_by="tester", project_id="p1"): now = time.time() conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat)" " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)", (task_id, project_id, "do work", status, created_by, now, now)) conn.commit() finally: conn.close() def _member(self, user_id, role): conn = sqlite3.connect(self.db_path) try: conn.execute( "CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY," " username TEXT UNIQUE, api_key_hash TEXT," " created_at REAL)") conn.execute( "CREATE TABLE IF NOT EXISTS project_members(" "project_id TEXT, user_id TEXT, role TEXT," " PRIMARY KEY(project_id, user_id))") conn.execute( "INSERT OR REPLACE INTO project_members" "(project_id, user_id, role) VALUES('p1', ?, ?)", (user_id, role)) conn.execute( "INSERT OR IGNORE INTO users(id, username, api_key_hash," " created_at) VALUES(?, ?, '', 0)", (user_id, user_id)) conn.commit() finally: conn.close() def _worker(self): return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV) def _followup_bodies(self): conn = sqlite3.connect(self.db_path) try: rows = conn.execute( "SELECT body FROM task_followups" " WHERE task_id = 'PM' ORDER BY created_at").fetchall() return [row[0] for row in rows] finally: conn.close() def _automations(self): conn = sqlite3.connect(self.db_path) try: try: rows = conn.execute( "SELECT name, prompt, interval_seconds, run_once" " FROM automations").fetchall() except sqlite3.OperationalError: # 一个自动化都没建过时表还不存在，不是失败 return [] return [tuple(row) for row in rows] finally: conn.close() def test_deliver_automation_create_and_bind(self): worker = self._worker() sender = worker.get_task("PM") future = time.time() + 3600 worker._deliver_marker_tools(sender, { "status": "completed", "task_creates": [], "task_polls": [], "wait_threads": [], "task_cancels": [], "task_retries": [], "goal_converts": [], "automation_creates": [{ "name": None, "prompt": "看 yyy", "interval_seconds": 86400, "first_run_at": future, "once": False, "bind": "heartbeat", "repo_url": None}], }) autos = self._automations() self.assertEqual(len(autos), 1) self.assertEqual(autos[0][1], "看 yyy") conn = sqlite3.connect(self.db_path) try: binding = conn.execute( "SELECT task_id, mode FROM automation_bindings" ).fetchone() bodies = self._followup_bodies() finally: conn.close() self.assertEqual(tuple(binding), ("PM", "heartbeat")) self.assertTrue(any("已创建自动化" in b for b in bodies)) def test_deliver_automation_past_first_run_rejected(self): worker = self._worker() sender = worker.get_task("PM") worker._deliver_marker_tools(sender, { "status": "completed", "task_creates": [], "task_polls": [], "wait_threads": [], "task_cancels": [], "task_retries": [], "goal_converts": [], "automation_creates": [{ "name": "old", "prompt": "看旧", "interval_seconds": 86400, "first_run_at": time.time() - 3600, "once": False, "bind": None, "repo_url": None}], }) self.assertEqual(self._automations(), []) bodies = self._followup_bodies() self.assertTrue(any("已过去" in b for b in bodies)) def test_deliver_goal_convert_queues_after_report(self): worker = self._worker() sender = worker.get_task("PM") worker._deliver_marker_tools(sender, { "status": "completed", "task_creates": [], "task_polls": [], "wait_threads": [], "task_cancels": [], "task_retries": [], "goal_converts": [{"goal": "收 xxx"}], "automation_creates": [{ "name": "daily", "prompt": "看 yyy", "interval_seconds": 86400, "first_run_at": None, "once": False, "bind": None, "repo_url": None}], }) bodies = self._followup_bodies() # 报告先、/goal 后：排空先跑报告轮、再转 goal self.assertTrue(any("已创建自动化" in b for b in bodies)) self.assertEqual(bodies[-1], "/goal 收 xxx") self.assertEqual(len(self._automations()), 1) def test_deliver_goal_task_ignored(self): conn = sqlite3.connect(self.db_path) try: conn.execute( "UPDATE tasks SET goal_mode = 1 WHERE id = 'PM'") conn.commit() finally: conn.close() worker = self._worker() sender = worker.get_task("PM") worker._deliver_marker_tools(sender, { "status": "completed", "task_creates": [], "task_polls": [], "wait_threads": [], "task_cancels": [], "task_retries": [], "goal_converts": [{"goal": "收 xxx"}], "automation_creates": [], }) self.assertEqual(self._followup_bodies(), []) if __name__ == "__main__": unittest.main()