diff --git a/AGENTS.md b/AGENTS.md index 5b07e25..1299d94 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -131,6 +131,7 @@ See `wiki-query` and `wiki-export` skills for how the filter is applied. - **Merge, do not overwrite.** Automatic identifier extraction may add terms, but it must not discard curated registry metadata. Registry values are the durable source for future index rebuilds and remote MCP lookup. - **Prefer precise identifiers in search ranking.** Exact or prefix matches in structured part-number and series fields should rank above incidental filename substrings such as timestamps or archive hashes. - **Keep remote ingest gated.** MCP ingest tools are write-capable and must remain disabled unless the service config explicitly enables the `ingest` tool group via `OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest`, `mcp_service.sh tools enable ingest`, or `mcp_service.sh start --enable-ingest`. Uploaded files and direct text notes must stage outside the vault under the MCP runtime root; `wiki-ingest` is responsible for accepting, archiving, and indexing sources inside the vault. Use `ingest_text_note` only for short remote-agent thoughts/notes; it must wrap the text as a temporary markdown upload and then use the fixed `/wiki-ingest` job path. Remote agents should use `create_ingest_upload_ticket` and the returned `upload_url` plus `X-Obsidian-Wiki-Upload-Token` for HTTP multipart uploads; bearer-token uploads remain only for local/admin compatibility. Prefer `start_job=false`, then `preview_upload_metadata`, then `start_ingest_job` after the preview looks correct. `validate_upload` and `preview_upload_metadata` may report exact sha256 duplicates, but must not claim fuzzy dedupe. Upload tickets are consumed only by authorized `POST` uploads, the upload endpoint must enforce configured host/origin restrictions, and oversized request bodies must fail with HTTP `413`. Do not expose base64 or chunked upload tools in the MCP tool surface. They may launch local OpenCode only through the fixed `/wiki-ingest` prompt; never expose arbitrary prompt or shell execution through MCP. Remote ingest responses must not expose local runtime paths; internal job records may keep paths for the fixed local prompt. Remote ingest must keep the OpenCode readiness preflight in front of job launch, and `check_ingest_opencode_readiness(run_smoke_test=true)` should stay available for operators to verify that OpenCode can load the project `wiki-ingest` skill before accepting uploads. +- **Keep remote ingest status authoritative.** Preserve the `queued`, `preflight`, `agent_running`, `verifying`, and `complete` phases. Compute `process_alive` only from process-supervisor ownership while `phase=agent_running`; do not infer semantic activity from logs or elapsed time. Treat `MCP_INGEST_OK` as permission to begin deterministic verification, not as final success. Final `state=succeeded` requires registry, manifest, file-index, configured archive, and declared output-path checks to pass. A failed verification must finish with `state=failed`, preserve partial outputs for diagnosis, and report compact path-redacted failures. - **Keep remote query gated.** MCP query tools must remain disabled unless the service config explicitly enables the `query` tool group via `OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,query`, `mcp_service.sh tools enable query`, or `mcp_service.sh start --enable-query`. They may launch local OpenCode only through the fixed `/wiki-query` prompt, with no arbitrary prompt or shell execution. `ask_wiki` is the preferred remote-agent query entrypoint; the async `start_wiki_query`/status/result tools remain available for long jobs. `public_only` defaults to false, meaning remote answers may use the full local vault visibility; enable the tool group only for trusted clients and set `public_only=true` per call for filtered user-facing answers. Remote query responses may expose log tails but must not expose local log file paths, and structured fields such as `answer_markdown`, `citations`, `warnings`, and `confidence` must remain backward-compatible with the legacy `answer` field. - **Keep OpenCode model choice server-side.** MCP clients must not pass arbitrary model names through ingest/query tools. Configure local defaults in `.runtime/obsidian-wiki-mcp.env`: ingest uses `OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/gpt-5.5` with `OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=high`; query uses `OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/gpt-5.5` with `OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=minimal`. diff --git a/README.md b/README.md index 75d6fdf..385727b 100644 --- a/README.md +++ b/README.md @@ -497,7 +497,11 @@ The endpoint accepts optional `body_text`, `body_title`, `bundle_id`, `source_ty Run `check_ingest_opencode_readiness` before the first remote upload. Use `run_smoke_test: true` when you want OpenCode itself to prove it can load the project `wiki-ingest` skill; otherwise the static preflight is also run automatically before every `start_ingest_job`. -Uploads and text notes land under `.runtime/mcp-ingest/uploads/`, outside the Obsidian vault, and are treated only as temporary source staging. Job state and logs live under `.runtime/mcp-ingest/jobs/`. Remote upload and job responses expose upload/job ids, hashes, states, and bounded result text, but omit local runtime file paths; the internal job record still keeps paths so the fixed local OpenCode prompt can ingest only the uploaded files. Completion is accepted only from the final JSON text event containing `MCP_INGEST_OK`; `MCP_INGEST_FAILED`, a missing marker, a nonzero exit, timeout, or cancellation marks the job failed/cancelled. `get_mcp_capabilities` includes `server_version`, `schema_version`/`capabilities_schema_version`, `tools_version`, file-index freshness, ingest queue state, and known limitations so remote agents can detect stale client instructions. Its `recommended_first_steps` only names currently enabled tool groups. The MCP server refuses to enable ingest if `OBSIDIAN_MCP_INGEST_RUNTIME_ROOT` points inside the vault; let `wiki-ingest` decide how to archive sources into `_sources/archive/`, update registries, and rebuild the file index. +Uploads and text notes land under `.runtime/mcp-ingest/uploads/`, outside the Obsidian vault, and are treated only as temporary source staging. Job state and logs live under `.runtime/mcp-ingest/jobs/`. Remote upload and job responses expose upload/job ids, hashes, states, and bounded result text, but omit local runtime file paths; the internal job record still keeps paths so the fixed local OpenCode prompt can ingest only the uploaded files. + +Ingest status has authoritative `phase` values: `queued`, `preflight`, `agent_running`, `verifying`, and `complete`. `phase_started_at` marks the current phase boundary, while `process_alive` is true only when the process supervisor still owns a live process in `agent_running`; the server does not infer vague activity from logs or elapsed time. The final JSON text event must contain `MCP_INGEST_OK` before verification begins. The server then strictly verifies each source against `_meta/source-registry.jsonl`, `.manifest.json`, `_meta/file-assets.sqlite`, the archive when `WIKI_ARCHIVE_SOURCES=true`, and every declared related/catalog/created/updated path. Only a passed verification produces `state=succeeded`; failed verification produces `state=failed` even when the agent reported success. Agent output and any partial vault writes are preserved for diagnosis and are not rolled back automatically. `MCP_INGEST_FAILED`, a missing marker, a nonzero exit, timeout, cancellation, readiness failure, or protocol failure skips verification and finishes with `verification.status=not_run`. + +`get_mcp_capabilities` includes `server_version`, `schema_version`/`capabilities_schema_version`, `tools_version`, file-index freshness, ingest queue state, authoritative phase support, strict post-run verification, and known limitations so remote agents can detect stale client instructions. Its `recommended_first_steps` only names currently enabled tool groups. The MCP server refuses to enable ingest if `OBSIDIAN_MCP_INGEST_RUNTIME_ROOT` points inside the vault; let `wiki-ingest` decide how to archive sources into `_sources/archive/`, update registries, and rebuild the file index. Ingest and query launch OpenCode with a fixed prompt and `--auto`, but the MCP service does not enforce a shell sandbox and prompt restrictions are not a security boundary. There is no direct arbitrary-prompt or arbitrary-shell MCP tool, yet a client able to influence vault/upload content must still be trusted. Keep these tool groups behind a trusted network boundary and rotate the token if it has been shared too broadly. diff --git a/obsidian_wiki/mcp_ingest.py b/obsidian_wiki/mcp_ingest.py index 0baf1f2..5073848 100644 --- a/obsidian_wiki/mcp_ingest.py +++ b/obsidian_wiki/mcp_ingest.py @@ -21,6 +21,7 @@ from threading import RLock from typing import Any from obsidian_wiki.file_assets import clean_config_value, db_path_for_vault, registry_path_for_vault +from obsidian_wiki.mcp_ingest_verification import verify_ingest_outputs from obsidian_wiki.mcp_jobs import ProcessOutcome, ProcessSupervisor, extract_final_answer, parse_marked_result DEFAULT_INGEST_MAX_FILE_BYTES = 100 * 1024 * 1024 @@ -533,16 +534,20 @@ class IngestManager: job_id = uuid.uuid4().hex job_dir = self.jobs_root / job_id job_dir.mkdir(parents=True, exist_ok=False) + created_at = utc_now() job = { "job_id": job_id, "state": "queued", + "phase": "queued", + "phase_started_at": created_at, + "verification": {"status": "not_run"}, "upload_ids": upload_ids, "bundle": bundle, "files": [self._public_upload(upload, expose_path=True) for upload in uploads], "note": self._safe_optional_text(note), "queue_sequence": time.time_ns(), - "created_at": utc_now(), - "updated_at": utc_now(), + "created_at": created_at, + "updated_at": created_at, "started_at": None, "finished_at": None, "exit_code": None, @@ -590,6 +595,10 @@ class IngestManager: return { "job_id": status["job_id"], "state": status["state"], + "phase": status["phase"], + "phase_started_at": status["phase_started_at"], + "process_alive": status["process_alive"], + "verification": status["verification"], "summary": self._summary_for_status(status), "files": status["files"], "exit_code": status.get("exit_code"), @@ -602,10 +611,21 @@ class IngestManager: job = self._load_job(job_id) if job["state"] == "queued": job["state"] = "cancelled" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() job["finished_at"] = utc_now() job["updated_at"] = utc_now() self._write_job(job) return self._public_job(job) + if job["state"] == "running" and self._job_phase(job) == "preflight": + job["state"] = "cancelled" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() + job["finished_at"] = utc_now() + job["updated_at"] = utc_now() + job["error"] = "cancelled during OpenCode readiness preflight" + self._write_job(job) + return self._public_job(job) if job["state"] == "running": self._supervisor.cancel(job_id) return self._public_job(self._load_job(job_id)) @@ -627,6 +647,8 @@ class IngestManager: for job in self._iter_jobs(): if job.get("state") == "queued": job["state"] = "cancelled" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() job["error"] = "ingest manager is shutting down" job["finished_at"] = utc_now() job["updated_at"] = utc_now() @@ -731,10 +753,35 @@ class IngestManager: return def _launch_job_locked(self, job: dict[str, Any]) -> bool: - readiness = self.check_opencode_readiness() + phase_started_at = utc_now() + job["state"] = "running" + job["phase"] = "preflight" + job["phase_started_at"] = phase_started_at + job["started_at"] = job.get("started_at") or phase_started_at + job["updated_at"] = phase_started_at + self._write_job(job) + self._lock.release() + try: + try: + readiness = self.check_opencode_readiness() + except Exception as exc: + readiness = { + "ok": False, + "static_preflight": False, + "runtime_smoke_test": "not_run", + "warnings": [], + "errors": [f"{type(exc).__name__}: {exc}"], + } + finally: + self._lock.acquire() + job = self._load_job(job["job_id"]) + if job.get("state") != "running" or self._job_phase(job) != "preflight": + return False job["preflight"] = readiness if not readiness["ok"]: job["state"] = "failed" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() job["error"] = "OpenCode readiness failed: " + "; ".join(readiness["errors"]) job["finished_at"] = utc_now() job["updated_at"] = utc_now() @@ -754,8 +801,6 @@ class IngestManager: f"MCP ingest {job['job_id']}", prompt, ] - job["state"] = "running" - job["started_at"] = utc_now() job["updated_at"] = utc_now() self._write_job(job) try: @@ -770,6 +815,8 @@ class IngestManager: ) except (OSError, RuntimeError, ValueError) as exc: job["state"] = "failed" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() job["error"] = str(exc) job["finished_at"] = utc_now() job["updated_at"] = utc_now() @@ -778,6 +825,9 @@ class IngestManager: current = self._load_job(job["job_id"]) if current.get("state") == "running": current["pid"] = process.pid + current["phase"] = "agent_running" + current["phase_started_at"] = utc_now() + current["updated_at"] = utc_now() self._write_job(current) return True @@ -789,6 +839,8 @@ class IngestManager: return if job.get("state") == "running": job["state"] = "failed" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() job["exit_code"] = outcome.returncode job["error"] = f"ingest completion handling failed: {type(error).__name__}: {error}" job["finished_at"] = utc_now() @@ -797,6 +849,7 @@ class IngestManager: self._start_next_job_locked() def _handle_process_outcome(self, outcome: ProcessOutcome) -> None: + verification_inputs: dict[str, Any] | None = None with self._lock: try: job = self._load_job(outcome.job_id) @@ -811,7 +864,6 @@ class IngestManager: failure_marker=INGEST_FAILURE_MARKER, ) job["exit_code"] = outcome.returncode - job["finished_at"] = utc_now() job["updated_at"] = utc_now() job["result_text"] = result_text if outcome.cancelled: @@ -827,8 +879,57 @@ class IngestManager: job["state"] = "failed" job["error"] = protocol_error else: + job["phase"] = "verifying" + job["phase_started_at"] = utc_now() + job["verification"] = {"status": "running"} + job["error"] = "" + verification_inputs = { + "vault_path": self.vault, + "uploads": [dict(item) for item in job["files"]], + "job_started_at": job.get("started_at"), + "archive_sources": parse_bool( + os.environ.get("WIKI_ARCHIVE_SOURCES"), + default=False, + ), + } + if verification_inputs is None: + job["finished_at"] = utc_now() + job["phase"] = "complete" + job["phase_started_at"] = utc_now() + self._write_job(job) + if verification_inputs is None: + self._start_next_job_locked() + return + + try: + verification = verify_ingest_outputs(**verification_inputs) + except Exception as exc: + verification = { + "status": "failed", + "checked_at": utc_now(), + "failures": [{"code": "verification_error"}], + "warnings": [f"{type(exc).__name__}: {exc}"], + "outputs": {"source_count": len(verification_inputs["uploads"])}, + } + + with self._lock: + try: + job = self._load_job(outcome.job_id) + except IngestError: + return + if job.get("state") != "running" or self._job_phase(job) != "verifying": + return + job["verification"] = verification + job["finished_at"] = utc_now() + job["phase"] = "complete" + job["phase_started_at"] = utc_now() + job["updated_at"] = utc_now() + if verification.get("status") == "passed": job["state"] = "succeeded" job["error"] = "" + else: + job["state"] = "failed" + job["error"] = "deterministic ingest verification failed" self._write_job(job) self._start_next_job_locked() @@ -961,6 +1062,9 @@ class IngestManager: for job in self._iter_jobs(): if job.get("state") in {"queued", "running"}: job["state"] = "failed" + job["phase"] = "complete" + job["phase_started_at"] = utc_now() + job["verification"] = {"status": "not_run"} job["error"] = "interrupted by MCP service restart" job["finished_at"] = utc_now() job["updated_at"] = utc_now() @@ -1110,6 +1214,29 @@ class IngestManager: seen.add(key) return duplicates + def _job_phase(self, job: dict[str, Any]) -> str: + phase = str(job.get("phase") or "") + if phase in {"queued", "preflight", "agent_running", "verifying", "complete"}: + return phase + state = str(job.get("state") or "") + if state == "queued": + return "queued" + if state == "running": + return "agent_running" + return "complete" + + def _job_verification(self, job: dict[str, Any]) -> dict[str, Any]: + verification = job.get("verification") + if not isinstance(verification, dict): + return {"status": "not_run"} + status = str(verification.get("status") or "not_run") + if status not in {"not_run", "running", "passed", "failed"}: + status = "not_run" + return { + **self._redact_public_value(verification), + "status": status, + } + def _public_job(self, job: dict[str, Any]) -> dict[str, Any]: output = str(job.get("result_text") or extract_final_answer(job["log_path"])) if output.startswith((INGEST_SUCCESS_MARKER, INGEST_FAILURE_MARKER)): @@ -1120,9 +1247,25 @@ class IngestManager: ) output = public_text or error output = self._redact_paths(output) + phase = self._job_phase(job) + phase_started_at = job.get("phase_started_at") + if not phase_started_at: + if phase == "queued": + phase_started_at = job.get("created_at") + elif phase == "agent_running": + phase_started_at = job.get("started_at") or job.get("updated_at") + else: + phase_started_at = job.get("finished_at") or job.get("updated_at") return { "job_id": job["job_id"], "state": job["state"], + "phase": phase, + "phase_started_at": phase_started_at, + "process_alive": ( + phase == "agent_running" + and self._supervisor.is_running(str(job["job_id"])) + ), + "verification": self._job_verification(job), "bundle": self._public_bundle(job["bundle"]) if job.get("bundle") else None, "files": [self._public_upload(file_meta) for file_meta in job["files"]], "created_at": job["created_at"], diff --git a/obsidian_wiki/mcp_ingest_verification.py b/obsidian_wiki/mcp_ingest_verification.py new file mode 100644 index 0000000..2f446c0 --- /dev/null +++ b/obsidian_wiki/mcp_ingest_verification.py @@ -0,0 +1,236 @@ +"""Deterministic post-run verification for remote MCP ingest jobs.""" + +from __future__ import annotations + +import json +import re +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from obsidian_wiki.file_assets import db_path_for_vault, registry_path_for_vault + + +SHA256_RE = re.compile(r"^(?:sha256:)?([0-9a-fA-F]{64})$") +DECLARED_PATH_FIELDS = ( + "related_pages", + "catalog_paths", + "pages_created", + "pages_updated", +) + + +def utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def verify_ingest_outputs( + *, + vault_path: str | Path, + uploads: list[dict[str, Any]], + job_started_at: str | None, + archive_sources: bool, +) -> dict[str, Any]: + """Verify durable ingest artifacts without trusting agent-reported success.""" + + vault = Path(vault_path).expanduser().resolve() + registry_records, registry_warnings = _load_registry(vault) + manifest_sources, manifest_warnings = _load_manifest_sources(vault) + failures: list[dict[str, str]] = [] + warnings = [*registry_warnings, *manifest_warnings] + outputs = { + "source_count": len(uploads), + "registry_count": 0, + "archive_count": 0, + "manifest_count": 0, + "indexed_count": 0, + } + + indexed_digests, index_warning = _load_indexed_digests(vault) + if index_warning: + warnings.append(index_warning) + + for upload in uploads: + upload_id = str(upload.get("upload_id") or "") + digest = _normalize_digest(upload.get("sha256")) + if digest is None: + failures.append({"code": "invalid_upload_sha256", "upload_id": upload_id}) + continue + source_id = f"sha256:{digest}" + + registry_record = next( + ( + record + for record in registry_records + if str(record.get("source_id") or "").lower() == source_id + ), + None, + ) + if registry_record is None: + failures.append({"code": "registry_missing", "upload_id": upload_id}) + else: + outputs["registry_count"] += 1 + + if archive_sources: + archive_path = str((registry_record or {}).get("archive_path") or "") + if not archive_path or not _vault_path_exists(vault, archive_path): + failures.append({"code": "archive_missing", "upload_id": upload_id}) + else: + outputs["archive_count"] += 1 + + upload_path = str(upload.get("path") or "") + manifest_record = _matching_manifest_record( + manifest_sources, + upload_path=upload_path, + source_id=source_id, + job_started_at=job_started_at, + ) + if manifest_record is None: + failures.append({"code": "manifest_missing", "upload_id": upload_id}) + else: + outputs["manifest_count"] += 1 + + if digest in indexed_digests: + outputs["indexed_count"] += 1 + else: + failures.append({"code": "index_missing", "upload_id": upload_id}) + + declared_paths = _declared_paths(registry_record, manifest_record) + for declared_path in declared_paths: + if not _vault_path_exists(vault, declared_path): + failures.append({"code": "declared_path_missing", "upload_id": upload_id}) + break + + return { + "status": "passed" if not failures else "failed", + "checked_at": utc_now(), + "failures": failures, + "warnings": warnings, + "outputs": outputs, + } + + +def _normalize_digest(value: Any) -> str | None: + match = SHA256_RE.fullmatch(str(value or "").strip()) + return match.group(1).lower() if match else None + + +def _load_registry(vault: Path) -> tuple[list[dict[str, Any]], list[str]]: + path = registry_path_for_vault(vault) + if not path.is_file(): + return [], [] + records: list[dict[str, Any]] = [] + malformed = 0 + for line in path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + malformed += 1 + continue + if isinstance(record, dict): + records.append(record) + warnings = ["source registry contains malformed records"] if malformed else [] + return records, warnings + + +def _load_manifest_sources(vault: Path) -> tuple[dict[str, Any], list[str]]: + path = vault / ".manifest.json" + if not path.is_file(): + return {}, [] + try: + manifest = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {}, ["manifest could not be parsed"] + sources = manifest.get("sources") if isinstance(manifest, dict) else None + return (sources if isinstance(sources, dict) else {}), [] + + +def _load_indexed_digests(vault: Path) -> tuple[set[str], str | None]: + db_path = db_path_for_vault(vault) + if not db_path.is_file(): + return set(), None + try: + with sqlite3.connect(str(db_path)) as conn: + rows = conn.execute("SELECT sha256 FROM file_assets").fetchall() + except sqlite3.Error: + return set(), "file asset index could not be queried" + return { + digest + for row in rows + if row and (digest := _normalize_digest(row[0])) is not None + }, None + + +def _matching_manifest_record( + sources: dict[str, Any], + *, + upload_path: str, + source_id: str, + job_started_at: str | None, +) -> dict[str, Any] | None: + for source_path, value in sources.items(): + if not isinstance(value, dict): + continue + original_path = str(value.get("original_path") or "") + if str(source_path) != upload_path and original_path != upload_path: + continue + hashes = { + str(value.get("content_hash") or "").lower(), + str(value.get("source_id") or "").lower(), + } + if source_id not in hashes: + continue + if not _is_fresh(value.get("ingested_at"), job_started_at): + continue + return value + return None + + +def _is_fresh(ingested_at: Any, job_started_at: str | None) -> bool: + if not job_started_at: + return bool(ingested_at) + ingested = _parse_timestamp(ingested_at) + started = _parse_timestamp(job_started_at) + return ingested is not None and started is not None and ingested >= started + + +def _parse_timestamp(value: Any) -> datetime | None: + text = str(value or "").strip() + if not text: + return None + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _declared_paths(*records: dict[str, Any] | None) -> set[str]: + paths: set[str] = set() + for record in records: + if not isinstance(record, dict): + continue + for field in DECLARED_PATH_FIELDS: + value = record.get(field) + if isinstance(value, str): + paths.add(value) + elif isinstance(value, list): + paths.update(str(item) for item in value if str(item).strip()) + return paths + + +def _vault_path_exists(vault: Path, relative_path: str) -> bool: + candidate = Path(relative_path) + if candidate.is_absolute(): + return False + resolved = (vault / candidate).resolve() + try: + resolved.relative_to(vault) + except ValueError: + return False + return resolved.exists() diff --git a/obsidian_wiki/mcp_server.py b/obsidian_wiki/mcp_server.py index 97d2313..869cfa0 100644 --- a/obsidian_wiki/mcp_server.py +++ b/obsidian_wiki/mcp_server.py @@ -382,6 +382,14 @@ def create_server( "preferred_text_tool": "ingest_text_note" if enable_ingest else None, "supports_bundle_upload": enable_ingest, "supports_body_text": enable_ingest, + "authoritative_job_phases": ( + ["queued", "preflight", "agent_running", "verifying", "complete"] + if enable_ingest + else [] + ), + "process_alive_source": "process_supervisor" if enable_ingest else None, + "supports_post_run_verification": enable_ingest, + "semantic_activity_inference": False, "multipart_optional_fields": ( [ "body_text", diff --git a/tests/test_mcp_ingest.py b/tests/test_mcp_ingest.py index 71b0478..acde0a3 100644 --- a/tests/test_mcp_ingest.py +++ b/tests/test_mcp_ingest.py @@ -4,6 +4,7 @@ import json import os import subprocess import tempfile +import threading import time import unittest from pathlib import Path @@ -423,12 +424,313 @@ class McpIngestTests(unittest.TestCase): log_path = Path(self.manager._load_job(job["job_id"])["log_path"]) log_path.write_text(self._json_text_event("MCP_INGEST_OK\nIngested source.pdf"), encoding="utf-8") popen.returncode = 0 - self.manager.refresh_jobs() + with patch( + "obsidian_wiki.mcp_ingest.verify_ingest_outputs", + return_value=self._passed_verification(1), + ): + self.manager.refresh_jobs() status = self.manager.get_job_status(job["job_id"]) self.assertEqual(status["state"], "succeeded") self.assertEqual(status["exit_code"], 0) self.assertNotIn("path", status["files"][0]) + def test_job_reports_authoritative_preflight_and_agent_running_phases(self): + upload = self._complete_upload("source.pdf", b"source data") + popen = FakePopen(returncode=None) + observed = {} + + def readiness_during_launch(*, run_smoke_test=False): + internal = next(iter(self.manager._iter_jobs())) + observed["state"] = internal["state"] + observed["phase"] = internal["phase"] + observed["verification"] = internal["verification"] + lock_probe = [] + + def acquire_manager_lock(): + acquired = self.manager._lock.acquire(timeout=0.2) + lock_probe.append(acquired) + if acquired: + self.manager._lock.release() + + thread = threading.Thread(target=acquire_manager_lock) + thread.start() + thread.join() + observed["status_lock_available"] = lock_probe == [True] + return { + "ok": True, + "errors": [], + "warnings": [], + "static_preflight": True, + "runtime_smoke_test": "skipped", + } + + with ( + patch.object(self.manager, "check_opencode_readiness", side_effect=readiness_during_launch), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen), + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + self.assertEqual(observed["state"], "running") + self.assertEqual(observed["phase"], "preflight") + self.assertEqual(observed["verification"]["status"], "not_run") + self.assertTrue(observed["status_lock_available"]) + self.assertEqual(status["state"], "running") + self.assertEqual(status["phase"], "agent_running") + self.assertTrue(status["phase_started_at"]) + self.assertTrue(status["process_alive"]) + self.assertEqual(status["verification"]["status"], "not_run") + + def test_queued_and_cancelled_job_phases_are_authoritative(self): + first = self._complete_upload("first.pdf", b"first") + second = self._complete_upload("second.pdf", b"second") + popen = FakePopen(returncode=None) + with ( + patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen), + ): + self.manager.start_ingest_job([first["upload_id"]]) + queued = self.manager.start_ingest_job([second["upload_id"]]) + + self.assertEqual(queued["state"], "queued") + self.assertEqual(queued["phase"], "queued") + self.assertFalse(queued["process_alive"]) + self.assertEqual(queued["verification"]["status"], "not_run") + + cancelled = self.manager.cancel_ingest_job(queued["job_id"]) + self.assertEqual(cancelled["state"], "cancelled") + self.assertEqual(cancelled["phase"], "complete") + self.assertFalse(cancelled["process_alive"]) + self.assertEqual(cancelled["verification"]["status"], "not_run") + + def test_preflight_failure_completes_without_verification(self): + upload = self._complete_upload("source.pdf", b"source") + (self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink() + with ( + patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock, + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + self.assertEqual(status["state"], "failed") + self.assertEqual(status["phase"], "complete") + self.assertFalse(status["process_alive"]) + self.assertEqual(status["verification"]["status"], "not_run") + popen_mock.assert_not_called() + + def test_preflight_job_can_be_cancelled_while_readiness_runs(self): + upload = self._complete_upload("source.pdf", b"source") + cancelled = {} + + def cancel_during_readiness(*, run_smoke_test=False): + internal = next(iter(self.manager._iter_jobs())) + + def cancel_job(): + cancelled.update(self.manager.cancel_ingest_job(internal["job_id"])) + + thread = threading.Thread(target=cancel_job) + thread.start() + thread.join() + return { + "ok": True, + "errors": [], + "warnings": [], + "static_preflight": True, + "runtime_smoke_test": "skipped", + } + + with ( + patch.object(self.manager, "check_opencode_readiness", side_effect=cancel_during_readiness), + patch( + "obsidian_wiki.mcp_ingest.subprocess.Popen", + return_value=FakePopen(returncode=None), + ) as popen_mock, + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + self.assertEqual(cancelled["state"], "cancelled") + self.assertEqual(status["state"], "cancelled") + self.assertEqual(status["phase"], "complete") + self.assertEqual(status["verification"]["status"], "not_run") + popen_mock.assert_not_called() + + def test_unexpected_preflight_error_finishes_job(self): + upload = self._complete_upload("source.pdf", b"source") + with ( + patch.object( + self.manager, + "check_opencode_readiness", + side_effect=RuntimeError("readiness crashed"), + ), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock, + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + self.assertEqual(status["state"], "failed") + self.assertEqual(status["phase"], "complete") + self.assertEqual(status["verification"]["status"], "not_run") + self.assertIn("readiness crashed", status["error"]) + popen_mock.assert_not_called() + + def test_legacy_job_records_derive_phase_and_verification_defaults(self): + upload = self._complete_upload("source.pdf", b"source") + (self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink() + with patch( + "obsidian_wiki.mcp_ingest.subprocess.run", + side_effect=self._fake_static_readiness_run, + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + internal = self.manager._load_job(status["job_id"]) + internal.pop("phase", None) + internal.pop("phase_started_at", None) + internal.pop("verification", None) + self.manager._write_job(internal) + + legacy = self.manager.get_ingest_job_status(status["job_id"]) + self.assertEqual(legacy["phase"], "complete") + self.assertEqual(legacy["verification"]["status"], "not_run") + self.assertFalse(legacy["process_alive"]) + + def test_service_restart_completes_interrupted_job_without_verification(self): + upload = self._complete_upload("source.pdf", b"source") + (self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink() + with patch( + "obsidian_wiki.mcp_ingest.subprocess.run", + side_effect=self._fake_static_readiness_run, + ): + status = self.manager.start_ingest_job([upload["upload_id"]]) + + internal = self.manager._load_job(status["job_id"]) + internal["state"] = "running" + internal["phase"] = "verifying" + internal["verification"] = {"status": "running"} + internal["finished_at"] = None + self.manager._write_job(internal) + self.manager._mark_interrupted_jobs() + + restarted = self.manager.get_ingest_job_status(status["job_id"]) + self.assertEqual(restarted["state"], "failed") + self.assertEqual(restarted["phase"], "complete") + self.assertEqual(restarted["verification"]["status"], "not_run") + self.assertIn("service restart", restarted["error"]) + + def test_success_marker_runs_verification_before_succeeding(self): + upload = self._complete_upload("source.pdf", b"source") + popen = FakePopen(returncode=None) + observed = {} + passed = { + "status": "passed", + "checked_at": "2026-07-16T10:01:00Z", + "failures": [], + "warnings": [], + "outputs": { + "source_count": 1, + "registry_count": 1, + "archive_count": 1, + "manifest_count": 1, + "indexed_count": 1, + }, + } + + def verify_while_observable(**kwargs): + internal = self.manager._load_job(job["job_id"]) + public = self.manager._public_job(internal) + observed["phase"] = public["phase"] + observed["process_alive"] = public["process_alive"] + observed["verification"] = public["verification"] + observed["uploads"] = kwargs["uploads"] + return passed + + with ( + patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen), + ): + job = self.manager.start_ingest_job([upload["upload_id"]]) + log_path = Path(self.manager._load_job(job["job_id"])["log_path"]) + log_path.write_text( + self._json_text_event("MCP_INGEST_OK\nIngested source.pdf"), + encoding="utf-8", + ) + popen.returncode = 0 + with patch( + "obsidian_wiki.mcp_ingest.verify_ingest_outputs", + side_effect=verify_while_observable, + ): + self.manager.refresh_jobs() + + status = self.manager.get_ingest_job_status(job["job_id"]) + self.assertEqual(observed["phase"], "verifying") + self.assertFalse(observed["process_alive"]) + self.assertEqual(observed["verification"]["status"], "running") + self.assertEqual(observed["uploads"][0]["path"], str(self._upload_payload_path(upload["upload_id"]))) + self.assertEqual(status["state"], "succeeded") + self.assertEqual(status["phase"], "complete") + self.assertFalse(status["process_alive"]) + self.assertEqual(status["verification"], passed) + + def test_failed_verification_overrides_agent_success_and_preserves_output(self): + upload = self._complete_upload("source.pdf", b"source") + popen = FakePopen(returncode=None) + failed = { + "status": "failed", + "checked_at": "2026-07-16T10:01:00Z", + "failures": [{"code": "manifest_missing", "upload_id": upload["upload_id"]}], + "warnings": [], + "outputs": { + "source_count": 1, + "registry_count": 1, + "archive_count": 1, + "manifest_count": 0, + "indexed_count": 1, + }, + } + with ( + patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen), + ): + job = self.manager.start_ingest_job([upload["upload_id"]]) + log_path = Path(self.manager._load_job(job["job_id"])["log_path"]) + log_path.write_text( + self._json_text_event("MCP_INGEST_OK\nCreated references/source.md"), + encoding="utf-8", + ) + popen.returncode = 0 + with patch( + "obsidian_wiki.mcp_ingest.verify_ingest_outputs", + return_value=failed, + ): + self.manager.refresh_jobs() + + result = self.manager.get_ingest_job_result(job["job_id"]) + self.assertEqual(result["state"], "failed") + self.assertEqual(result["phase"], "complete") + self.assertEqual(result["verification"], failed) + self.assertIn("verification failed", result["error"]) + self.assertIn("Created references/source.md", result["opencode_output_tail"]) + + def test_protocol_failure_skips_verification(self): + upload = self._complete_upload("source.pdf", b"source") + popen = FakePopen(returncode=None) + with ( + patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run), + patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen), + ): + job = self.manager.start_ingest_job([upload["upload_id"]]) + log_path = Path(self.manager._load_job(job["job_id"])["log_path"]) + log_path.write_text( + self._json_text_event("MCP_INGEST_FAILED\nagent failed"), + encoding="utf-8", + ) + popen.returncode = 0 + with patch("obsidian_wiki.mcp_ingest.verify_ingest_outputs") as verify_mock: + self.manager.refresh_jobs() + + status = self.manager.get_ingest_job_status(job["job_id"]) + self.assertEqual(status["state"], "failed") + self.assertEqual(status["phase"], "complete") + self.assertEqual(status["verification"]["status"], "not_run") + verify_mock.assert_not_called() + def test_ingest_result_rejects_skill_failure_with_zero_exit_code(self): upload = self._complete_upload("source.pdf", b"source data") popen = FakePopen(returncode=None) @@ -464,8 +766,12 @@ class McpIngestTests(unittest.TestCase): encoding="utf-8", ) popen.returncode = 0 - result = self.manager.get_ingest_job_result(job["job_id"]) - status = self.manager.get_ingest_job_status(job["job_id"]) + with patch( + "obsidian_wiki.mcp_ingest.verify_ingest_outputs", + return_value=self._passed_verification(1), + ): + result = self.manager.get_ingest_job_result(job["job_id"]) + status = self.manager.get_ingest_job_status(job["job_id"]) self.assertEqual(result["state"], "succeeded") self.assertEqual(result["opencode_output_tail"], "Ingest complete") @@ -679,6 +985,21 @@ class McpIngestTests(unittest.TestCase): raise subprocess.TimeoutExpired(args, kwargs.get("timeout")) raise AssertionError(f"unexpected subprocess.run args: {args}") + def _passed_verification(self, source_count): + return { + "status": "passed", + "checked_at": "2026-07-16T10:01:00Z", + "failures": [], + "warnings": [], + "outputs": { + "source_count": source_count, + "registry_count": source_count, + "archive_count": source_count, + "manifest_count": source_count, + "indexed_count": source_count, + }, + } + def _json_text_event(self, text): return json.dumps( { diff --git a/tests/test_mcp_ingest_verification.py b/tests/test_mcp_ingest_verification.py new file mode 100644 index 0000000..a016c70 --- /dev/null +++ b/tests/test_mcp_ingest_verification.py @@ -0,0 +1,155 @@ +import hashlib +import json +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from obsidian_wiki.mcp_ingest_verification import verify_ingest_outputs + + +class McpIngestVerificationTests(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.root = Path(self.tempdir.name) + self.vault = self.root / "vault" + self.runtime = self.root / "runtime" + self.vault.mkdir() + self.runtime.mkdir() + (self.vault / "_meta").mkdir() + self.payload = b"verified source" + self.digest = hashlib.sha256(self.payload).hexdigest() + self.upload_path = self.runtime / "source.pdf" + self.upload_path.write_bytes(self.payload) + self.upload = { + "upload_id": "upload-1", + "path": str(self.upload_path), + "sha256": f"sha256:{self.digest}", + } + self.started_at = "2026-07-16T10:00:00Z" + self.archive_path = "_sources/archive/project/2026-07/source.pdf" + self.registry_record = { + "source_id": f"sha256:{self.digest}", + "archive_path": self.archive_path, + "related_pages": ["references/source.md"], + "catalog_paths": ["_sources/catalog/source.md"], + } + self.manifest_record = { + "ingested_at": "2026-07-16T10:01:00Z", + "content_hash": f"sha256:{self.digest}", + "source_id": f"sha256:{self.digest}", + "original_path": str(self.upload_path), + "archive_path": self.archive_path, + "pages_created": ["references/source.md"], + "pages_updated": [], + "catalog_paths": ["_sources/catalog/source.md"], + } + self._write_complete_outputs() + + def tearDown(self): + self.tempdir.cleanup() + + def _write_complete_outputs(self): + archive = self.vault / self.archive_path + archive.parent.mkdir(parents=True, exist_ok=True) + archive.write_bytes(self.payload) + for relative_path in ("references/source.md", "_sources/catalog/source.md"): + path = self.vault / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("verified", encoding="utf-8") + registry = self.vault / "_meta" / "source-registry.jsonl" + registry.write_text(json.dumps(self.registry_record) + "\n", encoding="utf-8") + manifest = {"version": 1, "sources": {str(self.upload_path): self.manifest_record}} + (self.vault / ".manifest.json").write_text( + json.dumps(manifest), + encoding="utf-8", + ) + db_path = self.vault / "_meta" / "file-assets.sqlite" + with sqlite3.connect(db_path) as conn: + conn.execute("CREATE TABLE file_assets (sha256 TEXT NOT NULL)") + conn.execute("INSERT INTO file_assets (sha256) VALUES (?)", (self.digest,)) + + def _verify(self): + return verify_ingest_outputs( + vault_path=self.vault, + uploads=[self.upload], + job_started_at=self.started_at, + archive_sources=True, + ) + + def test_passes_when_all_deterministic_outputs_exist(self): + result = self._verify() + + self.assertEqual(result["status"], "passed") + self.assertEqual(result["failures"], []) + self.assertEqual( + result["outputs"], + { + "source_count": 1, + "registry_count": 1, + "archive_count": 1, + "manifest_count": 1, + "indexed_count": 1, + }, + ) + self.assertTrue(result["checked_at"]) + + def test_fails_when_registry_record_is_missing(self): + (self.vault / "_meta" / "source-registry.jsonl").write_text("", encoding="utf-8") + + result = self._verify() + + self.assertEqual(result["status"], "failed") + self.assertIn("registry_missing", {item["code"] for item in result["failures"]}) + + def test_fails_when_required_archive_is_missing(self): + (self.vault / self.archive_path).unlink() + + result = self._verify() + + self.assertEqual(result["status"], "failed") + self.assertIn("archive_missing", {item["code"] for item in result["failures"]}) + + def test_fails_when_manifest_record_is_missing_or_stale(self): + self.manifest_record["ingested_at"] = "2026-07-16T09:59:59Z" + manifest = {"version": 1, "sources": {str(self.upload_path): self.manifest_record}} + (self.vault / ".manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + + result = self._verify() + + self.assertEqual(result["status"], "failed") + self.assertIn("manifest_missing", {item["code"] for item in result["failures"]}) + + def test_fails_when_file_asset_index_row_is_missing(self): + with sqlite3.connect(self.vault / "_meta" / "file-assets.sqlite") as conn: + conn.execute("DELETE FROM file_assets") + + result = self._verify() + + self.assertEqual(result["status"], "failed") + self.assertIn("index_missing", {item["code"] for item in result["failures"]}) + + def test_fails_when_declared_output_path_is_missing(self): + (self.vault / "references" / "source.md").unlink() + + result = self._verify() + + self.assertEqual(result["status"], "failed") + self.assertIn("declared_path_missing", {item["code"] for item in result["failures"]}) + + def test_archive_check_is_skipped_when_archiving_is_disabled(self): + (self.vault / self.archive_path).unlink() + + result = verify_ingest_outputs( + vault_path=self.vault, + uploads=[self.upload], + job_started_at=self.started_at, + archive_sources=False, + ) + + self.assertEqual(result["status"], "passed") + self.assertEqual(result["outputs"]["archive_count"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server_remote.py b/tests/test_mcp_server_remote.py index 4a8e5c5..f456b84 100644 --- a/tests/test_mcp_server_remote.py +++ b/tests/test_mcp_server_remote.py @@ -1055,6 +1055,16 @@ class RemoteMcpServerTests(unittest.TestCase): self.assertEqual(capabilities["ingest"]["preferred_text_tool"], "ingest_text_note") self.assertTrue(capabilities["ingest"]["supports_bundle_upload"]) self.assertTrue(capabilities["ingest"]["supports_body_text"]) + self.assertEqual( + capabilities["ingest"]["authoritative_job_phases"], + ["queued", "preflight", "agent_running", "verifying", "complete"], + ) + self.assertEqual( + capabilities["ingest"]["process_alive_source"], + "process_supervisor", + ) + self.assertTrue(capabilities["ingest"]["supports_post_run_verification"]) + self.assertFalse(capabilities["ingest"]["semantic_activity_inference"]) self.assertIn("body_text", capabilities["ingest"]["multipart_optional_fields"]) self.assertIn("source_type", capabilities["ingest"]["multipart_optional_fields"]) self.assertIn("validate_upload", capabilities["ingest"]["tools"])