hansjone/oclaw
1
0
Fork 0
mirror of https://github.com/hansjone/oclaw.git synced 2026-10-10 04:00:44 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

重构主控编排与运行时预热链路，统一工作区提示词/专家调度协议并补齐 wiki 记忆注入与写回闭环。

Browse source 
同时收敛启动与运维脚本默认行为（含 wiki worker）、更新 Admin 可观测性与相关测试，降低首轮时延并提高运行稳定性。

Made-with: Cursor
This commit is contained in:
oliver 2026-04-26 08:34:33 +08:00
parent 4a23b715a2
commit dbbe3add6a
14438 changed files with 2693620 additions and 2546 deletions
Download patch file Download diff file Expand all files Collapse all files

260
tests/test_admin_skills_api.py
View file

@ -6,6 +6,7 @@ import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
from fastapi.testclient import TestClient
@ -87,6 +88,22 @@ class AdminSkillsApiTests(unittest.TestCase):
self.assertEqual(r.status_code, 200, r.text)
self.assertTrue((r.json() or {}).get("ok"))
def test_skills_uninstall(self) -> None:
c = self.client.post(
"/admin/api/skills/create",
json={"name": "to_remove_skill", "description": "remove me", "body_markdown": "# remove"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
u = self.client.post("/admin/api/skills/uninstall", json={"name": "to_remove_skill"}, headers=self._h())
self.assertEqual(u.status_code, 200, u.text)
self.assertTrue((u.json() or {}).get("ok"))
ls = self.client.get("/admin/api/skills", headers=self._h())
self.assertEqual(ls.status_code, 200, ls.text)
items = (ls.json() or {}).get("items") or []
self.assertFalse(any(str(x.get("name") or "") == "to_remove_skill" for x in items))
def test_skills_binding_get_and_save(self) -> None:
c = self.client.post(
"/admin/api/skills/create",
@ -118,6 +135,22 @@ class AdminSkillsApiTests(unittest.TestCase):
self.assertTrue(sb.get("enabled"))
self.assertIn("bind_demo_skill", (sb.get("mapping") or {}).get("generalist", []))
def test_skills_mode_get_and_save(self) -> None:
g = self.client.get("/admin/api/skills/mode", headers=self._h())
self.assertEqual(g.status_code, 200, g.text)
gb = g.json() or {}
self.assertTrue(gb.get("ok"))
s = self.client.post(
"/admin/api/skills/mode",
json={"prompt_in_system": True, "toolcall_enabled": False},
headers=self._h(),
)
self.assertEqual(s.status_code, 200, s.text)
sb = s.json() or {}
self.assertTrue(sb.get("ok"))
self.assertTrue(bool(sb.get("prompt_in_system")))
self.assertFalse(bool(sb.get("toolcall_enabled")))
def test_skills_effective_dashboard(self) -> None:
c = self.client.post(
"/admin/api/skills/create",
@ -134,7 +167,7 @@ class AdminSkillsApiTests(unittest.TestCase):
"generalist": [],
"ops": [],
"image": [],
"memory_curator": [],
"memory": [],
},
},
headers=self._h(),
@ -168,6 +201,231 @@ class AdminSkillsApiTests(unittest.TestCase):
self.assertEqual(r.status_code, 200, r.text)
self.assertTrue((r.json() or {}).get("ok"))
def test_workspace_skill_create_test_run_and_failure_feedback(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_demo_skill", "description": "workspace demo", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
body = c.json() or {}
self.assertTrue(body.get("ok"))
r1 = self.client.post(
"/admin/api/skills/test-run",
json={"name": "ws_demo_skill", "args": {"hello": "world"}},
headers=self._h(),
)
self.assertEqual(r1.status_code, 200, r1.text)
result1 = ((r1.json() or {}).get("result") or {})
self.assertTrue(bool(result1.get("ok")))
self.assertIn("result", result1)
# fs_write is disabled by template; write-like arg should be rejected and return classified failure.
r2 = self.client.post(
"/admin/api/skills/test-run",
json={"name": "ws_demo_skill", "args": {"output_path": "../escape.txt"}},
headers=self._h(),
)
self.assertEqual(r2.status_code, 200, r2.text)
result2 = ((r2.json() or {}).get("result") or {})
self.assertFalse(bool(result2.get("ok")))
self.assertIn(str(result2.get("error_code") or ""), {"path_restricted", "runtime_error"})
def test_skills_self_check_endpoint(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_selfcheck_skill", "description": "workspace selfcheck", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
r = self.client.get("/admin/api/skills/self-check", headers=self._h())
self.assertEqual(r.status_code, 200, r.text)
body = r.json() or {}
self.assertTrue(body.get("ok"))
self.assertGreaterEqual(int(body.get("skills_total") or 0), 1)
self.assertGreaterEqual(int(body.get("executable_total") or 0), 1)
self.assertIn("invalid_runtime_entries", body)
self.assertIn("classification_counts", body)
def test_skills_self_check_with_execution_and_classification(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_selfcheck_exec_skill", "description": "workspace selfcheck exec", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
skill_dir = self.skills_root / "_workspace" / "ws_selfcheck_exec_skill"
run_py = skill_dir / "scripts" / "run.py"
run_py.write_text("print('x' * 5000)\n", encoding="utf-8")
skill_md = skill_dir / "SKILL.md"
content = skill_md.read_text(encoding="utf-8")
content = content.replace('"permissions": {"fs_write": false, "net": false, "process": true}', '"permissions": {"fs_write": false, "net": false, "process": true}, "max_output_bytes": 1024')
skill_md.write_text(content, encoding="utf-8")
r = self.client.get("/admin/api/skills/self-check?include_execution=true", headers=self._h())
self.assertEqual(r.status_code, 200, r.text)
body = r.json() or {}
self.assertTrue(body.get("ok"))
self.assertTrue(bool(body.get("execution_checked")))
self.assertGreaterEqual(int(body.get("execution_checked_total") or 0), 1)
counts = body.get("classification_counts") if isinstance(body.get("classification_counts"), dict) else {}
self.assertGreaterEqual(int(counts.get("output_truncated") or 0), 1)
def test_market_install_uses_adapter_and_installs(self) -> None:
pkg = Path(self._tmp.name) / "pkg3"
d = pkg / "demo3"
d.mkdir(parents=True, exist_ok=True)
(d / "SKILL.md").write_text("name: market_adapter_skill\ndescription: demo\n", encoding="utf-8")
z = Path(self._tmp.name) / "demo3.zip"
with zipfile.ZipFile(z, "w") as zf:
zf.write(d / "SKILL.md", arcname="demo3/SKILL.md")
class _FakeAdapter:
provider = "clawhub"
def search(self, query: str, *, limit: int = 20):
_ = (query, limit)
return [{"slug": "owner/demo3", "name": "demo3"}]
def detail(self, slug: str):
_ = slug
return {"slug": "owner/demo3", "latestVersion": "1.0.0"}
def resolve_archive_url(self, *, slug: str, version: str | None = None):
_ = (slug, version)
return z.resolve().as_uri(), "1.0.0"
with patch("oclaw.interfaces.admin.skills_api.get_market_adapter", return_value=_FakeAdapter()):
r = self.client.post(
"/admin/api/skills/market/install",
json={"slug": "owner/demo3"},
headers=self._h(),
)
self.assertEqual(r.status_code, 200, r.text)
rb = r.json() or {}
self.assertTrue(rb.get("ok"))
installed_name = str(((rb.get("result") or {}).get("name") or "")).strip()
self.assertTrue(installed_name)
ls = self.client.get("/admin/api/skills", headers=self._h())
self.assertEqual(ls.status_code, 200, ls.text)
items = (ls.json() or {}).get("items") or []
self.assertTrue(any(str(x.get("name") or "") == installed_name for x in items))
def test_workspace_skill_real_task_style_roundtrip(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_task_skill", "description": "task style", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
# Replace template runtime script with a task-like implementation.
skill_script = self.skills_root / "_workspace" / "ws_task_skill" / "scripts" / "run.py"
skill_script.write_text(
"import json\n"
"import sys\n\n"
"def main():\n"
" payload = json.loads(sys.stdin.read() or '{}')\n"
" args = payload.get('args') or {}\n"
" text = str(args.get('text') or '')\n"
" words = [w for w in text.strip().split(' ') if w]\n"
" out = {\n"
" 'ok': True,\n"
" 'word_count': len(words),\n"
" 'upper': text.upper(),\n"
" 'contains_number': any(ch.isdigit() for ch in text),\n"
" }\n"
" print(json.dumps(out, ensure_ascii=False))\n\n"
"if __name__ == '__main__':\n"
" main()\n",
encoding="utf-8",
)
r = self.client.post(
"/admin/api/skills/test-run",
json={"name": "ws_task_skill", "args": {"text": "oclaw skill 2026"}},
headers=self._h(),
)
self.assertEqual(r.status_code, 200, r.text)
result = ((r.json() or {}).get("result") or {})
self.assertTrue(bool(result.get("ok")))
parsed = result.get("result") if isinstance(result.get("result"), dict) else {}
self.assertEqual(int(parsed.get("word_count") or 0), 3)
self.assertEqual(str(parsed.get("upper") or ""), "OCLAW SKILL 2026")
self.assertTrue(bool(parsed.get("contains_number")))
def test_workspace_skill_failure_classification_timeout(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_timeout_skill", "description": "timeout test", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
skill_dir = self.skills_root / "_workspace" / "ws_timeout_skill"
skill_md = skill_dir / "SKILL.md"
content = skill_md.read_text(encoding="utf-8")
content = content.replace('"permissions": {"fs_write": false, "net": false, "process": true}', '"permissions": {"fs_write": false, "net": false, "process": true}, "timeout_s": 1')
skill_md.write_text(content, encoding="utf-8")
run_py = skill_dir / "scripts" / "run.py"
run_py.write_text(
"import json\nimport sys\nimport time\n\n"
"def main():\n"
" _ = json.loads(sys.stdin.read() or '{}')\n"
" time.sleep(2)\n"
" print(json.dumps({'ok': True}, ensure_ascii=False))\n\n"
"if __name__ == '__main__':\n"
" main()\n",
encoding="utf-8",
)
r = self.client.post(
"/admin/api/skills/test-run",
json={"name": "ws_timeout_skill", "args": {}},
headers=self._h(),
)
self.assertEqual(r.status_code, 200, r.text)
result = ((r.json() or {}).get("result") or {})
self.assertFalse(bool(result.get("ok")))
self.assertEqual(str(result.get("error_code") or ""), "timeout")
def test_workspace_skill_failure_classification_output_limit(self) -> None:
c = self.client.post(
"/admin/api/skills/create-workspace",
json={"name": "ws_output_skill", "description": "output limit test", "runtime_type": "python"},
headers=self._h(),
)
self.assertEqual(c.status_code, 200, c.text)
self.assertTrue((c.json() or {}).get("ok"))
skill_dir = self.skills_root / "_workspace" / "ws_output_skill"
skill_md = skill_dir / "SKILL.md"
content = skill_md.read_text(encoding="utf-8")
content = content.replace('"permissions": {"fs_write": false, "net": false, "process": true}', '"permissions": {"fs_write": false, "net": false, "process": true}, "max_output_bytes": 128')
skill_md.write_text(content, encoding="utf-8")
run_py = skill_dir / "scripts" / "run.py"
run_py.write_text(
"import json\nimport sys\n\n"
"def main():\n"
" _ = json.loads(sys.stdin.read() or '{}')\n"
" print('x' * 4096)\n\n"
"if __name__ == '__main__':\n"
" main()\n",
encoding="utf-8",
)
r = self.client.post(
"/admin/api/skills/test-run",
json={"name": "ws_output_skill", "args": {}},
headers=self._h(),
)
self.assertEqual(r.status_code, 200, r.text)
result = ((r.json() or {}).get("result") or {})
self.assertTrue("stdout" in result)
self.assertLessEqual(len(str(result.get("stdout") or "")), 1024)
if __name__ == "__main__":
unittest.main()

29
tests/test_admin_tool_policy_oclaw_retry_codes.py
View file

@ -131,6 +131,35 @@ class AdminToolPolicyOclawRetryCodesTests(unittest.TestCase):
self.assertEqual(str(detail.get("code") or ""), "invalid_retryable_error_codes")
self.assertEqual(list(detail.get("unknown_retryable_error_codes") or []), ["typo_code_x"])
def test_tool_policy_save_syncs_run_command_env_gate(self) -> None:
token = self._login()
headers = {"authorization": f"Bearer {token}"}
payload = {
"turn_max_tool_workers": 8,
"turn_max_tool_rounds": 8,
"turn_max_context_messages": 80,
"sse_queue_maxsize": 2000,
"tool_log_max_chars": 200000,
"enable_mcp_tools": True,
"enable_plugin_tools": False,
"enable_run_command": True,
"tool_llm_message_max_chars": 0,
"mcp_filesystem_extra_roots": "",
"mcp_env_allowlist": "",
"oclaw_retryable_error_codes": "provider_timeout, context_overflow",
"oclaw_retry_codes_strict_mode": False,
"wecom_longconn_workers": 2,
"wecom_longconn_inbound_queue_maxsize": 200,
}
r = self.client.post("/admin/api/tool-policy", json=payload, headers=headers)
self.assertEqual(r.status_code, 200)
self.assertEqual(os.environ.get("AIA_ENABLE_RUN_COMMAND"), "1")
payload["enable_run_command"] = False
r2 = self.client.post("/admin/api/tool-policy", json=payload, headers=headers)
self.assertEqual(r2.status_code, 200)
self.assertEqual(os.environ.get("AIA_ENABLE_RUN_COMMAND"), "0")
if __name__ == "__main__":
unittest.main()

61
tests/test_agent_context_loader.py Normal file
View file

@ -0,0 +1,61 @@
from __future__ import annotations
from pathlib import Path
import pytest
from oclaw.runtime.agent_context import loader as loader_mod
def _set_project_root(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setattr(loader_mod, "PROJECT_ROOT", tmp_path)
def test_build_role_system_context_reads_runtime_workspaces(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "main"
ws.mkdir(parents=True, exist_ok=True)
(ws / "SOUL.md").write_text("main soul", encoding="utf-8")
(ws / "ROLE_SYSTEM.md").write_text("main role system", encoding="utf-8")
out = loader_mod.build_role_system_context("manager")
assert "# SOUL" in out
assert "main soul" in out
assert "# ROLE_SYSTEM" in out
assert "main role system" in out
def test_build_role_system_context_cache_invalidates_on_file_change(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "ops"
ws.mkdir(parents=True, exist_ok=True)
role_file = ws / "ROLE_SYSTEM.md"
role_file.write_text("v1", encoding="utf-8")
out1 = loader_mod.build_role_system_context("ops")
assert "v1" in out1
role_file.write_text("v2", encoding="utf-8")
out2 = loader_mod.build_role_system_context("ops")
assert "v2" in out2
def test_build_role_system_context_renders_template_vars(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "ops"
ws.mkdir(parents=True, exist_ok=True)
(ws / "ROLE_SYSTEM.md").write_text("experts={{EXPERTS_HINT}}", encoding="utf-8")
out = loader_mod.build_role_system_context("ops", template_vars={"EXPERTS_HINT": "ops,generalist"})
assert "experts=ops,generalist" in out
def test_build_role_system_context_template_var_isolated_by_cache_key(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "ops"
ws.mkdir(parents=True, exist_ok=True)
(ws / "ROLE_SYSTEM.md").write_text("x={{A}}", encoding="utf-8")
a = loader_mod.build_role_system_context("ops", template_vars={"A": "one"})
b = loader_mod.build_role_system_context("ops", template_vars={"A": "two"})
assert "x=one" in a
assert "x=two" in b

34
tests/test_fastapi_startup_prebuild.py Normal file
View file

@ -0,0 +1,34 @@
from __future__ import annotations
from fastapi import FastAPI
from oclaw.interfaces.http import fastapi_app as app_mod
def test_run_startup_hooks_runs_prebuild_warmup(monkeypatch) -> None:
class DummyStore:
def revoke_all_auth_sessions(self) -> int:
return 0
monkeypatch.setattr(app_mod, "SqliteStore", lambda _p: DummyStore())
monkeypatch.setattr(app_mod, "db_path", lambda: "dummy.sqlite")
monkeypatch.setattr(app_mod, "prepare_gateway_plugin_bootstrap", lambda **kwargs: {"ok": True})
monkeypatch.setattr(app_mod, "resolve_runtime_config", lambda: {})
monkeypatch.setattr(app_mod, "skill_runtime_diagnostics", lambda: {"skills_root": "/tmp", "skills_total": 0})
monkeypatch.setattr(app_mod, "warm_expert_workspace_cache", lambda: None)
monkeypatch.setattr(app_mod, "_relocate_root_scan_artifacts", lambda: None)
monkeypatch.setattr(app_mod, "initialize_hooks_runtime", lambda **kwargs: None)
monkeypatch.setattr(app_mod, "trigger_hook_event", lambda **kwargs: None)
monkeypatch.setattr(app_mod, "_resolve_startup_workspace_dirs", lambda _cfg: [("default", "/tmp/ws")])
called: dict[str, object] = {"scheduler": 0}
def _run_prewarm(**kwargs):
called["prewarm"] = kwargs
return {"ok": True, "elapsed_ms": 1, "freeze": {"frozen": True}}
monkeypatch.setattr(app_mod, "run_runtime_prewarm", _run_prewarm)
monkeypatch.setattr(app_mod, "_spawn_periodic_prewarm_loop", lambda: called.__setitem__("scheduler", 1))
monkeypatch.setattr(app_mod, "_prewarm_interval_seconds", lambda: 600)
app_mod._run_startup_hooks(FastAPI())
assert "prewarm" in called
assert called["scheduler"] == 1

4
tests/test_gateway_server_methods.py
View file

@ -271,9 +271,9 @@ def test_skills_status_requires_store_then_lists_skills() -> None:
assert ok["ok"] is True
payload = ok["payload"] or {}
assert isinstance(payload.get("skills"), list)
# Repo ships oclaw/runtime/skills/{main,coding,social}
# Skill names come from SKILL.md frontmatter (not directory names).
names = {str(x.get("name")) for x in payload["skills"] if isinstance(x, dict)}
assert "main" in names
assert "weather" in names
def test_system_set_heartbeats_requires_boolean() -> None:

16
tests/test_memory_tool_aliases.py Normal file
View file

@ -0,0 +1,16 @@
from __future__ import annotations
from oclaw.runtime.tools.catalog import materialize_tool_specs
def test_memory_tools_only_expose_canonical_names() -> None:
specs = materialize_tool_specs(expert="memory")
names = {str(x.name or "") for x in specs}
assert "memory_wiki_status" in names
assert "memory_wiki_get" in names
assert "memory_wiki_search" in names
assert "memory_wiki_lint" in names
assert "memory_wiki_apply" in names
assert all(not str(x).startswith("memory_curator_") for x in names)

182
tests/test_oclaw_gateway_trace.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import json
from types import SimpleNamespace
import pytest
from oclaw.platform.llm.chat_models import LLMResponse
@ -223,18 +224,39 @@ def test_gateway_comprehensive_mode_manager_first_selects_specialist(monkeypatch
class _ManagerModel:
def chat(self, _messages, _tools, *, on_token=None):
return LLMResponse(content='{"route":{"specialist":"image","reason":"needs image edits"}}', tool_calls=[])
return LLMResponse(
content='{"route":{"specialist":"image","reason":"needs image edits"},"dispatch":{"instruction_text":"Please edit the image background."}}',
tool_calls=[],
)
class _Exec:
def __init__(self, model=None):
self.model = model
self.tools = None
self.tools = object()
self.system_prompt = ""
monkeypatch.setattr(
"oclaw.runtime.gateway.get_manager_prompt_prebuild",
lambda **kwargs: {
"manager_context": "manager",
"allowed_fixed": ("generalist", "ops", "image", "memory"),
"allowed_fixed_quoted": '"generalist", "ops", "image", "memory"',
},
)
captured: dict = {}
def _run_agent_core(**kwargs):
data = kwargs.get("data")
captured["exec_text"] = getattr(getattr(data, "msg", None), "text", None)
return SimpleNamespace(outcome=SimpleNamespace(final_text="specialist_answer"))
monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core)
chosen: dict[str, str] = {}
def _factory(sid: str) -> object:
chosen["sid"] = sid
return _Exec()
return _Exec(model=object())
gw = OclawGateway(store=Store())
msg = StandardMessage(
@ -251,9 +273,9 @@ def test_gateway_comprehensive_mode_manager_first_selects_specialist(monkeypatch
assert out.interaction_mode == "comprehensive"
assert out.selected_specialist == "image"
assert chosen.get("sid") == "image"
assert captured.get("exec_text") == "Please edit the image background."
def test_gateway_comprehensive_mode_dynamic_agent_fallback_success(monkeypatch: pytest.MonkeyPatch) -> None:
def test_gateway_comprehensive_mode_has_manager_final_pass(monkeypatch: pytest.MonkeyPatch) -> None:
class Store:
def get_setting(self, _k: str) -> str:
return ""
@ -262,23 +284,39 @@ def test_gateway_comprehensive_mode_dynamic_agent_fallback_success(monkeypatch:
return None
class _ManagerModel:
def __init__(self) -> None:
self.calls = 0
def chat(self, _messages, _tools, *, on_token=None):
return LLMResponse(
content=(
'{"route":{"specialist":"finance","reason":"requires finance domain"},'
'"dynamic_agent":{"name":"finance","system_prompt":"You are a finance specialist.","tool_policy":{"allow_tags":["read"],"allow_tools":["read_file"]},"reason":"dynamic_finance"}}'
),
tool_calls=[],
)
self.calls += 1
if self.calls == 1:
return LLMResponse(
content='{"route":{"specialist":"generalist","reason":"general"},"dispatch":{"instruction_text":"Analyze the finance report and provide key points."}}',
tool_calls=[],
)
return LLMResponse(content="final_from_manager", tool_calls=[])
class _Exec:
def __init__(self, model=None):
self.model = model
self.tools = None
self.tools = object()
self.system_prompt = ""
monkeypatch.setattr("oclaw.runtime.gateway.build_ephemeral_executor", lambda *args, **kwargs: _Exec())
monkeypatch.setattr(
"oclaw.runtime.gateway.get_manager_prompt_prebuild",
lambda **kwargs: {
"manager_context": "manager",
"allowed_fixed": ("generalist", "ops", "image", "memory"),
"allowed_fixed_quoted": '"generalist", "ops", "image", "memory"',
},
)
monkeypatch.setattr(
"oclaw.runtime.gateway.run_agent_core",
lambda **kwargs: SimpleNamespace(outcome=SimpleNamespace(final_text="specialist_answer")),
)
gw = OclawGateway(store=Store())
model = _ManagerModel()
msg = StandardMessage(
session_id="sid-d1",
tenant_id="t1",
@ -289,12 +327,13 @@ def test_gateway_comprehensive_mode_dynamic_agent_fallback_success(monkeypatch:
attachments=[],
metadata={"interaction_mode": "comprehensive", "selected_specialist": "ops"},
)
out = gw.handle_turn(msg=msg, lang="en", executor=_Exec(model=_ManagerModel()))
out = gw.handle_turn(msg=msg, lang="en", executor=_Exec(model=model))
assert out.interaction_mode == "comprehensive"
assert out.selected_specialist == "finance"
assert out.selected_specialist == "generalist"
assert out.reply_text == "final_from_manager"
def test_gateway_comprehensive_mode_dynamic_agent_invalid_falls_back_generalist() -> None:
def test_gateway_comprehensive_mode_dynamic_agent_dispatches_instruction_only(monkeypatch: pytest.MonkeyPatch) -> None:
class Store:
def get_setting(self, _k: str) -> str:
return ""
@ -306,8 +345,9 @@ def test_gateway_comprehensive_mode_dynamic_agent_invalid_falls_back_generalist(
def chat(self, _messages, _tools, *, on_token=None):
return LLMResponse(
content=(
'{"route":{"specialist":"finance","reason":"requires finance domain"},'
'"dynamic_agent":{"name":"finance","system_prompt":"<tool_call>bad</tool_call>","tool_policy":{"allow_tags":["read"]},"reason":"bad_prompt"}}'
'{"route":{"specialist":"dyn:sql","reason":"needs ad-hoc expert"},'
'"dispatch":{"instruction_text":"Write a SQL query to compute daily active users."},'
'"dynamic_agent":{"name":"dyn:sql","system_prompt":"You are a SQL expert.","tool_policy":{"allow_tags":[],"allow_tools":[]},"reason":"dynamic"}}'
),
tool_calls=[],
)
@ -315,29 +355,113 @@ def test_gateway_comprehensive_mode_dynamic_agent_invalid_falls_back_generalist(
class _Exec:
def __init__(self, model=None):
self.model = model
self.tools = None
self.tools = object()
self.system_prompt = ""
picked: dict[str, str] = {}
monkeypatch.setattr(
"oclaw.runtime.gateway.get_manager_prompt_prebuild",
lambda **kwargs: {
"manager_context": "manager",
"allowed_fixed": ("generalist", "ops", "image", "memory"),
"allowed_fixed_quoted": '"generalist", "ops", "image", "memory"',
},
)
monkeypatch.setattr("oclaw.runtime.gateway.build_ephemeral_executor", lambda *args, **kwargs: _Exec(model=object()))
def _factory(sid: str) -> object:
picked["sid"] = sid
return _Exec()
captured: dict = {}
def _run_agent_core(**kwargs):
data = kwargs.get("data")
captured["exec_text"] = getattr(getattr(data, "msg", None), "text", None)
return SimpleNamespace(outcome=SimpleNamespace(final_text="dynamic_specialist_answer"))
monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core)
gw = OclawGateway(store=Store())
msg = StandardMessage(
session_id="sid-d2",
session_id="sid-dyn-1",
tenant_id="t1",
user_id="u1",
role="user",
channel="admin_chat",
text="analyze finance report",
text="How do I compute DAU from events table?",
attachments=[],
metadata={"interaction_mode": "comprehensive", "selected_specialist": "ops"},
metadata={"interaction_mode": "comprehensive"},
)
out = gw.handle_turn(msg=msg, lang="en", executor=_Exec(model=_ManagerModel()), specialist_executor_factory=_factory)
out = gw.handle_turn(msg=msg, lang="en", executor=_Exec(model=_ManagerModel()))
assert out.interaction_mode == "comprehensive"
assert out.selected_specialist == "dyn:sql"
assert captured.get("exec_text") == "Write a SQL query to compute daily active users."
def test_gateway_comprehensive_mode_ignores_manager_self_and_dispatches_specialist(monkeypatch: pytest.MonkeyPatch) -> None:
class Store:
def get_setting(self, _k: str) -> str:
return ""
def add_trace_event(self, **_kwargs: object) -> None:
return None
class _ManagerModel:
def __init__(self) -> None:
self.calls = 0
def chat(self, _messages, _tools, *, on_token=None):
self.calls += 1
if self.calls == 1:
return LLMResponse(
content='{"route":{"kind":"manager_self","specialist":"generalist","reason":"can answer directly"},"dispatch":{"instruction_text":"请直接给用户简明答案。"}}',
tool_calls=[],
)
return LLMResponse(content="manager_self_final", tool_calls=[])
class _Exec:
def __init__(self, model=None):
self.model = model
self.tools = object()
self.system_prompt = ""
monkeypatch.setattr(
"oclaw.runtime.gateway.get_manager_prompt_prebuild",
lambda **kwargs: {
"manager_context": "manager",
"allowed_fixed": ("generalist", "ops", "image", "memory"),
"allowed_fixed_quoted": '"generalist", "ops", "image", "memory"',
},
)
captured: dict[str, str] = {}
def _run_agent_core(**kwargs):
data = kwargs["data"]
captured["exec_text"] = str(getattr(data.msg, "text", ""))
class _Outcome:
final_text = "specialist_result"
class _Out:
outcome = _Outcome()
return _Out()
monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core)
gw = OclawGateway(store=Store())
msg = StandardMessage(
session_id="sid-self-1",
tenant_id="t1",
user_id="u1",
role="user",
channel="admin_chat",
text="请直接回答这个简单问题",
attachments=[],
metadata={"interaction_mode": "comprehensive"},
)
out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec(model=_ManagerModel()))
assert out.interaction_mode == "comprehensive"
assert out.selected_specialist == "generalist"
assert picked.get("sid") == "generalist"
assert captured.get("exec_text") == "请直接给用户简明答案。"
assert out.reply_text == "manager_self_final"
def test_gateway_command_hook_uses_parsed_command_and_context(monkeypatch: pytest.MonkeyPatch) -> None:

37
tests/test_oclaw_generalist_executor.py
View file

@ -17,12 +17,33 @@ def test_build_gateway_executor_defaults_to_generalist(tmp_path: Path) -> None:
def test_build_gateway_executor_generalist_run_command_is_disabled_by_default(tmp_path: Path) -> None:
os.environ.pop("AIA_ENABLE_RUN_COMMAND", None)
store = SqliteStore(str(tmp_path / "ops.sqlite"))
ex = build_gateway_executor(store, lang="zh", specialist="generalist")
run_tool = next((t for t in ex.tools.list() if t.name == "run_command"), None)
assert run_tool is not None
out = run_tool.handler({"command": "echo hi"})
assert isinstance(out, dict)
assert out.get("ok") is False
assert out.get("error") == "disabled"
os.environ["OPS_ASSISTANT_DB_PATH"] = str(tmp_path / "ops.sqlite")
try:
store = SqliteStore(str(tmp_path / "ops.sqlite"))
ex = build_gateway_executor(store, lang="zh", specialist="generalist")
run_tool = next((t for t in ex.tools.list() if t.name == "run_command"), None)
assert run_tool is not None
out = run_tool.handler({"command": "echo hi"})
assert isinstance(out, dict)
assert out.get("ok") is False
assert out.get("error") == "disabled"
finally:
os.environ.pop("OPS_ASSISTANT_DB_PATH", None)
def test_build_gateway_executor_generalist_run_command_prefers_db_setting(tmp_path: Path) -> None:
os.environ["AIA_ENABLE_RUN_COMMAND"] = "0"
os.environ["OPS_ASSISTANT_DB_PATH"] = str(tmp_path / "ops.sqlite")
try:
store = SqliteStore(str(tmp_path / "ops.sqlite"))
store.set_setting("AIA_ENABLE_RUN_COMMAND", "1")
ex = build_gateway_executor(store, lang="zh", specialist="generalist")
run_tool = next((t for t in ex.tools.list() if t.name == "run_command"), None)
assert run_tool is not None
out = run_tool.handler({"command": "echo hi"})
assert isinstance(out, dict)
assert out.get("ok") is True
finally:
os.environ.pop("AIA_ENABLE_RUN_COMMAND", None)
os.environ.pop("OPS_ASSISTANT_DB_PATH", None)

18
tests/test_oclaw_gmail_watcher_gates.py Normal file
View file

@ -0,0 +1,18 @@
from __future__ import annotations
from oclaw.runtime.hooks.gmail_watcher import start_gmail_watcher
def test_start_gmail_watcher_top_level_hooks_disabled() -> None:
r = start_gmail_watcher({"hooks": {"enabled": False, "internal": {"enabled": True}}})
assert r.started is False and r.reason == "hooks not enabled"
def test_start_gmail_watcher_internal_disabled() -> None:
r = start_gmail_watcher({"hooks": {"internal": {"enabled": False}}})
assert r.started is False and r.reason == "hooks not enabled"
def test_start_gmail_watcher_no_account() -> None:
r = start_gmail_watcher({"hooks": {"internal": {"enabled": True}, "gmail": {}}})
assert r.started is False and r.reason == "no gmail account configured"

99
tests/test_oclaw_gmail_watcher_lifecycle.py Normal file
View file

@ -0,0 +1,99 @@
from __future__ import annotations
from dataclasses import dataclass, field
from oclaw.runtime.hooks.gmail_watcher import GmailWatcherResult
from oclaw.runtime.hooks.gmail_watcher_lifecycle import start_gmail_watcher_with_logs
@dataclass
class _Log:
infos: list[str] = field(default_factory=list)
warns: list[str] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
def info(self, msg: str) -> None:
self.infos.append(msg)
def warn(self, msg: str) -> None:
self.warns.append(msg)
def error(self, msg: str) -> None:
self.errors.append(msg)
def test_gmail_lifecycle_logs_startup_success() -> None:
log = _Log()
start_gmail_watcher_with_logs(
cfg={},
log=log,
starter=lambda _cfg: GmailWatcherResult(started=True),
)
assert log.infos == ["gmail watcher started"]
assert not log.warns
assert not log.errors
def test_gmail_lifecycle_logs_actionable_reason() -> None:
log = _Log()
start_gmail_watcher_with_logs(
cfg={},
log=log,
starter=lambda _cfg: GmailWatcherResult(started=False, reason="auth failed"),
)
assert log.warns == ["gmail watcher not started: auth failed"]
def test_gmail_lifecycle_suppresses_expected_reason() -> None:
log = _Log()
start_gmail_watcher_with_logs(
cfg={},
log=log,
starter=lambda _cfg: GmailWatcherResult(started=False, reason="hooks not enabled"),
)
assert not log.warns
def test_gmail_lifecycle_supports_skip_callback_oclaw_env(monkeypatch) -> None:
monkeypatch.setenv("OCLAW_SKIP_GMAIL_WATCHER", "1")
monkeypatch.delenv("OPENCLAW_SKIP_GMAIL_WATCHER", raising=False)
log = _Log()
called = {"skip": 0}
start_gmail_watcher_with_logs(
cfg={},
log=log,
on_skipped=lambda: called.__setitem__("skip", called["skip"] + 1),
starter=lambda _cfg: GmailWatcherResult(started=True),
)
assert called["skip"] == 1
assert not log.infos
def test_gmail_lifecycle_supports_skip_callback(monkeypatch) -> None:
monkeypatch.delenv("OCLAW_SKIP_GMAIL_WATCHER", raising=False)
monkeypatch.setenv("OPENCLAW_SKIP_GMAIL_WATCHER", "1")
log = _Log()
called = {"skip": 0}
start_gmail_watcher_with_logs(
cfg={},
log=log,
on_skipped=lambda: called.__setitem__("skip", called["skip"] + 1),
starter=lambda _cfg: GmailWatcherResult(started=True),
)
assert called["skip"] == 1
assert not log.infos
def test_gmail_lifecycle_logs_startup_error() -> None:
log = _Log()
def _raise(_cfg):
raise RuntimeError("boom")
start_gmail_watcher_with_logs(
cfg={},
log=log,
starter=_raise,
)
assert log.errors == ["gmail watcher failed to start: boom"]

825
tests/test_oclaw_hooks_bundled_parity.py Normal file
View file

@ -0,0 +1,825 @@
from __future__ import annotations
import asyncio
import importlib.util
import shutil
import pytest
from pathlib import Path
from oclaw.runtime.hooks.internal_hooks import clear_hooks, create_hook_event, trigger_hook
from oclaw.runtime.hooks.loader import load_internal_hooks
def _load_module(module_path: Path):
spec = importlib.util.spec_from_file_location(f"testmod_{module_path.stem}", str(module_path))
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def test_load_internal_hooks_does_not_duplicate_handlers_on_reload(tmp_path: Path) -> None:
hooks_root = tmp_path / "hooks"
demo = hooks_root / "demo"
demo.mkdir(parents=True)
(demo / "HOOK.md").write_text(
"---\n"
"name: demo\n"
"description: demo hook\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
"---\n",
encoding="utf-8",
)
(demo / "handler.py").write_text(
"def handle(event):\n"
" ctx = event.context\n"
" ctx['calls'] = int(ctx.get('calls') or 0) + 1\n",
encoding="utf-8",
)
cfg: dict = {"hooks": {"internal": {"enabled": True}}}
clear_hooks()
assert load_internal_hooks(cfg, workspace_dir=str(tmp_path), bundled_hooks_dir=str(hooks_root)) == 1
assert load_internal_hooks(cfg, workspace_dir=str(tmp_path), bundled_hooks_dir=str(hooks_root)) == 1
ctx: dict = {"calls": 0}
async def _run() -> None:
ev = create_hook_event("command", "new", "agent:main:main", context=ctx)
await trigger_hook(ev)
asyncio.run(_run())
assert ctx["calls"] == 1
def test_bootstrap_extra_files_only_injects_allowed_basenames(tmp_path: Path) -> None:
ws = tmp_path / "workspace"
ws.mkdir(parents=True)
(ws / "AGENTS.md").write_text("# agents\n", encoding="utf-8")
(ws / "README.md").write_text("# readme\n", encoding="utf-8")
handler_path = (
Path(__file__).resolve().parents[1]
/ "runtime"
/ "hooks"
/ "bundled"
/ "bootstrap-extra-files"
/ "handler.py"
)
mod = _load_module(handler_path)
event = create_hook_event(
"agent",
"bootstrap",
"agent:main:main",
context={
"workspaceDir": str(ws),
"bootstrapFiles": [],
"cfg": {
"hooks": {
"internal": {
"entries": {
"bootstrap-extra-files": {"enabled": True, "paths": ["**/*.md"]},
}
}
}
},
},
)
mod.handle(event)
files = event.context.get("bootstrapFiles") or []
assert isinstance(files, list)
names = {str((item or {}).get("name")) for item in files if isinstance(item, dict)}
assert "AGENTS.md" in names
assert "README.md" not in names
def test_workspace_hook_discovery_supports_package_manifest_hooks(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_hook_entries_from_dir
from oclaw.runtime.hooks.hook_types import ensure_entry_dict
hooks_root = tmp_path / "hooks"
pkg = hooks_root / "pack-a"
declared = pkg / "nested" / "demo-hook"
declared.mkdir(parents=True)
(pkg / "package.json").write_text(
"{"
"\"openclaw\": {"
"\"hooks\": [\"nested/demo-hook\", \"../outside\"]"
"}"
"}",
encoding="utf-8",
)
(declared / "HOOK.md").write_text(
"---\n"
"name: nested-demo\n"
"description: nested hook\n"
"enabled: true\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
"---\n",
encoding="utf-8",
)
(declared / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
entries = load_hook_entries_from_dir(str(hooks_root), source="oclaw-managed")
assert len(entries) == 1
row = ensure_entry_dict(entries[0])
hook = row["hook"]
assert hook["name"] == "nested-demo"
assert hook["source"] == "oclaw-managed"
assert row.get("invocation", {}).get("enabled") is True
def test_hook_frontmatter_metadata_normalization() -> None:
from oclaw.runtime.hooks.frontmatter import parse_frontmatter, resolve_oclaw_metadata
content = (
"---\n"
"name: demo\n"
"description: demo\n"
"metadata:\n"
" oclaw:\n"
" always: true\n"
" hookKey: demo-key\n"
" export: custom_export\n"
" events: [\"command:new\", \"command:reset\", 123]\n"
" os: [\"darwin\", \"linux\", \"\"]\n"
" requires:\n"
" bins: [\"git\", \"\"]\n"
" anyBins: [\"python\", \"node\"]\n"
" env: [\"OPENAI_API_KEY\"]\n"
" config: [\"workspace.dir\"]\n"
" install:\n"
" - id: bund\n"
" kind: bundled\n"
" - id: bad\n"
" kind: unknown\n"
"---\n"
)
fm = parse_frontmatter(content).frontmatter
md = resolve_oclaw_metadata(fm)
assert isinstance(md, dict)
assert md.get("always") is True
assert md.get("hookKey") == "demo-key"
assert md.get("export") == "custom_export"
assert md.get("events") == ["command:new", "command:reset"]
assert md.get("os") == ["darwin", "linux"]
assert md.get("requires", {}).get("bins") == ["git"]
assert md.get("requires", {}).get("anyBins") == ["python", "node"]
assert md.get("requires", {}).get("env") == ["OPENAI_API_KEY"]
assert md.get("requires", {}).get("config") == ["workspace.dir"]
assert md.get("install") == [{"kind": "bundled", "id": "bund"}]
def test_hook_invocation_policy_disables_loading_even_when_config_enabled(tmp_path: Path) -> None:
from oclaw.runtime.hooks.loader import load_internal_hooks
from oclaw.runtime.hooks.internal_hooks import clear_hooks, create_hook_event, trigger_hook
hooks_root = tmp_path / "hooks"
demo = hooks_root / "demo"
demo.mkdir(parents=True)
(demo / "HOOK.md").write_text(
"---\n"
"name: demo\n"
"description: demo hook\n"
"enabled: false\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
"---\n",
encoding="utf-8",
)
(demo / "handler.py").write_text(
"def handle(event):\n"
" event.context['calls'] = int(event.context.get('calls') or 0) + 1\n",
encoding="utf-8",
)
cfg = {"hooks": {"internal": {"enabled": True, "entries": {"demo": {"enabled": True}}}}}
clear_hooks()
assert load_internal_hooks(cfg, workspace_dir=str(tmp_path), bundled_hooks_dir=str(hooks_root)) == 0
ctx: dict = {"calls": 0}
async def _run() -> None:
ev = create_hook_event("command", "new", "agent:main:main", context=ctx)
await trigger_hook(ev)
asyncio.run(_run())
assert ctx["calls"] == 0
def test_should_include_hook_respects_requires_env_and_config(monkeypatch) -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry_raw = {
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {
"events": ["command:new"],
"requires": {"env": ["DEMO_TOKEN"], "config": ["workspace.dir"]},
},
"invocation": {"enabled": True},
}
entry = ensure_hook_entry(entry_raw)
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
monkeypatch.delenv("DEMO_TOKEN", raising=False)
assert should_include_hook(entry=entry, config=cfg) is False
monkeypatch.setenv("DEMO_TOKEN", "x")
assert should_include_hook(entry=entry, config=cfg) is True
def test_should_include_hook_respects_os_allowlist() -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry = ensure_hook_entry({
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "os": ["amigaos"]},
"invocation": {"enabled": True},
})
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
assert should_include_hook(entry=entry, config=cfg) is False
def test_should_include_hook_requires_bins_all_must_exist() -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry = ensure_hook_entry({
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "requires": {"bins": ["definitely_missing_binary_123"]}},
"invocation": {"enabled": True},
})
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
assert should_include_hook(entry=entry, config=cfg) is False
def test_should_include_hook_requires_any_bins_at_least_one() -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry_ok = ensure_hook_entry({
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "requires": {"anyBins": ["python"]}},
"invocation": {"enabled": True},
})
entry_fail = ensure_hook_entry({
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "requires": {"anyBins": ["missing_bin_abc", "missing_bin_xyz"]}},
"invocation": {"enabled": True},
})
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
assert should_include_hook(entry=entry_ok, config=cfg) is True
assert should_include_hook(entry=entry_fail, config=cfg) is False
def test_should_include_hook_requires_config_path_truthy() -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry = ensure_hook_entry({
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "requires": {"config": ["workspace.dir"]}},
"invocation": {"enabled": True},
})
cfg_false = {"workspace": {"dir": ""}, "hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
cfg_true = {"workspace": {"dir": "/tmp/ws"}, "hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
assert should_include_hook(entry=entry, config=cfg_false) is False
assert should_include_hook(entry=entry, config=cfg_true) is True
def test_parse_hook_manifest_core_roundtrip() -> None:
from oclaw.runtime.hooks.hook_manifest_core import parse_hook_manifest
fm = {
"name": "demo-hook",
"description": "demo description",
"enabled": False,
"metadata": {
"oclaw": {
"events": ["command:new", "command:reset", ""],
"always": True,
"hookKey": "demo-key",
"requires": {"bins": ["git"], "anyBins": ["python"], "env": ["X_TOKEN"]},
"install": [{"kind": "bundled", "id": "bundled"}],
}
},
}
parsed = parse_hook_manifest(frontmatter=fm, default_name="fallback")
assert parsed.name == "demo-hook"
assert parsed.description == "demo description"
assert parsed.invocation_enabled is False
md = parsed.metadata.as_dict()
assert md.get("events") == ["command:new", "command:reset"]
assert md.get("always") is True
assert md.get("hookKey") == "demo-key"
assert md.get("requires", {}).get("bins") == ["git"]
assert md.get("requires", {}).get("anyBins") == ["python"]
assert md.get("requires", {}).get("env") == ["X_TOKEN"]
assert md.get("install") == [{"kind": "bundled", "id": "bundled"}]
def test_compat_wrappers_accept_dict_entries() -> None:
from oclaw.runtime.hooks.config import should_include_hook_compat
from oclaw.runtime.hooks.policy import resolve_hook_entries_compat
entry = {
"hook": {"name": "demo", "description": "", "source": "oclaw-managed", "filePath": "", "baseDir": "", "handlerPath": ""},
"metadata": {"events": ["command:new"]},
"invocation": {"enabled": True},
"frontmatter": {},
}
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
assert should_include_hook_compat(entry=entry, config=cfg) is True
out = resolve_hook_entries_compat([entry])
assert isinstance(out, list) and len(out) == 1
assert out[0].get("hook", {}).get("name") == "demo"
def test_ensure_hook_entry_normalizes_unknown_source() -> None:
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
row = {"hook": {"name": "x", "source": "unknown-source"}}
entry = ensure_hook_entry(row)
assert entry.hook.source == "oclaw-managed"
def test_should_include_hook_remote_eligibility_overrides_local_runtime() -> None:
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.hook_types import ensure_hook_entry
entry = ensure_hook_entry(
{
"hook": {"name": "demo", "source": "oclaw-managed"},
"metadata": {"events": ["command:new"], "requires": {"bins": ["git"]}},
"invocation": {"enabled": True},
}
)
cfg = {"hooks": {"internal": {"entries": {"demo": {"enabled": True}}}}}
called: dict[str, int] = {"hasBin": 0}
def _has_bin(_bin: str) -> bool:
called["hasBin"] += 1
return False
ok = should_include_hook(
entry=entry,
config=cfg,
eligibility={"remote": {"hasBin": _has_bin, "hasAnyBin": lambda bins: False}},
)
assert ok is False
assert called["hasBin"] == 1
def test_workspace_discovers_enabled_plugin_hook_dirs(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_workspace_hook_entries
ws = tmp_path / "ws"
plugin_root = ws / ".openclaw" / "extensions" / "sample-bundle"
hook_dir = plugin_root / "hooks" / "bundle-hook"
hook_dir.mkdir(parents=True)
(plugin_root / ".codex-plugin").mkdir(parents=True)
(plugin_root / ".codex-plugin" / "plugin.json").write_text(
'{"name":"Sample Bundle","hooks":"hooks"}',
encoding="utf-8",
)
(hook_dir / "HOOK.md").write_text(
"---\n"
"name: bundle-hook\n"
"description: bundle hook\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
"---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
entries = load_workspace_hook_entries(
str(ws),
config={"hooks": {"internal": {"enabled": True}}, "plugins": {"entries": {"sample-bundle": {"enabled": True}}}},
)
found = [e for e in entries if e.hook.name == "bundle-hook" and e.hook.source == "oclaw-plugin"]
assert found
assert found[0].hook.pluginId == "sample-bundle"
def test_workspace_skips_disabled_plugin_hook_dirs(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_workspace_hook_entries
ws = tmp_path / "ws"
plugin_root = ws / ".openclaw" / "extensions" / "sample-bundle"
hook_dir = plugin_root / "hooks" / "bundle-hook"
hook_dir.mkdir(parents=True)
(plugin_root / ".codex-plugin").mkdir(parents=True)
(plugin_root / ".codex-plugin" / "plugin.json").write_text(
'{"name":"Sample Bundle","hooks":"hooks"}',
encoding="utf-8",
)
(hook_dir / "HOOK.md").write_text(
"---\nname: bundle-hook\ndescription: bundle hook\nmetadata:\n oclaw:\n events: [\"command:new\"]\n---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
entries = load_workspace_hook_entries(
str(ws),
config={"hooks": {"internal": {"enabled": True}}, "plugins": {"entries": {"sample-bundle": {"enabled": False}}}},
)
assert all(not (e.hook.name == "bundle-hook" and e.hook.source == "oclaw-plugin") for e in entries)
def test_legacy_internal_hook_handler_loads_from_workspace_relative_path(tmp_path: Path) -> None:
from oclaw.runtime.hooks.internal_hooks import (
clear_hooks,
create_hook_event,
set_hooks_enabled,
trigger_hook,
)
from oclaw.runtime.hooks.loader import load_internal_hooks
(tmp_path / "hooks").mkdir(parents=True, exist_ok=True)
legacy_path = tmp_path / "legacy_handler.py"
legacy_path.write_text(
"def legacy(event):\n"
" event.context['legacy_called'] = True\n",
encoding="utf-8",
)
cfg = {
"hooks": {
"internal": {
"enabled": True,
"handlers": [{"event": "command:new", "module": "legacy_handler.py", "export": "legacy"}],
}
}
}
clear_hooks()
set_hooks_enabled(True)
assert load_internal_hooks(cfg, workspace_dir=str(tmp_path), bundled_hooks_dir=str(tmp_path / "hooks")) >= 1
async def _run() -> dict:
ctx: dict = {"seed": True}
ev = create_hook_event("command", "new", "agent:main:main", context=ctx)
await trigger_hook(ev)
return ev.context
ctx = asyncio.run(_run())
assert ctx.get("legacy_called") is True
def test_legacy_internal_hook_handler_rejects_path_escape(tmp_path: Path) -> None:
from oclaw.runtime.hooks.loader import load_internal_hooks
(tmp_path / "hooks").mkdir(parents=True, exist_ok=True)
cfg = {
"hooks": {
"internal": {
"enabled": True,
"handlers": [{"event": "command:new", "module": "../outside.py", "export": "legacy"}],
}
}
}
assert load_internal_hooks(cfg, workspace_dir=str(tmp_path), bundled_hooks_dir=str(tmp_path / "hooks")) == 0
def test_build_workspace_hook_status_reports_summary(tmp_path: Path) -> None:
from oclaw.runtime.hooks.hooks_status import build_workspace_hook_status
ws = tmp_path / "ws"
ok_dir = ws / "hooks" / "ok-hook"
blocked_dir = ws / "hooks" / "blocked-hook"
ok_dir.mkdir(parents=True)
blocked_dir.mkdir(parents=True)
(ok_dir / "HOOK.md").write_text(
"---\nname: ok-hook\ndescription: ok\nmetadata:\n oclaw:\n events: [\"command:new\"]\n---\n",
encoding="utf-8",
)
(ok_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
(blocked_dir / "HOOK.md").write_text(
"---\nname: blocked-hook\ndescription: blocked\nenabled: false\nmetadata:\n oclaw:\n events: [\"command:new\"]\n---\n",
encoding="utf-8",
)
(blocked_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
report = build_workspace_hook_status(
str(ws),
config={
"hooks": {
"internal": {
"enabled": True,
"entries": {"ok-hook": {"enabled": True}, "blocked-hook": {"enabled": True}},
}
}
},
)
summary = report.get("summary") or {}
hooks = report.get("hooks") or []
assert summary.get("discovered_total") == 2
assert summary.get("enabled_by_config_total") == 1
assert summary.get("loadable_total") == 1
names = {str(h.get("name")) for h in hooks if isinstance(h, dict)}
assert "ok-hook" in names and "blocked-hook" in names
def test_build_workspace_hook_status_includes_install_options_and_missing_bins(tmp_path: Path) -> None:
from oclaw.runtime.hooks.hooks_status import build_workspace_hook_status
ws = tmp_path / "ws"
hook_dir = ws / "hooks" / "installable-hook"
hook_dir.mkdir(parents=True)
(hook_dir / "HOOK.md").write_text(
"---\n"
"name: installable-hook\n"
"description: hook with install metadata\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
" requires:\n"
" bins: [\"definitely_missing_binary_123\"]\n"
" install:\n"
" - id: npm-install\n"
" kind: npm\n"
" label: Install via npm\n"
" bins: [\"definitely_missing_binary_123\"]\n"
"---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
report = build_workspace_hook_status(
str(ws),
config={"hooks": {"internal": {"enabled": True, "entries": {"installable-hook": {"enabled": True}}}}},
)
summary = report.get("summary") or {}
hooks = report.get("hooks") or []
assert summary.get("missing_bins_total") == 1
row = next((h for h in hooks if isinstance(h, dict) and h.get("name") == "installable-hook"), {})
assert "definitely_missing_binary_123" in list(row.get("missing_bins") or [])
install_options = list(row.get("install_options") or [])
assert install_options and install_options[0].get("id") == "npm-install"
suggestion = row.get("install_suggestion") or {}
assert suggestion.get("id") == "npm-install"
def test_build_workspace_hook_status_selects_best_install_suggestion(tmp_path: Path) -> None:
from oclaw.runtime.hooks.hooks_status import build_workspace_hook_status
ws = tmp_path / "ws"
hook_dir = ws / "hooks" / "smart-install-hook"
hook_dir.mkdir(parents=True)
(hook_dir / "HOOK.md").write_text(
"---\n"
"name: smart-install-hook\n"
"description: hook with multiple install options\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
" requires:\n"
" bins: [\"missing_a\", \"missing_b\"]\n"
" install:\n"
" - id: option-single\n"
" kind: npm\n"
" bins: [\"missing_a\"]\n"
" - id: option-better\n"
" kind: npm\n"
" bins: [\"missing_a\", \"missing_b\"]\n"
"---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
report = build_workspace_hook_status(
str(ws),
config={"hooks": {"internal": {"enabled": True, "entries": {"smart-install-hook": {"enabled": True}}}}},
)
row = next((h for h in list(report.get("hooks") or []) if h.get("name") == "smart-install-hook"), {})
suggestion = row.get("install_suggestion") or {}
assert suggestion.get("id") == "option-better"
def test_hook_dir_prefers_handler_py_over_ts_and_sh(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_hook_entries_from_dir
h = tmp_path / "h"
h.mkdir()
(h / "HOOK.md").write_text(
"---\n"
"name: multi\n"
"description: multi\n"
"metadata:\n"
" oclaw:\n"
" events: [\"x:y\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.py").write_text("def handle(e):\n return None\n", encoding="utf-8")
(h / "handler.ts").write_text("export default () => undefined\n", encoding="utf-8")
(h / "handler.sh").write_text("#!/bin/sh\ntrue\n", encoding="utf-8")
entries = load_hook_entries_from_dir(str(tmp_path), source="oclaw-bundled")
assert len(entries) == 1
assert entries[0].hook.handlerPath.endswith("handler.py")
def test_hook_dir_selects_ts_over_sh_when_no_py(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_hook_entries_from_dir
h = tmp_path / "h2"
h.mkdir()
(h / "HOOK.md").write_text(
"---\n"
"name: ts-or-sh\n"
"description: ts or sh\n"
"metadata:\n"
" oclaw:\n"
" events: [\"x:y\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.sh").write_text("#!/bin/sh\ntrue\n", encoding="utf-8")
(h / "handler.ts").write_text("export default () => undefined\n", encoding="utf-8")
entries = load_hook_entries_from_dir(str(tmp_path), source="oclaw-bundled")
assert len(entries) == 1
assert entries[0].hook.handlerPath.endswith("handler.ts")
def test_hook_dir_prefers_mjs_over_sh_and_bash(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_hook_entries_from_dir
h = tmp_path / "h4"
h.mkdir()
(h / "HOOK.md").write_text(
"---\n"
"name: mjs\n"
"description: mjs\n"
"metadata:\n"
" oclaw:\n"
" events: [\"x:y\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.mjs").write_text("export default () => undefined\n", encoding="utf-8")
(h / "handler.bash").write_text("#!/bin/bash\ntrue\n", encoding="utf-8")
(h / "handler.sh").write_text("#!/bin/sh\ntrue\n", encoding="utf-8")
entries = load_hook_entries_from_dir(str(tmp_path), source="oclaw-bundled")
assert len(entries) == 1
assert entries[0].hook.handlerPath.endswith("handler.mjs")
def test_hook_dir_selects_sh_when_only_shell(tmp_path: Path) -> None:
from oclaw.runtime.hooks.workspace import load_hook_entries_from_dir
h = tmp_path / "h3"
h.mkdir()
(h / "HOOK.md").write_text(
"---\n"
"name: sh-only\n"
"description: sh\n"
"metadata:\n"
" oclaw:\n"
" events: [\"x:y\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.sh").write_text("#!/bin/sh\ntrue\n", encoding="utf-8")
entries = load_hook_entries_from_dir(str(tmp_path), source="oclaw-bundled")
assert len(entries) == 1
assert entries[0].hook.handlerPath.endswith("handler.sh")
@pytest.mark.skipif(shutil.which("bash") is None, reason="bash not available")
def test_sh_hook_merges_stdout_context(tmp_path: Path) -> None:
from oclaw.runtime.hooks.internal_hooks import clear_hooks, create_hook_event, trigger_hook
from oclaw.runtime.hooks.loader import load_internal_hooks
hooks = tmp_path / "hooks"
h = hooks / "sh-hook"
h.mkdir(parents=True)
(h / "HOOK.md").write_text(
"---\n"
"name: sh-hook\n"
"description: sh\n"
"metadata:\n"
" oclaw:\n"
" events: [\"probe:test\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.sh").write_text(
"#!/usr/bin/env bash\n"
"python - <<'PY'\n"
"import json, sys\n"
"e = json.load(sys.stdin)\n"
"c = e.get('context') or {}\n"
"c['from_sh'] = 1\n"
"print(json.dumps({'context': c}))\n"
"PY\n",
encoding="utf-8",
)
clear_hooks()
n = load_internal_hooks(
{"hooks": {"internal": {"enabled": True, "entries": {"sh-hook": {"enabled": True}}}}},
str(tmp_path),
bundled_hooks_dir=str(hooks),
)
assert n == 1
ctx: dict = {"a": 1}
async def _go() -> None:
ev = create_hook_event("probe", "test", "sess-1", context=ctx)
await trigger_hook(ev)
asyncio.run(_go())
assert ctx.get("from_sh") == 1
@pytest.mark.skipif(shutil.which("node") is None, reason="node not available")
def test_mjs_module_hook_mutates_context(tmp_path: Path) -> None:
from oclaw.runtime.hooks.internal_hooks import clear_hooks, create_hook_event, trigger_hook
from oclaw.runtime.hooks.loader import load_internal_hooks
hooks = tmp_path / "hooks"
h = hooks / "jmod"
h.mkdir(parents=True)
(h / "HOOK.md").write_text(
"---\n"
"name: jmod\n"
"description: mjs mod\n"
"metadata:\n"
" oclaw:\n"
" events: [\"probe2:test2\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.mjs").write_text(
"export default (e) => { e.context = e.context || {}; e.context.mjs = 1; };\n",
encoding="utf-8",
)
clear_hooks()
n = load_internal_hooks(
{"hooks": {"internal": {"enabled": True, "entries": {"jmod": {"enabled": True}}}}},
str(tmp_path),
bundled_hooks_dir=str(hooks),
)
assert n == 1
ctx: dict = {}
async def _go() -> None:
ev = create_hook_event("probe2", "test2", "s2", context=ctx)
await trigger_hook(ev)
asyncio.run(_go())
assert ctx.get("mjs") == 1
@pytest.mark.skipif(shutil.which("node") is None, reason="node not available")
def test_mjs_script_mode_hook_merges_stdout(tmp_path: Path) -> None:
from oclaw.runtime.hooks.internal_hooks import clear_hooks, create_hook_event, trigger_hook
from oclaw.runtime.hooks.loader import load_internal_hooks
hooks = tmp_path / "hooks2"
h = hooks / "jscr"
h.mkdir(parents=True)
(h / "HOOK.md").write_text(
"---\n"
"name: jscr\n"
"description: mjs script\n"
"metadata:\n"
" oclaw:\n"
" events: [\"probe3:test3\"]\n"
" hookMode: script\n"
"---\n",
encoding="utf-8",
)
(h / "handler.mjs").write_text(
"import { readFileSync } from 'node:fs';\n"
"const e = JSON.parse(readFileSync(0, 'utf-8'));\n"
"const c = e.context || {};\n"
"c.jscript = 1;\n"
"console.log(JSON.stringify({ context: c }));\n",
encoding="utf-8",
)
clear_hooks()
n = load_internal_hooks(
{"hooks": {"internal": {"enabled": True, "entries": {"jscr": {"enabled": True}}}}},
str(tmp_path),
bundled_hooks_dir=str(hooks),
)
assert n == 1
ctx: dict = {}
async def _go() -> None:
ev = create_hook_event("probe3", "test3", "s3", context=ctx)
await trigger_hook(ev)
asyncio.run(_go())
assert ctx.get("jscript") == 1

68
tests/test_oclaw_hooks_config_toggle_cli.py Normal file
View file

@ -0,0 +1,68 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from oclaw.runtime.hooks.user_config_hooks import apply_hook_entry_enabled, load_storage_config_document, save_storage_config_document
def test_apply_hook_entry_enabled_roundtrip(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("OCLAW_CONFIG_PATH", str(tmp_path / "c.json"))
doc: dict = {"other": 1}
apply_hook_entry_enabled(doc, "my-hook", True, ensure_internal_hooks_enabled=True)
assert doc["hooks"]["internal"]["enabled"] is True
assert doc["hooks"]["internal"]["entries"]["my-hook"]["enabled"] is True
apply_hook_entry_enabled(doc, "my-hook", False)
assert doc["hooks"]["internal"]["entries"]["my-hook"]["enabled"] is False
save_storage_config_document(doc)
loaded = json.loads((tmp_path / "c.json").read_text(encoding="utf-8"))
assert loaded["other"] == 1
assert loaded["hooks"]["internal"]["entries"]["my-hook"]["enabled"] is False
def test_hooks_enable_disable_workspace_hook(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("OCLAW_CONFIG_PATH", str(tmp_path / "cfg.json"))
(tmp_path / "cfg.json").write_text(
json.dumps(
{
"hooks": {
"internal": {
"enabled": True,
"entries": {"toggle-hook": {"enabled": False}},
}
}
},
indent=2,
),
encoding="utf-8",
)
ws = tmp_path / "ws"
h = ws / "hooks" / "toggle-hook"
h.mkdir(parents=True)
(h / "HOOK.md").write_text(
"---\n"
"name: toggle-hook\n"
"description: t\n"
"metadata:\n"
" oclaw:\n"
" events: [\"probe:toggle\"]\n"
"---\n",
encoding="utf-8",
)
(h / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_disable, _cmd_hooks_enable
assert _cmd_hooks_enable(argparse.Namespace(workspace=str(ws), name="toggle-hook")) == 0
doc = json.loads((tmp_path / "cfg.json").read_text(encoding="utf-8"))
assert doc["hooks"]["internal"]["entries"]["toggle-hook"]["enabled"] is True
assert _cmd_hooks_disable(argparse.Namespace(workspace=str(ws), name="toggle-hook")) == 0
doc2 = json.loads((tmp_path / "cfg.json").read_text(encoding="utf-8"))
assert doc2["hooks"]["internal"]["entries"]["toggle-hook"]["enabled"] is False
def test_load_storage_empty_when_missing(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("OCLAW_CONFIG_PATH", str(tmp_path / "missing.json"))
assert load_storage_config_document() == {}

96
tests/test_oclaw_hooks_eligibility_metadata.py Normal file
View file

@ -0,0 +1,96 @@
from __future__ import annotations
import platform
from oclaw.runtime.hooks.config import should_include_hook
from oclaw.runtime.hooks.eligibility_from_metadata import hook_eligibility_from_message_metadata
from oclaw.runtime.hooks.hook_types import HookEntry, HookInvocation, HookRef
def _entry(*, name: str = "x", bins: tuple[str, ...] = (), os_list: tuple[str, ...] = ()) -> HookEntry:
md: dict = {
"events": ["a:b"],
"requires": {"bins": list(bins)},
}
if os_list:
md["os"] = list(os_list)
return HookEntry(
hook=HookRef(
name=name,
description="",
source="oclaw-workspace",
pluginId=None,
filePath="/x/HOOK.md",
baseDir="/x",
handlerPath="/x/handler.py",
),
frontmatter={},
metadata=md,
invocation=HookInvocation(enabled=True),
)
def test_hook_eligibility_from_metadata_none_when_absent() -> None:
assert hook_eligibility_from_message_metadata(None) is None
assert hook_eligibility_from_message_metadata({}) is None
assert hook_eligibility_from_message_metadata({"hookEligibility": {}}) is None
assert hook_eligibility_from_message_metadata({"hookEligibility": {"remote": {}}}) is None
def test_hook_eligibility_from_metadata_bins_present_predicates() -> None:
elig = hook_eligibility_from_message_metadata(
{"hookEligibility": {"remote": {"binsPresent": ["git", "node"]}}},
)
assert elig is not None
remote = elig.get("remote") or {}
assert remote["hasBin"]("git") is True
assert remote["hasBin"]("node") is True
assert remote["hasBin"]("missing-tool-xyz") is False
assert remote["hasAnyBin"](["a", "git"]) is True
assert remote["hasAnyBin"](["a", "b"]) is False
def test_hook_eligibility_from_metadata_platforms_lowercased() -> None:
elig = hook_eligibility_from_message_metadata(
{"hookEligibility": {"remote": {"platforms": ["Darwin", "LINUX"]}}},
)
assert elig is not None
assert elig["remote"]["platforms"] == ["darwin", "linux"]
def test_hook_eligibility_from_metadata_note_only() -> None:
elig = hook_eligibility_from_message_metadata({"hookEligibility": {"remote": {"note": " agent-a "}}})
assert elig is not None
assert elig["remote"]["note"] == "agent-a"
def test_should_include_hook_respects_bins_present_remote() -> None:
entry = _entry(name="needs-git", bins=("git",))
cfg = {"hooks": {"internal": {"enabled": True, "entries": {"needs-git": {"enabled": True}}}}}
assert should_include_hook(entry=entry, config=cfg, eligibility=None) is (True if __import__("shutil").which("git") else False)
elig = hook_eligibility_from_message_metadata(
{"hookEligibility": {"remote": {"binsPresent": ["fantasy-bin-oclaw-test"]}}},
)
assert elig is not None
assert should_include_hook(entry=entry, config=cfg, eligibility=elig) is False
elig_ok = hook_eligibility_from_message_metadata({"hookEligibility": {"remote": {"binsPresent": ["git"]}}})
assert elig_ok is not None
assert should_include_hook(entry=entry, config=cfg, eligibility=elig_ok) is True
def test_should_include_hook_remote_platforms_satisfy_os_allowlist_when_local_differs() -> None:
entry = _entry(name="linux-only", bins=(), os_list=("linux",))
cfg = {"hooks": {"internal": {"enabled": True, "entries": {"linux-only": {"enabled": True}}}}}
cur = platform.system().lower()
if cur == "linux":
assert should_include_hook(entry=entry, config=cfg, eligibility=None) is True
else:
assert should_include_hook(entry=entry, config=cfg, eligibility=None) is False
elig = hook_eligibility_from_message_metadata({"hookEligibility": {"remote": {"platforms": ["linux"]}}})
assert elig is not None
# OpenClaw semantics: remote platform list can satisfy hook ``os`` even when this process runs elsewhere.
assert should_include_hook(entry=entry, config=cfg, eligibility=elig) is True

73
tests/test_oclaw_hooks_runtime.py
View file

@ -3,7 +3,12 @@ from __future__ import annotations
import json
from pathlib import Path
from oclaw.runtime.hooks_runtime import resolve_runtime_config
from oclaw.runtime.hooks.eligibility_from_metadata import hook_eligibility_from_message_metadata
from oclaw.runtime.hooks_runtime import (
_reset_hooks_runtime_state_for_test,
initialize_hooks_runtime,
resolve_runtime_config,
)
def test_resolve_runtime_config_from_env_json(monkeypatch) -> None:
@ -32,3 +37,69 @@ def test_resolve_runtime_config_from_config_path(monkeypatch, tmp_path: Path) ->
== 99
)
def test_initialize_hooks_runtime_passes_remote_eligibility(monkeypatch, tmp_path: Path) -> None:
_reset_hooks_runtime_state_for_test()
empty_bundled = tmp_path / "empty-bundled"
empty_bundled.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr("oclaw.runtime.hooks_runtime.runtime_hooks_bundled_root", lambda: str(empty_bundled))
monkeypatch.setattr("oclaw.runtime.hooks.merge_skill_hook_dirs.discover_workspace_skill_manifests", lambda: ())
ws = tmp_path / "workspace"
hook_dir = ws / "hooks" / "demo"
hook_dir.mkdir(parents=True, exist_ok=True)
(hook_dir / "HOOK.md").write_text(
"---\n"
"name: demo\n"
"description: demo\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
" requires:\n"
" bins: [\"git\"]\n"
"---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
cfg = {"hooks": {"internal": {"enabled": True, "entries": {"demo": {"enabled": True}}}}}
loaded = initialize_hooks_runtime(
cfg=cfg,
workspace_dir=str(ws),
eligibility={"remote": {"hasBin": lambda _bin: False, "hasAnyBin": lambda _bins: False}},
)
assert loaded == 0
def test_initialize_hooks_runtime_loads_when_metadata_bins_present_reports_requirement(
monkeypatch, tmp_path: Path
) -> None:
_reset_hooks_runtime_state_for_test()
empty_bundled = tmp_path / "empty-bundled"
empty_bundled.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr("oclaw.runtime.hooks_runtime.runtime_hooks_bundled_root", lambda: str(empty_bundled))
monkeypatch.setattr("oclaw.runtime.hooks.merge_skill_hook_dirs.discover_workspace_skill_manifests", lambda: ())
ws = tmp_path / "workspace"
hook_dir = ws / "hooks" / "demo"
hook_dir.mkdir(parents=True, exist_ok=True)
(hook_dir / "HOOK.md").write_text(
"---\n"
"name: demo\n"
"description: demo\n"
"metadata:\n"
" oclaw:\n"
" events: [\"command:new\"]\n"
" requires:\n"
" bins: [\"oclaw_fake_hook_bin\"]\n"
"---\n",
encoding="utf-8",
)
(hook_dir / "handler.py").write_text("def handle(event):\n return None\n", encoding="utf-8")
cfg = {"hooks": {"internal": {"enabled": True, "entries": {"demo": {"enabled": True}}}}}
elig = hook_eligibility_from_message_metadata(
{"hookEligibility": {"remote": {"binsPresent": ["oclaw_fake_hook_bin"]}}},
)
assert elig is not None
loaded = initialize_hooks_runtime(cfg=cfg, workspace_dir=str(ws), eligibility=elig)
assert loaded == 1

29
tests/test_oclaw_hooks_runtime_gmail_init.py Normal file
View file

@ -0,0 +1,29 @@
from __future__ import annotations
from pathlib import Path
from oclaw.runtime.hooks_runtime import _reset_hooks_runtime_state_for_test, initialize_hooks_runtime
def test_initialize_hooks_runtime_invokes_gmail_lifecycle(monkeypatch, tmp_path: Path) -> None:
_reset_hooks_runtime_state_for_test()
calls: list[object] = []
def _cap(*, cfg, log, on_skipped=None, starter=None):
calls.append(cfg)
monkeypatch.setattr(
"oclaw.runtime.hooks.gmail_watcher_lifecycle.start_gmail_watcher_with_logs",
_cap,
)
bundled = tmp_path / "bundled"
bundled.mkdir()
monkeypatch.setattr("oclaw.runtime.hooks_runtime.runtime_hooks_bundled_root", lambda: str(bundled))
monkeypatch.setattr("oclaw.runtime.hooks.merge_skill_hook_dirs.discover_workspace_skill_manifests", lambda: ())
ws = tmp_path / "ws"
ws.mkdir()
cfg = {"hooks": {"internal": {"enabled": True}}}
initialize_hooks_runtime(cfg=cfg, workspace_dir=str(ws))
assert len(calls) == 1
assert isinstance(calls[0], dict)

92
tests/test_oclaw_operations_hooks_cmd.py Normal file
View file

@ -0,0 +1,92 @@
from __future__ import annotations
import argparse
import json
from io import StringIO
import sys
def test_build_hooks_status_report_empty_workspace(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
from oclaw.runtime.operations.hooks_cmd import build_hooks_status_report
r = build_hooks_status_report(str(tmp_path), config={"hooks": {"internal": {"enabled": True}}})
assert "summary" in r and "hooks" in r
assert r["workspace_dir"] == str(tmp_path)
def test_hooks_list_json(capsys, tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_list
ns = argparse.Namespace(workspace=str(tmp_path), json=True, eligible=False, verbose=False)
assert _cmd_hooks_list(ns) == 0
out = capsys.readouterr().out
obj = json.loads(out)
assert obj["workspace_dir"] == str(tmp_path)
assert "hooks" in obj
def test_hooks_check_text(capsys, tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_check
ns = argparse.Namespace(workspace=str(tmp_path), json=False)
assert _cmd_hooks_check(ns) == 0
assert "Hooks status" in capsys.readouterr().out
def test_hooks_info_missing(capsys, tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_info
ns = argparse.Namespace(workspace=str(tmp_path), name="nonexistent-hook-xyz", json=False)
assert _cmd_hooks_info(ns) == 1
assert "not found" in capsys.readouterr().out.lower()
def test_hooks_install_deprecated_exits_2(capsys) -> None:
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_install
ns = argparse.Namespace(spec="foo@1.0.0", link=False, pin=False)
assert _cmd_hooks_install(ns) == 2
err = capsys.readouterr().err
assert "deprecated" in err.lower() and "foo@1.0.0" in err
def test_hooks_install_missing_spec_exits_1(capsys) -> None:
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_install
assert _cmd_hooks_install(argparse.Namespace(spec="", link=False, pin=False)) == 1
assert "missing" in capsys.readouterr().err.lower()
def test_hooks_update_deprecated_exits_2(capsys) -> None:
from oclaw.runtime.operations.hooks_cmd import _cmd_hooks_update
assert _cmd_hooks_update(argparse.Namespace(hook_id="pack-a", all=False, dry_run=True)) == 2
err = capsys.readouterr().err
assert "deprecated" in err.lower()
def test_main_hooks_install_invocation(capsys) -> None:
from oclaw.runtime.operations import main as cli_main
code = cli_main(["hooks", "install", "some-spec@1"])
assert code == 2
assert "deprecated" in capsys.readouterr().err.lower()
def test_main_hooks_list_invocation(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
from oclaw.runtime.operations import main as cli_main
old = sys.stdout
buf = StringIO()
try:
sys.stdout = buf
code = cli_main(["hooks", "list", "--json", "--workspace", str(tmp_path)])
finally:
sys.stdout = old
assert code == 0
obj = json.loads(buf.getvalue())
assert "hooks" in obj

49
tests/test_oclaw_project_context_prompt.py
View file

@ -14,15 +14,13 @@ class _Store:
def test_build_project_context_block_reads_workspace_bootstrap(monkeypatch, tmp_path: Path) -> None:
ws = tmp_path / "oclaw" / "runtime" / "assets" / "agent_workspaces" / "workspace-main"
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "main"
ws.mkdir(parents=True, exist_ok=True)
(ws / "AGENTS.md").write_text("agents rules", encoding="utf-8")
(ws / "SOUL.md").write_text("soul voice", encoding="utf-8")
(ws / "TOOLS.md").write_text("tooling note", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
out = build_project_context_block(store=_Store())
assert "[project_context]" in out
assert "[AGENTS.md]" in out
assert "[SOUL.md]" in out
assert "[TOOLS.md]" in out
def test_build_project_context_block_empty_when_no_files(monkeypatch, tmp_path: Path) -> None:
@ -32,40 +30,17 @@ def test_build_project_context_block_empty_when_no_files(monkeypatch, tmp_path:
def test_build_project_context_block_fallback_to_project_root(monkeypatch, tmp_path: Path) -> None:
(tmp_path / "AGENTS.md").write_text("legacy root agents", encoding="utf-8")
(tmp_path / "TOOLS.md").write_text("legacy root tools", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
out = build_project_context_block(store=_Store())
assert "[AGENTS.md]" in out
assert "legacy root agents" in out
def test_build_project_context_block_legacy_workspace_fallback(monkeypatch, tmp_path: Path) -> None:
ws = tmp_path / "oclaw" / "workspace"
ws.mkdir(parents=True, exist_ok=True)
(ws / "AGENTS.md").write_text("legacy workspace agents", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
out = build_project_context_block(store=_Store())
assert "[AGENTS.md]" in out
assert "legacy workspace agents" in out
def test_build_project_context_block_workspace_main_fallback(monkeypatch, tmp_path: Path) -> None:
ws = tmp_path / "oclaw" / "workspace-main"
ws.mkdir(parents=True, exist_ok=True)
(ws / "AGENTS.md").write_text("legacy workspace-main agents", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
out = build_project_context_block(store=_Store())
assert "[AGENTS.md]" in out
assert "legacy workspace-main agents" in out
assert "[TOOLS.md]" in out
assert "legacy root tools" in out
def test_build_project_context_block_triggers_bootstrap_for_each_root(monkeypatch, tmp_path: Path) -> None:
ws_agent = tmp_path / "oclaw" / "runtime" / "assets" / "agent_workspaces" / "workspace-main"
ws_main = tmp_path / "oclaw" / "workspace-main"
ws_agent = tmp_path / "oclaw" / "runtime" / "workspaces" / "main"
ws_agent.mkdir(parents=True, exist_ok=True)
ws_main.mkdir(parents=True, exist_ok=True)
(ws_agent / "AGENTS.md").write_text("agent root", encoding="utf-8")
(ws_main / "AGENTS.md").write_text("main root", encoding="utf-8")
(ws_agent / "TOOLS.md").write_text("agent root", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
calls: list[dict] = []
@ -83,16 +58,16 @@ def test_build_project_context_block_triggers_bootstrap_for_each_root(monkeypatc
workspaces = [str((c.get("context") or {}).get("workspaceDir") or "") for c in calls]
norm = {w.replace("\\", "/") for w in workspaces}
assert str(ws_agent).replace("\\", "/") in norm
assert str(ws_main).replace("\\", "/") in norm
assert str(tmp_path).replace("\\", "/") in norm
def test_project_context_bootstrap_includes_agent_id_when_config_matches(monkeypatch, tmp_path: Path) -> None:
ws_agent = tmp_path / "oclaw" / "runtime" / "assets" / "agent_workspaces" / "workspace-main"
ws_agent = tmp_path / "oclaw" / "runtime" / "workspaces" / "main"
ws_social = tmp_path / "oclaw" / "workspace-social"
ws_agent.mkdir(parents=True, exist_ok=True)
ws_social.mkdir(parents=True, exist_ok=True)
(ws_agent / "AGENTS.md").write_text("agent root", encoding="utf-8")
(ws_social / "AGENTS.md").write_text("social root", encoding="utf-8")
(ws_agent / "TOOLS.md").write_text("agent root", encoding="utf-8")
(ws_social / "TOOLS.md").write_text("social root", encoding="utf-8")
monkeypatch.setattr("oclaw.runtime.project_context_prompt.PROJECT_ROOT", tmp_path)
calls: list[dict] = []

6
tests/test_oclaw_role_prompt_paths.py
View file

@ -7,6 +7,6 @@ def test_specialist_prompts_loaded_from_oclaw_tree() -> None:
ops = default_system_prefix_for_specialist("ops", "zh")
gen = default_system_prefix_for_specialist("generalist", "zh")
img = default_system_prefix_for_specialist("image", "zh")
assert "网络运维专家" in ops
assert "通识专家" in gen
assert "图像处理专家" in img
assert "运维专家（ops specialist）" in ops
assert "通识专家（generalist specialist）" in gen
assert "图像专家（image specialist）" in img

3
tests/test_oclaw_router.py
View file

@ -107,8 +107,7 @@ def test_interaction_mode_normalization_supports_legacy_values() -> None:
def test_requested_specialist_normalization_defaults_to_generalist() -> None:
assert normalize_requested_specialist("ops") == "ops"
assert normalize_requested_specialist("image") == "image"
assert normalize_requested_specialist("memory_curator") == "memory_curator"
assert normalize_requested_specialist("memory-curator") == "memory_curator"
assert normalize_requested_specialist("memory") == "memory"
assert normalize_requested_specialist("unknown") == "generalist"

73
tests/test_oclaw_system_prompt.py
View file

@ -2,6 +2,8 @@ from __future__ import annotations
from pathlib import Path
import pytest
from oclaw.runtime.system_prompt import build_oclaw_executor_system_prompt
from oclaw.runtime.types import OclawMemoryContext
from oclaw.platform.persistence.sqlite_store import SqliteStore
@ -30,7 +32,8 @@ def test_build_oclaw_executor_system_prompt_includes_skills_block(tmp_path: Path
memory_context=OclawMemoryContext(),
lang="zh",
)
assert "<available_skills>" in out
assert "## 技能（skills）：" in out
assert '- name:"' in out
def test_build_oclaw_executor_system_prompt_without_tools_has_no_skills_block(tmp_path: Path) -> None:
@ -44,7 +47,9 @@ def test_build_oclaw_executor_system_prompt_without_tools_has_no_skills_block(tm
memory_context=OclawMemoryContext(),
lang="zh",
)
assert "<available_skills>" not in out
assert "## 技能（skills）：" not in out
assert "不会自动执行技能 `scripts/` 目录下的文件" in out
assert "scripts/" in out
def test_build_oclaw_executor_system_prompt_with_tools_has_skills(tmp_path: Path) -> None:
@ -59,4 +64,66 @@ def test_build_oclaw_executor_system_prompt_with_tools_has_skills(tmp_path: Path
memory_context=OclawMemoryContext(),
lang="zh",
)
assert "<available_skills>" in out
assert "## 技能（skills）：" in out
assert '- name:"' in out
def test_executor_static_prompt_cache_invalidates_on_settings_change(monkeypatch: pytest.MonkeyPatch) -> None:
from oclaw.runtime import system_prompt as sp
class DummyStore:
def __init__(self) -> None:
self.values: dict[str, str] = {}
def get_setting(self, key: str) -> str:
return str(self.values.get(key, ""))
class DummyReg:
pass
calls = {"n": 0}
def _skills_block(**kwargs) -> str:
_ = kwargs
calls["n"] += 1
return "skills-block"
monkeypatch.setattr(sp, "expert_workspace_signature_token", lambda: ("sig",))
monkeypatch.setattr(sp, "build_project_context_block", lambda **kwargs: "")
monkeypatch.setattr(sp, "build_skills_catalog_block", _skills_block)
monkeypatch.setattr(
sp,
"render_runtime_prompt",
lambda prompt_id, variables, strict: f"{prompt_id}\n{variables.get('skills_catalog') or ''}",
)
store = DummyStore()
reg = DummyReg()
_ = sp.get_executor_prompt_static(
store=store,
tools=reg, # type: ignore[arg-type]
base_url="",
base_system="base",
workspace_dir=None,
skill_binding_role="generalist",
)
_ = sp.get_executor_prompt_static(
store=store,
tools=reg, # type: ignore[arg-type]
base_url="",
base_system="base",
workspace_dir=None,
skill_binding_role="generalist",
)
assert calls["n"] == 1
store.values["AIA_SKILL_DISABLED_NAMES"] = "[\"weather\"]"
_ = sp.get_executor_prompt_static(
store=store,
tools=reg, # type: ignore[arg-type]
base_url="",
base_system="base",
workspace_dir=None,
skill_binding_role="generalist",
)
assert calls["n"] == 2

6
tests/test_prompt_loader.py
View file

@ -27,9 +27,9 @@ def test_render_prompt_success() -> None:
def test_load_runtime_prompt_doc_with_frontmatter() -> None:
doc = load_runtime_prompt_doc("roles/specialists/ops/system.md")
assert doc.frontmatter.get("title") == "oclaw_role_specialist_ops_system"
assert "你是网络运维专家" in doc.body
doc = load_runtime_prompt_doc("runtime/default_system.zh.md")
assert doc.frontmatter.get("title") == "default_runtime_system_zh"
assert "你是一个通用 AI 助手" in doc.body
def test_render_runtime_prompt_success() -> None:

103
tests/test_prompt_prebuild.py Normal file
View file

@ -0,0 +1,103 @@
from __future__ import annotations
from typing import Any
from oclaw.runtime import prompt_prebuild as pp
class _DummyStore:
def __init__(self, settings: dict[str, str] | None = None):
self._settings = dict(settings or {})
def get_setting(self, key: str) -> str:
return str(self._settings.get(key, ""))
def test_get_manager_prompt_prebuild_includes_structured_skills(monkeypatch) -> None:
monkeypatch.setattr(pp, "discover_specialist_ids", lambda: ("generalist", "ops", "image", "memory"))
monkeypatch.setattr(
pp,
"list_experts",
lambda: [
{"id": "generalist", "files": {"ROLE_SYSTEM.md": "General specialist for broad tasks."}},
{"id": "ops", "files": {"ROLE_SYSTEM.md": "Ops specialist for runtime and services."}},
],
)
monkeypatch.setattr(pp, "expert_workspace_signature_token", lambda: ("sig",))
def _ctx(_role: str, template_vars: dict[str, Any] | None = None) -> str:
return f"CTX\n{str((template_vars or {}).get('MANAGER_DYNAMIC_EXPERTS_HINT') or '')}"
monkeypatch.setattr(pp, "build_role_system_context", _ctx)
out = pp.get_manager_prompt_prebuild(
store=_DummyStore(),
registry=object(),
base_url="",
memory_enabled=True,
)
assert "generalist" in str(out.get("allowed_fixed") or "")
assert "- generalist:" in str(out.get("manager_context") or "")
assert "General specialist for broad tasks." in str(out.get("manager_context") or "")
def test_warm_startup_prompt_prebuild_warms_all_roles(monkeypatch) -> None:
monkeypatch.setattr(
pp,
"get_manager_prompt_prebuild",
lambda **_k: {
"manager_context": "manager_ctx",
"allowed_fixed": ("generalist", "ops"),
"allowed_fixed_quoted": '"generalist", "ops"',
},
)
monkeypatch.setattr(pp, "discover_specialist_ids", lambda: ("generalist", "ops"))
monkeypatch.setattr(pp, "build_role_system_context", lambda role, template_vars=None: f"{role}_ctx")
captured: dict[str, Any] = {}
def _warm(**kwargs):
captured.update(kwargs)
return {"roles_warmed": len(kwargs.get("role_base_systems") or {})}
monkeypatch.setattr(pp, "warm_executor_prompt_cache", _warm)
out = pp.warm_startup_prompt_prebuild(
store=_DummyStore(),
registry=object(),
base_url="",
memory_enabled=True,
)
assert out["ok"] is True
role_map = captured.get("role_base_systems") if isinstance(captured, dict) else {}
assert isinstance(role_map, dict)
assert "manager" in role_map
assert "generalist" in role_map
assert "ops" in role_map
def test_runtime_prewarm_prompts_snapshot_returns_roles(monkeypatch) -> None:
monkeypatch.setattr(pp, "discover_specialist_ids", lambda: ("generalist", "ops"))
monkeypatch.setattr(
pp,
"get_manager_prompt_prebuild",
lambda **_k: {
"manager_context": "manager_ctx",
},
)
monkeypatch.setattr(pp, "build_role_system_context", lambda role, template_vars=None: f"{role}_ctx")
monkeypatch.setattr(pp, "get_executor_prompt_static", lambda **kwargs: f"exec::{kwargs.get('skill_binding_role')}")
monkeypatch.setattr(pp, "default_registry", lambda **kwargs: object())
out = pp.runtime_prewarm_prompts_snapshot(store=_DummyStore())
assert out["ok"] is True
prompts = out.get("prompts") or {}
assert "manager" in prompts
assert "generalist" in prompts
assert "ops" in prompts
assert prompts["manager"].get("system_prompt") == "exec::manager"
assert prompts["generalist"].get("system_prompt") == "exec::generalist"
assert prompts["ops"].get("system_prompt") == "exec::ops"
assert "manager_system_prompt" not in prompts["manager"]
assert "executor_system_prompt" not in prompts["manager"]
assert "manager_user_scaffold" not in prompts["manager"]

23
tests/test_runtime_prewarm_api.py Normal file
View file

@ -0,0 +1,23 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from oclaw.interfaces.admin import routes as routes_mod
def test_runtime_prewarm_routes_registered_and_guarded() -> None:
app = FastAPI()
app.include_router(routes_mod.build_admin_router())
client = TestClient(app, raise_server_exceptions=False)
paths = set(app.openapi().get("paths", {}).keys())
assert "/admin/api/runtime/prewarm" in paths
assert "/admin/api/runtime/prewarm/status" in paths
assert "/admin/api/runtime/prewarm/prompts" in paths
r1 = client.get("/admin/api/runtime/prewarm/status")
r2 = client.post("/admin/api/runtime/prewarm", json={"mode": "sync", "reason": "test"})
r3 = client.get("/admin/api/runtime/prewarm/prompts")
assert r1.status_code == 401
assert r2.status_code == 401
assert r3.status_code == 401

32
tests/test_skill_executor_trace.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from types import SimpleNamespace
from oclaw.runtime import skill_executor as skill_executor_mod
from oclaw.runtime.skill_executor import SkillExecutionContext, SkillExecutor
from oclaw.runtime.tools.base import ToolRegistry, ToolSpec
@ -47,3 +48,34 @@ def test_skill_executor_emits_skill_trace_events() -> None:
assert "skill_selected" in et
assert "skill_executed" in et
def test_skill_executor_emits_internal_hook_events(monkeypatch) -> None:
calls: list[tuple[str, str]] = []
def _cap(**kwargs): # noqa: ANN003
t = str(kwargs.get("event_type") or "")
a = str(kwargs.get("action") or "")
calls.append((t, a))
monkeypatch.setattr(skill_executor_mod, "trigger_hook_event", _cap)
store = _DummyStore()
reg = ToolRegistry(
[
ToolSpec(
name="echo_skill",
description="echo",
parameters={"type": "object", "properties": {"x": {"type": "string"}}},
handler=lambda a: {"ok": True, "x": a.get("x")},
read_only=True,
)
]
)
ex = SkillExecutor()
uses = [SimpleNamespace(id="call_1", name="echo_skill", arguments={"x": "1"})]
ex.execute_skill_uses(
ctx=SkillExecutionContext(store=store, tools=reg, session_id="s1", trace_id="t1", workspace_dir="/tmp/ws"),
assistant_msg_id=1,
skill_uses=uses,
)
assert calls == [("skill", "before"), ("skill", "after")]

2
tests/test_skill_role_binding_catalog.py
View file

@ -68,7 +68,7 @@ def test_collect_unfiltered_when_binding_disabled(tmp_path: Path, monkeypatch) -
skill_binding_role="generalist",
)
names = {e[0] for e in entries}
assert "solo-skill" not in names
assert "solo-skill" in names
def test_normalize_drops_unknown_skills(tmp_path: Path) -> None:

8
tests/test_skills_prompt.py
View file

@ -3,15 +3,13 @@ from __future__ import annotations
from oclaw.runtime.skills_prompt import format_skills_for_prompt
def test_format_skills_for_prompt_xml_shape() -> None:
def test_format_skills_for_prompt_line_shape() -> None:
out = format_skills_for_prompt(
[("demo", "desc", "/tmp/skills/demo/SKILL.md")],
max_chars=50_000,
)
assert "<available_skills>" in out
assert "</available_skills>" in out
assert "<name>demo</name>" in out
assert "read_file" in out
assert "## 技能（skills）：" in out
assert '- name:"demo", description:"desc", path:"/tmp/skills/demo/SKILL.md"' in out
def test_format_skills_for_prompt_truncates_by_budget() -> None:

30
tests/test_tool_wire_freeze_switch.py Normal file
View file

@ -0,0 +1,30 @@
from __future__ import annotations
from oclaw.runtime import direct_loop as dl
class _DummyStore:
def __init__(self, settings: dict[str, str] | None = None):
self.settings = dict(settings or {})
def get_setting(self, key: str) -> str:
return str(self.settings.get(key, ""))
def test_tool_wire_freeze_default_on(monkeypatch) -> None:
monkeypatch.setattr(dl, "_prepare_llm_tools", lambda **kwargs: [])
monkeypatch.setenv("AIA_TOOL_WIRE_FROZEN_ON_STARTUP", "")
store = _DummyStore()
_ = dl.warm_tool_wire_cache(store=store, tools=object(), base_url="", roles=["manager"])
st = dl.tool_wire_freeze_status(store=store)
assert st["enabled"] is True
assert st["frozen"] is True
def test_tool_wire_freeze_disabled_by_setting(monkeypatch) -> None:
monkeypatch.setattr(dl, "_prepare_llm_tools", lambda **kwargs: [])
store = _DummyStore({"AIA_TOOL_WIRE_FROZEN_ON_STARTUP": "0"})
_ = dl.warm_tool_wire_cache(store=store, tools=object(), base_url="", roles=["manager"])
st = dl.tool_wire_freeze_status(store=store)
assert st["enabled"] is False
assert st["frozen"] is False

62
tests/test_wiki_auto_inject_hook.py
View file

@ -185,6 +185,68 @@ def test_handle_store_only_mode_skips_injection(monkeypatch, tmp_path: Path) ->
assert str(meta.get("skip_reason") or "") == "memory_mode_store_only"
def test_handle_respects_manager_gate_off(monkeypatch, tmp_path: Path) -> None:
mod = _load_handler_module()
wiki = tmp_path / "wiki"
wiki.mkdir(parents=True, exist_ok=True)
(wiki / "ops.md").write_text("router vlan baseline\n", encoding="utf-8")
monkeypatch.setattr(
mod,
"_load_config",
lambda: {
"plugins": {
"entries": {
"memory-wiki": {
"wiki_root": str(wiki),
"auto": {"enabled": True, "inject": {"max_chars": 800, "top_k": 3}},
}
}
}
},
)
event = types.SimpleNamespace(
type="llm",
action="before_prompt_build",
context={"userText": "router vlan", "need_wiki_inject": False, "prepend_system_context": ""},
)
mod.handle(event)
assert str(event.context.get("prepend_system_context") or "") == ""
meta = event.context.get("wiki_inject_meta") or {}
assert bool(meta.get("enabled")) is False
assert str(meta.get("skip_reason") or "") == "manager_gate_off"
def test_handle_requires_wiki_query_when_manager_gate_on(monkeypatch, tmp_path: Path) -> None:
mod = _load_handler_module()
wiki = tmp_path / "wiki"
wiki.mkdir(parents=True, exist_ok=True)
(wiki / "ops.md").write_text("router vlan baseline\n", encoding="utf-8")
monkeypatch.setattr(
mod,
"_load_config",
lambda: {
"plugins": {
"entries": {
"memory-wiki": {
"wiki_root": str(wiki),
"auto": {"enabled": True, "inject": {"max_chars": 800, "top_k": 3}},
}
}
}
},
)
event = types.SimpleNamespace(
type="llm",
action="before_prompt_build",
context={"userText": "router vlan", "need_wiki_inject": True, "prepend_system_context": ""},
)
mod.handle(event)
assert str(event.context.get("prepend_system_context") or "") == ""
meta = event.context.get("wiki_inject_meta") or {}
assert bool(meta.get("enabled")) is False
assert str(meta.get("skip_reason") or "") == "manager_wiki_query_missing"
def test_collect_snippets_prioritizes_merged_turns(tmp_path: Path) -> None:
mod = _load_handler_module()
wiki = tmp_path / "wiki"

101
tests/test_workspaces_experts.py Normal file
View file

@ -0,0 +1,101 @@
from __future__ import annotations
from pathlib import Path
import pytest
from oclaw.runtime.workspaces import experts as experts_mod
from oclaw.runtime.agents.specialists import discover_specialist_ids
def _set_project_root(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setattr(experts_mod, "PROJECT_ROOT", tmp_path)
def test_list_experts_reads_runtime_workspaces(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "qa"
ws.mkdir(parents=True, exist_ok=True)
(ws / "SOUL.md").write_text("qa soul", encoding="utf-8")
rows = experts_mod.list_experts()
ids = {str(x.get("id") or "") for x in rows}
assert "qa" in ids
def test_create_expert_requires_soul(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
with pytest.raises(ValueError, match="soul_required"):
experts_mod.create_expert(expert_id="qa", files={})
def test_create_expert_rejects_legacy_file_names(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
with pytest.raises(ValueError, match="unsupported_file_name"):
experts_mod.create_expert(expert_id="qa", files={"SOUL.md": "ok", "AGENTS.md": "legacy"})
def test_create_update_delete_expert_flow(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
created = experts_mod.create_expert(
expert_id="qa",
files={"SOUL.md": "init soul", "ROLE_SYSTEM.md": "role system"},
)
assert created["id"] == "qa"
experts_mod.update_expert_files(expert_id="qa", files={"SOUL.md": "updated soul"})
rows = experts_mod.list_experts()
qa = next((x for x in rows if str(x.get("id") or "") == "qa"), {})
files = qa.get("files") if isinstance(qa, dict) else {}
assert "updated soul" in str((files or {}).get("SOUL.md") or "")
experts_mod.delete_expert("qa")
rows2 = experts_mod.list_experts()
assert all(str(x.get("id") or "") != "qa" for x in rows2)
def test_delete_builtin_expert_is_protected(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces" / "main"
ws.mkdir(parents=True, exist_ok=True)
(ws / "SOUL.md").write_text("main soul", encoding="utf-8")
with pytest.raises(ValueError, match="builtin_expert_protected"):
experts_mod.delete_expert("main")
def test_discover_specialists_reads_workspaces(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces"
(ws / "main").mkdir(parents=True, exist_ok=True)
(ws / "generalist").mkdir(parents=True, exist_ok=True)
(ws / "generalist" / "ROLE_SYSTEM.md").write_text("g", encoding="utf-8")
(ws / "ops").mkdir(parents=True, exist_ok=True)
(ws / "ops" / "ROLE_SYSTEM.md").write_text("o", encoding="utf-8")
(ws / "qa").mkdir(parents=True, exist_ok=True)
(ws / "qa" / "SOUL.md").write_text("q", encoding="utf-8")
got = set(discover_specialist_ids())
assert "qa" in got
assert "main" not in got
def test_build_expert_catalog_block_contains_dynamic_experts(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces"
(ws / "main").mkdir(parents=True, exist_ok=True)
(ws / "qa").mkdir(parents=True, exist_ok=True)
(ws / "qa" / "SOUL.md").write_text("QA expert soul", encoding="utf-8")
(ws / "qa" / "ROLE_SYSTEM.md").write_text("Handle test plans and regressions", encoding="utf-8")
block = experts_mod.build_expert_catalog_block(include_main=False, per_field_limit=80, max_total_chars=500)
assert "qa" in block
assert "role_system=" in block
assert "soul=" in block
def test_discover_specialists_cache_invalidates_after_create(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_set_project_root(monkeypatch, tmp_path)
ws = tmp_path / "oclaw" / "runtime" / "workspaces"
(ws / "main").mkdir(parents=True, exist_ok=True)
(ws / "generalist").mkdir(parents=True, exist_ok=True)
(ws / "generalist" / "SOUL.md").write_text("g", encoding="utf-8")
before = set(discover_specialist_ids())
assert "qa" not in before
experts_mod.create_expert(expert_id="qa", files={"SOUL.md": "qa soul"})
after = set(discover_specialist_ids())
assert "qa" in after

53
tests/test_workspaces_prompt_contract.py Normal file
View file

@ -0,0 +1,53 @@
from __future__ import annotations
from pathlib import Path
from oclaw.platform.config.paths import PROJECT_ROOT
def _read(path: Path) -> str:
return path.read_text(encoding="utf-8")
def test_main_dynamic_placeholder_only_in_role_system() -> None:
root = (PROJECT_ROOT / "oclaw" / "runtime" / "workspaces" / "main").resolve()
role_system = _read(root / "ROLE_SYSTEM.md")
soul = _read(root / "SOUL.md")
assert "{{MANAGER_DYNAMIC_EXPERTS_HINT}}" in role_system
assert "{{" not in soul
for name in ("AGENTS.md", "IDENTITY.md", "USER.md"):
p = root / name
if p.exists():
assert "{{" not in _read(p)
def test_main_hard_routing_rule_only_in_role_system() -> None:
root = (PROJECT_ROOT / "oclaw" / "runtime" / "workspaces" / "main").resolve()
role_system = _read(root / "ROLE_SYSTEM.md")
assert "回退 `generalist`" in role_system
for name in ("AGENTS.md", "IDENTITY.md", "USER.md", "SOUL.md"):
p = root / name
if p.exists():
assert "回退 `generalist`" not in _read(p)
def test_every_workspace_has_role_system() -> None:
root = (PROJECT_ROOT / "oclaw" / "runtime" / "workspaces").resolve()
for item in root.iterdir():
if not item.is_dir():
continue
if item.name.startswith("__"):
continue
assert (item / "ROLE_SYSTEM.md").exists(), f"missing ROLE_SYSTEM.md for {item.name}"
def test_workspace_legacy_prompt_files_removed() -> None:
root = (PROJECT_ROOT / "oclaw" / "runtime" / "workspaces").resolve()
legacy = ("AGENTS.md", "IDENTITY.md", "USER.md")
for item in root.iterdir():
if not item.is_dir():
continue
if item.name.startswith("__"):
continue
for name in legacy:
assert not (item / name).exists(), f"legacy prompt file still exists: {item.name}/{name}"