Add a drain-and-restore path for a live serve upgrade (MN-REQ-06.14)

docs/README.md

@@ -68,6 +68,7 @@ Multitask MUST for this product. Index: [`operations/README.md`](operations/READ
 | [`operations/product-gateway-contract.md`](operations/product-gateway-contract.md) | Product gateway on memnet-mcp (MN-REQ-06.12): route by product, house, or session |
 | [`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 and restore a live serve without dropping sessions (MN-REQ-06.14) |
 | [`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).

docs/operations/README.md

@@ -12,6 +12,7 @@ Agent operating doctrine for this product (not domain recipes).
 | [`product-gateway-contract.md`](product-gateway-contract.md) | Product gateway on memnet-mcp (MN-REQ-06.12): per-product route to one owning serve |
 | [`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, restore, and `memnet-upgrade` so a serve swap keeps sessions (MN-REQ-06.14) |
 | [`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).

docs/operations/safe-upgrade.md

@@ -0,0 +1,112 @@
+# Safe serve upgrade
+
+Upgrade `memnet-serve` without silently dropping a loaded session (MN-REQ-06.14). The agent loop does not change: cue, then `pin_map`, then `mutate`. This is a patch-level cut on 0.19. Do not bump the package version from this procedure.
+
+The manual procedure this replaces is a side-by-side virtualenv: install the new version, save every session, stop the old serve, start the new serve on the same port, reload, and check counts. Anything that failed to save was easy to miss. The drain below names every session it could not snapshot and refuses a ready-to-stop result unless you pass an explicit override.
+
+## Order
+
+Deploy client tolerance **before** the serve swap.
+
+1. Install the new `memnet-mcp` (and the droplet gateway, if this host is the gateway) so `serve_draining` and a brief connection refusal are retried. Leave the gateway's `pinned_version` on the version that is still running.
+2. Confirm `MEMNET_UPGRADE_RETRY_S` is unset or about `30` on those client processes. `0` disables retry.
+3. Only then run `memnet-upgrade` against the serve.
+
+During the gap, new `session_open` calls receive `@ERR: serve_draining|retry_after_s=<seconds>` and clients back off. After the old process has exited and before the new one listens, connections are refused. That refusal is retried for the same window. A command the serve already accepted is not retried on timeout.
+
+Downtime that remains: a few seconds while the port is closed, covered by that retry. Sessions keep their ids, CapsPolicy ACL bindings, TTL expiry clock, and house (the session tag map).
+
+## What the serve does
+
+Admin credential: `MEMNET_ADMIN_TOKEN` (same rule as the usage report). Unset is `@ERR: admin_unconfigured`. A mismatch is `@ERR: admin_denied`. This command is not an agent MCP tool.
+
+```bash
+memnet admin upgrade-prepare --state-dir "$MEMNET_STATE_DIR"
+```
+
+Envelope form: `{"upgrade_prepare": true, "admin_token": "<token>"}`.
+
+Drain behaviour:
+
+- Refuse new `session_open` with `serve_draining`.
+- Finish commands already in flight, then refuse further commands.
+- Snapshot every loaded session with the lossless snapshot writer.
+- Write `upgrade-manifest.json` and `upgrade-snapshots/` under `MEMNET_STATE_DIR` (default `~/.local/state/memnet`). The manifest records session ids, row and edge counts, sha256 checksums, the serve version, and snapshot format `1`.
+- An ACL and clock passport is an extra `# upgrade-passport` line. Older v1 loaders skip `#` lines, so a rollback can still read the graph.
+- If any session raises `snapshot_unsaveable` (or any other save error), the command exits non-zero, lists that session in `upgrade-blocked.json`, and does **not** write a ready manifest. The serve keeps accepting work. Pass `--allow-unsaved` only when you accept dropping those named sessions.
+- Do not stop the process unless stderr contains `@STAT: upgrade_prepare|ready|`.
+
+The new process reads the manifest on startup, reloads each snapshot, and checks counts and checksums. It prints:
+
+```text
+@STAT: upgrade_restore|ok|<n>|failed|<m>
+```
+
+Snapshot files stay on disk until `memnet admin upgrade-retire` after a clean restore report. A format this process does not support, or a checksum or parse failure, exits `3` and does not delete or rewrite the files. Restarting before retire replays the upgrade snapshots (writes made after the restore are not in those files). Retire as soon as the stat line is clean.
+
+## Helper
+
+`memnet-upgrade` encodes the side-by-side venv procedure. It refuses to start unless `--clients-ready` is set.
+
+```bash
+memnet-upgrade \
+  --clients-ready \
+  --new-python /opt/memnet/venv-new/bin/python \
+  --new-exec "/opt/memnet/venv-new/bin/memnet serve --host 127.0.0.1 --port 18765" \
+  --state-dir /var/lib/memnet \
+  --unit /etc/systemd/system/memnet-serve.service \
+  --restart-unit memnet-serve
+```
+
+Steps:
+
+1. Preflight: import `memnet` with the new interpreter and check `SUPPORTED_SNAPSHOT_FORMATS` against the current manifest (or format `1` when no manifest exists yet).
+2. `admin upgrade-prepare` on the running serve.
+3. Copy the unit to `memnet-serve.service.bak` and replace `ExecStart=`.
+4. If `--gateway-config` and `--product` are set, copy the config to `*.bak` and set that product's `pinned_version` to the new version **after** the old serve is drained and **before** the new process listens.
+5. Restart the unit. The new process restores and writes `upgrade-restore.json`.
+6. If `failed` is not `0`, write the unit backup and the config backup back and restart again.
+
+`--state-dir` must be the same directory as the unit's `Environment=MEMNET_STATE_DIR`. The serve reads that variable at startup; the prepare command also receives `--state-dir`.
+
+Build the new venv beside the old one. Do not overwrite it until the restore has verified.
+
+```bash
+python -m venv /opt/memnet/venv-new
+/opt/memnet/venv-new/bin/pip install "memnet-llm==<new>"
+```
+
+Use the index you trust. Do not put tokens or session ids in the unit file. Keep `MEMNET_ADMIN_TOKEN` in a root-only `EnvironmentFile`.
+
+## rpi5-syson
+
+Local engine on the Pi:
+
+| Process | Port | Unit (adjust to the host) |
+|---------|------|---------------------------|
+| `memnet serve` | `18765` | `memnet-serve.service` |
+| `memnet-mcp` streamable-http | `18766` | `memnet-mcp-http.service` |
+
+Upgrade the MCP unit first (retry code, same serve version pin). Then `memnet-upgrade` the serve unit on `18765` with `MEMNET_STATE_DIR` on the Pi disk. Leave `18766` up so clients retry against the serve gap.
+
+## Endleaf engine serve
+
+The Endleaf product backend is the serve named in the droplet gateway registry (host and port live in that config, not in this repo). Use that unit's port and state directory. The drain and restore are the same command. Do not point the gateway at a second backend to "move" live sessions.
+
+## Droplet gateway
+
+The gateway process is `memnet-mcp --transport gateway` with `MEMNET_GATEWAY_CONFIG`. Ship the retry-capable build first, with `pinned_version` still equal to the running engine. Then, in the same `memnet-upgrade` invocation that swaps the engine, pass:
+
+```bash
+--gateway-config /etc/memnet/gateway.json \
+--product endleaf \
+--restart-unit memnet-gateway
+```
+
+The pin changes only after drain succeeds, and it is restored from `gateway.json.bak` if the engine restore fails. Restart the gateway after the pin write so it stops caching the previous version (`version_cache_s`). No plaintext credentials in git. Hashes stay in the config file on the droplet.
+
+## Rollback
+
+A failed verification restores the previous `ExecStart=` and the previous pin, then restarts. Snapshot files are still in the state directory. The old venv loads format v1 snapshots (`#` passport lines are ignored). That reload keeps the session id and the graph. The old loader rebases the TTL clock from `ttl_minutes` and does not read ACL bindings from the passport; re-grant those on the old serve if you stay there. The new serve's restore path keeps the id, the ACL bindings, the expiry instant, and the house.
+
+Neighbourhood reserves are not part of the snapshot. Take a fresh reserve after the new serve is up.

parts/common/memnet/memnet/cli.py

@@ -528,6 +528,71 @@ def admin_usage_report(
         raise typer.Exit(code)
 
 
+@admin_app.command("upgrade-prepare")
+def admin_upgrade_prepare(
+    token: Annotated[
+        str | None,
+        typer.Option(
+            "--token",
+            help="Admin token presented by the caller (do not log). Not MEMNET_ADMIN_TOKEN.",
+        ),
+    ] = None,
+    allow_unsaved: Annotated[
+        bool,
+        typer.Option(
+            "--allow-unsaved",
+            help="Reach ready-to-stop even when a session could not be snapshotted.",
+        ),
+    ] = False,
+    state_dir: Annotated[
+        str | None,
+        typer.Option("--state-dir", help="Manifest directory (default MEMNET_STATE_DIR)."),
+    ] = None,
+) -> None:
+    """Snapshot every loaded session and write an upgrade manifest. Not agent MCP."""
+    from pathlib import Path
+
+    from memnet.admin_usage import caller_token
+    from memnet.upgrade import prepare_upgrade
+
+    presented = token if token else caller_token()
+    directory = Path(state_dir) if state_dir else None
+    try:
+        result = prepare_upgrade(presented, allow_unsaved=allow_unsaved, directory=directory)
+    except MemNetError as exc:
+        _handle_error(exc)
+        return
+    sys.stdout.write(result.stdout_json())
+    sys.stderr.write(result.stat + "\n")
+    if result.exit_code:
+        sys.stderr.write(f"@ERR: upgrade_blocked|unsaved|{len(result.unsaved)}\n")
+        raise typer.Exit(result.exit_code)
+
+
+@admin_app.command("upgrade-retire")
+def admin_upgrade_retire(
+    token: Annotated[
+        str | None,
+        typer.Option("--token", help="Admin token presented by the caller (do not log)."),
+    ] = None,
+    state_dir: Annotated[str | None, typer.Option("--state-dir")] = None,
+) -> None:
+    """Stop replaying an upgrade manifest after a verified restore. Keeps the files."""
+    from pathlib import Path
+
+    from memnet.admin_usage import authenticate, caller_token
+    from memnet.upgrade import retire_manifest
+
+    presented = token if token else caller_token()
+    try:
+        authenticate(presented)
+        retire_manifest(Path(state_dir) if state_dir else None)
+    except MemNetError as exc:
+        _handle_error(exc)
+    emit_stdout('{"ok": true, "retired": true}')
+    emit_stderr("@STAT: upgrade_retire|1|-")
+
+
 @session_app.command("expire-status")
 def session_expire_status() -> None:
     """Booleans for expire-save (serve_status). Path redacted; no sids."""

parts/common/memnet/memnet/local_ipc_gateway.py

@@ -78,6 +78,9 @@ def run_ipc_serve(path: str | None = None) -> None:
     sock_path = resolve_ipc_path(path)
     os.environ["MEMNET_IPC_SOCKET"] = sock_path
     os.environ["MEMNET_SERVE_INTERNAL"] = "1"
+    from memnet.upgrade import startup_restore_or_exit
+
+    startup_restore_or_exit()
     try:
         from memnet.cheap_llm_import_guard import maybe_install_cheap_llm_import_guard
 

parts/common/memnet/memnet/serve.py

@@ -88,6 +88,14 @@ def _handle_request(payload: dict[str, Any]) -> dict[str, Any]:
     token_s = admin_token if isinstance(admin_token, str) else None
     if payload.get("admin_usage") is True:
         return usage_report_envelope(token_s)
+    if payload.get("upgrade_prepare") is True:
+        from memnet.upgrade import upgrade_prepare_envelope
+
+        return upgrade_prepare_envelope(token_s, allow_unsaved=bool(payload.get("allow_unsaved")))
+    if payload.get("upgrade_retire") is True:
+        from memnet.upgrade import upgrade_retire_envelope
+
+        return upgrade_retire_envelope(token_s)
 
     argv = payload.get("args", [])
     if not isinstance(argv, list):
@@ -99,26 +107,39 @@ def _handle_request(payload: dict[str, Any]) -> dict[str, Any]:
             "stdout": "",
             "stderr": "@ERR: bad_request|stdin must be a string\n",
         }
+    from memnet.upgrade import drain_gate, draining_envelope
+
+    counted, refuse = drain_gate.begin(argv)
+    if refuse:
+        return draining_envelope()
     os.environ["MEMNET_SERVE_INTERNAL"] = "1"
     from memnet.cli import app
     from memnet.output import capture_request_stdio
 
     code = 0
+    stdout = ""
+    stderr = ""
     token_ctx = set_caller_token(token_s)
-    with capture_request_stdio(stdin_text if isinstance(stdin_text, str) else None) as (out, err):
-        try:
-            result = app(argv, prog_name="memnet", standalone_mode=False)
-            if isinstance(result, int) and result != 0:
-                code = result
-        except SystemExit as exc:
-            code = int(exc.code) if isinstance(exc.code, int) else 1
-        except Exception as exc:
-            code = 1
-            err.write(f"@ERR: internal|{type(exc).__name__}: {exc}\n")
-        finally:
-            reset_caller_token(token_ctx)
-        stdout = out.getvalue()
-        stderr = err.getvalue()
+    try:
+        with capture_request_stdio(stdin_text if isinstance(stdin_text, str) else None) as (
+            out,
+            err,
+        ):
+            try:
+                result = app(argv, prog_name="memnet", standalone_mode=False)
+                if isinstance(result, int) and result != 0:
+                    code = result
+            except SystemExit as exc:
+                code = int(exc.code) if isinstance(exc.code, int) else 1
+            except Exception as exc:
+                code = 1
+                err.write(f"@ERR: internal|{type(exc).__name__}: {exc}\n")
+            finally:
+                reset_caller_token(token_ctx)
+            stdout = out.getvalue()
+            stderr = err.getvalue()
+    finally:
+        drain_gate.end(counted)
     return {"exit_code": code, "stdout": stdout, "stderr": stderr}
 
 
@@ -232,6 +253,9 @@ def run_serve(host: str | None = None, port: int | None = None) -> None:
     port = port or serve_port()
     validate_serve_bind_host(host)
     os.environ["MEMNET_SERVE_INTERNAL"] = "1"
+    from memnet.upgrade import startup_restore_or_exit
+
+    startup_restore_or_exit()
     try:
         from memnet.admin_usage import mark_serve_start
 

parts/common/memnet/memnet/session.py

@@ -318,6 +318,9 @@ def open_session(
     caps: Caps | None = None,
     product: str | None = None,
 ) -> SessionStore:
+    from memnet.upgrade import refuse_new_session
+
+    refuse_new_session()
     caps = caps or Caps()
     purge_expired(caps)
     if count_sessions(caps) >= caps.max_sessions:

parts/common/memnet/memnet/snapshot.py

@@ -336,6 +336,7 @@ def load_snapshot(
     ttl_minutes: int | None = None,
     keep_id: bool = False,
     hide_path: bool = False,
+    preserve_clocks: bool = False,
 ) -> SessionStore:
     caps = caps or Caps()
     purge_expired(caps)
@@ -371,6 +372,7 @@ def load_snapshot(
         caps=caps,
         ttl_minutes=ttl_minutes,
         keep_id=keep_id,
+        preserve_clocks=preserve_clocks,
     )
 
 
@@ -380,6 +382,7 @@ def load_snapshot_text(
     caps: Caps | None = None,
     ttl_minutes: int | None = None,
     keep_id: bool = False,
+    preserve_clocks: bool = False,
 ) -> SessionStore:
     caps = caps or Caps()
     meta, map_lines, rel_lines, rec_lines = _parse_sections(split_snapshot_lines(text))
@@ -395,11 +398,16 @@ def load_snapshot_text(
     ttl = ttl_minutes if ttl_minutes is not None else meta.ttl_minutes
     if ttl < 1 or ttl > 1440:
         raise MemNetError("bad_ttl", "ttl must be 1..1440")
-    expires = now + timedelta(minutes=ttl)
+    if preserve_clocks:
+        created_at = meta.created_at
+        expires_at = meta.expires_at
+    else:
+        created_at = now.isoformat().replace("+00:00", "Z")
+        expires_at = (now + timedelta(minutes=ttl)).isoformat().replace("+00:00", "Z")
     new_meta = SessionMeta(
         session_id=session_id,
-        created_at=now.isoformat().replace("+00:00", "Z"),
-        expires_at=expires.isoformat().replace("+00:00", "Z"),
+        created_at=created_at,
+        expires_at=expires_at,
         ttl_minutes=ttl,
         has_writes=meta.has_writes,
         modified_at=meta.modified_at,