diff --git a/docs/README.md b/docs/README.md index 5c9072ab..e6849d4a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -69,6 +69,7 @@ Multitask MUST for this product. Index: [`operations/README.md`](operations/READ | [`operations/cluster-route-vs-slice-hand-carry.md`](operations/cluster-route-vs-slice-hand-carry.md) | Two named moves: ClusterRoute vs SliceHandCarry (#191 / #47 cousin; inventOnly) | | [`operations/admin-usage-report.md`](operations/admin-usage-report.md) | Admin-only serve usage JSON for a product-gate admin MCP (opaque alias; not agent MCP) | | [`operations/safe-upgrade.md`](operations/safe-upgrade.md) | Drain, restart, and restore a live serve inside memnet serve (MN-REQ-06.14) | +| [`operations/backup.md`](operations/backup.md) | Hourly checkpoint of loaded sessions, restic to R2, weekly restore check (MN-REQ-06.15) | | [`operations/one-session-per-document.md`](operations/one-session-per-document.md) | One MemNet session per document over loopback serve (no MCP front) | Product skill: [`.cursor/skills/memnet-reference/`](../.cursor/skills/memnet-reference/). SysML trail: MN-REQ-12 → [`sysml-models/outputs/multitask-case-study.md`](../sysml-models/outputs/multitask-case-study.md). diff --git a/docs/operations/README.md b/docs/operations/README.md index bdafc5f1..dd4e97ce 100644 --- a/docs/operations/README.md +++ b/docs/operations/README.md @@ -13,6 +13,7 @@ Agent operating doctrine for this product (not domain recipes). | [`cluster-route-vs-slice-hand-carry.md`](cluster-route-vs-slice-hand-carry.md) | Two named moves: ClusterRoute vs SliceHandCarry (#191 / #47 cousin; inventOnly) | | [`admin-usage-report.md`](admin-usage-report.md) | Admin-only serve usage JSON (opaque alias; not agent MCP; MN-REQ-06.11) | | [`safe-upgrade.md`](safe-upgrade.md) | Drain and restore a live serve from inside memnet serve (MN-REQ-06.14) | +| [`backup.md`](backup.md) | Hourly checkpoint of loaded sessions, restic to R2, weekly checkpoint-verify (MN-REQ-06.15) | | [`one-session-per-document.md`](one-session-per-document.md) | Product gate: one serve session per document over loopback (no MCP); 0.19.18 probe | Application pattern for `modelbasedPrj-*` / `SysMLEdgePrj-*`: [`../application-notes/system/llm-system-dev-multitask.md`](../application-notes/system/llm-system-dev-multitask.md). Index: [`../README.md`](../README.md). diff --git a/docs/operations/backup.md b/docs/operations/backup.md new file mode 100644 index 00000000..6e506532 --- /dev/null +++ b/docs/operations/backup.md @@ -0,0 +1,53 @@ +# Hourly checkpoint backup + +A backup of snapshot files only sees sessions a product has already saved. Loaded sessions stay in RAM until something calls `session_save`, and not every product does. `memnet admin checkpoint` copies every loaded session to disk without draining the serve and without refusing new work (MN-REQ-06.15). + +The command is admin-only. It is not an agent MCP tool. `MEMNET_ADMIN_TOKEN` is the serve credential (unset is `@ERR: admin_unconfigured`; a mismatch is `@ERR: admin_denied`). The caller presents that same string with `--token`. Do not log the token, and do not commit it. + +Each loaded session is locked only while its graph is serialised. Other sessions keep serving. The copy uses the lossless snapshot path and is written atomically (temp file, fsync, rename) to: + +```text +$MEMNET_STATE_DIR/checkpoint/.snap +``` + +`checkpoint-manifest.json` in that directory records each saved session's row count, edge count, and checksum, plus the serve version, the snapshot format version, and a timestamp. The row shape matches the upgrade manifest. A session that cannot be saved is named on stderr (`snapshot_unsaveable` and the reason) and skipped. The command still writes the others. The exit code is non-zero when any session was skipped. Stdout is: + +```text +@STAT: checkpoint|ok||skipped||bytes| +``` + +A checkpoint does not slide TTL or expiry clocks, and it does not change session ids, ACL, or live state. It does not retire a manifest and it does not load a snapshot into the serve. Files for sessions that are no longer loaded are removed. A skipped session does not keep an older snap, so the directory matches the manifest. + +## Hourly: checkpoint, then restic to R2 + +Set the repository and the R2 credentials in the job environment. R2 speaks the S3 API. Do not write the key values into git or into this file. + +```text +RESTIC_REPOSITORY=s3:https://.r2.cloudflarestorage.com/ +AWS_ACCESS_KEY_ID +AWS_SECRET_ACCESS_KEY +MEMNET_STATE_DIR +MEMNET_ADMIN_TOKEN +``` + +```bash +memnet admin checkpoint \ + --token "$MEMNET_ADMIN_TOKEN" \ + --state-dir "$MEMNET_STATE_DIR" +restic backup "$MEMNET_STATE_DIR/checkpoint" +``` + +Run that once an hour. Restic the checkpoint directory when the command exits 0 (`skipped|0`). A non-zero exit means a loaded session was skipped; leave the previous restic snapshot as the latest good copy and alert on the named session. + +## Weekly: restore, then checkpoint-verify + +Restore the latest restic snapshot onto a scratch directory (not the live state directory). Then check that copy with the throwaway loader: + +```bash +restic restore latest --target "$RESTORE_DIR" +memnet admin checkpoint-verify \ + --token "$MEMNET_ADMIN_TOKEN" \ + "$RESTORE_DIR$MEMNET_STATE_DIR/checkpoint" +``` + +Stdout is `@STAT: checkpoint_verify|ok||failed|`. The helper loads each snapshot into a throwaway in-process store and checks row counts, edge counts, and checksums against the manifest. A corrupt or missing file fails the check and leaves the files in place. It does not load those sessions into the live serve. A non-zero exit means the restored copy is not a good backup. diff --git a/parts/common/memnet/memnet/checkpoint.py b/parts/common/memnet/memnet/checkpoint.py new file mode 100644 index 00000000..48e228fe --- /dev/null +++ b/parts/common/memnet/memnet/checkpoint.py @@ -0,0 +1,350 @@ +"""Admin checkpoint of every loaded session (MN-REQ-06.15). + +Not an agent MCP tool. Copies each loaded session through the lossless +snapshot path without draining the serve, refusing new opens, sliding +TTL, or loading anything back into the live registry. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import threading +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from memnet import __version__ +from memnet.admin_usage import authenticate +from memnet.exceptions import MemNetError +from memnet.registry import list_entries +from memnet.session import iso_timestamp +from memnet.upgrade import snapshot_format_version, state_dir + +CHECKPOINT_DIRNAME = "checkpoint" +MANIFEST_NAME = "checkpoint-manifest.json" +MANIFEST_FORMAT = 1 +_PHASE_OPEN = "open" +_checkpoint_lock = threading.Lock() + + +def _fsync_fd(fd: int) -> None: + os.fsync(fd) + + +def atomic_write_bytes(path: Path, data: bytes) -> None: + """Write ``data`` via a temp file, fsync, and rename. + + A failure before rename leaves the destination untouched and removes + the temp file. The destination is never a partial write. + """ + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + try: + with tmp.open("wb") as handle: + handle.write(data) + handle.flush() + _fsync_fd(handle.fileno()) + os.replace(tmp, path) + except OSError: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + raise + try: + dir_fd = os.open(path.parent, os.O_RDONLY) + except OSError: + return + try: + try: + _fsync_fd(dir_fd) + except OSError: + return + finally: + os.close(dir_fd) + + +def _safe_name(session_id: str) -> str: + cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in session_id) + return cleaned or "session" + + +def _edge_count(store: Any) -> int: + total = 0 + for hid in store.write_order: + rec = store._by_hid.get(hid) + if rec is not None and rec.tag == "EDG": + total += 1 + return total + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _stat(saved: int, skipped: int, nbytes: int) -> str: + return f"@STAT: checkpoint|ok|{saved}|skipped|{skipped}|bytes|{nbytes}" + + +def _verify_stat(ok: int, failed: int) -> str: + return f"@STAT: checkpoint_verify|ok|{ok}|failed|{failed}" + + +def checkpoint_dir(directory: Path | None = None) -> Path: + root = directory or state_dir() + return root / CHECKPOINT_DIRNAME + + +def _refuse_if_draining() -> None: + from memnet.upgrade import drain_gate + + if drain_gate.phase != _PHASE_OPEN: + raise MemNetError( + "serve_draining", + f"retry_after_s={drain_gate.retry_after_s}", + exit_code=2, + ) + + +def _serialise(entry: Any) -> tuple[bytes, int, int, dict[str, Any]]: + """Copy one session. The session lock covers serialise only.""" + from memnet.session import SessionStore + from memnet.snapshot import snapshot_text + + sid = entry.meta.session_id + with entry.lock: + ss = SessionStore(sid, entry.store.caps) + text = snapshot_text(ss) + rows = entry.store.row_count_non_law() + edges = _edge_count(entry.store) + meta = { + "expires_at": entry.meta.expires_at, + "created_at": entry.meta.created_at, + "ttl_minutes": entry.meta.ttl_minutes, + "house_tags": sorted(entry.tag_map.tags.keys()), + "product": entry.meta.product, + } + return text.encode("utf-8"), rows, edges, meta + + +@dataclass +class CheckpointResult: + saved: int + skipped: list[dict[str, str]] + nbytes: int + manifest_path: Path + stat: str + exit_code: int + + def stdout_text(self) -> str: + return self.stat + "\n" + + def stderr_text(self) -> str: + lines = [] + for row in self.skipped: + lines.append(f"@ERR: {row['code']}|{row['session_id']}|{row['detail']}") + return ("\n".join(lines) + "\n") if lines else "" + + +def run_checkpoint( + presented_token: str | None, + *, + directory: Path | None = None, +) -> CheckpointResult: + """Snapshot every loaded session. Unsaveable sessions are skipped.""" + authenticate(presented_token) + with _checkpoint_lock: + return _checkpoint_locked(directory) + + +def _checkpoint_locked(directory: Path | None) -> CheckpointResult: + _refuse_if_draining() + dest = checkpoint_dir(directory) + dest.mkdir(parents=True, exist_ok=True) + saved_rows: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] + for entry in list_entries(): + sid = entry.meta.session_id + try: + payload, rows, edges, meta = _serialise(entry) + except MemNetError as exc: + skipped.append({"session_id": sid, "code": exc.code, "detail": exc.message}) + continue + except OSError as exc: + skipped.append( + {"session_id": sid, "code": "snapshot_io_error", "detail": type(exc).__name__} + ) + continue + name = f"{_safe_name(sid)}.snap" + path = dest / name + try: + atomic_write_bytes(path, payload) + except OSError as exc: + skipped.append( + {"session_id": sid, "code": "snapshot_io_error", "detail": type(exc).__name__} + ) + continue + digest = _sha256(payload) + saved_rows.append( + { + "session_id": sid, + "file": name, + "rows": rows, + "edges": edges, + "checksum_sha256": digest, + "expires_at": meta["expires_at"], + "created_at": meta["created_at"], + "ttl_minutes": meta["ttl_minutes"], + "house_tags": meta["house_tags"], + "product": meta["product"], + "saved": True, + } + ) + loaded = {item.meta.session_id for item in list_entries()} + saved_rows = [row for row in saved_rows if row["session_id"] in loaded] + keep = {row["file"] for row in saved_rows} + nbytes = 0 + for row in saved_rows: + snap = dest / row["file"] + if snap.is_file(): + nbytes += snap.stat().st_size + for path in dest.glob("*.snap"): + if path.name not in keep: + path.unlink() + for path in dest.glob("*.tmp"): + path.unlink() + manifest = { + "manifest_format": MANIFEST_FORMAT, + "snapshot_format": snapshot_format_version(), + "serve_version": __version__, + "timestamp": iso_timestamp(), + "sessions": saved_rows, + "unsaved": skipped, + } + manifest_path = dest / MANIFEST_NAME + atomic_write_bytes( + manifest_path, + (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"), + ) + stat = _stat(len(saved_rows), len(skipped), nbytes) + return CheckpointResult( + saved=len(saved_rows), + skipped=skipped, + nbytes=nbytes, + manifest_path=manifest_path, + stat=stat, + exit_code=2 if skipped else 0, + ) + + +def checkpoint_envelope(token: str | None) -> dict[str, Any]: + """Serve wire op ``{checkpoint: true, admin_token}``.""" + try: + result = run_checkpoint(token) + except MemNetError as exc: + return { + "exit_code": exc.exit_code, + "stdout": "", + "stderr": f"@ERR: {exc.code}|{exc.message}\n", + } + return { + "exit_code": result.exit_code, + "stdout": result.stdout_text(), + "stderr": result.stderr_text(), + } + + +def _fail(failed: list[dict[str, str]], name: str, code: str, detail: str) -> None: + failed.append({"file": name, "code": code, "detail": detail}) + + +def _snap_name(file_field: str) -> str | None: + if not file_field or file_field != Path(file_field).name: + return None + if file_field in {".", ".."} or "/" in file_field or "\\" in file_field: + return None + return file_field + + +@dataclass +class VerifyResult: + ok: int + failed: list[dict[str, str]] + stat: str + exit_code: int + + def stdout_text(self) -> str: + return self.stat + "\n" + + def stderr_text(self) -> str: + lines = [f"@ERR: {row['code']}|{row['file']}|{row['detail']}" for row in self.failed] + return ("\n".join(lines) + "\n") if lines else "" + + +def verify_checkpoint(directory: Path, presented_token: str | None) -> VerifyResult: + """Load each snap in ``directory`` into a throwaway store and check the manifest.""" + authenticate(presented_token) + failed: list[dict[str, str]] = [] + ok = 0 + manifest_path = directory / MANIFEST_NAME + if not manifest_path.is_file(): + _fail(failed, MANIFEST_NAME, "checkpoint_manifest_missing", "missing manifest") + stat = _verify_stat(0, len(failed)) + return VerifyResult(ok=0, failed=failed, stat=stat, exit_code=2) + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + _fail(failed, MANIFEST_NAME, "checkpoint_manifest", type(exc).__name__) + stat = _verify_stat(0, len(failed)) + return VerifyResult(ok=0, failed=failed, stat=stat, exit_code=2) + if not isinstance(manifest, dict): + _fail(failed, MANIFEST_NAME, "checkpoint_manifest", "manifest must be an object") + stat = _verify_stat(0, len(failed)) + return VerifyResult(ok=0, failed=failed, stat=stat, exit_code=2) + sessions = manifest.get("sessions") or [] + if not isinstance(sessions, list): + _fail(failed, MANIFEST_NAME, "checkpoint_manifest", "sessions must be a list") + sessions = [] + listed: set[str] = set() + from memnet.snapshot import materialise_snapshot_text + + for row in sessions: + if not isinstance(row, dict): + _fail(failed, "", "checkpoint_manifest", "bad row") + continue + name = _snap_name(str(row.get("file") or "")) + label = name or str(row.get("session_id") or "") + if name is None: + _fail(failed, label, "checkpoint_path", "snapshot file must be a single name") + continue + listed.add(name) + path = directory / name + if not path.is_file(): + _fail(failed, name, "checkpoint_missing", "missing snapshot file") + continue + data = path.read_bytes() + want = str(row.get("checksum_sha256") or "") + if _sha256(data) != want: + _fail(failed, name, "checkpoint_checksum", "checksum mismatch") + continue + try: + entry = materialise_snapshot_text(data.decode("utf-8")) + except (MemNetError, UnicodeError) as exc: + code = exc.code if isinstance(exc, MemNetError) else "checkpoint_corrupt" + detail = exc.message if isinstance(exc, MemNetError) else type(exc).__name__ + _fail(failed, name, code, detail) + continue + rows = entry.store.row_count_non_law() + edges = _edge_count(entry.store) + if rows != int(row.get("rows", -1)) or edges != int(row.get("edges", -1)): + _fail(failed, name, "checkpoint_counts", "row or edge count mismatch") + continue + ok += 1 + if directory.is_dir(): + for path in sorted(directory.glob("*.snap")): + if path.name not in listed: + _fail(failed, path.name, "checkpoint_extra", "snapshot not in the manifest") + stat = _verify_stat(ok, len(failed)) + return VerifyResult(ok=ok, failed=failed, stat=stat, exit_code=2 if failed else 0) diff --git a/parts/common/memnet/memnet/cli.py b/parts/common/memnet/memnet/cli.py index f53be7c8..4cf2a5ad 100644 --- a/parts/common/memnet/memnet/cli.py +++ b/parts/common/memnet/memnet/cli.py @@ -569,6 +569,66 @@ def admin_upgrade_prepare( raise typer.Exit(result.exit_code) +@admin_app.command("checkpoint") +def admin_checkpoint( + token: Annotated[ + str | None, + typer.Option( + "--token", + help="Admin token presented by the caller (do not log). Not MEMNET_ADMIN_TOKEN.", + ), + ] = None, + state_dir: Annotated[ + str | None, + typer.Option("--state-dir", help="State directory (default MEMNET_STATE_DIR)."), + ] = None, +) -> None: + """Save every loaded session without draining. Not an agent MCP tool.""" + from memnet.admin_usage import caller_token + from memnet.checkpoint import run_checkpoint + + presented = token if token else caller_token() + directory = Path(state_dir) if state_dir else None + try: + result = run_checkpoint(presented, directory=directory) + except MemNetError as exc: + _handle_error(exc) + return + sys.stdout.write(result.stdout_text()) + if result.stderr_text(): + sys.stderr.write(result.stderr_text()) + if result.exit_code: + raise typer.Exit(result.exit_code) + + +@admin_app.command("checkpoint-verify") +def admin_checkpoint_verify( + directory: Annotated[Path, typer.Argument(help="Checkpoint directory to check.")], + token: Annotated[ + str | None, + typer.Option( + "--token", + help="Admin token presented by the caller (do not log). Not MEMNET_ADMIN_TOKEN.", + ), + ] = None, +) -> None: + """Check a checkpoint directory in a throwaway store. Not an agent MCP tool.""" + from memnet.admin_usage import caller_token + from memnet.checkpoint import verify_checkpoint + + presented = token if token else caller_token() + try: + result = verify_checkpoint(directory, presented) + except MemNetError as exc: + _handle_error(exc) + return + sys.stdout.write(result.stdout_text()) + if result.stderr_text(): + sys.stderr.write(result.stderr_text()) + if result.exit_code: + raise typer.Exit(result.exit_code) + + @session_app.command("expire-status") def session_expire_status() -> None: """Booleans for expire-save (serve_status). Path redacted; no sids.""" diff --git a/parts/common/memnet/memnet/serve.py b/parts/common/memnet/memnet/serve.py index 09e1c468..90ea6613 100644 --- a/parts/common/memnet/memnet/serve.py +++ b/parts/common/memnet/memnet/serve.py @@ -92,6 +92,10 @@ def _handle_request(payload: dict[str, Any]) -> dict[str, Any]: from memnet.upgrade import upgrade_prepare_envelope return upgrade_prepare_envelope(token_s, allow_unsaved=bool(payload.get("allow_unsaved"))) + if payload.get("checkpoint") is True: + from memnet.checkpoint import checkpoint_envelope + + return checkpoint_envelope(token_s) argv = payload.get("args", []) if not isinstance(argv, list): diff --git a/parts/common/memnet/memnet/snapshot.py b/parts/common/memnet/memnet/snapshot.py index 0a6b80ea..d7c931b6 100644 --- a/parts/common/memnet/memnet/snapshot.py +++ b/parts/common/memnet/memnet/snapshot.py @@ -425,3 +425,30 @@ def load_snapshot_text( ) register(session_id, entry) return SessionStore(session_id, caps) + + +def materialise_snapshot_text(text: str, *, caps: Caps | None = None) -> SessionEntry: + """Parse a snapshot into a store that is not registered. + + Checkpoint verify uses this throwaway store. It does not mint an id, + slide clocks, or touch the live registry. + """ + caps = caps or Caps() + meta, map_lines, rel_lines, rec_lines = _parse_sections(split_snapshot_lines(text)) + tag_map = load_persisted_map_from_lines(map_lines, caps) + relations = _parse_relations(rel_lines) + if not relations: + from memnet.session import _seed_relations + + relations = set(_seed_relations()) + store = MemStore(tag_map, caps) + used_nicks: set[str] = set() + for line in rec_lines: + rec = parse_line(line, tag_map, caps, used_nicks=used_nicks) + store.upsert(rec, relations=relations) + return SessionEntry( + meta=meta, + tag_map=tag_map, + store=store, + relations=relations, + ) diff --git a/sysml-models/models/deploy.sysml b/sysml-models/models/deploy.sysml index fa5b2343..fa5f6e7d 100644 --- a/sysml-models/models/deploy.sysml +++ b/sysml-models/models/deploy.sysml @@ -1360,6 +1360,62 @@ package MemNet { satisfy MN_REQ_00_MissionBridge::MN_REQ_07_McpAgentBoundary::MN_REQ_07_3_NoDomainProductTools; } + part def CmdAdminCheckpoint { + doc /* + CLI/serve: memnet admin checkpoint. Envelope + {checkpoint:true, admin_token}. NOT an agent MCP tool. + Does not drain and does not refuse a new session open. + */ + attribute requiresStoreKeyId : Boolean = false; + attribute implemented : Boolean = true; + attribute productCommand : Boolean = false; + attribute onAgentMcp : Boolean = false; + } + + part def CmdAdminCheckpointVerify { + doc /* + CLI/serve: memnet admin checkpoint-verify . Loads + snapshots into a throwaway store and checks the manifest. + NOT an agent MCP tool. Does not load the live registry. + */ + attribute requiresStoreKeyId : Boolean = false; + attribute implemented : Boolean = true; + attribute productCommand : Boolean = false; + attribute onAgentMcp : Boolean = false; + } + + part def ServeCheckpoint { + doc /* + Admin checkpoint of every loaded session (MN-REQ-06.15). + Per-session lock, lossless atomic snapshot under + MEMNET_STATE_DIR/checkpoint. Unsaveable sessions are + named and skipped. Does not drain, retire, or load. + */ + attribute implemented : Boolean = true; + attribute noSemVerBump : Boolean = true; + attribute onAgentMcp : Boolean = false; + attribute adminCredEnv : String = "MEMNET_ADMIN_TOKEN"; + attribute refusesNewSessionOpen : Boolean = false; + attribute drainsServe : Boolean = false; + attribute sessionLockOnly : Boolean = true; + attribute usesLosslessSnapshot : Boolean = true; + attribute atomicWrite : Boolean = true; + attribute skipsUnsaveable : Boolean = true; + attribute unsaveableFatal : Boolean = false; + attribute exitNonZeroWhenSkipped : Boolean = true; + attribute removesStale : Boolean = true; + attribute preservesSessionId : Boolean = true; + attribute preservesAcl : Boolean = true; + attribute preservesTtlClock : Boolean = true; + attribute retiresOrLoads : Boolean = false; + attribute statCheckpoint : String = "checkpoint"; + attribute statCheckpointVerify : String = "checkpoint_verify"; + + satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_15_ServeCheckpoint; + satisfy MN_REQ_00_MissionBridge::MN_REQ_01_SessionLifecycle::MN_REQ_01_9_SnapshotLosslessRoundTrip; + satisfy MN_REQ_00_MissionBridge::MN_REQ_07_McpAgentBoundary::MN_REQ_07_3_NoDomainProductTools; + } + part def TcpServeBridge { doc /* Migration / multi-process fallback: length-prefixed JSON over TCP @@ -1369,6 +1425,7 @@ package MemNet { path redacted) for MemNetUsageDashboard. adminUsage = MN-REQ-06.11 admin JSON (opaque aliases; not 06.5). safeUpgrade = MN-REQ-06.14 drain, manifest, and startup restore. + serveCheckpoint = MN-REQ-06.15 per-session checkpoint; no drain. Many mn_* sessions in one registry. Not a projectId map. Not a second graph. Admin report MUST NOT emit those mn_* ids. */ @@ -1379,6 +1436,7 @@ package MemNet { port adminReportOut : AdminUsageReportOutPort; part adminUsage : AdminUsageReport; part safeUpgrade : SafeServeUpgrade; + part serveCheckpoint : ServeCheckpoint; attribute host : String = "127.0.0.1"; attribute listenPort : Integer = 18765; attribute requestOutputIsolated : Boolean = true; @@ -1388,6 +1446,7 @@ package MemNet { satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_11_AdminServeUsageReport; satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_13_ServeRequestIsolation; satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_14_SafeServeUpgrade; + satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_15_ServeCheckpoint; satisfy MN_REQ_00_MissionBridge::MN_REQ_12_MultitaskMode::MN_REQ_12_2_SharedStoreTransport; } @@ -1722,6 +1781,8 @@ package MemNet { part catalogSnap : CmdCatalogSnap; part adminUsageReport : CmdAdminUsageReport; part upgradePrepare : CmdAdminUpgradePrepare; + part checkpoint : CmdAdminCheckpoint; + part checkpointVerify : CmdAdminCheckpointVerify; attribute leftoverReadGetNested : Boolean = false; attribute leftoverNewNested : Boolean = false; attribute leftoverAssignedIdMapNested : Boolean = false; @@ -1739,10 +1800,12 @@ package MemNet { attribute noRequiredIdTools : Boolean = true; attribute adminUsageOnAgentMcp : Boolean = false; attribute upgradePrepareOnAgentMcp : Boolean = false; + attribute checkpointOnAgentMcp : Boolean = false; satisfy MN_REQ_00_MissionBridge::MN_REQ_01_SessionLifecycle::MN_REQ_01_2_SessionLifecycleOps; satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_11_AdminServeUsageReport; satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_14_SafeServeUpgrade; + satisfy MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_15_ServeCheckpoint; satisfy MN_REQ_00_MissionBridge::MN_REQ_01_SessionLifecycle::MN_REQ_01_4_SaveSessionSnapshot; satisfy MN_REQ_00_MissionBridge::MN_REQ_01_SessionLifecycle::MN_REQ_01_5_LoadSessionSnapshot; satisfy MN_REQ_00_MissionBridge::MN_REQ_01_SessionLifecycle::MN_REQ_01_6_OptionalKeepSessionId; diff --git a/sysml-models/models/implementation.sysml b/sysml-models/models/implementation.sysml index 534d1b04..a74d235d 100644 --- a/sysml-models/models/implementation.sysml +++ b/sysml-models/models/implementation.sysml @@ -127,6 +127,11 @@ package MemNetImplementation { attribute path : String = "parts/common/memnet/memnet/upgrade.py"; attribute pyName : String = "memnet.upgrade"; } + part def ServeCheckpointMod { + doc /* Admin checkpoint of loaded sessions without drain (MN-REQ-06.15). */ + attribute path : String = "parts/common/memnet/memnet/checkpoint.py"; + attribute pyName : String = "memnet.checkpoint"; + } part def CliFacadeMod { attribute path : String = "parts/common/memnet/memnet/cli.py"; attribute pyName : String = "memnet.cli"; @@ -279,6 +284,7 @@ package MemNetImplementation { part serveDaemonMod : ServeDaemonMod; part adminUsageMod : AdminUsageReportMod; part safeUpgradeMod : SafeUpgradeMod; + part serveCheckpointMod : ServeCheckpointMod; part cliMod : CliFacadeMod; part sessionsMod : SessionLifecycleMod; part storeMod : GraphStoreMod; @@ -365,6 +371,11 @@ package MemNetImplementation { end logical ::> memNetSystem.core.transport.tcp.safeUpgrade; end code ::> safeUpgradeMod; } + allocation serveCheckpointToMod : SoftwareAllocate { + doc /* Per-session checkpoint snaps on memnet-serve. Not agent MCP. */ + end logical ::> memNetSystem.core.transport.tcp.serveCheckpoint; + end code ::> serveCheckpointMod; + } allocation cliToMod : SoftwareAllocate { end logical ::> memNetSystem.core.cli; end code ::> cliMod; diff --git a/sysml-models/models/requirements.sysml b/sysml-models/models/requirements.sysml index 8d10367c..db7a4b4b 100644 --- a/sysml-models/models/requirements.sysml +++ b/sysml-models/models/requirements.sysml @@ -37,6 +37,7 @@ MN-REQ-06.12 Product gateway on memnet-mcp: per-product route to owning serve (not mn_tip_) MN-REQ-06.13 Serve request isolation: each command's stdout, stderr, and exit_code MN-REQ-06.14 Safe serve upgrade inside memnet serve: drain, manifest, restore, retire; no silent session loss + MN-REQ-06.15 Admin checkpoint of every loaded session: lossless snaps, no drain, no traffic refuse MN-REQ-11.17 Catalog Snap / model Snap — session strata (0.15) MN-REQ-11.17.1 Cross-cut satisfy locators on the catalog; SysMLEdge pin_map is a different tool (not this catalog) @@ -1188,6 +1189,58 @@ package MemNetRequirements { */ attribute requirementId : String = "MN-REQ-06.14"; } + + requirement def MN_REQ_06_15_ServeCheckpoint { + doc /* + SHALL checkpoint every loaded session to disk without + draining the serve and without refusing traffic. + Patch-level c on 0.19. Same usage method (cue then + pin_map, mutate). No SemVer bump. + Admin credential MEMNET_ADMIN_TOKEN (same gate as + MN-REQ-06.11 and MN-REQ-06.14; unset + admin_unconfigured; mismatch admin_denied). + Command `memnet admin checkpoint` and envelope + {checkpoint:true, admin_token}. Restore-test helper + `memnet admin checkpoint-verify `. Neither is an + agent MCP tool (onAgentMcp=false). + For each loaded session, take that session's lock only + long enough to serialise a consistent copy. Other + sessions keep serving. A new session open SHALL NOT + be refused (refusesNewSessionOpen=false). The serve + SHALL NOT enter quiesce (drainsServe=false). + Write the copy through the lossless snapshot path + (MN-REQ-01.9), atomically (temp file, fsync, rename), + to /checkpoint/.snap. + Overwrite the previous checkpoint for that session. + Remove checkpoint files for sessions that are no + longer loaded. A skipped session SHALL NOT keep an + older snap in that directory: the directory after a + run holds one snap per session saved in that run. + Write checkpoint-manifest.json in that directory. + Reuse the upgrade manifest row (session, rows, edges, + checksum) plus serve version, snapshot format version, + and a timestamp. + A session that cannot be saved (snapshot_unsaveable + and the reason) SHALL be named and skipped. It SHALL + NOT fail the other sessions. The exit code SHALL be + non-zero when any session was skipped. Print + @STAT: checkpoint|ok|n|skipped|m|bytes|b. + SHALL NOT slide TTL or expiry clocks, and SHALL NOT + change session ids, CapsPolicy ACL, or live state. + A checkpoint SHALL NOT retire a manifest and SHALL + NOT load a snapshot into the live registry. + checkpoint-verify loads every snapshot in the given + directory into a throwaway in-process store and + checks row counts, edge counts, and checksums against + the manifest. It prints + @STAT: checkpoint_verify|ok|n|failed|m. + A corrupt or missing snapshot file SHALL fail that + check. The weekly restore test runs this helper on a + restic-restored copy. No real session id in docs or + tests. + */ + attribute requirementId : String = "MN-REQ-06.15"; + } } // ----- MCP agent boundary ----- @@ -2222,6 +2275,8 @@ package MemNetRequirements { : MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_13_ServeRequestIsolation; requirement safeServeUpgradeReq : MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_14_SafeServeUpgrade; + requirement serveCheckpointReq + : MN_REQ_00_MissionBridge::MN_REQ_06_ProcessBoundary::MN_REQ_06_15_ServeCheckpoint; requirement mcpAgentBoundaryReq : MN_REQ_00_MissionBridge::MN_REQ_07_McpAgentBoundary; diff --git a/sysml-models/models/verify.sysml b/sysml-models/models/verify.sysml index fb110f75..3f2a7d18 100644 --- a/sysml-models/models/verify.sysml +++ b/sysml-models/models/verify.sysml @@ -54,6 +54,10 @@ TcpServeBridge; admin drain; manifest; startup restore; retire after a clean restore; no client retry; no silent session loss; no SemVer bump. + Serve checkpoint (MN-VER-06-S13): ServeCheckpoint on + TcpServeBridge; admin credential; per-session lock; lossless + atomic snaps; skip unsaveable; no drain; no live load; no + SemVer bump. Optional id nickname TARGET (MN-VER-02-S01): GraphElement identity; CREATE () legal; HiddenStoreHandle off-wire; 0.9 leftover_by_id store. TARGET command list cue/pattern only (MN-VER-07-S01): no required-id @@ -1685,6 +1689,50 @@ package MemNetVerification { } } + verification def MN_VER_06_S13_ServeCheckpoint { + doc /* + MN-REQ-06.15 — checkpoint every loaded session without + draining or refusing traffic. Per-session lock only. + Lossless atomic snapshot. Unsaveable sessions are named + and skipped. Exit is non-zero when any were skipped. + TTL, id, ACL, and live state stay put. A checkpoint does + not retire or load. checkpoint-verify checks a restored + directory. Not an agent MCP tool. No SemVer bump. + */ + attribute verificationId : String = "MN-VER-06-S13"; + + subject tcp : TcpServeBridge; + subject cli : CliFacade; + + objective serveCheckpoint { + verify serveCheckpointReq; + require constraint { + tcp.serveCheckpoint.implemented == true + and tcp.serveCheckpoint.noSemVerBump == true + and tcp.serveCheckpoint.onAgentMcp == false + and tcp.serveCheckpoint.adminCredEnv == "MEMNET_ADMIN_TOKEN" + and tcp.serveCheckpoint.refusesNewSessionOpen == false + and tcp.serveCheckpoint.drainsServe == false + and tcp.serveCheckpoint.sessionLockOnly == true + and tcp.serveCheckpoint.usesLosslessSnapshot == true + and tcp.serveCheckpoint.atomicWrite == true + and tcp.serveCheckpoint.skipsUnsaveable == true + and tcp.serveCheckpoint.unsaveableFatal == false + and tcp.serveCheckpoint.exitNonZeroWhenSkipped == true + and tcp.serveCheckpoint.removesStale == true + and tcp.serveCheckpoint.preservesSessionId == true + and tcp.serveCheckpoint.preservesAcl == true + and tcp.serveCheckpoint.preservesTtlClock == true + and tcp.serveCheckpoint.retiresOrLoads == false + and tcp.serveCheckpoint.statCheckpoint == "checkpoint" + and tcp.serveCheckpoint.statCheckpointVerify == "checkpoint_verify" + and cli.checkpointOnAgentMcp == false + and cli.checkpoint.onAgentMcp == false + and cli.checkpointVerify.onAgentMcp == false + } + } + } + verification def MN_VER_04_S06_HousekeepEndpointIdentity { doc /* MN-REQ-04.12 — housekeep resolves edge endpoints by hid and @@ -2252,6 +2300,8 @@ package MemNetVerification { : MN_VER_06_S11_ServeRequestIsolation; verification safeServeUpgradeVerify : MN_VER_06_S12_SafeServeUpgrade; + verification serveCheckpointVerify + : MN_VER_06_S13_ServeCheckpoint; verification housekeepEndpointIdentityVerify : MN_VER_04_S06_HousekeepEndpointIdentity; } diff --git a/sysml-models/outputs/README.md b/sysml-models/outputs/README.md index 6fdfe53c..0fc4714e 100644 --- a/sysml-models/outputs/README.md +++ b/sysml-models/outputs/README.md @@ -32,6 +32,7 @@ Keep product-canon and GQL application studies. Do not restore leftover `NEW` mi | [usage-dashboard-case-study.md](usage-dashboard-case-study.md) | Human look at serve n/max + housekeep; manage = Memnetor+Devicor | MN-REQ-06.5; MN-VER-06-S02; HTTP parked | | [admin-usage-report-case-study.md](admin-usage-report-case-study.md) | Admin-only serve usage JSON; opaque alias; not agent MCP | MN-REQ-06.11; MN-VER-06-S09 | | [safe-upgrade-case-study.md](safe-upgrade-case-study.md) | Drain, manifest, and startup restore so a serve swap does not drop sessions | MN-REQ-06.14; MN-VER-06-S12 | +| [checkpoint-case-study.md](checkpoint-case-study.md) | Checkpoint every loaded session without draining; hourly restic backup | MN-REQ-06.15; MN-VER-06-S13 | | [device-fleet-one-mcp-case-study.md](device-fleet-one-mcp-case-study.md) | Device MemNet services; one MemNet MCP (tip/ops) at the droplet; product face is sysmledge | MN-REQ-06.6; MN-VER-06-S03; tip≠face; not #47 | | [tip-memnet-access-portal-case-study.md](tip-memnet-access-portal-case-study.md) | Admin invite, Google login, Bearer for keyed tip MCP; unauthenticated WWW MemNet refused | MN-REQ-06.8; MN-VER-06-S05; tip≠face; portal sidecar | | [lan-mcp-front-case-study.md](lan-mcp-front-case-study.md) | One MCP catalogue over N LAN serves; ClusterRoute; registry owner; not shipped | MN-REQ-06.9; MN-VER-06-S07; tip≠face; inventOnly #191 | diff --git a/sysml-models/outputs/checkpoint-case-study.md b/sysml-models/outputs/checkpoint-case-study.md new file mode 100644 index 00000000..99e51b9a --- /dev/null +++ b/sysml-models/outputs/checkpoint-case-study.md @@ -0,0 +1,17 @@ +# Case study: serve checkpoint (MN-REQ-06.15) + +**Story:** An hourly off-host backup only sees what a product has already saved. Some products never call `session_save`, so a live session can be missing from the backup. The checkpoint copies every loaded session to disk while the serve keeps accepting work. + +**Requirement:** MN-REQ-06.15 (`serveCheckpointReq`). **Verify:** MN-VER-06-S13. + +| | | +|--|--| +| Part | `ServeCheckpoint` on `TcpServeBridge` | +| Commands | `CmdAdminCheckpoint` and `CmdAdminCheckpointVerify` on `CliFacade`; not on `McpFacade` | +| Snapshot | lossless writer, atomic temp file plus fsync plus rename | +| Directory | `$MEMNET_STATE_DIR/checkpoint/` plus `checkpoint-manifest.json` | +| Flag | `implemented=true`; `noSemVerBump=true`; `drainsServe=false`; `retiresOrLoads=false` | + +The admin credential is `MEMNET_ADMIN_TOKEN`. One session lock covers serialising that session only. Other sessions keep serving, and a new open is not refused. An unsaveable session is named (`snapshot_unsaveable` and the reason) and skipped; the exit code is non-zero. TTL, session id, ACL, and live state stay put. `checkpoint-verify` checks a restic-restored copy in a throwaway store and prints `@STAT: checkpoint_verify|ok|n|failed|m`. + +Teach: [`docs/operations/backup.md`](../../docs/operations/backup.md). diff --git a/sysml-models/outputs/product-nest-one-page.md b/sysml-models/outputs/product-nest-one-page.md index f13e79a5..cfddf7ac 100644 --- a/sysml-models/outputs/product-nest-one-page.md +++ b/sysml-models/outputs/product-nest-one-page.md @@ -27,6 +27,7 @@ ARCHIVE LOOK MemNetArchive (models/archive.sysml) — leftover_* / OPS LOOK MemNetUsageDashboard — human look only; not agent wire SERVE ADMIN LOOK AdminUsageReport on TcpServeBridge — admin JSON; opaque alias; not agent MCP (MN-REQ-06.11) SERVE UPGRADE SafeServeUpgrade on TcpServeBridge — drain, manifest, restore, retire (MN-REQ-06.14) +SERVE CHECKPOINT ServeCheckpoint on TcpServeBridge — loaded sessions to disk, no drain (MN-REQ-06.15) OPS FLEET MemNetOpsFleet — device MemNet services; one MemNet MCP at droplet (tip/ops; tip≠face; product face is sysmledge) OPS ACCESS TipMemNetAccessPortal — Szu-Wei invite, Google login, Bearer for keyed tip MCP at droplet WWW (tip≠face; not sysmledge; portal sidecar; look-only service/client status) OPS LAN FRONT MemNetLanMcpFront — ClusterRoute: one MCP catalogue over N LAN serves (inventOnly #191; tip≠face; not shipped) diff --git a/sysml-models/outputs/ssot-to-code-allocate-map.md b/sysml-models/outputs/ssot-to-code-allocate-map.md index 6150d781..6ca94aee 100644 --- a/sysml-models/outputs/ssot-to-code-allocate-map.md +++ b/sysml-models/outputs/ssot-to-code-allocate-map.md @@ -20,6 +20,7 @@ | tcpDaemonToMod | `…transport.tcp` | `serveDaemonMod` | `parts/common/memnet/memnet/serve.py` | `memnet.serve` | true | | adminUsageToMod | `…transport.tcp.adminUsage` | `adminUsageMod` | `parts/common/memnet/memnet/admin_usage.py` | `memnet.admin_usage` | true | | safeUpgradeToMod | `…transport.tcp.safeUpgrade` | `safeUpgradeMod` | `parts/common/memnet/memnet/upgrade.py` | `memnet.upgrade` | true | +| serveCheckpointToMod | `…transport.tcp.serveCheckpoint` | `serveCheckpointMod` | `parts/common/memnet/memnet/checkpoint.py` | `memnet.checkpoint` | true | | cliToMod | `memNetSystem.core.cli` | `cliMod` | `parts/common/memnet/memnet/cli.py` | `memnet.cli` | true | | sessionsToMod | `…sessions` | `sessionsMod` | `parts/common/memnet/memnet/session_lifecycle.py` | `memnet.session_lifecycle` | true | | storeToMod | `…sessions.store` | `storeMod` | `parts/common/memnet/memnet/graph_store.py` | `memnet.graph_store` | true | diff --git a/tests/test_checkpoint.py b/tests/test_checkpoint.py new file mode 100644 index 00000000..b0931957 --- /dev/null +++ b/tests/test_checkpoint.py @@ -0,0 +1,316 @@ +"""Admin checkpoint of loaded sessions without draining (MN-REQ-06.15).""" + +from __future__ import annotations + +import json +import threading +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from memnet.checkpoint import ( + MANIFEST_NAME, + atomic_write_bytes, + run_checkpoint, + verify_checkpoint, +) +from memnet.cli import app +from memnet.in_process_engine import run_argv +from memnet.registry import count, get_entry +from memnet.serve import _handle_request +from memnet.session import close_session, open_session +from memnet.snapshot import SNAPSHOT_MAGIC, materialise_snapshot_text +from memnet.upgrade import drain_gate + +TOKEN = "test-admin-token" +runner = CliRunner() + + +def _mutate(sid: str, text: str) -> None: + raw = run_argv(["mutate", "--stdin", "--session", sid], stdin=text) + assert raw["exit_code"] == 0, raw["stderr"] + + +def _edge_poison(sid: str) -> None: + entry = get_entry(sid) + assert entry is not None + for hid in entry.store.write_order: + rec = entry.store._by_hid.get(hid) + if rec is not None and rec.tag == "EDG": + rec.fields["not_a_field"] = "x" + return + raise AssertionError("expected an edge to poison") + + +def _edge_endpoints_resolve(entry) -> None: + for hid in entry.store.write_order: + rec = entry.store._by_hid.get(hid) + if rec is None or rec.tag != "EDG": + continue + assert rec.fields.get("src") in entry.store._by_hid + assert rec.fields.get("dist") in entry.store._by_hid + + +def test_checkpoint_during_other_session_mutates(memnet_temp, monkeypatch, tmp_path, schema_file): + monkeypatch.setenv("MEMNET_ADMIN_TOKEN", TOKEN) + held = open_session(map_file=str(schema_file), ttl_minutes=2) + other = open_session(map_file=str(schema_file), ttl_minutes=2) + _mutate(held.session_id, "CREATE (:TSK {id: 'TSK_held', goal: 'quiet', status: 'open'})\n") + _mutate( + other.session_id, + "CREATE (:TSK {id: 'TSK_a', goal: 'kept', status: 'open'})\n" + "CREATE (:TSK {id: 'TSK_b', goal: 'kept', status: 'open'})\n" + "MATCH (a {id: 'TSK_a'}), (b {id: 'TSK_b'}) " + "CREATE (a)-[:owns {id: 'E_keep'}]->(b)\n", + ) + held.enable_acl() + held.grant_caller("caller-demo", can_pin_map=True, can_mutate=False) + held_entry = get_entry(held.session_id) + assert held_entry is not None + expires = held_entry.meta.expires_at + created = held_entry.meta.created_at + before_rows = get_entry(other.session_id).store.row_count_non_law() + release = threading.Event() + holding = threading.Event() + + def _hold() -> None: + with held_entry.lock: + holding.set() + assert release.wait(timeout=5) + + holder = threading.Thread(target=_hold) + holder.start() + assert holding.wait(timeout=2) + box: dict = {} + + def _run() -> None: + box["result"] = run_checkpoint(TOKEN, directory=tmp_path) + + worker = threading.Thread(target=_run) + worker.start() + assert worker.is_alive() + _mutate(other.session_id, "CREATE (:TSK {id: 'TSK_during', goal: 'live', status: 'open'})\n") + other_expires = get_entry(other.session_id).meta.expires_at + assert worker.is_alive() + fresh = open_session(map_file=str(schema_file)) + assert fresh.session_id + assert worker.is_alive() + counted, refuse = drain_gate.begin(["session", "open"]) + assert refuse is None + drain_gate.end(counted) + assert drain_gate.phase == "open" + release.set() + worker.join(timeout=5) + holder.join(timeout=2) + assert not worker.is_alive() + result = box["result"] + assert result.exit_code == 0 + assert result.skipped == [] + assert result.stat == f"@STAT: checkpoint|ok|{result.saved}|skipped|0|bytes|{result.nbytes}" + assert result.saved >= 2 + back = get_entry(held.session_id) + assert back is not None + assert back.meta.session_id == held.session_id + assert back.meta.expires_at == expires + assert back.meta.created_at == created + assert back.acl is not None and back.acl.enabled is True + assert back.acl.callers["caller-demo"].can_mutate is False + assert get_entry(other.session_id).meta.expires_at == other_expires + manifest = json.loads((tmp_path / "checkpoint" / MANIFEST_NAME).read_text(encoding="utf-8")) + assert manifest["manifest_format"] == 1 + assert manifest["snapshot_format"] == 1 + assert manifest["serve_version"] + assert manifest["timestamp"].endswith("Z") + assert manifest["unsaved"] == [] + other_row = next(row for row in manifest["sessions"] if row["session_id"] == other.session_id) + snap = materialise_snapshot_text( + (tmp_path / "checkpoint" / other_row["file"]).read_text(encoding="utf-8") + ) + assert snap.store.row_count_non_law() == other_row["rows"] + assert other_row["rows"] >= before_rows + _edge_endpoints_resolve(snap) + held_row = next(row for row in manifest["sessions"] if row["session_id"] == held.session_id) + held_snap = materialise_snapshot_text( + (tmp_path / "checkpoint" / held_row["file"]).read_text(encoding="utf-8") + ) + assert held_snap.store.row_count_non_law() == held_entry.store.row_count_non_law() + assert count() >= 3 + + +def test_unsaveable_session_is_skipped(memnet_temp, monkeypatch, tmp_path, schema_file): + monkeypatch.setenv("MEMNET_ADMIN_TOKEN", TOKEN) + good = open_session(map_file=str(schema_file)) + bad = open_session(map_file=str(schema_file)) + _mutate(good.session_id, "CREATE (:TSK {id: 'TSK_good', goal: 'kept', status: 'open'})\n") + _mutate( + bad.session_id, + "CREATE (:TSK {id: 'TSK_a', goal: 'a', status: 'open'})\n" + "CREATE (:TSK {id: 'TSK_b', goal: 'b', status: 'open'})\n" + "MATCH (a {id: 'TSK_a'}), (b {id: 'TSK_b'}) " + "CREATE (a)-[:owns {id: 'E_bad'}]->(b)\n", + ) + first = run_checkpoint(TOKEN, directory=tmp_path) + assert first.exit_code == 0 + assert first.saved == 2 + manifest_text = (tmp_path / "checkpoint" / MANIFEST_NAME).read_text(encoding="utf-8") + first_manifest = json.loads(manifest_text) + bad_name = next( + row["file"] for row in first_manifest["sessions"] if row["session_id"] == bad.session_id + ) + assert (tmp_path / "checkpoint" / bad_name).is_file() + expires = get_entry(bad.session_id).meta.expires_at + _edge_poison(bad.session_id) + result = run_checkpoint(TOKEN, directory=tmp_path) + assert result.exit_code == 2 + assert result.saved == 1 + assert len(result.skipped) == 1 + assert result.skipped[0]["code"] == "snapshot_unsaveable" + assert result.skipped[0]["session_id"] == bad.session_id + assert result.skipped[0]["detail"] + assert result.stat.startswith("@STAT: checkpoint|ok|1|skipped|1|bytes|") + assert not (tmp_path / "checkpoint" / bad_name).exists() + manifest = json.loads((tmp_path / "checkpoint" / MANIFEST_NAME).read_text(encoding="utf-8")) + saved_ids = {row["session_id"] for row in manifest["sessions"]} + assert good.session_id in saved_ids + assert bad.session_id not in saved_ids + assert manifest["unsaved"][0]["code"] == "snapshot_unsaveable" + assert get_entry(bad.session_id) is not None + assert get_entry(bad.session_id).meta.expires_at == expires + assert get_entry(good.session_id).meta.expires_at + + +def test_atomic_write_never_leaves_a_partial_file(tmp_path, monkeypatch): + dest = tmp_path / "session.snap" + original = (SNAPSHOT_MAGIC + "\nCOMPLETE-OLD\n").encode("utf-8") + dest.write_bytes(original) + + def _boom(_fd: int) -> None: + raise OSError("fsync failed") + + monkeypatch.setattr("memnet.checkpoint._fsync_fd", _boom) + with pytest.raises(OSError): + atomic_write_bytes(dest, b"PARTIAL-NEW-BYTES") + assert dest.read_bytes() == original + assert b"PARTIAL" not in dest.read_bytes() + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_stale_checkpoint_removed_after_close(memnet_temp, monkeypatch, tmp_path, schema_file): + monkeypatch.setenv("MEMNET_ADMIN_TOKEN", TOKEN) + keep = open_session(map_file=str(schema_file)) + gone = open_session(map_file=str(schema_file)) + _mutate(keep.session_id, "CREATE (:TSK {id: 'TSK_keep', goal: 'stay', status: 'open'})\n") + _mutate(gone.session_id, "CREATE (:TSK {id: 'TSK_gone', goal: 'drop', status: 'open'})\n") + first = run_checkpoint(TOKEN, directory=tmp_path) + assert first.saved == 2 + dest = tmp_path / "checkpoint" + manifest = json.loads((dest / MANIFEST_NAME).read_text(encoding="utf-8")) + gone_name = next( + row["file"] for row in manifest["sessions"] if row["session_id"] == gone.session_id + ) + assert (dest / gone_name).is_file() + close_session(gone.session_id) + second = run_checkpoint(TOKEN, directory=tmp_path) + assert second.exit_code == 0 + assert second.saved == 1 + assert not (dest / gone_name).exists() + again = json.loads((dest / MANIFEST_NAME).read_text(encoding="utf-8")) + ids = {row["session_id"] for row in again["sessions"]} + assert ids == {keep.session_id} + assert list(dest.glob("*.snap")) == [dest / again["sessions"][0]["file"]] + + +def test_checkpoint_verify_good_corrupt_and_missing( + memnet_temp, monkeypatch, tmp_path, schema_file +): + monkeypatch.setenv("MEMNET_ADMIN_TOKEN", TOKEN) + ss = open_session(map_file=str(schema_file)) + _mutate(ss.session_id, "CREATE (:TSK {id: 'TSK_v', goal: 'verify', status: 'open'})\n") + result = run_checkpoint(TOKEN, directory=tmp_path) + assert result.exit_code == 0 + dest = tmp_path / "checkpoint" + live = count() + good = verify_checkpoint(dest, TOKEN) + assert good.exit_code == 0 + assert good.ok == 1 + assert good.failed == [] + assert good.stat == "@STAT: checkpoint_verify|ok|1|failed|0" + assert count() == live + snap = next(dest.glob("*.snap")) + original = snap.read_bytes() + snap.write_bytes(original[:40] + b"XXXX" + original[44:]) + corrupt = verify_checkpoint(dest, TOKEN) + assert corrupt.exit_code != 0 + assert corrupt.ok == 0 + assert corrupt.failed + assert corrupt.failed[0]["code"] == "checkpoint_checksum" + assert corrupt.stat == "@STAT: checkpoint_verify|ok|0|failed|1" + assert snap.read_bytes() != original + assert count() == live + snap.write_bytes(original) + snap.unlink() + missing = verify_checkpoint(dest, TOKEN) + assert missing.exit_code != 0 + assert any(row["code"] == "checkpoint_missing" for row in missing.failed) + assert missing.stat.startswith("@STAT: checkpoint_verify|ok|0|failed|") + assert count() == live + + +def test_admin_token_required(memnet_temp, monkeypatch, tmp_path, schema_file): + ss = open_session(map_file=str(schema_file)) + _mutate(ss.session_id, "CREATE (:TSK {id: 'TSK_tok', goal: 'gate', status: 'open'})\n") + monkeypatch.delenv("MEMNET_ADMIN_TOKEN", raising=False) + denied = _handle_request({"checkpoint": True, "admin_token": "anything"}) + assert denied["exit_code"] == 2 + assert "admin_unconfigured" in denied["stderr"] + assert denied["stdout"] == "" + monkeypatch.setenv("MEMNET_ADMIN_TOKEN", TOKEN) + monkeypatch.setenv("MEMNET_STATE_DIR", str(tmp_path / "wire")) + wrong = _handle_request({"checkpoint": True, "admin_token": "wrong"}) + assert wrong["exit_code"] == 2 + assert "admin_denied" in wrong["stderr"] + missing = _handle_request({"checkpoint": True}) + assert missing["exit_code"] == 2 + assert "admin_denied" in missing["stderr"] + cli_missing = runner.invoke(app, ["admin", "checkpoint", "--state-dir", str(tmp_path)]) + assert cli_missing.exit_code == 2 + assert "admin_denied" in cli_missing.stderr + verify_missing = runner.invoke(app, ["admin", "checkpoint-verify", str(tmp_path)]) + assert verify_missing.exit_code == 2 + assert "admin_denied" in verify_missing.stderr + ok = _handle_request({"checkpoint": True, "admin_token": TOKEN}) + assert ok["exit_code"] == 0, ok + assert ok["stdout"].startswith("@STAT: checkpoint|ok|") + assert "skipped|0|" in ok["stdout"] + cli = runner.invoke( + app, + ["admin", "checkpoint", "--token", TOKEN, "--state-dir", str(tmp_path / "cli")], + ) + assert cli.exit_code == 0, cli.stderr + assert cli.stdout.startswith("@STAT: checkpoint|ok|") + checked = runner.invoke( + app, + [ + "admin", + "checkpoint-verify", + "--token", + TOKEN, + str(tmp_path / "cli" / "checkpoint"), + ], + ) + assert checked.exit_code == 0, checked.stderr + assert checked.stdout.startswith("@STAT: checkpoint_verify|ok|") + + +def test_backup_doc_has_no_session_id_or_secret(): + text = (Path(__file__).resolve().parents[1] / "docs/operations/backup.md").read_text( + encoding="utf-8" + ) + assert "mn_" not in text + assert "memnet admin checkpoint" in text + assert "checkpoint-verify" in text + assert "MEMNET_ADMIN_TOKEN" in text + assert "restic" in text + assert "R2" in text + assert "test-admin-token" not in text diff --git a/tests/test_sysml_checkpoint.py b/tests/test_sysml_checkpoint.py new file mode 100644 index 00000000..55b76b5d --- /dev/null +++ b/tests/test_sysml_checkpoint.py @@ -0,0 +1,66 @@ +"""Honesty-c: MN-REQ-06.15 serve checkpoint is modelled before the code.""" + +from __future__ import annotations + +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +MODELS = ROOT / "sysml-models" / "models" +DEPLOY = MODELS / "deploy.sysml" +REQUIREMENTS = MODELS / "requirements.sysml" +VERIFY = MODELS / "verify.sysml" +IMPLEMENTATION = MODELS / "implementation.sysml" +STUDY = ROOT / "sysml-models" / "outputs" / "checkpoint-case-study.md" +NEST = ROOT / "sysml-models" / "outputs" / "product-nest-one-page.md" +TEACH = ROOT / "docs" / "operations" / "backup.md" +MAP = ROOT / "sysml-models" / "outputs" / "ssot-to-code-allocate-map.md" + + +def test_checkpoint_parts(): + text = DEPLOY.read_text(encoding="utf-8") + assert "part def ServeCheckpoint" in text + assert "part def CmdAdminCheckpoint" in text + assert "part def CmdAdminCheckpointVerify" in text + assert "part serveCheckpoint : ServeCheckpoint" in text + assert "attribute drainsServe : Boolean = false" in text + assert "attribute retiresOrLoads : Boolean = false" in text + assert "attribute refusesNewSessionOpen : Boolean = false" in text + mcp = text.split("part def McpFacade", 1)[1].split("part def ", 1)[0] + assert "part checkpoint" not in mcp + gateway = text.split("part def MemNetProductGateway", 1)[1].split("part def ", 1)[0] + assert "MN_REQ_06_15_ServeCheckpoint" not in gateway + + +def test_requirement_verify_allocate(): + req = REQUIREMENTS.read_text(encoding="utf-8") + assert "requirement def MN_REQ_06_15_ServeCheckpoint" in req + assert 'attribute requirementId : String = "MN-REQ-06.15"' in req + assert "serveCheckpointReq" in req + assert "checkpoint|ok|n|skipped|m|bytes|b" in req + assert "snapshot_unsaveable" in req + ver = VERIFY.read_text(encoding="utf-8") + assert "verification def MN_VER_06_S13_ServeCheckpoint" in ver + assert "verify serveCheckpointReq" in ver + assert "tcp.serveCheckpoint.drainsServe == false" in ver + assert "tcp.serveCheckpoint.retiresOrLoads == false" in ver + impl = IMPLEMENTATION.read_text(encoding="utf-8") + assert "part def ServeCheckpointMod" in impl + assert "allocation serveCheckpointToMod" in impl + assert "memnet.checkpoint" in impl + ledger = MAP.read_text(encoding="utf-8") + assert "| serveCheckpointToMod |" in ledger + + +def test_outputs_and_docs(): + study = STUDY.read_text(encoding="utf-8") + assert "MN-REQ-06.15" in study + assert "MN-VER-06-S13" in study + nest = NEST.read_text(encoding="utf-8") + assert "ServeCheckpoint" in nest + teach = TEACH.read_text(encoding="utf-8") + assert "memnet admin checkpoint" in teach + assert "checkpoint-verify" in teach + assert "MEMNET_ADMIN_TOKEN" in teach + assert "restic" in teach + assert "mn_" not in teach + assert "mn_" not in study