phase: 187_sync_peak_memory_line

**Phase 187 — final verification pass: all criteria met, repo 100% green.**

Verified (all four tasks already in `complete/`)
- `app/core/memory.py`: `rss_mb` / `rss_peak_mb` / `cgroup_peak_mb` / `fmt_mb`, pure stdlib, never-raise, **100%** coverage, 24 branch tests.
- Lifecycle lines append-only: `sync: started … rss_mb=`, all three terminals `… rss_mb= rss_peak_mb=` (failed = sanitized one-liner beside the surviving `logger.exception`), `import: summary … rss_mb=`; prefixes byte-identical.
- `GET /api/sync/status`: additive `rss_mb`/`rss_peak_mb` (`int|null`), keyed off the ONE `_sample_rss` (run start, phase transitions, forced terminals, 5 s-throttled hook), `peak >= current` invariant, anonymous 403.
- `sources.js` `fmtSyncPeakClause`: ` · peak <N> MB` trailing the counts line / "Sync cancelled", nothing when null, no new control/live region.

Gates (exact commands)
- `uv run pytest --cov=app --cov-report=term-missing` → **4570 passed**, TOTAL **99%**, `app/api/sync.py` **100%**, `app/core/memory.py` **100%**
- `uv run ruff check .` → clean · `uv run pyright` → **0 errors**
- `uv run pytest tests/e2e/test_sync_peak_memory_visible.py -v --no-cov` → **5 passed in isolation** (ran twice)
- All 27 touched E2E families re-run **in isolation**, all green (progress_box 8, cancel 3, status_freshness 4, summary_visibility 3, sync_button 3, session_reset 4, git_freeze 2, git_log_stream_dates 5, PDF/image suites, …)

Completion criteria: every checkbox ✅ except the commit + phase-dir move, which belong to the harness.

Documentation
- README reviewed in full (pages, commands, config tables cross-checked against `app/config.py` — 85 env rows, all defaults match; phase-187 ops section, greps, process-vs-cgroup caveat): consistent, no changes needed.
- `AGENTS.md`: **fixed one pre-existing inconsistency** — rule 6 claimed the no-CDN test covers "the index page"; it actually parametrizes 10 pages (all but `/doc-edit.html`).

Notable
- Two failures early in the pass were self-inflicted by running the E2E module concurrently with the full suite on the shared Postgres (TRUNCATE deadlock + one order-dependent `test_overview` flake); both vanish under the documented serial isolation — no code defect, no change made.
- **PLAN §9 still needs the owner's reconciliation** (phase 187 adds the sync memory fields + two status keys; PLAN is never agent-edited — AGENTS rule 10 carries the spec). `TODO.md` holds an unrelated owner note, left untouched.

Next pending phase: `188_chunks_hnsw_index`.

tests/unit/test_core_memory.py

