diff --git a/docs/generated/release-truth.json b/docs/generated/release-truth.json index 05507e7e..9a49ed11 100644 --- a/docs/generated/release-truth.json +++ b/docs/generated/release-truth.json @@ -134,7 +134,7 @@ "console_entrypoints": 8, "mcp_tools": 50, "ops_cli_commands": 5, - "pytest_test_functions": 3795 + "pytest_test_functions": 3803 }, "feature_profile_matrix": { "capture_hook": [ diff --git a/docs/generated/release-truth.md b/docs/generated/release-truth.md index b07647e6..c9f56220 100644 --- a/docs/generated/release-truth.md +++ b/docs/generated/release-truth.md @@ -13,7 +13,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`. - Main CLI commands: **118** - Operations CLI commands: **5** - Console entrypoints: **8** -- Pytest source test functions: **3795** +- Pytest source test functions: **3803** ## MCP tools diff --git a/memorymaster/core/service.py b/memorymaster/core/service.py index 5ff14591..c9cc5d4d 100644 --- a/memorymaster/core/service.py +++ b/memorymaster/core/service.py @@ -1318,9 +1318,9 @@ def _build_query_statuses(self, include_stale: bool, include_conflicted: bool, i def _query_legacy_mode(self, query_text: str, limit: int, statuses: list[str], normalized_scopes: list[str] | None, include_sensitive: bool, requesting_agent: str | None, record_accesses: bool = True) -> list[dict[str, Any]]: """Query using legacy retrieval mode.""" - legacy = self._legacy_candidates( - query_text, limit, statuses, normalized_scopes - ) + conversational = " OR " in query_text + candidate_limit = max(limit * 12, 60) if conversational else limit + legacy = self._legacy_candidates(query_text, candidate_limit, statuses, normalized_scopes) if not include_sensitive: legacy = [claim for claim in legacy if not is_sensitive_claim(claim)] # Visibility: filter out private claims from other agents @@ -1328,7 +1328,7 @@ def _query_legacy_mode(self, query_text: str, limit: int, statuses: list[str], n ranked_rows = rank_claim_rows( query_text, legacy, - mode="legacy", + mode="hybrid" if conversational else "legacy", limit=limit, vector_hook=None, ) diff --git a/scripts/run_v47_operational_acceptance.py b/scripts/run_v47_operational_acceptance.py new file mode 100644 index 00000000..0b1754ff --- /dev/null +++ b/scripts/run_v47_operational_acceptance.py @@ -0,0 +1,527 @@ +"""Fail-closed v4.7 operational acceptance gate. + +Exit codes intentionally match the fleet steward gate: 0 accepted, 1 failed, +3 incomplete because a scheduled observation is not yet due. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import sqlite3 +import subprocess +import sys +import time +import urllib.error +import urllib.request +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Any, Callable, Iterable + + +REPO_ROOT = Path(__file__).resolve().parents[1] +try: + sys.path.remove(str(REPO_ROOT)) +except ValueError: + pass +sys.path.insert(0, str(REPO_ROOT)) + + +EXPECTED_VERSION = "4.7.0" +TARGET_HUMAN_ID = "mm-8aef" +TARGET_QUERY = "why does wezterm cli time out from Node but not from bash" +ACTIVE_OBSERVATION_STATES = ("pending", "leased", "retryable", "blocked") +TASK_NAMES = ( + "MemoryMaster-Dreaming", + "MemoryMasterSteward", + "MemoryMaster-MCP-HTTP-Hermes", + "MemoryMaster-Checkpoint-Daily", + "MemoryMaster-Checkpoint-Weekly", +) +CHECKPOINT_TASKS = ( + "MemoryMaster-Checkpoint-Daily", + "MemoryMaster-Checkpoint-Weekly", +) + + +class Verdict(str, Enum): + PASS = "PASS" + FAIL = "FAIL" + NOT_YET_DUE = "NOT-YET-DUE" + + +@dataclass(frozen=True, slots=True) +class CheckResult: + name: str + verdict: Verdict + detail: str + due_at: str | None = None + + +@dataclass(frozen=True, slots=True) +class GateConfig: + db: Path + runtime_python: Path + base_url: str + receipt_file: Path + session_hook: Path + expected_version: str = EXPECTED_VERSION + retrieval_samples: int = 5 + retrieval_p95_seconds: float = 2.0 + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_time(value: str | None) -> datetime | None: + if not value: + return None + normalized = value.strip().replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(normalized) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.astimezone() + return parsed.astimezone(timezone.utc) + + +def _connect_ro(db: Path) -> sqlite3.Connection: + connection = sqlite3.connect(f"file:{db.as_posix()}?mode=ro", uri=True, timeout=30) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA query_only=ON") + connection.execute("PRAGMA busy_timeout=30000") + return connection + + +def _discover_db(explicit: str | None) -> Path: + candidates: list[Path] = [] + if explicit: + candidates.append(Path(explicit).expanduser()) + env_db = os.environ.get("MEMORYMASTER_DEFAULT_DB") + if env_db: + candidates.append(Path(env_db).expanduser()) + candidates.append(Path.cwd() / "memorymaster.db") + try: + common = subprocess.run( + ["git", "rev-parse", "--path-format=absolute", "--git-common-dir"], + capture_output=True, + text=True, + check=True, + timeout=5, + ).stdout.strip() + candidates.append(Path(common).resolve().parent / "memorymaster.db") + except (OSError, subprocess.SubprocessError): + pass + for candidate in candidates: + if candidate.is_file(): + return candidate.resolve() + return candidates[0].resolve() + + +def _python_version(executable: Path) -> str | None: + try: + run = subprocess.run( + [str(executable), "-c", "import memorymaster; print(memorymaster.__version__)"], + capture_output=True, + text=True, + timeout=15, + check=False, + ) + except OSError: + return None + return run.stdout.strip() if run.returncode == 0 else None + + +def _discover_runtime_python(explicit: str | None, expected: str) -> Path: + if explicit: + return Path(explicit).expanduser().resolve() + env_python = os.environ.get("MEMORYMASTER_RUNTIME_PYTHON") + if env_python: + return Path(env_python).expanduser().resolve() + root = Path.home() / ".memorymaster" / "runtime" + suffix = Path("Scripts/python.exe") if os.name == "nt" else Path("bin/python") + candidates = sorted(root.glob(f"*/{suffix.as_posix()}"), key=lambda p: p.stat().st_mtime, reverse=True) + return next((item for item in candidates if _python_version(item) == expected), Path(sys.executable)) + + +def check_identity_and_db(config: GateConfig) -> CheckResult: + try: + import memorymaster + + source_version = memorymaster.__version__ + runtime_version = _python_version(config.runtime_python) + with _connect_ro(config.db) as connection: + migration = connection.execute("SELECT MAX(version) FROM schema_versions").fetchone()[0] + quick = [str(row[0]) for row in connection.execute("PRAGMA quick_check")] + except (OSError, sqlite3.Error) as exc: + return CheckResult("condition_1_identity_db", Verdict.FAIL, f"probe_error={type(exc).__name__}") + ok = ( + source_version == config.expected_version + and runtime_version == config.expected_version + and int(migration or 0) >= 21 + and quick == ["ok"] + ) + detail = ( + f"source={source_version} runtime={runtime_version or 'unavailable'} " + f"migration={migration} quick_check={','.join(quick[:3])}" + ) + return CheckResult("condition_1_identity_db", Verdict.PASS if ok else Verdict.FAIL, detail) + + +def _graph_lineage_errors(connection: sqlite3.Connection) -> int: + query = """ + SELECT COUNT(*) FROM graph_observation_supports gos + LEFT JOIN graph_observations go ON go.observation_claim_id=gos.observation_claim_id + LEFT JOIN claims sc ON sc.id=gos.supporting_claim_id + LEFT JOIN evidence_items ei ON ei.id=gos.evidence_item_id + LEFT JOIN source_items si ON si.id=gos.source_item_id + LEFT JOIN entity_edges ee ON ee.source_id=gos.source_entity_id + AND ee.target_id=gos.target_entity_id AND ee.relation=gos.relation + WHERE go.observation_claim_id IS NULL OR sc.id IS NULL OR ei.id IS NULL + OR si.id IS NULL OR ee.source_id IS NULL + """ + return int(connection.execute(query).fetchone()[0]) + + +def _current_graph_diagnostics(db: Path) -> tuple[int, int, int]: + from collections import defaultdict + + from memorymaster.knowledge.graph_observation_repository import _observation_support + from memorymaster.knowledge.graph_observations import discover_components + from memorymaster.knowledge.ontology import load_ontology + + ontology = load_ontology() + relations = tuple(sorted(ontology.relations)) + marks = ",".join("?" for _ in relations) + with _connect_ro(db) as connection: + scope_count = int(connection.execute( + "SELECT COUNT(DISTINCT scope) FROM entity_edge_supports" + ).fetchone()[0]) + rows = connection.execute(f""" + SELECT ees.supporting_claim_id AS claim_id, + cel.evidence_item_id AS evidence_id, + e.source_item_id, ees.source_entity_id, ees.relation, + ees.target_entity_id, ees.ontology_version, c.scope, + c.tenant_id, c.confidence, s.occurred_at + FROM entity_edge_supports ees + JOIN claims c ON c.id=ees.supporting_claim_id + JOIN claim_evidence_links cel ON cel.claim_id=c.id + JOIN evidence_items e ON e.id=cel.evidence_item_id + JOIN source_items s ON s.id=e.source_item_id + WHERE ees.scope=c.scope AND c.status='confirmed' + AND c.visibility<>'sensitive' + AND COALESCE(c.claim_type, '') NOT IN ('observation','skill','summary') + AND COALESCE(c.source_agent, '')<>'memorymaster-graph-observer' + AND s.retired_at IS NULL AND s.sensitivity='none' AND e.sensitivity='none' + AND ees.ontology_version=? AND ees.relation IN ({marks}) + ORDER BY c.scope, c.tenant_id, cel.evidence_item_id, c.id, + ees.source_entity_id, ees.relation, ees.target_entity_id + """, (ontology.version, *relations)).fetchall() + symmetric = frozenset(name for name, definition in ontology.relations.items() if definition.symmetric) + grouped: dict[tuple[str, str | None], list[Any]] = defaultdict(list) + for row in rows: + support = _observation_support(row, symmetric) + grouped[(support.scope, support.tenant_id)].append(support) + components = diagnostics = 0 + for (scope, tenant_id), supports in grouped.items(): + result = discover_components(supports, scope=scope, tenant_id=tenant_id) + components += len(result.components) + diagnostics += len(result.diagnostics) + return scope_count, components, diagnostics + + +def check_graph_observations(config: GateConfig) -> CheckResult: + try: + with _connect_ro(config.db) as connection: + marks = ",".join("?" for _ in ACTIVE_OBSERVATION_STATES) + backlog = int(connection.execute( + f"SELECT COUNT(*) FROM graph_observation_jobs WHERE status IN ({marks})", + ACTIVE_OBSERVATION_STATES, + ).fetchone()[0]) + completed = int(connection.execute( + "SELECT COUNT(*) FROM graph_observation_jobs WHERE stage='discover' AND status='completed'" + ).fetchone()[0]) + observations = int(connection.execute("SELECT COUNT(*) FROM graph_observations").fetchone()[0]) + lineage_errors = _graph_lineage_errors(connection) + scopes, eligible, diagnostics = _current_graph_diagnostics(config.db) + except (OSError, sqlite3.Error, RuntimeError, ValueError) as exc: + return CheckResult("condition_2_graph_observations", Verdict.FAIL, f"probe_error={type(exc).__name__}") + empty_proven = observations > 0 or (completed > 0 and scopes > 0 and eligible == 0) + ok = backlog == 0 and lineage_errors == 0 and empty_proven + detail = ( + f"backlog={backlog} completed_discovery={completed} observations={observations} " + f"current_scopes={scopes} eligible_components={eligible} diagnostics={diagnostics} " + f"lineage_errors={lineage_errors} empty_proven={str(empty_proven).lower()}" + ) + return CheckResult("condition_2_graph_observations", Verdict.PASS if ok else Verdict.FAIL, detail) + + +def _profile_marker_emitted(hook: Path, db: Path) -> bool: + if not hook.is_file(): + return False + env = dict(os.environ) + env["MEMORYMASTER_DEFAULT_DB"] = str(db) + try: + run = subprocess.run( + [sys.executable, str(hook)], + input="{}", + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + env=env, + timeout=30, + check=False, + ) + payload = json.loads(run.stdout) + context = payload.get("hookSpecificOutput", {}).get("additionalContext", "") + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): + return False + return run.returncode == 0 and "MemoryMaster compiled user profile" in str(context) + + +def check_compiled_profile(config: GateConfig) -> CheckResult: + try: + with _connect_ro(config.db) as connection: + active_run = int(connection.execute( + "SELECT COUNT(*) FROM compiled_profile_runs WHERE status='completed'" + ).fetchone()[0]) + facts = connection.execute( + "SELECT COUNT(*), MIN(independent_sessions), MIN(support_count) " + "FROM compiled_profile_facts WHERE status='active'" + ).fetchone() + mismatch = int(connection.execute(""" + SELECT COUNT(*) FROM compiled_profile_facts f + LEFT JOIN ( + SELECT fact_id, COUNT(*) supports, COUNT(DISTINCT session_id) sessions + FROM compiled_profile_supports GROUP BY fact_id + ) s ON s.fact_id=f.id + WHERE f.status='active' AND ( + f.support_count<>COALESCE(s.supports,0) + OR f.independent_sessions<>COALESCE(s.sessions,0) + ) + """).fetchone()[0]) + marker = _profile_marker_emitted(config.session_hook, config.db) + except (OSError, sqlite3.Error) as exc: + return CheckResult("condition_3_compiled_profile", Verdict.FAIL, f"probe_error={type(exc).__name__}") + count, min_sessions, min_supports = map(lambda value: int(value or 0), facts) + ok = active_run == 1 and count > 0 and min_sessions >= 2 and min_supports >= 2 and mismatch == 0 and marker + detail = ( + f"active_run={active_run} active_facts={count} min_sessions={min_sessions} " + f"min_supports={min_supports} support_mismatches={mismatch} session_marker={str(marker).lower()}" + ) + return CheckResult("condition_3_compiled_profile", Verdict.PASS if ok else Verdict.FAIL, detail) + + +def _http_status(url: str) -> int | None: + try: + with urllib.request.urlopen(url, timeout=5) as response: # noqa: S310 - fixed local operator URL by default + return int(response.status) + except (OSError, urllib.error.URLError): + return None + + +def _task_state() -> dict[str, dict[str, Any]]: + if os.name != "nt": + return {} + quoted = ",".join(f"'{name}'" for name in TASK_NAMES) + command = ( + "function fmt($v){if($null -eq $v){return ''}; return $v.ToString('o')}; " + f"$names=@({quoted}); $rows=@(foreach($n in $names){{" + "$t=Get-ScheduledTask -TaskName $n -ErrorAction SilentlyContinue; if($t){" + "$i=Get-ScheduledTaskInfo -TaskName $n; [pscustomobject]@{name=$n;enabled=[bool]$t.Settings.Enabled;" + "state=[string]$t.State;last_run=(fmt $i.LastRunTime);next_run=(fmt $i.NextRunTime);" + "last_result=[int64]$i.LastTaskResult}}}); $rows|ConvertTo-Json -Compress" + ) + run = subprocess.run( + ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", command], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=20, + check=False, + ) + if run.returncode != 0 or not run.stdout.strip(): + return {} + payload = json.loads(run.stdout) + rows = payload if isinstance(payload, list) else [payload] + return {str(row["name"]): row for row in rows} + + +def _read_receipts(path: Path) -> list[dict[str, Any]]: + if not path.is_file(): + return [] + rows: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + rows.append(value) + return rows + + +def checkpoint_result( + task_name: str, + state: dict[str, Any] | None, + receipts: Iterable[dict[str, Any]], + now: datetime, +) -> CheckResult: + label = "condition_4_checkpoint_" + ("daily" if task_name.endswith("Daily") else "weekly") + if not state or not state.get("enabled"): + return CheckResult(label, Verdict.FAIL, "scheduled task missing or disabled") + last_run = _parse_time(str(state.get("last_run") or "")) + never_ran = last_run is None or last_run.year < 2001 + due = _parse_time(str(state.get("next_run") or "")) if never_ran else last_run + valid = [ + row for row in receipts + if row.get("task") == task_name + and row.get("work_performed") is True + and str(row.get("result", "")).lower() == "pass" + and _parse_time(str(row.get("completed_at") or "")) is not None + ] + if valid: + return CheckResult(label, Verdict.PASS, f"real_work_receipts={len(valid)}") + if never_ran and due is not None and now < due: + due_text = due.isoformat() + return CheckResult(label, Verdict.NOT_YET_DUE, "first natural fire has not occurred", due_text) + due_text = due.isoformat() if due is not None else "unknown" + return CheckResult(label, Verdict.FAIL, f"no real-work receipt after due={due_text}", due_text) + + +def check_runtime(config: GateConfig, *, now: datetime | None = None) -> list[CheckResult]: + health = _http_status(config.base_url.rstrip("/") + "/healthz") + ready = _http_status(config.base_url.rstrip("/") + "/readyz") + try: + tasks = _task_state() + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): + tasks = {} + enabled = all(name in tasks and bool(tasks[name].get("enabled")) for name in TASK_NAMES) + base_ok = health == 200 and ready == 200 and enabled + base = CheckResult( + "condition_4_runtime", + Verdict.PASS if base_ok else Verdict.FAIL, + f"healthz={health} readyz={ready} enabled_tasks={sum(name in tasks and bool(tasks[name].get('enabled')) for name in TASK_NAMES)}/{len(TASK_NAMES)}", + ) + receipts = _read_receipts(config.receipt_file) + stamp = now or _utc_now() + checkpoints = [checkpoint_result(name, tasks.get(name), receipts, stamp) for name in CHECKPOINT_TASKS] + return [base, *checkpoints] + + +def _percentile_95(samples: list[float]) -> float: + ordered = sorted(samples) + return ordered[max(0, math.ceil(0.95 * len(ordered)) - 1)] + + +def _retrieval_once(db: Path, query: str, limit: int) -> list[str]: + from memorymaster.core.service import MemoryService + from memorymaster.recall.planner import RetrievalRequest, build_retrieval_plan + + service = MemoryService(str(db), workspace_root=db.parent, read_only=True) + plan = build_retrieval_plan(RetrievalRequest(query_text=query, limit=limit, trust_mode="trusted")) + rows = service.query_rows( + query_text=plan.search_text, + limit=plan.limit, + include_stale=False, + include_conflicted=False, + include_candidates=False, + retrieval_mode=plan.effective_mode, + allow_sensitive=False, + scope_allowlist=None, + record_accesses=False, + ) + return [str(row["claim"].human_id or "") for row in rows] + + +def check_retrieval( + config: GateConfig, + *, + retrieve: Callable[[Path, str, int], list[str]] = _retrieval_once, +) -> CheckResult: + timings: list[float] = [] + rankings: list[list[str]] = [] + try: + for _ in range(config.retrieval_samples): + started = time.perf_counter() + rankings.append(retrieve(config.db, TARGET_QUERY, 5)) + timings.append(time.perf_counter() - started) + except Exception as exc: # noqa: BLE001 - gate converts probe failures into a typed verdict + return CheckResult("condition_5_natural_language_retrieval", Verdict.FAIL, f"probe_error={type(exc).__name__}") + p95 = _percentile_95(timings) + target_hits = sum(TARGET_HUMAN_ID in ranking for ranking in rankings) + ok = target_hits == len(rankings) and p95 <= config.retrieval_p95_seconds + detail = ( + f"target={TARGET_HUMAN_ID} hits={target_hits}/{len(rankings)} " + f"p95_seconds={p95:.3f} budget_seconds={config.retrieval_p95_seconds:.3f} " + f"last_top5={','.join(rankings[-1])}" + ) + return CheckResult("condition_5_natural_language_retrieval", Verdict.PASS if ok else Verdict.FAIL, detail) + + +def exit_code(results: Iterable[CheckResult]) -> int: + verdicts = {result.verdict for result in results} + if Verdict.FAIL in verdicts: + return 1 + if Verdict.NOT_YET_DUE in verdicts: + return 3 + return 0 + + +def run_gate(config: GateConfig, *, now: datetime | None = None) -> list[CheckResult]: + results = [ + check_identity_and_db(config), + check_graph_observations(config), + check_compiled_profile(config), + ] + results.extend(check_runtime(config, now=now)) + results.append(check_retrieval(config)) + return results + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--db") + parser.add_argument("--runtime-python") + parser.add_argument("--base-url", default=os.environ.get("MEMORYMASTER_HTTP_URL", "http://127.0.0.1:8765")) + parser.add_argument("--receipt-file", default=str(Path.home() / ".memorymaster" / "checkpoints" / "work-receipts.jsonl")) + parser.add_argument("--session-hook", default=str(Path.home() / ".claude" / "hooks" / "memorymaster-session-start.py")) + parser.add_argument("--retrieval-samples", type=int, default=5) + parser.add_argument("--json", action="store_true") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + db = _discover_db(args.db) + config = GateConfig( + db=db, + runtime_python=_discover_runtime_python(args.runtime_python, EXPECTED_VERSION), + base_url=args.base_url, + receipt_file=Path(args.receipt_file).expanduser(), + session_hook=Path(args.session_hook).expanduser(), + retrieval_samples=max(1, min(20, args.retrieval_samples)), + ) + results = run_gate(config) + code = exit_code(results) + if args.json: + print(json.dumps({"exit_code": code, "checks": [{**asdict(item), "verdict": item.verdict.value} for item in results]}, indent=2)) + else: + for item in results: + due = f" due={item.due_at}" if item.due_at else "" + print(f"{item.name}: {item.verdict.value}{due} - {item.detail}") + label = {0: "ACCEPTED", 1: "FAILED", 3: "INCOMPLETE"}[code] + print(f"overall: {label} (exit {code})") + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_natural_language_retrieval_regression.py b/tests/test_natural_language_retrieval_regression.py new file mode 100644 index 00000000..f0f4b4d2 --- /dev/null +++ b/tests/test_natural_language_retrieval_regression.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from memorymaster.core.models import Claim +from memorymaster.core.service import MemoryService +from memorymaster.recall.planner import RetrievalRequest, build_retrieval_plan + + +QUERY = "why does wezterm cli time out from Node but not from bash" + + +def _claim(claim_id: int, text: str, human_id: str) -> Claim: + return Claim( + id=claim_id, + text=text, + idempotency_key=None, + normalized_text=None, + claim_type="fact", + subject=None, + predicate=None, + object_value=None, + scope="project:wezbridge", + volatility="low", + status="confirmed", + confidence=0.765, + pinned=False, + supersedes_claim_id=None, + replaced_by_claim_id=None, + created_at="2026-08-13T00:00:00+00:00", + updated_at="2026-08-13T00:00:00+00:00", + last_validated_at=None, + archived_at=None, + human_id=human_id, + tier="core", + ) + + +class _Store: + def __init__(self, claims: list[Claim]) -> None: + self.claims = claims + self.limits: list[int] = [] + + def list_claims(self, *, limit: int, **_kwargs) -> list[Claim]: + self.limits.append(limit) + return self.claims[:limit] + + +def test_fast_natural_language_path_returns_mm_8aef_in_top_five() -> None: + distractors = [ + _claim(index, f"WezTerm operational note number {index}", f"mm-decoy-{index}") + for index in range(1, 8) + ] + target = _claim( + 128576, + "Duplicate wezterm-gui processes cause wezterm CLI ETIMEDOUT from Node while bash stays fast.", + "mm-8aef", + ) + store = _Store([*distractors, target]) + service = MemoryService.__new__(MemoryService) + service.store = store + service.tenant_id = None + plan = build_retrieval_plan(RetrievalRequest(query_text=QUERY, limit=5)) + + rows = service._query_legacy_mode( + plan.search_text, + 5, + ["confirmed"], + None, + True, + None, + record_accesses=False, + ) + + assert [row["claim"].human_id for row in rows][:1] == ["mm-8aef"] + assert min(store.limits) >= 60 + + +def test_short_keyword_path_keeps_legacy_candidate_bound() -> None: + store = _Store([_claim(1, "wezterm note", "mm-short")]) + service = MemoryService.__new__(MemoryService) + service.store = store + service.tenant_id = None + + rows = service._query_legacy_mode( + "wezterm", 5, ["confirmed"], None, True, None, record_accesses=False + ) + + assert [row["claim"].human_id for row in rows] == ["mm-short"] + assert store.limits == [5] diff --git a/tests/test_v47_operational_acceptance.py b/tests/test_v47_operational_acceptance.py new file mode 100644 index 00000000..8b522044 --- /dev/null +++ b/tests/test_v47_operational_acceptance.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import importlib.util +import sys +from datetime import datetime, timezone +from pathlib import Path + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "run_v47_operational_acceptance.py" + + +def _module(): + spec = importlib.util.spec_from_file_location("v47_acceptance", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _config(module, tmp_path: Path, *, samples: int = 3): + return module.GateConfig( + db=tmp_path / "memory.db", + runtime_python=Path("python"), + base_url="http://127.0.0.1:8765", + receipt_file=tmp_path / "receipts.jsonl", + session_hook=tmp_path / "hook.py", + retrieval_samples=samples, + ) + + +def test_exit_codes_are_fail_closed() -> None: + module = _module() + passed = module.CheckResult("pass", module.Verdict.PASS, "ok") + pending = module.CheckResult("pending", module.Verdict.NOT_YET_DUE, "waiting") + failed = module.CheckResult("failed", module.Verdict.FAIL, "broken") + + assert module.exit_code([passed]) == 0 + assert module.exit_code([passed, pending]) == 3 + assert module.exit_code([passed, pending, failed]) == 1 + + +def test_gate_binds_imports_to_its_repository() -> None: + module = _module() + + assert Path(module.sys.path[0]).resolve() == module.REPO_ROOT + + +def test_checkpoint_is_pending_until_due_then_fails_without_receipt() -> None: + module = _module() + state = { + "enabled": True, + "last_run": "1999-11-30T00:00:00-03:00", + "next_run": "2026-08-14T11:25:00-03:00", + } + before = datetime(2026, 8, 14, 14, 0, tzinfo=timezone.utc) + after = datetime(2026, 8, 14, 15, 0, tzinfo=timezone.utc) + + pending = module.checkpoint_result("MemoryMaster-Checkpoint-Daily", state, [], before) + failed = module.checkpoint_result("MemoryMaster-Checkpoint-Daily", state, [], after) + + assert pending.verdict is module.Verdict.NOT_YET_DUE + assert pending.due_at == "2026-08-14T14:25:00+00:00" + assert failed.verdict is module.Verdict.FAIL + + +def test_checkpoint_pass_requires_real_work_receipt() -> None: + module = _module() + state = { + "enabled": True, + "last_run": "2026-08-14T11:25:02-03:00", + "next_run": "2026-08-15T11:25:00-03:00", + } + transport_only = { + "task": "MemoryMaster-Checkpoint-Daily", + "work_performed": False, + "result": "pass", + "completed_at": "2026-08-14T14:30:00Z", + } + real_work = {**transport_only, "work_performed": True} + now = datetime(2026, 8, 14, 15, 0, tzinfo=timezone.utc) + + assert module.checkpoint_result(state=state, task_name=transport_only["task"], receipts=[transport_only], now=now).verdict is module.Verdict.FAIL + assert module.checkpoint_result(state=state, task_name=real_work["task"], receipts=[real_work], now=now).verdict is module.Verdict.PASS + + +def test_retrieval_condition_requires_every_hit_and_latency_budget(tmp_path: Path, monkeypatch) -> None: + module = _module() + config = _config(module, tmp_path) + times = iter((0.0, 0.5, 1.0, 1.4, 2.0, 2.6)) + monkeypatch.setattr(module.time, "perf_counter", lambda: next(times)) + + passed = module.check_retrieval(config, retrieve=lambda *_: ["mm-8aef"]) + + assert passed.verdict is module.Verdict.PASS + assert "hits=3/3" in passed.detail + assert "p95_seconds=0.600" in passed.detail + + +def test_retrieval_condition_fails_when_target_missing(tmp_path: Path) -> None: + module = _module() + config = _config(module, tmp_path, samples=1) + + failed = module.check_retrieval(config, retrieve=lambda *_: ["mm-other"]) + + assert failed.verdict is module.Verdict.FAIL + assert "hits=0/1" in failed.detail