| """会話ごとの自由な仕事。本物の頭脳・本番データは使わない。""" |
| import contextlib |
| import datetime |
| import importlib.util |
| import json |
| import os |
| from pathlib import Path |
| import sys |
| import tempfile |
| import threading |
| import unittest |
| from unittest import mock |
| from urllib.request import Request, urlopen |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| sys.path.insert(0, str(ROOT / "kernel")) |
| import chats |
| import conversation |
| import loop |
| import server |
|
|
| spec = importlib.util.spec_from_file_location("test_work_hako", ROOT / "dougu/hako.py") |
| hako = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(hako) |
|
|
|
|
|
|
| def _dougu_jiyuu(): |
| """他の試験が本番 ~/LocalAI_mirror/kernel を sys.path に足すと、本番の jiyuu が読まれる。写しの dougu/jiyuu.py を必ず使う。""" |
| want = str(ROOT / "dougu" / "jiyuu.py") |
| mod = sys.modules.get("jiyuu") |
| if mod is None or getattr(mod, "__file__", "") != want: |
| sys.modules.pop("jiyuu", None) |
| sys.path.insert(0, str(ROOT / "dougu")) |
| try: |
| import jiyuu as mod |
| finally: |
| sys.path.remove(str(ROOT / "dougu")) |
| mod.gate.hako = mod._hako |
| return mod |
|
|
| class ConversationWorkTests(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.root = Path(self.tmp.name).resolve() |
| self.old = chats.DIR, chats.DB, chats._READY |
| chats.DIR, chats.DB, chats._READY = str(self.root / "legacy"), str(self.root / "chats.sqlite3"), False |
| self.cid = chats.create()["id"] |
|
|
| def tearDown(self): |
| chats.DIR, chats.DB, chats._READY = self.old |
| self.tmp.cleanup() |
|
|
| def test_activity_isolated_bounded_and_reopened(self): |
| other = chats.create()["id"] |
| self.assertEqual(chats.load(other)["動き"], []) |
| for i in range(207): |
| chats.add_activity(self.cid, {"type": "note", "text": str(i)}) |
| chats.add_activity(other, {"text": "別の会話"}) |
| chats._READY = False |
| activity = chats.load(self.cid)["動き"] |
| self.assertEqual(len(activity), 200) |
| self.assertEqual(activity[0]["text"], "7") |
| self.assertEqual(chats.load(other)["動き"][0]["text"], "別の会話") |
| chats.clear_activity(self.cid) |
| self.assertEqual(chats.activities(self.cid), []) |
| self.assertEqual(len(chats.activities(other)), 1) |
|
|
| def fill(self, pairs=6): |
| for i in range(pairs): |
| chats.add_turn(self.cid, "user", f"頼み{i}") |
| chats.add_turn(self.cid, "bot", f"答え{i}") |
|
|
| def test_compact_replaces_previous_summary_and_keeps_recent_pairs(self): |
| self.fill() |
| chats.update_state(self.cid, summary="前回の要約", folders=[str(self.root)]) |
| summarize = mock.Mock(return_value="新しい要約") |
| conversation.compact(chats, self.cid, summarize) |
| c = chats.load(self.cid) |
| self.assertEqual(c["要約"], "新しい要約") |
| self.assertEqual([t["文"] for t in c["やりとり"]], [f"{role}{i}" for i in range(3, 6) for role in ("頼み", "答え")]) |
| self.assertEqual(summarize.call_args.args[1], "前回の要約") |
| self.assertEqual(conversation.messages(c)[0]["role"], "system") |
| self.assertIn("新しい要約", conversation.messages(c)[0]["content"]) |
| self.assertEqual(c["触ってよいフォルダ"], [str(self.root)]) |
| self.fill(2) |
| conversation.compact(chats, self.cid, lambda old, previous: previous + "を更新") |
| self.assertEqual(chats.load(self.cid)["要約"], "新しい要約を更新") |
|
|
| def test_compact_failure_or_concurrent_message_does_not_discard_turns(self): |
| self.fill() |
| before = chats.load(self.cid) |
| with self.assertRaises(ValueError): |
| conversation.compact(chats, self.cid, lambda *_: "") |
| self.assertEqual(chats.load(self.cid), before) |
| def changed(*_): |
| chats.add_turn(self.cid, "user", "追加") |
| return "要約" |
| with self.assertRaises(ValueError): |
| conversation.compact(chats, self.cid, changed) |
| self.assertEqual(len(chats.load(self.cid)["やりとり"]), 13) |
| self.assertEqual(chats.load(self.cid)["要約"], "") |
|
|
| def test_cancelled_compact_and_branch_preserve_context(self): |
| self.fill() |
| stop = threading.Event() |
| stop.set() |
| with self.assertRaises(ValueError): |
| conversation.compact(chats, self.cid, lambda *_: "要約", stop=stop) |
| self.assertEqual(len(chats.load(self.cid)["やりとり"]), 12) |
| chats.update_state(self.cid, summary="決定", folders=[str(self.root)], loop={"running": True}) |
| branch = chats.fork(self.cid) |
| self.assertEqual(branch["要約"], "決定") |
| self.assertEqual(branch["触ってよいフォルダ"], [str(self.root)]) |
| self.assertEqual(branch["loop"], {}) |
|
|
| def test_stream_connects_mock_tools_to_the_correct_conversation(self): |
| import queue |
| sys.path.append(str(ROOT / "dougu")) |
| jiyuu = _dougu_jiyuu() |
| import kyoudou |
| import teachers |
| old_settings = server.CTX["設定"] |
| server.CTX["設定"] = {"モード": "本番", "輪": "新"} |
| chats.update_state(self.cid, summary="決定を引き継ぐ", folders=[str(self.root)]) |
| captured = {} |
| def answer(text, on_event, **args): |
| captured.update(args) |
| on_event({"type": "tool_start", "id": "1", "name": "read", "label": "読む"}) |
| on_event({"type": "tool_end", "id": "1", "ok": True, "summary": "確認"}) |
| return "完了" |
| try: |
| with contextlib.ExitStack() as stack: |
| stack.enter_context(mock.patch.object(server, "_gakushuu_busy", side_effect=contextlib.nullcontext)) |
| stack.enter_context(mock.patch.object(server, "_temoto_tsukau", side_effect=contextlib.nullcontext)) |
| stack.enter_context(mock.patch.object(server, "moderu_youi", return_value=(True, "mock"))) |
| stack.enter_context(mock.patch.object(teachers, "mado_settei")) |
| stack.enter_context(mock.patch.object(kyoudou, "is_shortcut", return_value=False)) |
| stack.enter_context(mock.patch.object(jiyuu, "kotaeru", side_effect=answer)) |
| result = server._run_request("読む", self.cid, queue.Queue(), threading.Event(), "kyoudou", []) |
| self.assertTrue(result["出力"].endswith("完了")) |
| self.assertEqual(captured["settei"]["要約"], "決定を引き継ぐ") |
| self.assertEqual(captured["settei"]["触ってよいフォルダ"], [str(self.root)]) |
| self.assertEqual([e["type"] for e in chats.activities(self.cid)], ["tool_start", "tool_end"]) |
| self.assertEqual(chats.load(chats.create()["id"])["動き"], []) |
| finally: |
| server.CTX["設定"] = old_settings |
|
|
| def test_token_meter_uses_props_tokenize_and_fallback(self): |
| self.fill(1) |
| def api(path, payload=None): |
| return {"/props": {"default_generation_settings": {"n_ctx": 100}}, |
| "/apply-template": {"prompt": "会話"}, "/tokenize": {"tokens": list(range(38))}}[path] |
| meter = conversation.Meter(api) |
| measured = meter.measure(chats.load(self.cid)) |
| self.assertEqual((measured["使った"], measured["上限"], measured["割合"]), (38, 100, 38)) |
| self.assertFalse(measured["推定"]) |
| missing = mock.Mock(side_effect=OSError("未提供")) |
| result = conversation.Meter(missing).measure(chats.load(self.cid), {"コンテキスト上限": 1234}) |
| self.assertEqual(result["上限"], 1234) |
| self.assertTrue(result["推定"]) |
| self.assertGreater(result["使った"], 0) |
|
|
| def make_loop(self, runner, lesson=None): |
| self.clock = [datetime.datetime(2026, 10, 4, 12).timestamp()] |
| return loop.Manager(chats, runner, now=lambda: self.clock[0], lesson=lesson) |
|
|
| def test_loop_rounds_learn_failures_interval_limit_and_restore(self): |
| prompts = [] |
| def runner(cid, prompt, stop): |
| prompts.append(prompt) |
| if len(prompts) == 1: |
| raise ValueError("試みAは失敗") |
| return {"ok": True, "result": "方法Bを確認", "next": "方法C"} |
| lessons = mock.Mock() |
| manager = self.make_loop(runner, lessons) |
| manager.start(self.cid, "技術を作る", 1800, maximum=3) |
| manager.tick() |
| manager.tick() |
| self.assertEqual(len(prompts), 1) |
| self.clock[0] += 1800 |
| restored = loop.Manager(chats, runner, now=lambda: self.clock[0], lesson=lessons) |
| restored.tick() |
| self.assertIn("試みAは失敗", prompts[1]) |
| self.clock[0] += 1800 |
| restored.tick() |
| self.assertIn("方法C", prompts[2]) |
| state = chats.get_state(self.cid)["loop"] |
| self.assertFalse(state["running"]) |
| self.assertEqual(state["round"], 3) |
| self.assertEqual(lessons.call_count, 2) |
|
|
| def test_loop_stop_during_round_and_night_pause(self): |
| manager = None |
| def runner(cid, prompt, stop): |
| manager.stop(cid) |
| self.assertTrue(stop.is_set()) |
| return {"ok": True, "result": "停止"} |
| manager = self.make_loop(runner) |
| manager.start(self.cid, "お題", 10) |
| manager.tick() |
| self.assertFalse(chats.get_state(self.cid)["loop"]["running"]) |
| manager.start(self.cid, "次のお題", 10) |
| self.clock[0] = datetime.datetime(2026, 10, 5, 1).timestamp() |
| manager.tick() |
| state = chats.get_state(self.cid)["loop"] |
| self.assertEqual(state["round"], 0) |
| self.assertTrue(state["running"]) |
| self.assertEqual(datetime.datetime.fromtimestamp(state["next"]).hour, 7) |
|
|
| def test_loop_inflight_restart_is_recorded_and_not_repeated(self): |
| runner = mock.Mock(return_value={"ok": False, "result": "失敗"}) |
| manager = self.make_loop(runner) |
| manager.start(self.cid, "お題") |
| state = chats.get_state(self.cid)["loop"] |
| state.update(inflight=True, round=1) |
| chats.update_state(self.cid, loop=state) |
| manager.tick() |
| self.assertIn("再起動で中断", runner.call_args.args[1]) |
| self.assertEqual(chats.get_state(self.cid)["loop"]["round"], 2) |
|
|
| def test_loop_command_parser(self): |
| self.assertEqual(loop.parse("/loop 30m 新しい技術を作って")["interval"], 1800) |
| self.assertEqual(loop.parse("/loop お題")["interval"], 600) |
| self.assertTrue(loop.parse("/loop 止める")["stop"]) |
| for text in ("/loop", "/loop 0s お題", "/loop 30m"): |
| with self.assertRaises(ValueError): |
| loop.parse(text) |
|
|
| def test_api_folders_and_activity_are_scoped_and_commands_listed(self): |
| old_settings, old_id = server.CTX["設定"], server.CTX.get("会話id") |
| server.CTX["設定"] = {"モード": "試験"} |
| httpd = server.Server(("127.0.0.1", 0), server.Handler) |
| thread = threading.Thread(target=httpd.serve_forever, daemon=True) |
| thread.start() |
| def api(path, body=None): |
| request = Request("http://127.0.0.1:%d%s" % (httpd.server_address[1], path), |
| data=None if body is None else json.dumps(body).encode(), |
| headers={"X-Token": server.TOKEN, "Content-Type": "application/json"}) |
| with urlopen(request) as response: |
| return json.load(response) |
| try: |
| api("/chat/folders", {"id": self.cid, "folders": [str(self.root)]}) |
| chats.add_activity(self.cid, {"type": "note", "text": "本人の会話"}) |
| self.assertEqual(api("/ugoki")["動き"], []) |
| self.assertEqual(api("/ugoki?cid=" + self.cid)["動き"][0]["text"], "本人の会話") |
| with mock.patch.object(server._CONTEXT_METER, "measure", return_value={"使った": 1}): |
| view = api("/chat/context?cid=" + self.cid) |
| self.assertEqual(view["触ってよいフォルダ"], [str(self.root)]) |
| self.assertIn("/compact", [x["名"] for x in api("/commands")]) |
| self.assertIn("/loop", [x["名"] for x in api("/commands")]) |
| with mock.patch.object(server, "_LOOPS", self.make_loop(mock.Mock())): |
| server._special_request("/loop お題", self.cid, threading.Event()) |
| api("/chat/loop/stop", {"id": self.cid}) |
| self.assertFalse(chats.get_state(self.cid)["loop"]["running"]) |
| finally: |
| httpd.shutdown() |
| httpd.server_close() |
| thread.join(2) |
| server.CTX["設定"], server.CTX["会話id"] = old_settings, old_id |
|
|
|
|
| class SandboxWorkTests(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.home = Path(self.tmp.name).resolve() / "home" |
| self.work = self.home / "Documents" |
| self.work.mkdir(parents=True) |
|
|
| def tearDown(self): |
| self.tmp.cleanup() |
|
|
| def test_parent_selection_never_allows_private_paths_or_symlinks(self): |
| private = [".claude", ".codex", ".ssh", ".claude.json", ".groq.env", ".future.env", |
| "LocalAI_mirror/kernel", "LocalAI_mirror/koukai/dougu/hako.py", "LocalAI_mirror/koukai/kernel/kyoudou.py"] |
| for path in private: |
| for write in (False, True): |
| self.assertFalse(hako.check_path(str(self.home / path), write=write, paths=[str(self.home)], home=str(self.home)), path) |
| self.assertTrue(hako.check_path(str(self.work / "new"), True, [str(self.work)], str(self.home))) |
| self.assertFalse(hako.check_path(str(self.home / "outside"), True, [str(self.work)], str(self.home))) |
| self.assertFalse(hako.check_path(str(self.home / "Documents-other/new"), True, [str(self.work)], str(self.home))) |
| (self.work / "escape").symlink_to(self.home / ".claude") |
| self.assertFalse(hako.check_path(str(self.work / "escape/new"), True, [str(self.work)], str(self.home))) |
|
|
| def test_mac_profile_limits_writes_and_denies_private_reads(self): |
| profile = hako.build_profile("戻せる", home=str(self.home), allowed_roots=[str(self.work)]) |
| self.assertNotIn("(allow file-write*)", profile) |
| self.assertIn(f'(subpath "{self.work}")', profile) |
| for name in (".claude", ".codex", ".ssh", "LocalAI_mirror/kernel"): |
| self.assertIn(f'(deny file-read* (subpath "{self.home / name}"))', profile) |
| self.assertIn(f'(deny file-write* (subpath "{self.home / name}"))', profile) |
| self.assertIn("(allow lsopen)", profile) |
|
|
| def test_linux_readonly_root_and_home_parent_exclusions(self): |
| secret = self.home / ".nvidia.env" |
| secret.write_text("fixture") |
| (self.home / ".ssh").mkdir() |
| with mock.patch.object(hako.shutil, "which", return_value="/usr/bin/bwrap"): |
| argv = hako._bwrap_argv("true", risk="戻せる", env={"HOME": str(self.home)}, allowed_roots=[str(self.home)]) |
| triples = [argv[i:i+3] for i in range(len(argv)-2)] |
| self.assertIn(["--ro-bind", "/", "/"], triples) |
| self.assertIn(["--bind", str(self.work), str(self.work)], triples) |
| self.assertNotIn(["--bind", str(self.home), str(self.home)], triples) |
| self.assertIn(["--ro-bind", "/dev/null", str(secret)], triples) |
| self.assertIn(["--tmpfs", str(self.home / ".ssh"), "--remount-ro"], triples) |
|
|
| def test_jiyuu_uses_new_sandbox_and_guards_direct_write(self): |
| sys.path.append(str(ROOT / "dougu")) |
| jiyuu = _dougu_jiyuu() |
| self.assertTrue(hasattr(jiyuu.gate.hako, "scope")) |
| with jiyuu.gate.hako.scope([str(self.work)]): |
| result = jiyuu._run("write", {"path": str(self.home / "outside"), "content": "禁止"}, "戻せる", "mock") |
| self.assertFalse(result["ok"]) |
| result = jiyuu._run("write", {"path": str(self.work / "allowed"), "content": "許可"}, "戻せる", "mock") |
| self.assertTrue(result["ok"]) |
| self.assertEqual((self.work / "allowed").read_text(), "許可") |
| with mock.patch.object(jiyuu.gate.hako, "_sandbox_executable", return_value="sandbox-exec"): |
| argv = jiyuu.gate.hako.command_argv("true", risk="戻せる") |
| self.assertIn(f'(subpath "{self.work}")', argv[2]) |
| self.assertNotIn("(allow file-write*)", argv[2]) |
|
|
| def test_jiyuu_summary_is_first_and_full_history_is_preserved(self): |
| sys.path.append(str(ROOT / "dougu")) |
| jiyuu = _dougu_jiyuu() |
| with contextlib.ExitStack() as stack: |
| for name in ("_system", "_knowledge_hint", "_memory_hint", "_gakushuu_hint", "_kanjou_hint", "_skill_hint", "_user_context"): |
| stack.enter_context(mock.patch.object(jiyuu, name, return_value="")) |
| stack.enter_context(mock.patch.object(jiyuu, "_record")) |
| ask = stack.enter_context(mock.patch.object(jiyuu, "_ask", return_value={"content": "答えです。"})) |
| stack.enter_context(mock.patch.object(jiyuu.gate, "TOMERU", None)) |
| jiyuu.kotaeru("頼み", rireki=[{"role": "user", "text": "長"*1500}] * 8, |
| settei={"要約": "要約の決定", "触ってよいフォルダ": [str(self.work)]}) |
| messages = ask.call_args.args[0] |
| self.assertEqual(messages[0]["content"], "") |
| self.assertIn("要約の決定", messages[1]["content"]) |
| self.assertEqual(sum(m.get("content") == "長"*1500 for m in messages), 8) |
|
|
| def test_loop_cannot_leave_background_commands_running(self): |
| sys.path.append(str(ROOT / "dougu")) |
| jiyuu = _dougu_jiyuu() |
| with mock.patch.object(jiyuu, "_job") as job: |
| result = jiyuu._run("sh", {"command": "true", "background": True}, "戻せる", "mock", settei={"反復": True}) |
| self.assertFalse(result["ok"]) |
| job.assert_not_called() |
|
|
| def test_shell_variable_expansion_cannot_bypass_write_scope(self): |
| sys.path.append(str(ROOT / "dougu")) |
| jiyuu = _dougu_jiyuu() |
| with mock.patch.dict(os.environ, {"KOUKAI_TEST_OUTSIDE": str(self.work / "outside")}): |
| with jiyuu.gate.hako.scope([str(Path.home())]): |
| with mock.patch.object(jiyuu.gate, "_write_file") as write: |
| result = jiyuu._run("write", {"path": "$KOUKAI_TEST_OUTSIDE", "content": "禁止"}, "戻せる", "mock") |
| self.assertFalse(result["ok"]) |
| write.assert_not_called() |
|
|
