Codex
1dd5d8d753
Remove order compatibility fields, routing notices and migration references from Wiki source, skills and documentation. Bump capability schema to v6 and validate the remaining Wiki query response contract. Move retired runtime configuration, database and completed job archives to the independent order project, uninstall legacy units and clear stale bytecode. Keep Git history for provenance. Validation: 173 Wiki tests pass; live independent order MCP authentication, queries and write rejection pass after cleanup. No integration filenames, config keys or source references remain in the Wiki working tree. BREAKING CHANGE: query results and capability metadata no longer contain legacy business-related fields. Wiki consumers must use the v6 schema.
479 lines
21 KiB
Python
479 lines
21 KiB
Python
import subprocess
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from obsidian_wiki.mcp_query import QueryError, QueryManager
|
|
|
|
|
|
class McpQueryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name)
|
|
self.vault = self.root / "vault"
|
|
self.project = self.root / "project"
|
|
self.runtime = self.project / ".runtime" / "mcp-query"
|
|
self.vault.mkdir()
|
|
self.project.mkdir()
|
|
self._write_opencode_skill_fixture(self.project)
|
|
self.manager = QueryManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.runtime,
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
max_query_chars=80,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.manager.close()
|
|
self.tempdir.cleanup()
|
|
|
|
def test_check_opencode_readiness_static_success(self):
|
|
with patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(result["static_preflight"])
|
|
self.assertEqual(result["runtime_smoke_test"], "skipped")
|
|
self.assertEqual(result["opencode_version"], "1.17.13")
|
|
self.assertFalse(result["errors"])
|
|
self.assertTrue(any(path.endswith(".agents/skills/wiki-query/SKILL.md") for path in result["checked_paths"]))
|
|
self.assertEqual(run_mock.call_count, 2)
|
|
|
|
def test_check_opencode_readiness_reports_missing_skill_files(self):
|
|
missing_project = self.root / "missing-project"
|
|
missing_project.mkdir()
|
|
(missing_project / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
manager = QueryManager(
|
|
vault_path=self.vault,
|
|
project_root=missing_project,
|
|
runtime_root=self.root / "runtime-missing",
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
)
|
|
|
|
with patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run):
|
|
result = manager.check_opencode_readiness()
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertFalse(result["static_preflight"])
|
|
self.assertIn(".agents/skills/wiki-query/SKILL.md", "\n".join(result["errors"]))
|
|
self.assertIn(".agents/skills/llm-wiki/SKILL.md", "\n".join(result["errors"]))
|
|
|
|
def test_check_opencode_readiness_smoke_test_success(self):
|
|
self.manager.opencode_model = "openai/gpt-5.5"
|
|
self.manager.opencode_variant = "minimal"
|
|
with patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_smoke_success_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["runtime_smoke_test"], "passed")
|
|
self.assertEqual(run_mock.call_count, 3)
|
|
smoke_args = run_mock.call_args_list[-1].args[0]
|
|
self.assertEqual(smoke_args[:4], ["/usr/bin/opencode", "run", "--format", "json"])
|
|
self.assertEqual(smoke_args[smoke_args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(smoke_args[smoke_args.index("--variant") + 1], "minimal")
|
|
self.assertIn("--auto", smoke_args)
|
|
self.assertIn("--dir", smoke_args)
|
|
self.assertIn("MCP query readiness smoke test", smoke_args)
|
|
|
|
def test_query_rejects_empty_oversized_and_invalid_mode(self):
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("")
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("x" * 81)
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("what do I know?", mode="full")
|
|
|
|
def test_close_still_closes_supervisor_when_queued_job_write_fails(self):
|
|
queued = {"state": "queued"}
|
|
with (
|
|
patch.object(self.manager, "_iter_jobs", return_value=[queued]),
|
|
patch.object(self.manager, "_write_job", side_effect=OSError("disk full")),
|
|
patch.object(self.manager._supervisor, "close") as supervisor_close,
|
|
):
|
|
with self.assertRaises(OSError):
|
|
self.manager.close()
|
|
|
|
supervisor_close.assert_called_once_with()
|
|
|
|
def test_start_query_job_constructs_restricted_opencode_command(self):
|
|
self.manager.opencode_model = "openai/gpt-5.5"
|
|
self.manager.opencode_variant = "minimal"
|
|
popen = FakePopen()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen) as popen_mock,
|
|
):
|
|
job = self.manager.start_query_job(
|
|
"what do I know about Mini FAKRA?",
|
|
mode="index_only",
|
|
public_only=True,
|
|
note="remote agent asked",
|
|
)
|
|
|
|
args = popen_mock.call_args.args[0]
|
|
self.assertEqual(args[:5], ["/usr/bin/opencode", "run", "--format", "json", "--auto"])
|
|
self.assertEqual(args[args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(args[args.index("--variant") + 1], "minimal")
|
|
self.assertIn("--dir", args)
|
|
self.assertEqual(args[args.index("--dir") + 1], str(self.project))
|
|
self.assertIn("--title", args)
|
|
self.assertIn(f"MCP query {job['job_id']}", args)
|
|
prompt = args[-1]
|
|
self.assertIn("/wiki-query", prompt)
|
|
self.assertIn("what do I know about Mini FAKRA?", prompt)
|
|
self.assertIn("index-only", prompt)
|
|
self.assertIn("public only", prompt)
|
|
self.assertIn("normal wiki-query log entry is allowed", prompt)
|
|
self.assertIn("wiki-ingest, wiki-update, wiki-rebuild", prompt)
|
|
self.assertIn("Do not run arbitrary shell commands", prompt)
|
|
self.assertNotIn("shell commands, file writes", prompt)
|
|
self.assertNotIn("rm -rf", prompt)
|
|
|
|
def test_query_result_returns_opencode_output_tail(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("what do I know?")
|
|
|
|
status = self.manager.get_query_job_status(job["job_id"])
|
|
self.assertNotIn("log_path", status)
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
self._json_text_event("MCP_QUERY_OK\nBased on the wiki:\nAnswer body"),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(set(result), {
|
|
"job_id", "state", "answer", "answer_markdown", "citations", "used_wiki",
|
|
"data_freshness", "warnings", "confidence", "mode", "public_only",
|
|
"exit_code", "opencode_output_tail", "error",
|
|
})
|
|
self.assertEqual(result["state"], "succeeded")
|
|
self.assertIn("Based on the wiki", result["answer"])
|
|
self.assertEqual(result["answer_markdown"], result["answer"])
|
|
self.assertEqual(result["citations"], [])
|
|
self.assertTrue(result["used_wiki"])
|
|
self.assertEqual(result["data_freshness"], [])
|
|
self.assertEqual(result["confidence"], "high")
|
|
self.assertEqual(result["warnings"], [])
|
|
self.assertIn("Answer body", result["opencode_output_tail"])
|
|
|
|
def test_query_result_extracts_structured_metadata_from_output(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("customer situation?")
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
self._json_text_event(
|
|
"MCP_QUERY_OK\n"
|
|
"Based on [[Mini FAKRA]] and [[BJD]].\n"
|
|
"Wiki sources updated through 2026-06-02.\n"
|
|
"Answer body"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["answer_markdown"], result["answer"])
|
|
self.assertEqual(result["citations"], ["Mini FAKRA", "BJD"])
|
|
self.assertTrue(result["used_wiki"])
|
|
self.assertEqual(result["data_freshness"], ["2026-06-02"])
|
|
self.assertEqual(result["confidence"], "high")
|
|
|
|
def test_query_result_marks_failed_jobs_low_confidence_with_warnings(self):
|
|
popen = FakePopen(returncode=1)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("what failed?")
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(self._json_text_event("MCP_QUERY_FAILED\nTraceback: failed"), encoding="utf-8")
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "failed")
|
|
self.assertEqual(result["confidence"], "low")
|
|
self.assertTrue(any("failed" in warning.lower() for warning in result["warnings"]))
|
|
|
|
def test_query_result_rejects_skill_failure_with_zero_exit_code(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("what failed?")
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(self._json_text_event("MCP_QUERY_FAILED\nSKILL_LOAD_FAILED: missing"), encoding="utf-8")
|
|
popen.returncode = 0
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "failed")
|
|
self.assertEqual(result["confidence"], "low")
|
|
self.assertIn("SKILL_LOAD_FAILED", result["error"])
|
|
|
|
def test_query_result_accepts_only_final_answer_event_and_hides_paths(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("safe answer?")
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
'{"type":"tool_use","part":{"state":{"output":"/home/private/secret"}}}\n'
|
|
+ self._json_text_event("MCP_QUERY_OK\nBased on [[BJD]]."),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "succeeded")
|
|
self.assertEqual(result["answer_markdown"], "Based on [[BJD]].")
|
|
self.assertNotIn("/home/private", result["opencode_output_tail"])
|
|
status = self.manager.get_query_job_status(job["job_id"])
|
|
self.assertNotIn("pid", status)
|
|
self.assertNotIn("checked_paths", status["preflight"])
|
|
|
|
internal = self.manager._load_job(job["job_id"])
|
|
internal["error"] = f"failed while reading {self.project / 'secret.txt'}"
|
|
internal["preflight"]["warnings"] = [f"check {self.runtime / 'private.log'}"]
|
|
self.manager._write_job(internal)
|
|
redacted = self.manager.get_query_job_status(job["job_id"])
|
|
serialized = json.dumps(redacted)
|
|
self.assertNotIn(str(self.project), serialized)
|
|
self.assertNotIn(str(self.runtime), serialized)
|
|
self.assertIn("<project>", serialized)
|
|
self.assertIn("<runtime>", serialized)
|
|
|
|
def test_ask_wiki_returns_timeout_payload_with_job_id(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
result = self.manager.ask_wiki(
|
|
"slow question?",
|
|
timeout_seconds=0,
|
|
poll_interval_seconds=1,
|
|
)
|
|
|
|
self.assertEqual(result["state"], "running")
|
|
self.assertRegex(result["job_id"], r"^[0-9a-f]{32}$")
|
|
self.assertEqual(result["next_poll_after_seconds"], 1)
|
|
self.assertIn("timed out", " ".join(result["warnings"]).lower())
|
|
|
|
def test_jobs_queue_one_at_a_time_and_cancel_queued_job(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
first = self.manager.start_query_job("first query")
|
|
second = self.manager.start_query_job("second query")
|
|
|
|
self.assertEqual(self.manager.get_query_job_status(first["job_id"])["state"], "running")
|
|
self.assertEqual(self.manager.get_query_job_status(second["job_id"])["state"], "queued")
|
|
|
|
cancelled = self.manager.cancel_query_job(second["job_id"])
|
|
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
|
|
def test_close_cancels_running_and_queued_jobs_and_releases_runtime(self):
|
|
popen = FakePopen(returncode=None)
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
first = self.manager.start_query_job("first query")
|
|
second = self.manager.start_query_job("second query")
|
|
self.manager.close()
|
|
|
|
self.assertEqual(self.manager._load_job(first["job_id"])["state"], "cancelled")
|
|
self.assertEqual(self.manager._load_job(second["job_id"])["state"], "cancelled")
|
|
replacement = QueryManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.runtime,
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
)
|
|
replacement.close()
|
|
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("after close")
|
|
|
|
def test_completed_job_advances_queue_without_status_polling(self):
|
|
first_process = FakePopen(returncode=None)
|
|
second_process = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", side_effect=[first_process, second_process]),
|
|
):
|
|
first = self.manager.start_query_job("first query")
|
|
second = self.manager.start_query_job("second query")
|
|
first_log = Path(self.manager._load_job(first["job_id"])["log_path"])
|
|
first_log.write_text(self._json_text_event("MCP_QUERY_OK\nfirst answer"), encoding="utf-8")
|
|
first_process.returncode = 0
|
|
|
|
deadline = time.monotonic() + 1
|
|
while time.monotonic() < deadline and self.manager._load_job(second["job_id"])["state"] != "running":
|
|
time.sleep(0.01)
|
|
|
|
self.assertEqual(self.manager._load_job(first["job_id"])["state"], "succeeded")
|
|
self.assertEqual(self.manager._load_job(second["job_id"])["state"], "running")
|
|
|
|
def test_failed_preflight_does_not_block_later_queued_job(self):
|
|
first_process = FakePopen(returncode=None)
|
|
third_process = FakePopen(returncode=None)
|
|
failed_readiness = {"ok": False, "errors": ["skill missing"]}
|
|
successful_readiness = {"ok": True, "errors": []}
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", side_effect=[first_process, third_process]),
|
|
):
|
|
first = self.manager.start_query_job("first query")
|
|
second = self.manager.start_query_job("second query")
|
|
third = self.manager.start_query_job("third query")
|
|
first_log = Path(self.manager._load_job(first["job_id"])["log_path"])
|
|
first_log.write_text(self._json_text_event("MCP_QUERY_OK\nfirst answer"), encoding="utf-8")
|
|
with patch.object(
|
|
self.manager,
|
|
"check_opencode_readiness",
|
|
side_effect=[failed_readiness, successful_readiness],
|
|
):
|
|
completed = self.manager._supervisor._complete_naturally(first["job_id"], 0)
|
|
|
|
self.assertTrue(completed)
|
|
self.assertEqual(self.manager._load_job(second["job_id"])["state"], "failed")
|
|
self.assertEqual(self.manager._load_job(third["job_id"])["state"], "running")
|
|
|
|
def test_cleanup_runtime_prunes_old_terminal_query_jobs(self):
|
|
popen = FakePopen(returncode=0)
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("old query")
|
|
self.manager.refresh_jobs()
|
|
job_dir = self.manager.jobs_root / job["job_id"]
|
|
old_epoch = time.time() - 40 * 24 * 60 * 60
|
|
self._touch_tree(job_dir, old_epoch)
|
|
|
|
result = self.manager.cleanup_runtime(max_age_days=30)
|
|
|
|
self.assertEqual(result["old_jobs_deleted"], 1)
|
|
self.assertFalse(job_dir.exists())
|
|
|
|
def _touch_tree(self, path: Path, epoch: float) -> None:
|
|
for child in path.rglob("*"):
|
|
os.utime(child, (epoch, epoch))
|
|
os.utime(path, (epoch, epoch))
|
|
|
|
def _write_opencode_skill_fixture(self, root: Path) -> None:
|
|
(root / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
wiki_query = root / ".agents" / "skills" / "wiki-query"
|
|
llm_wiki = root / ".agents" / "skills" / "llm-wiki"
|
|
wiki_query.mkdir(parents=True)
|
|
llm_wiki.mkdir(parents=True)
|
|
(wiki_query / "SKILL.md").write_text(
|
|
"---\nname: wiki-query\ndescription: Answer questions from the wiki.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
(llm_wiki / "SKILL.md").write_text(
|
|
"---\nname: llm-wiki\ndescription: Shared wiki framework instructions.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def _fake_static_readiness_run(self, args, **_kwargs):
|
|
if args == ["/usr/bin/opencode", "--version"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout="1.17.13\n", stderr="")
|
|
if args == ["/usr/bin/opencode", "run", "--help"]:
|
|
return subprocess.CompletedProcess(
|
|
args,
|
|
0,
|
|
stdout="--format\n--auto\n--dir\n--title\n",
|
|
stderr="",
|
|
)
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_success_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
return subprocess.CompletedProcess(
|
|
args,
|
|
0,
|
|
stdout=self._json_text_event("WIKI_QUERY_SKILL_LOADED"),
|
|
stderr="",
|
|
)
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _json_text_event(self, text):
|
|
import json
|
|
|
|
return json.dumps(
|
|
{
|
|
"type": "text",
|
|
"part": {
|
|
"type": "text",
|
|
"text": text,
|
|
"metadata": {"openai": {"phase": "final_answer"}},
|
|
},
|
|
}
|
|
) + "\n"
|
|
|
|
|
|
class FakePopen:
|
|
def __init__(self, returncode=None):
|
|
self.returncode = returncode
|
|
self.pid = 12345
|
|
self.terminated = False
|
|
|
|
def poll(self):
|
|
return self.returncode
|
|
|
|
def wait(self, timeout=None):
|
|
deadline = time.monotonic() + (timeout or 0)
|
|
while self.returncode is None and (timeout is None or time.monotonic() < deadline):
|
|
time.sleep(0.01)
|
|
if self.returncode is None:
|
|
raise subprocess.TimeoutExpired("fake", timeout)
|
|
return self.returncode
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
self.returncode = -15
|
|
|
|
def kill(self):
|
|
self.returncode = -9
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|