@@ -0,0 +1,285 @@
+"""Unit tests: ``app.core.memory`` — process RSS helpers (phase 187, task 01).
+
+Every branch of the four helpers, with the filesystem stubbed through the
+module's path constants (``PROC_STATUS_PATH`` / ``CGROUP_PEAK_PATH`` →
+``tmp_path`` files, so ``open`` is never mocked globally) and ``resource``
+stubbed through ``memory._resource`` (the guarded-import binding):
+
+* the REAL ``/proc/self/status`` read on a Linux runner — ``rss_mb()`` is a
+  positive int and provably comes from the ``VmRSS`` parse, not the
+  fallback (the fallback is pinned to a sentinel that never surfaces);
+* the kB→MB floor arithmetic (1500 kB → 1 MB);
+* missing file, unparseable value, value-less ``VmRSS:`` line, and a file
+  without any ``VmRSS`` line — each falls back to ``rss_peak_mb()``;
+* the non-Linux path: ``/proc`` open raises AND ``getrusage`` is patched →
+  the proxy value; both unavailable → ``None``;
+* ``rss_peak_mb`` arithmetic, the no-``resource`` ``None``, and the
+  never-raise wrapper when ``getrusage`` itself raises;
+* ``cgroup_peak_mb`` present (bytes→MB), absent, garbage, empty → the
+  latter three are ``None``;
+* ``fmt_mb(None) == "?"``, ``fmt_mb(0) == "0"`` (0 is a value, not
+  missing), plain ints as digits;
+* the purity contract: stdlib imports only — no ``app.`` import — and the
+  docstring carries the "Pure stdlib" line the phase completion criteria
+  require.
+"""
+from __future__ import annotations
+
+import ast
+import importlib
+import inspect
+import sys
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+import app.core.memory as memory
+
+
+#: Stub whose ``getrusage`` answers with a fixed ``ru_maxrss`` (kB on
+#: Linux) so fallback assertions are deterministic.
+def _fake_resource(maxrss_kb: int) -> Any:
+    return SimpleNamespace(
+        RUSAGE_SELF=0,
+        getrusage=lambda who: SimpleNamespace(ru_maxrss=maxrss_kb),
+    )
+
+
+#: Stub whose ``getrusage`` raises — the never-raise contract probe.
+def _raising_resource(exc: Exception) -> Any:
+    def getrusage(who: int) -> Any:
+        raise exc
+
+    return SimpleNamespace(RUSAGE_SELF=0, getrusage=getrusage)
+
+
+# ---------------------------------------------------------------------------
+# rss_mb — the real /proc read (Linux) and the stubbed failure modes
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="needs /proc/self/status")
+def test_rss_mb_reads_real_vm_rss_on_linux(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Pin the FALLBACK to a negative sentinel: a positive int can therefore
+    # only come from the real VmRSS parse, not from getrusage.
+    monkeypatch.setattr(memory, "rss_peak_mb", lambda: -1)
+
+    value = memory.rss_mb()
+
+    assert value is not None
+    assert isinstance(value, int)
+    assert value > 0
+
+
+def test_rss_mb_parses_vm_rss_kb_flooring_to_mb(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    status = tmp_path / "status"
+    status.write_text("Name:\tpython\nVmRSS:\t   1500 kB\nVmSize:\t999999 kB\n")
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
+    monkeypatch.setattr(memory, "_resource", _fake_resource(999_999))
+
+    assert memory.rss_mb() == 1  # 1500 kB // 1024 — integer floor, not round
+
+
+def test_rss_mb_missing_file_falls_back_to_getrusage(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(tmp_path / "nope"))
+    monkeypatch.setattr(memory, "_resource", _fake_resource(9216))  # 9 MB
+
+    assert memory.rss_mb() == 9
+
+
+def test_rss_mb_unparseable_value_falls_back(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    status = tmp_path / "status"
+    status.write_text("VmRSS:\tnot-a-number kB\n")
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
+    monkeypatch.setattr(memory, "_resource", _fake_resource(2048))
+
+    assert memory.rss_mb() == 2
+
+
+def test_rss_mb_value_less_line_falls_back(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    # "VmRSS:" with no value: the split()[1] raises IndexError internally,
+    # the suppress turns it into the rss_peak_mb() proxy.
+    status = tmp_path / "status"
+    status.write_text("VmRSS:\n")
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
+    monkeypatch.setattr(memory, "_resource", _fake_resource(3072))
+
+    assert memory.rss_mb() == 3
+
+
+def test_rss_mb_file_without_vm_rss_line_falls_back(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    status = tmp_path / "status"
+    status.write_text("Name:\tpython\nVmSize:\t999999 kB\nVmHWM:\t888888 kB\n")
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
+    monkeypatch.setattr(memory, "_resource", _fake_resource(1024))
+
+    assert memory.rss_mb() == 1
+
+
+def test_rss_mb_returns_none_when_proc_and_resource_both_unavailable(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(tmp_path / "nope"))
+    monkeypatch.setattr(memory, "_resource", None)
+
+    assert memory.rss_mb() is None
+
+
+# ---------------------------------------------------------------------------
+# rss_peak_mb — arithmetic, the guarded import, the never-raise wrapper
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("maxrss_kb", "expected_mb"),
+    [(1024, 1), (1023, 0), (1500, 1), (4_294_967_296, 4_194_304)],
+)
+def test_rss_peak_mb_arithmetic(
+    monkeypatch: pytest.MonkeyPatch, maxrss_kb: int, expected_mb: int
+) -> None:
+    monkeypatch.setattr(memory, "_resource", _fake_resource(maxrss_kb))
+
+    assert memory.rss_peak_mb() == expected_mb
+
+
+def test_rss_peak_mb_none_without_resource_module(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(memory, "_resource", None)
+
+    assert memory.rss_peak_mb() is None
+
+
+def test_guarded_import_degrades_to_none_when_resource_is_missing(
+    tmp_path: Any,
+) -> None:
+    # The guarded-import branch itself: a platform WITHOUT the Unix-only
+    # `resource` module must still import this module, and every helper
+    # must answer None (no /proc, no proxy). `sys.modules[name] = None`
+    # makes `import resource` raise ImportError, so re-executing the
+    # module exercises the except arm for real (no pragma).
+    saved = sys.modules.get("resource")
+    sys.modules["resource"] = None  # type: ignore[assignment]
+    try:
+        importlib.reload(memory)
+
+        assert memory._resource is None
+        assert memory.rss_peak_mb() is None
+        memory.PROC_STATUS_PATH = str(tmp_path / "nope")  # reset by reload
+        assert memory.rss_mb() is None
+    finally:
+        if saved is not None:
+            sys.modules["resource"] = saved
+        else:  # pragma: no cover - non-Unix runners never had it
+            sys.modules.pop("resource", None)
+        importlib.reload(memory)  # restore the real bindings for later tests
+
+
+def test_module_reloads_with_real_resource_after_the_missing_module_probe() -> None:
+    # Sibling pin: the finally-reload above restored the module, so the
+    # guarded import took the happy arm again (order-independent: this
+    # only passes if _resource is the live module on this platform).
+    if not sys.platform.startswith("linux"):
+        pytest.skip("resource exists on Linux")
+
+    assert memory._resource is not None
+
+
+def test_rss_peak_mb_never_raises_when_getrusage_raises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(memory, "_resource", _raising_resource(OSError("boom")))
+
+    assert memory.rss_peak_mb() is None
+
+
+def test_rss_peak_mb_real_call_on_linux_yields_positive_int() -> None:
+    if not sys.platform.startswith("linux"):
+        pytest.skip("resource.getrusage is the Linux deploy path")
+    value = memory.rss_peak_mb()
+
+    assert value is not None
+    assert isinstance(value, int)
+    assert value > 0
+
+
+# ---------------------------------------------------------------------------
+# cgroup_peak_mb — present / absent / garbage / empty
+# ---------------------------------------------------------------------------
+
+
+def test_cgroup_peak_mb_present_converts_bytes_to_mb(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    # 2,306,867,200 bytes == 2200 MiB — the incident box's ~2.16 GB of
+    # cgroup page-cache-inflated "peak" (module docstring caveat).
+    peak = tmp_path / "memory.peak"
+    peak.write_text("2306867200\n")
+    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(peak))
+
+    assert memory.cgroup_peak_mb() == 2200
+
+
+def test_cgroup_peak_mb_absent_is_none(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(tmp_path / "memory.peak"))
+
+    assert memory.cgroup_peak_mb() is None
+
+
+@pytest.mark.parametrize("body", ["max", "", "   \n", "1048576 kB"])
+def test_cgroup_peak_mb_garbage_is_none(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any, body: str
+) -> None:
+    peak = tmp_path / "memory.peak"
+    peak.write_text(body)
+    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(peak))
+
+    assert memory.cgroup_peak_mb() is None
+
+
+# ---------------------------------------------------------------------------
+# fmt_mb — the "?" rendering
+# ---------------------------------------------------------------------------
+
+
+def test_fmt_mb_renders_none_as_question_mark_and_values_as_digits() -> None:
+    assert memory.fmt_mb(None) == "?"
+    assert memory.fmt_mb(0) == "0"  # 0 is a VALUE, not missing
+    assert memory.fmt_mb(512) == "512"
+
+
+# ---------------------------------------------------------------------------
+# Purity contract (completion criterion: stdlib only, no app. imports)
+# ---------------------------------------------------------------------------
+
+
+def test_module_imports_stdlib_only_and_declares_pure_stdlib() -> None:
+    tree = ast.parse(inspect.getsource(memory))
+    imported: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            imported.update(alias.name.split(".")[0] for alias in node.names)
+        elif isinstance(node, ast.ImportFrom):
+            if node.level:  # relative import — still inside the app package
+                imported.add("app")
+            elif node.module:
+                imported.add(node.module.split(".")[0])
+
+    assert "app" not in imported
+    # __future__, contextlib, types are static; `resource` is the guarded
+    # Unix-only stdlib import — everything must stay standard library.
+    assert imported <= {"__future__", "contextlib", "resource", "types"}, imported
+    assert "Pure stdlib" in (memory.__doc__ or "")

tests/unit/test_frontend_sync_upload.py

@@ -667,7 +667,12 @@ def test_settle_and_idle_labels_go_through_the_parts_writer() -> None:
     success = _fn(js, "applySyncSuccess")
     assert '"Synced"' in success and "fmtSyncTime(status.finished_at)" in success
     i_settle = success.find("settleSyncButton(")
-    i_counts = success.find("setSyncProgress(fmtSyncResult(status.detail))")
+    # Phase 187 (task 03): the clause builder rides the SAME writer call
+    # (the counts line + its trailing " · peak N MB", null rendering no
+    # clause), so the ordering pin reads the new argument list.
+    i_counts = success.find(
+        "setSyncProgress(fmtSyncResult(status.detail) + fmtSyncPeakClause(status))"
+    )
     assert -1 < i_settle < i_counts, (
         "the counts line (the box's result line) follows the settle label"
     )

tests/unit/test_importer.py

@@ -18,6 +18,7 @@ import base64
 import hashlib
 import logging
 import os
+import re
 import textwrap
 from collections.abc import Callable, Sequence
 from datetime import UTC, datetime
@@ -54,6 +55,11 @@ from tests.fakes import FakeEmbedder
 #: (``Settings().import_extension_set`` — now seventeen formats).
 DEFAULT_EXTS = frozenset({".md", ".markdown", ".txt", ".yaml", ".yml", ".json", ".py"})
 
+#: Phase 187 (task 02): the ONE rendering :func:`app.core.memory.fmt_mb`
+#: allows for an MB field — ``?`` when the platform cannot know, else a
+#: positive integer (never ``None``, never a float or a negative).
+MB = r"(?:\?|[1-9][0-9]*)"
+
 
 class _PoisonEmbedder(FakeEmbedder):
     """Fails (like a real endpoint) on any text containing 'poison'."""
@@ -1001,7 +1007,10 @@ def test_import_summary_log_line_includes_summary_counters(
     and the phase-151 ``pdf_images_deduped`` / ``pdf_images_blank_skipped``
     counters (the parser's per-PDF drop counts, next to
     ``pdf_pictorial_pages``) sit after ``dates_updated``, before
-    ``formats``."""
+    ``formats``. Phase 187 (task 02) APPENDS a trailing ``rss_mb=``
+    (LOCKED A2 — the memory the run sits at, ``?`` when unknowable); the
+    pre-phase text up to and including ``formats=…`` is byte-identical.
+    """
     s = ImportSummary()
     s.files, s.added, s.chunks, s.embed_batches = 3, 3, 5, 4
     s.summaries, s.summary_errors = 2, 1
@@ -1019,7 +1028,7 @@ def test_import_summary_log_line_includes_summary_counters(
     with caplog.at_level(logging.INFO, logger="app.importer"):
         s.log()
     line = caplog.records[-1].getMessage()
-    assert line == (
+    assert line.startswith(
         "import: summary files=3 added=3 updated=0 unchanged=0 pruned=0 errors=0 "
         "chunks=5 embed_batches=4 summaries=2 summary_errors=1 summary_backfilled=1 "
         "dates_updated=0 images_failed=0 svg_indexed=4 image_context_backfilled=2 "
@@ -1027,6 +1036,9 @@ def test_import_summary_log_line_includes_summary_counters(
         "pdf_images_deduped=3 pdf_images_blank_skipped=1 "
         "formats=yaml:2,md:1"
     )
+    # Phase 187 (task 02): the ONE appended field, `?` (unknowable) or a
+    # positive integer MB — never `None`, never a negative or a float.
+    assert re.fullmatch(rf".* rss_mb={MB}", line), line
 
 
 def test_prune_removes_files_now_excluded_by_format_filter(db, tmp_path: Path) -> None:
@@ -1082,7 +1094,10 @@ def test_formats_counter_counts_extensionless_name_token(
             for r in caplog.records
             if "import: summary files=" in r.getMessage()
         )
-        assert line.endswith("formats=dockerfile:1,md:1")
+        # Phase 187 (task 02): ``formats=`` is no longer the LAST field
+        # (`` rss_mb=…`` follows), but its position and value are intact.
+        assert " formats=dockerfile:1,md:1" in line
+        assert re.search(rf"formats=dockerfile:1,md:1 rss_mb={MB}$", line), line
         assert "unknown" not in line
     finally:
         _cleanup_source(db, root.name)

tests/unit/test_sync_button.py

@@ -203,6 +203,10 @@ RETIRED_ID = "-".join(("sync", "result"))  # built at runtime: 3.12+ folds
 
 HAVE_NODE = shutil.which("node") is not None
 
+#: Phase 187 (task 02): what :func:`app.core.memory.fmt_mb` may render for
+#: an MB field — ``?`` when the platform cannot know, else a positive int.
+MB = r"(?:\?|[1-9][0-9]*)"
+
 
 def _text(path: Path) -> str:
     assert path.is_file(), f"missing frontend file: {path}"
@@ -1112,9 +1116,13 @@ def test_sources_js_renders_result_line_and_banner_directly() -> None:
     for other subscribers."""
     js = _text(SOURCES_JS)
     success = _body(js, "applySyncSuccess")
-    assert "setSyncProgress(fmtSyncResult(status.detail))" in success, (
-        "the counts live in status.detail — a bare status renders all zeros"
-    )
+    # Phase 187 (task 03): the counts line keeps its builder and gains
+    # the TRAILING peak clause — a null pair appends the empty string, so
+    # every pre-phase line stays byte-identical.
+    assert (
+        "setSyncProgress(fmtSyncResult(status.detail) + fmtSyncPeakClause(status))"
+        in success
+    ), "the counts live in status.detail — a bare status renders all zeros"
     assert "loadTree()" in success, (
         "the catalog re-fetches live on a successful sync (phase 97: the tree)"
     )
@@ -1319,14 +1327,21 @@ def fresh_sync_status() -> Iterator[None]:
 
     Phase 134 (task 01): the module ``_cancel_event`` joins the reset —
     a stale (set) event from a previous test's cancelled run would
-    otherwise bleed into the next run's boundary checks."""
+    otherwise bleed into the next run's boundary checks.
+
+    Phase 187 (task 03): the memory-sampling guard (``_rss_sampled_at``)
+    joins the reset — it is module-global state beside ``_status``, and a
+    test that pins "the per-file hook sampled" has to know whether the
+    guard says "never read" or "still inside the 5 s window"."""
     sync_api._status = sync_api.SyncStatus()
     sync_api._task = None
     sync_api._cancel_event = asyncio.Event()
+    sync_api._rss_sampled_at = float("-inf")
     yield
     sync_api._status = sync_api.SyncStatus()
     sync_api._task = None
     sync_api._cancel_event = asyncio.Event()
+    sync_api._rss_sampled_at = float("-inf")
 
 
 class _GatedClone:
@@ -1567,9 +1582,11 @@ def test_idle_status_pins_full_shape_including_progress_keys(
 ) -> None:
     """Idle: the three phase-64 progress keys ride along as null/0/0,
     the two phase-129 chunk keys as 0/0, the two phase-152 image keys
-    as 0/0, the four phase-98 phase keys as null/0/0/0, and EVERY
-    pre-existing key is unchanged — the full response dict is pinned,
-    so the current UI and every existing consumer keep working."""
+    as 0/0, the four phase-98 phase keys as null/0/0/0, the two phase-187
+    memory keys as null/null (NO reading is not 0 — and the Sources page
+    renders no peak clause for it), and EVERY pre-existing key is
+    unchanged — the full response dict is pinned, so the current UI and
+    every existing consumer keep working."""
     assert sync_api.sync_status() == {
         "state": "idle",
         "started_at": None,
@@ -1592,6 +1609,10 @@ def test_idle_status_pins_full_shape_including_progress_keys(
         "current_summary": None,
         "summaries_done": 0,
         "summaries_total": 0,
+        # Phase 187 (task 03): the memory keys — null/null idle (never
+        # sampled: never 0, never an invented number).
+        "rss_mb": None,
+        "rss_peak_mb": None,
     }
 
 
@@ -2562,10 +2583,15 @@ def test_cancel_endpoint_stops_the_in_flight_run(
     # through the closure in flight is gone; the FILE counts above
     # keep their final values).
     assert s["images_done"] == 0 and s["images_total"] == 0
-    # PLAN §9: the ample-logging line for the converted cancel.
+    # PLAN §9: the ample-logging line for the converted cancel. Phase
+    # 187 (task 02) APPENDS the memory fields (LOCKED A2) — the pre-phase
+    # prefix stays byte-identical, the tail is `?` or a positive int.
     assert any(
-        r.getMessage()
-        == "sync: cancelled by user state=cancelled files_done=1 files_total=3"
+        re.fullmatch(
+            r"sync: cancelled by user state=cancelled files_done=1 files_total=3"
+            rf" rss_mb={MB} rss_peak_mb={MB}",
+            r.getMessage(),
+        )
         for r in caplog.records
     )
 
@@ -3091,7 +3117,9 @@ def test_sources_js_apply_sync_cancelled_is_neutral() -> None:
     js = _text(SOURCES_JS)
     body = _body(js, "applySyncCancelled")
     assert 'settleSyncButton("Sync sources")' in body
-    assert 'setSyncProgress("Sync cancelled")' in body
+    # Phase 187 (task 03): the neutral copy keeps its exact wording and
+    # the peak clause trails it (a null pair appends the empty string).
+    assert 'setSyncProgress("Sync cancelled" + fmtSyncPeakClause(status))' in body
     assert "hideSyncError()" in body
     assert "emitSyncStatus(status)" in body
     assert "showSyncModal" not in body, "cancel is intent — no modal"
@@ -3123,3 +3151,449 @@ def test_sources_js_section_header_documents_the_cancel_contract() -> None:
     assert "5. sync cancelled" in header, "the tree's cancelled branch (the 5th)"
     assert "8. both idle" in header, "the tree is now 8 branches"
     assert "#sync-cancel-btn" in header, "the control's contract is documented"
+
+
+# ---------- phase 187 (task 03): the run's memory on the status ----------
+#
+# The owner incident (2026-10-15): "how much RAM did that sync use?" was
+# answered by reading systemd unit-teardown lines out of three boots —
+# and the boot that killed the VM had none. Tasks 01/02 made the log
+# carry the number; task 03 puts the SAME pair on the status object, so
+# the page that started the run answers without a shell. The pins below
+# hold the three things that make that honest: the pair is ADDITIVE (the
+# key set is exactly the old set plus two), it is SAMPLED from ONE place
+# (forced at the run start / every phase transition / every terminal,
+# throttled to one read per 5 s in the per-file hook — LOCKED A4), and a
+# reading that does not exist stays null (never 0, never invented).
+
+
+#: The status payload's key set BEFORE phase 187 — the additive pin's
+#: left-hand side: the payload is exactly this set plus
+#: :data:`_MEMORY_KEYS`, nothing else, forever.
+_PRE_187_STATUS_KEYS = frozenset(
+    {
+        "state",
+        "started_at",
+        "finished_at",
+        "detail",
+        "error",
+        "current_file",
+        "files_done",
+        "files_total",
+        "current_chunk",
+        "chunks_total",
+        "images_done",
+        "images_total",
+        "phase",
+        "current_summary",
+        "summaries_done",
+        "summaries_total",
+    }
+)
+
+#: Phase 187 (task 03): the two additive memory keys.
+_MEMORY_KEYS = frozenset({"rss_mb", "rss_peak_mb"})
+
+
+def test_status_key_set_is_the_old_set_plus_the_memory_pair(
+    fresh_sync_status: None,
+) -> None:
+    """The key-set pin (D4: "Additive only — every pre-existing key
+    keeps its name, type, and meaning"): the payload is EXACTLY the
+    pre-phase-187 set ``∪ {rss_mb, rss_peak_mb}`` — a fifth sub-position
+    or a renamed field fails here even when the value assertions pass."""
+    keys = set(sync_api.sync_status())
+    assert keys == _PRE_187_STATUS_KEYS | _MEMORY_KEYS, keys
+    assert _MEMORY_KEYS.isdisjoint(_PRE_187_STATUS_KEYS)
+
+
+class _FakeClock:
+    """``sync_api.time`` stubbed: the sampler is the module's only
+    ``time.monotonic()`` reader, so the 5 s guard is assertable without
+    sleeping (the test moves the clock, never the wall)."""
+
+    def __init__(self, start: float = 10_000.0) -> None:
+        self.now = start
+
+    def monotonic(self) -> float:
+        return self.now
+
+
+class _StubMemory:
+    """The two ``app.core.memory`` reads as a scripted pair-per-read
+    source: sample *n* returns ``readings[n]`` (the last pair repeats)
+    and :attr:`reads` counts the samples, so "sampled at most once per
+    5 s" is a number and not a comment.
+
+    Both helpers are patched on ``app.api.sync`` (the module under test
+    imports them by name), which is also what proves the sampler reads
+    them from ONE place: a call counter that disagrees with
+    ``_status`` is a second, unsampled write path."""
+
+    def __init__(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        readings: Sequence[tuple[int | None, int | None]],
+    ) -> None:
+        self.readings = list(readings)
+        self.reads = 0
+        monkeypatch.setattr(sync_api, "rss_mb", self._current)
+        monkeypatch.setattr(sync_api, "rss_peak_mb", self._peak)
+
+    def _pair(self) -> tuple[int | None, int | None]:
+        return self.readings[min(self.reads, len(self.readings) - 1)]
+
+    def _current(self) -> int | None:
+        return self._pair()[0]
+
+    def _peak(self) -> int | None:
+        value = self._pair()[1]
+        self.reads += 1  # the PAIR is one sample — counted when it completes
+        return value
+
+
+def test_sample_rss_stores_both_readings_from_one_sample(
+    fresh_sync_status: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """``_sample_rss`` is the ONE writer of the pair: both fields land
+    from the same read, so the status and the lifecycle line can never
+    report two different instants."""
+    monkeypatch.setattr(sync_api, "time", _FakeClock())
+    memory = _StubMemory(monkeypatch, [(412, 1_936)])
+
+    sync_api._sample_rss(force=True)
+
+    assert sync_api._status.rss_mb == 412
+    assert sync_api._status.rss_peak_mb == 1_936
+    assert memory.reads == 1
+
+
+def test_sample_rss_throttles_the_unforced_call_and_force_always_reads(
+    fresh_sync_status: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """LOCKED A4 / D3, pinned on the guard itself: the FIRST call reads
+    (never sampled), the next one inside ``_RSS_SAMPLE_INTERVAL_S`` is a
+    NO-OP, one past the interval reads again, and ``force=True`` reads
+    regardless AND re-arms the window (which is what keeps the 27k-file
+    walk at ~600 reads)."""
+    clock = _FakeClock(10_000.0)
+    monkeypatch.setattr(sync_api, "time", clock)
+    memory = _StubMemory(
+        monkeypatch, [(100, 200), (101, 201), (102, 202), (103, 203), (104, 204)]
+    )
+    assert sync_api._RSS_SAMPLE_INTERVAL_S == 5.0, "the phase's sampling window"
+
+    sync_api._sample_rss()  # never sampled → reads
+    assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (100, 200)
+
+    clock.now += sync_api._RSS_SAMPLE_INTERVAL_S - 0.001
+    sync_api._sample_rss()  # inside the window → no-op
+    assert memory.reads == 1
+    assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (100, 200)
+
+    clock.now += 0.002  # the window closed
+    sync_api._sample_rss()
+    assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (101, 201)
+
+    sync_api._sample_rss(force=True)  # forced, immediately → reads
+    assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (102, 202)
+    sync_api._sample_rss()  # …and the forced read re-armed the window
+    assert memory.reads == 3, "the forced sample closed the window it opened"
+
+    clock.now += sync_api._RSS_SAMPLE_INTERVAL_S
+    sync_api._sample_rss()
+    assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (103, 203)
+    assert memory.reads == 4
+
+
+def test_sample_rss_keeps_none_as_unknown(
+    fresh_sync_status: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A platform that cannot know (no ``/proc``, no ``resource``) stays
+    ``None`` on the status — never 0, never a plausible number. The wire
+    form is JSON null and the UI renders no clause for it (the same
+    ``?``-is-not-0 rule :func:`app.core.memory.fmt_mb` holds for the
+    log)."""
+    monkeypatch.setattr(sync_api, "time", _FakeClock())
+    _StubMemory(monkeypatch, [(None, None)])
+
+    sync_api._sample_rss(force=True)
+
+    assert sync_api._status.rss_mb is None
+    assert sync_api._status.rss_peak_mb is None
+    body = sync_api.sync_status()
+    assert body["rss_mb"] is None and body["rss_peak_mb"] is None
+
+
+def test_sample_rss_never_publishes_a_peak_below_the_current_reading(
+    fresh_sync_status: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The pair's invariant, pinned at the ONE sampling point (task 04).
+
+    ``VmRSS`` and ``ru_maxrss`` are two kernel counters read at two
+    instants, both floored to whole MB, and the high-water refreshes
+    LAZILY — so a growing run really does report ``rss_mb`` above
+    ``rss_peak_mb`` (measured on this repo's own suite at 164 vs 162).
+    A "peak" under the process's own current footprint is a FALSE
+    number, not a conservative one, so the sampler lifts it to the
+    current reading: a lower bound on the truth, never an invention.
+    Without this the invariant every consumer (both lifecycle lines, the
+    status payload, the page's clause) now relies on would be a coin
+    flip — and the phase's own integration gate proved it."""
+    monkeypatch.setattr(sync_api, "time", _FakeClock())
+    _StubMemory(monkeypatch, [(412, 410)])  # the lazy-kernel case
+
+    sync_api._sample_rss(force=True)
+
+    assert sync_api._status.rss_mb == 412
+    assert sync_api._status.rss_peak_mb == 412, "the lagging peak is raised"
+    assert sync_api.sync_status()["rss_peak_mb"] == 412, "the wire agrees"
+
+
+def test_sample_rss_does_not_fabricate_an_unknown_peak(
+    fresh_sync_status: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The normalization's edge: an UNKNOWABLE peak stays ``None``.
+
+    Lifting a lagging reading is a bound on a real measurement; inventing
+    a peak from the current reading on a platform where
+    :func:`app.core.memory.rss_peak_mb` answered ``None`` would be the
+    invented number the phase forbids (``null`` renders no clause, ``0``
+    is a value). So the clamp needs BOTH readings to be real."""
+    monkeypatch.setattr(sync_api, "time", _FakeClock())
+    _StubMemory(monkeypatch, [(412, None)])
+
+    sync_api._sample_rss(force=True)
+
+    assert sync_api._status.rss_mb == 412
+    assert sync_api._status.rss_peak_mb is None
+
+
+def test_run_samples_at_start_phase_transitions_and_terminal(
+    fresh_sync_status: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The sampling POINTS, walked through the real runner: run start →
+    ``import`` → ``overview`` → ``summaries`` → the success terminal, one
+    forced read each, and the per-file progress hook's throttled call
+    adds NOTHING inside the window the run start just armed (LOCKED A4).
+
+    The last pair is the number the incident was about: the terminal
+    KEEPS it (the phase-64 keep-final-values convention) and the
+    ``sync: done`` line logs the very same pair the payload reports —
+    one reading, two surfaces, no disagreement between the page and the
+    log an operator could be argued with."""
+    clock = _FakeClock(10_000.0)
+    monkeypatch.setattr(sync_api, "time", clock)
+    # run start, import, overview, summaries, done (the hook is throttled).
+    memory = _StubMemory(
+        monkeypatch,
+        [(10, 20), (11, 21), (12, 22), (13, 23), (14, 4_321)],
+    )
+    clone = _GatedClone(threading.Event(), threading.Event())
+    fake_import = _GatedImport(
+        ImportSummary(files=3, added=1, unchanged=2),
+        threading.Event(),
+        threading.Event(),
+    )
+    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
+
+    with caplog.at_level(logging.INFO, logger="app.api.sync"):
+        thread, errors = _start_run()
+        try:
+            assert clone.started.wait(5.0), "the run never reached the clone phase"
+            # The run start sampled BEFORE any pipeline step: a run that
+            # fails in the clone already has a number on the status.
+            s = sync_api.sync_status()
+            assert s["state"] == "running"
+            assert (s["rss_mb"], s["rss_peak_mb"]) == (10, 20)
+            clone.release.set()
+            assert fake_import.started.wait(5.0), "the run never reached the import"
+            # Import phase: the transition forced the second read, and
+            # the hook's own call (this fake fires one file) is inside
+            # the window — still 2 reads, not 3 (A4 in numbers).
+            s = sync_api.sync_status()
+            assert s["phase"] == "import"
+            assert fake_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]
+            assert (s["rss_mb"], s["rss_peak_mb"]) == (11, 21)
+            assert memory.reads == 2, "the per-file hook never bypasses the guard"
+            fake_import.release.set()
+        finally:
+            clone.release.set()
+            fake_import.release.set()
+            thread.join(10.0)
+    assert errors == [], f"the runner raised: {errors!r}"
+
+    # The terminal pair is the FINAL reading, kept next to the counts,
+    # and it survives the run (the page shows it after the settle).
+    s = sync_api.sync_status()
+    assert s["state"] == "success"
+    assert (s["rss_mb"], s["rss_peak_mb"]) == (14, 4_321)
+    assert s["files_done"] == 1 and s["files_total"] == 3
+
+    done = [
+        r.getMessage()
+        for r in caplog.records
+        if r.getMessage().startswith("sync: done detail=")
+    ]
+    assert len(done) == 1, done
+    assert done[0].endswith("rss_mb=14 rss_peak_mb=4321"), done[0]
+
+
+def test_per_file_hook_samples_once_the_window_closes(
+    fresh_sync_status: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """The other half of D3: the number on the page MOVES. The hook
+    fires once per file, and once the 5 s window has closed the next
+    file's call re-reads — so a long walk reports the memory climbing
+    (PLAN §7.4's never-stale contract) at ~600 reads per 27k files
+    instead of 27k."""
+    clock = _FakeClock(10_000.0)
+    monkeypatch.setattr(sync_api, "time", clock)
+    memory = _StubMemory(monkeypatch, [(100, 200), (110, 210), (140, 260)])
+    clone = _GatedClone(threading.Event(), threading.Event())
+    fake_import = _GatedImport(
+        ImportSummary(files=1, added=1),
+        threading.Event(),
+        threading.Event(),
+    )
+    # The fake fires ONE file, parks, then returns — so the hook is
+    # called once mid-run and the test calls it a second time with the
+    # window closed (the second file of a real walk).
+    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
+
+    thread, errors = _start_run()
+    try:
+        assert clone.started.wait(5.0), "the run never reached the clone phase"
+        clone.release.set()
+        assert fake_import.started.wait(5.0), "the run never reached the import"
+        s = sync_api.sync_status()
+        assert s["phase"] == "import"
+        assert (s["rss_mb"], s["rss_peak_mb"]) == (110, 210)  # the transition
+        assert memory.reads == 2, "the hook's call inside the window read nothing"
+        clock.now += sync_api._RSS_SAMPLE_INTERVAL_S + 1.0
+        sync_api._sample_rss()  # the NEXT file's hook call
+        assert (sync_api._status.rss_mb, sync_api._status.rss_peak_mb) == (140, 260)
+        fake_import.release.set()
+    finally:
+        clone.release.set()
+        fake_import.release.set()
+        thread.join(10.0)
+    assert errors == [], f"the runner raised: {errors!r}"
+    assert memory.reads >= 3
+
+
+# ---------- phase 187 (task 03): the peak clause on the result line ------
+
+
+def test_sources_js_peak_clause_is_a_trailing_clause_with_a_null_guard() -> None:
+    """The frontend SOURCE pin (the browser behavior is
+    tests/e2e/test_sync_peak_memory_visible.py, phase 187 task 04):
+
+    * ONE builder, reading ``status.rss_peak_mb ?? status.rss_mb`` (the
+      high-water mark, the current reading as the fall-back);
+    * a non-number (null / undefined / garbage) renders the EMPTY string
+      — never ``peak ? MB``, never ``peak null MB``;
+    * it is APPENDED to the two result strings that already exist
+      (``fmtSyncResult(...)``, ``"Sync cancelled"``) through the single
+      box writer — no new control, no new live region, no replaced
+      state string;
+    * NOTHING else reads it: the running label, the failed box (the
+      banner + modal own the error, byte-identical per phase 150) and
+      the upload settles never grow a clause;
+    * the shell markup is untouched — the box is still one text span."""
+    js = _text(SOURCES_JS)
+    builder = _body(js, "fmtSyncPeakClause")
+    assert "status.rss_peak_mb ?? status.rss_mb" in builder
+    assert 'return "";' in builder, "null renders NOTHING"
+    assert "` · peak ${peak} MB`" in builder, "the clause's exact shape"
+
+    success = _body(js, "applySyncSuccess")
+    assert (
+        "setSyncProgress(fmtSyncResult(status.detail) + fmtSyncPeakClause(status))"
+        in success
+    )
+    cancelled = _body(js, "applySyncCancelled")
+    assert 'setSyncProgress("Sync cancelled" + fmtSyncPeakClause(status))' in cancelled
+
+    for untouched in (
+        "enterSyncRunningState",
+        "fmtSyncPhaseLabel",
+        "fmtSyncLabel",
+        "fmtSyncResult",
+        "setSyncLabel",
+        "setSyncProgress",
+        "applySyncFailure",
+        "applySyncIdle",
+    ):
+        assert "fmtSyncPeakClause" not in _body(js, untouched), (
+            f"{untouched} must not grow a clause — the terminal result "
+            "line is the only surface (D4)"
+        )
+
+    html = _text(SHELL_HTML)
+    assert '<span class="sync-progress-text" id="sync-progress-text"></span>' in html
+    assert 'id="sync-progress-peak"' not in html, "no new element for the clause"
+
+
+@pytest.mark.skipif(not HAVE_NODE, reason="node not available")
+def test_sources_js_peak_clause_renders_under_node() -> None:
+    """The REAL extracted ``fmtSyncResult`` + ``fmtSyncPeakClause`` under
+    node, composed exactly as ``applySyncSuccess`` composes them: the
+    counts line keeps its exact wording and the clause trails it
+    (``3 added · 2 unchanged · peak 1936 MB``); a null peak falls back to
+    the current reading; BOTH null (idle, a pre-phase-187 run, the
+    upload job, a platform that cannot know) render the counts line with
+    NO clause — never ``peak null MB``; ``0`` is a VALUE, not missing
+    (the ``fmt_mb(0) == "0"`` rule, mirrored); and a negative or
+    non-numeric value renders nothing."""
+    src = (
+        _body(_text(SOURCES_JS), "fmtSyncResult")
+        + "\n"
+        + _body(_text(SOURCES_JS), "fmtSyncPeakClause")
+    )
+    driver_expr = (
+        "const CASES = ["
+        "{ detail: { added: 3, unchanged: 2 }, rss_mb: 512, rss_peak_mb: 1936 },"
+        "{ detail: { added: 0, unchanged: 1 }, rss_mb: 512, rss_peak_mb: null },"
+        "{ detail: { added: 0, unchanged: 1 }, rss_mb: null, rss_peak_mb: null },"
+        "{ detail: { added: 1 } },"
+        "{ detail: { added: 2, pruned: 4 }, rss_mb: 0, rss_peak_mb: 0 },"
+        "{ detail: { added: 1 }, rss_peak_mb: -5 },"
+        "{ detail: { added: 1 }, rss_peak_mb: 'big' },"
+        "];\n"
+        "const LINES = CASES.map("
+        "  (s) => fmtSyncResult(s.detail) + fmtSyncPeakClause(s)\n"
+        ");\n"
+        "const CANCELLED = CASES.map("
+        "  (s) => 'Sync cancelled' + fmtSyncPeakClause(s)\n"
+        ");\n"
+        "console.log(JSON.stringify({ LINES, CANCELLED }));"
+    )
+    proc = subprocess.run(
+        ["node", "-e", f"{src}\n{driver_expr}"], capture_output=True, text=True, timeout=60
+    )
+    assert proc.returncode == 0, f"node failed: {proc.stderr}"
+    out = json.loads(proc.stdout)
+    assert out["LINES"] == [
+        "3 added · 2 unchanged · peak 1936 MB",
+        "0 added · 1 unchanged · peak 512 MB",  # the current-reading fallback
+        "0 added · 1 unchanged",  # both null → no clause at all
+        "1 added",  # idle / pre-phase / upload status: no keys, no clause
+        "2 added · 4 pruned · peak 0 MB",  # 0 is a value, not missing
+        "1 added",  # a negative reading is not a number to show
+        "1 added",  # garbage is not a number either
+    ], out
+    assert out["CANCELLED"] == [
+        "Sync cancelled · peak 1936 MB",
+        "Sync cancelled · peak 512 MB",
+        "Sync cancelled",
+        "Sync cancelled",
+        "Sync cancelled · peak 0 MB",
+        "Sync cancelled",
+        "Sync cancelled",
+    ], out