from __future__ import annotations import contextlib import importlib.util import io import json import sqlite3 import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "agent-continuity.py" SPEC = importlib.util.spec_from_file_location("agent_continuity_cli", SCRIPT) assert SPEC and SPEC.loader cli = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(cli) class AgentContinuityTests(unittest.TestCase): def run_cli(self, home: Path, *args: str) -> dict: stdout = io.StringIO() stderr = io.StringIO() with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = cli.main(["--home", str(home), *args]) self.assertEqual(code, 0, stderr.getvalue()) output = stdout.getvalue().strip() return json.loads(output) if output else {} def test_default_home_uses_agents_data_directory(self) -> None: with mock.patch.dict(cli.os.environ, {"AGENT_CONTINUITY_HOME": ""}, clear=False): with mock.patch.object(cli.Path, "home", return_value=Path("/tmp/test-agent-user")): store = cli.Store() expected_home = Path("/tmp/test-agent-user/.agents/data/agent-continutiy").resolve() self.assertEqual(store.home, expected_home) self.assertEqual(store.db_path, expected_home / "agent-continuity.db") def test_each_session_gets_a_unique_default_branch(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") first = self.run_cli(home, "create", "--title", "第一个会话") second = self.run_cli(home, "create", "--title", "第二个会话") self.assertNotEqual(first["branch_id"], second["branch_id"]) self.assertTrue(first["branch_id"].startswith("br_")) self.assertTrue(second["branch_id"].startswith("br_")) first_branches = self.run_cli(home, "branch-list", first["session_id"]) second_branches = self.run_cli(home, "branch-list", second["session_id"]) self.assertEqual(first_branches[0]["name"], "main") self.assertEqual(second_branches[0]["name"], "main") self.assertTrue(first_branches[0]["active"]) self.assertTrue(second_branches[0]["active"]) def test_new_session_coexists_with_legacy_main_branch(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) store = cli.Store(home=home) with store.connect() as conn: created_at = cli.now_iso() conn.execute( "INSERT INTO sessions(session_id, title, status, active_branch_id, metadata_json, created_at, updated_at) " "VALUES (?, ?, 'active', 'main', '{}', ?, ?)", ("legacy_session", "旧会话", created_at, created_at), ) conn.execute( "INSERT INTO branches(branch_id, session_id, parent_branch_id, name, status, metadata_json, created_at, updated_at) " "VALUES ('main', ?, NULL, 'main', 'active', '{}', ?, ?)", ("legacy_session", created_at, created_at), ) created = self.run_cli(home, "create", "--title", "新会话") self.assertNotEqual(created["branch_id"], "main") self.assertTrue(created["branch_id"].startswith("br_")) self.assertEqual( self.run_cli(home, "branch-list", "legacy_session")[0]["branch_id"], "main", ) def test_lifecycle_probe_resume_expand_and_export(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session = self.run_cli(home, "create", "--title", "恢复测试") session_id = session["session_id"] run = self.run_cli(home, "begin-run", session_id, "--agent", "test-agent") run_id = run["run_id"] self.run_cli(home, "append-user", run_id, "--text", "请设计 SQLite 恢复协议。") self.run_cli(home, "append-assistant", run_id, "--text", "先定义 checkpoint 和事件日志。") self.run_cli(home, "finish-run", run_id, "--status", "completed") self.run_cli( home, "checkpoint", session_id, "--summary-json", json.dumps( { "objective": "设计 SQLite 恢复协议", "completed": ["定义 checkpoint"], "next_action": "实现 CLI", } ), ) self.run_cli( home, "save-native-state", session_id, "--provider", "test-provider", "--kind", "opaque-compaction", "--payload-json", '{"opaque":true}', ) manifest = self.run_cli(home, "probe", session_id) self.assertEqual(manifest["coverage"]["objective"], True) resumed = self.run_cli(home, "resume", session_id, "--budget-tokens", "1200") self.assertEqual(len(resumed["messages"]), 2) expanded = self.run_cli(home, "expand", manifest["next_cursor"], "--query", "SQLite") self.assertEqual(len(expanded["items"]), 1) export_path = home / "export.json" self.run_cli(home, "export", session_id, "--output", str(export_path)) exported = json.loads(export_path.read_text(encoding="utf-8")) self.assertEqual(exported["format"], "agent-continuity-export") self.assertEqual(len(exported["messages"]), 2) self.assertIs(exported["native_states"][0]["portable"], False) verified = self.run_cli(home, "verify") self.assertEqual(verified["integrity"], "ok") def test_message_idempotency(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session_id = self.run_cli(home, "create", "--title", "幂等测试")["session_id"] run_id = self.run_cli(home, "begin-run", session_id, "--agent", "test")["run_id"] first = self.run_cli(home, "append-user", run_id, "--text", "same", "--idempotency-key", "u-1") second = self.run_cli(home, "append-user", run_id, "--text", "same", "--idempotency-key", "u-1") self.assertEqual(first["status"], "recorded") self.assertEqual(second["status"], "existing") verified = self.run_cli(home, "verify") self.assertEqual(verified["messages"], 1) def test_spool_fallback_and_recovery(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session_id = self.run_cli(home, "create", "--title", "spool 测试")["session_id"] with mock.patch.object(cli.Store, "connect", side_effect=sqlite3.OperationalError("offline")): run = self.run_cli(home, "begin-run", session_id, "--agent", "offline-agent") self.assertEqual(run["status"], "spooled") message = self.run_cli( home, "append-user", run["run_id"], "--session-id", session_id, "--text", "必须保留的用户消息", ) self.assertEqual(message["status"], "spooled") finished = self.run_cli( home, "finish-run", run["run_id"], "--session-id", session_id, "--status", "failed", ) self.assertEqual(finished["status"], "spooled") recovered = self.run_cli(home, "recover") self.assertEqual(recovered["processed_records"], 3) verified = self.run_cli(home, "verify") self.assertEqual(verified["messages"], 1) self.assertEqual(verified["pending_runs"], 0) self.assertEqual(verified["spool_files"], 0) def test_message_survives_missing_run(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session_id = self.run_cli(home, "create", "--title", "孤立消息测试")["session_id"] message = self.run_cli( home, "append-user", "run_missing", "--session-id", session_id, "--text", "即使 run 不可恢复也必须保留", ) self.assertEqual(message["status"], "spooled") recovered = self.run_cli(home, "recover") self.assertEqual(recovered["processed_records"], 1) verified = self.run_cli(home, "verify") self.assertEqual(verified["messages"], 1) def test_branch_lease_artifact_auto_checkpoint_and_backup(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session_id = self.run_cli(home, "create", "--title", "扩展功能测试")["session_id"] run = self.run_cli( home, "begin-run", session_id, "--agent", "test", "--lease-ttl-seconds", "60", ) self.run_cli(home, "append-user", run["run_id"], "--text", "实现 artifact 和分支") self.run_cli(home, "append-assistant", run["run_id"], "--text", "已完成基础实现") self.run_cli(home, "finish-run", run["run_id"], "--status", "completed") checkpoint = self.run_cli(home, "checkpoint", session_id) self.assertGreaterEqual(checkpoint["token_estimate"], 1) branches = self.run_cli(home, "branch-create", session_id, "--name", "实验分支") self.assertEqual(branches["copied_messages"], 2) self.assertTrue(branches["switched"]) self.assertEqual(len(self.run_cli(home, "branch-list", session_id)), 2) artifact_path = home / "artifact.txt" artifact_path.write_text("artifact", encoding="utf-8") artifact = self.run_cli( home, "artifact-add", session_id, "--kind", "text", "--path", str(artifact_path), ) self.assertTrue(artifact["sha256"]) exported_path = home / "export.json" self.run_cli(home, "export", session_id, "--output", str(exported_path)) exported = json.loads(exported_path.read_text(encoding="utf-8")) self.assertEqual(len(exported["artifacts"]), 1) self.run_cli(home, "config", "set", "retention.max_backups", "--value", "2") backup = self.run_cli(home, "backup") self.assertTrue(Path(backup["backup"]).is_file()) def test_command_wrapper_persists_visible_output(self) -> None: with tempfile.TemporaryDirectory() as raw_home: home = Path(raw_home) self.run_cli(home, "init") session_id = self.run_cli(home, "create", "--title", "wrapper 测试")["session_id"] runner = SCRIPT.parents[1] / "scripts" / "agent-continuity-run.py" result = subprocess.run( [ sys.executable, str(runner), "--home", str(home), "--session-id", session_id, "--agent", "test-wrapper", "--text", "hello", "--", sys.executable, "-c", "import sys; print('reply:' + sys.stdin.read())", ], capture_output=True, text=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("reply:hello", result.stdout) exported_path = home / "wrapper-export.json" self.run_cli(home, "export", session_id, "--output", str(exported_path)) messages = json.loads(exported_path.read_text(encoding="utf-8"))["messages"] self.assertEqual([message["role"] for message in messages], ["user", "assistant"]) if __name__ == "__main__": unittest.main()