feat: sleep-time janitor + scratch tier (items #3 and #5 part 2)

src/omind/cli.py

@@ -11,6 +11,8 @@
   * ``omind export`` — write the entire OMI dataset to a json or tar.gz bundle.
   * ``omind import`` — load an OMI dataset bundle back into a folder.
   * ``omind reindex`` — regenerate index.md under the inter-process write lock.
+  * ``omind maintain`` — sleep-time janitor: propose merges, refresh the index,
+    opt-in journal rollup / mesh sync. Safe by default (dry run).
   * ``omind quickstart`` — print the manual-wiring steps `setup` automates.
   * ``omind graph`` — query the [[wikilink]] knowledge graph (neighbors, path,
     orphans, dangling links, stats, export).
@@ -67,7 +69,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--version", action="version", version=f"omind {__version__}")
     sub = parser.add_subparsers(
         dest="command",
-        metavar="{help,setup,quickstart,serve,doctor,self-update,backup,ai,export,import,reindex,note,rollup,recover,hook}",
+        metavar="{help,setup,quickstart,serve,doctor,self-update,backup,ai,export,import,reindex,maintain,note,rollup,recover,hook}",
     )
 
     help_p = sub.add_parser(
@@ -313,6 +315,37 @@ def build_parser() -> argparse.ArgumentParser:
     )
     _add_vault_args(reindex)
 
+    maintain = sub.add_parser(
+        "maintain",
+        help="sleep-time janitor: propose merges, refresh the index, and "
+        "(opt-in) roll up journals or sync the mesh — safe by default",
+    )
+    maintain.add_argument(
+        "--apply",
+        action="store_true",
+        help="perform the safe maintenance (index refresh); consolidations stay "
+        "propose-only regardless",
+    )
+    maintain.add_argument(
+        "--sync",
+        action="store_true",
+        help="also run a mesh sync — the only fleet-propagating, irreversible "
+        "step, so it is opt-in and never in --apply",
+    )
+    maintain.add_argument(
+        "--rollup",
+        action="store_true",
+        help="also roll up eligible weeks of daily journals (a lossy history "
+        "squash — opt-in, never default)",
+    )
+    maintain.add_argument(
+        "--report-note",
+        action="store_true",
+        help="write the run report as a vault note (default: stdout + state dir "
+        "only, so the janitor doesn't litter the vault)",
+    )
+    _add_vault_args(maintain)
+
     convert = sub.add_parser(
         "convert",
         help="migrate an OMI vault to the Open Knowledge Format (OKF): give every "
@@ -1122,6 +1155,35 @@ def _run_import(args: argparse.Namespace) -> int:
     return 1 if (result.conflicts and not args.force) else 0
 
 
+def _run_maintain(args: argparse.Namespace) -> int:
+    from omind import maintain, mesh
+
+    omi_dir = (args.vault / args.folder).expanduser()
+    node_id = ""
+    if args.sync:
+        cfg = mesh.load_node_config(omi_dir)
+        if cfg is None:
+            print(
+                "--sync needs a mesh node — run `omind mesh init` first, or drop --sync",
+                file=sys.stderr,
+            )
+            return 1
+        node_id = cfg.node_id
+    report = maintain.run(
+        omi_dir,
+        node_id=node_id,
+        apply=args.apply,
+        sync=args.sync,
+        rollup=args.rollup,
+        report_note=args.report_note,
+    )
+    if report.refused:
+        return 1
+    if not args.apply and not args.sync and not args.rollup:
+        print("(dry run — pass --apply to act; nothing in the vault was changed)")
+    return 1 if report.aborted else 0
+
+
 def _run_reindex(args: argparse.Namespace) -> int:
     from omind import searchindex
     from omind.journal import migrate_journals
@@ -1729,6 +1791,8 @@ def main(argv: list[str] | None = None) -> int:
         return _run_ai(args)
     if args.command == "reindex":
         return _run_reindex(args)
+    if args.command == "maintain":
+        return _run_maintain(args)
     if args.command == "convert":
         return _run_convert(args)
     if args.command == "note":

src/omind/filelock.py

@@ -31,6 +31,15 @@ def lock_fd(fd: int) -> None:
         os.lseek(fd, 0, os.SEEK_SET)
         msvcrt.locking(fd, msvcrt.LK_LOCK, _REGION_BYTES)
 
+    def try_lock_fd(fd: int) -> bool:
+        """Take the exclusive lock without blocking; ``False`` if held elsewhere."""
+        os.lseek(fd, 0, os.SEEK_SET)
+        try:
+            msvcrt.locking(fd, msvcrt.LK_NBLCK, _REGION_BYTES)
+        except OSError:
+            return False
+        return True
+
     def unlock_fd(fd: int) -> None:
         """Release the lock taken by :func:`lock_fd`."""
         os.lseek(fd, 0, os.SEEK_SET)
@@ -43,6 +52,14 @@ def lock_fd(fd: int) -> None:
         """Block until this process holds the exclusive lock on ``fd``."""
         fcntl.flock(fd, fcntl.LOCK_EX)
 
+    def try_lock_fd(fd: int) -> bool:
+        """Take the exclusive lock without blocking; ``False`` if held elsewhere."""
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            return False
+        return True
+
     def unlock_fd(fd: int) -> None:
         """Release the lock taken by :func:`lock_fd`."""
         fcntl.flock(fd, fcntl.LOCK_UN)
@@ -89,6 +106,27 @@ def append_locked(path: Path, *, mode: int = 0o600) -> Iterator[int]:
         os.close(fd)
 
 
+@contextlib.contextmanager
+def try_exclusive(path: Path, *, mode: int = 0o600) -> Iterator[bool]:
+    """Try to take the exclusive lock on ``path`` WITHOUT blocking.
+
+    Yields ``True`` when this process took the lock (held for the block) and
+    ``False`` when another process already holds it — the primitive behind the
+    janitor's single-instance mutex and its "is a sync in flight?" probe, where
+    waiting is exactly the wrong behaviour. Same sibling-``.lock`` discipline as
+    :func:`exclusive`.
+    """
+    fd = os.open(path, os.O_RDWR | os.O_CREAT | _BINARY | _NOFOLLOW, mode)
+    acquired = try_lock_fd(fd)
+    try:
+        yield acquired
+    finally:
+        if acquired:
+            with contextlib.suppress(OSError):
+                unlock_fd(fd)
+        os.close(fd)
+
+
 @contextlib.contextmanager
 def exclusive(path: Path, *, mode: int = 0o600) -> Iterator[int]:
     """Open (creating) ``path`` read-write and hold the exclusive lock on it.

src/omind/maintain.py

@@ -0,0 +1,185 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 Aaron K. Clark
+"""Sleep-time janitor — ``omind maintain`` (2026-08-27 roundtable, item #3).
+
+One opt-in maintenance pass over a vault, built to the ratified invariants:
+
+* **Consolidation is propose-only, always.** ``maintain`` may surface merge
+  candidates but never applies them — propose-and-review is permanent, so even
+  ``--apply`` only proposes.
+* **Single janitor per vault.** A non-blocking flock mutex refuses a second
+  concurrent run rather than double-maintaining.
+* **Refuse while a mesh sync (or any vault write) is in flight** — an explicit
+  probe of the vault write-lock, not a guess.
+* **Fail-closed pipeline.** A failed step aborts the rest; in particular the
+  fleet-propagating ``--sync`` never runs after an earlier failure.
+* **GC / mesh-sync stays out of the default apply set** — it is the only step
+  that propagates fleet-wide and is irreversible, so it is opt-in via ``--sync``.
+* **Rollups are opt-in** (``--rollup``), never in the default set — a lossy
+  history squash is a deliberate choice, not a default.
+* **No per-run report notes in the vault.** The report goes to stdout and the
+  state dir; ``--report-note`` is the only way one lands in the vault.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import time
+from collections.abc import Callable
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+
+from omind import consolidate, filelock, journal, mesh, paths, searchindex
+from omind.store import LOCK_FILENAME, SCRATCH_SUFFIX, NoteFields, OmiStore
+
+#: Scratch-tier TTL (item #5 part 2): 7 days from LAST MODIFICATION (HAL9000's
+#: clock refinement — not creation). Expiry ARCHIVES, never deletes.
+_SCRATCH_TTL_DAYS = 7
+
+
+@dataclass
+class StepResult:
+    name: str
+    ok: bool
+    detail: str
+
+
+@dataclass
+class MaintainReport:
+    ran: bool = False
+    #: Non-empty when the janitor declined to run (mutex held, or sync in flight).
+    refused: str = ""
+    apply: bool = False
+    aborted: bool = False
+    steps: list[StepResult] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, object]:
+        return asdict(self)
+
+
+def _mesh_sync_in_flight(omi_dir: Path) -> bool:
+    """Whether the vault write-lock is held right now (a sync or a note write)."""
+    with filelock.try_exclusive(omi_dir / LOCK_FILENAME) as acquired:
+        return not acquired
+
+
+def _reindex_detail(omi_dir: Path) -> str:
+    """Refresh the derived index. ``refresh`` is incremental and live — it never
+    empties the index in place, so a concurrent searcher never sees a hole (the
+    build-and-swap invariant)."""
+    index = searchindex.shared(omi_dir)
+    if index is None:
+        return "index unavailable on this machine — skipped"
+    result = index.refresh()
+    if result is None:
+        return "index busy — skipped"
+    return f"{result.reindexed} reindexed, {result.removed} reaped, {result.notes} notes"
+
+
+def _expire_scratch(omi_dir: Path, *, apply: bool) -> str:
+    """Archive scratch notes untouched for the TTL. Machine-local, so this is
+    safe outside ``--sync``; archives (never deletes) via the soft-delete path."""
+    store = OmiStore(omi_dir)
+    cutoff = time.time() - _SCRATCH_TTL_DAYS * 86_400
+    expired: list[str] = []
+    for path in sorted(omi_dir.glob(f"*{SCRATCH_SUFFIX}")):
+        with contextlib.suppress(Exception):
+            if store.read_fields(path.name).disabled:
+                continue  # already archived
+            if path.stat().st_mtime >= cutoff:
+                continue
+            expired.append(path.name)
+            if apply:
+                store.disable_note(path.name)
+    verb = "archived" if apply else "would expire"
+    return f"{len(expired)} scratch note(s) {verb} (>{_SCRATCH_TTL_DAYS}d since last change)"
+
+
+def _sync_detail(omi_dir: Path, node_id: str) -> str:
+    report = mesh.sync(omi_dir, node_id, log=lambda *_: None)
+    return f"synced against {len(report.peers)} peer(s)"
+
+
+def _persist(omi_dir: Path, report: MaintainReport) -> None:
+    path = paths.maintain_state_path(omi_dir)
+    with contextlib.suppress(OSError):
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(report.to_dict(), indent=2), encoding="utf-8")
+
+
+def _write_report_note(omi_dir: Path, report: MaintainReport) -> None:
+    """Opt-in only: a vault note summarising the run. Off by default because a
+    report note is itself lint/dedup fodder that then replicates fleet-wide."""
+    lines = [f"- {s.name}: {'ok' if s.ok else 'FAILED'} — {s.detail}" for s in report.steps]
+    OmiStore(omi_dir).create_note(
+        NoteFields(
+            title="Maintenance Report",
+            summary="Latest `omind maintain` run.",
+            details="\n".join(lines) or "no steps ran",
+            tags=["omi", "maintenance"],
+        )
+    )
+
+
+def run(
+    omi_dir: Path | str,
+    *,
+    node_id: str = "",
+    apply: bool = False,
+    sync: bool = False,
+    rollup: bool = False,
+    report_note: bool = False,
+    log: Callable[[str], None] = print,
+) -> MaintainReport:
+    """Run one maintenance pass. Safe by default: with no flags it proposes and
+    reports, changing nothing in the vault."""
+    omi = Path(omi_dir).expanduser()
+    report = MaintainReport(apply=apply)
+
+    with filelock.try_exclusive(paths.maintain_lock_path(omi)) as got_mutex:
+        if not got_mutex:
+            report.refused = "another `omind maintain` is already running for this vault"
+            log(report.refused)
+            return report
+        if _mesh_sync_in_flight(omi):
+            report.refused = "a mesh sync or vault write is in flight — try again shortly"
+            log(report.refused)
+            return report
+        report.ran = True
+
+        def step(name: str, action: Callable[[], str]) -> None:
+            if report.aborted:
+                return
+            try:
+                detail = action()
+                report.steps.append(StepResult(name, True, detail))
+                log(f"[maintain] {name}: {detail}")
+            except Exception as exc:  # noqa: BLE001 — fail-closed: record and stop
+                report.steps.append(StepResult(name, False, f"aborted: {exc}"))
+                report.aborted = True
+                log(f"[maintain] {name} FAILED, aborting: {exc}")
+
+        # Always propose-only, in every mode (propose-and-review is permanent).
+        step(
+            "propose-consolidations",
+            lambda: f"{len(consolidate.propose(omi))} merge review plan(s) proposed"
+            " (propose-only; never auto-applied)",
+        )
+        # Scratch expiry runs in every mode: reports what would expire on a dry
+        # run, archives on --apply. Machine-local, so it is safe outside --sync.
+        step("expire-scratch", lambda: _expire_scratch(omi, apply=apply))
+        if apply:
+            step("reindex", lambda: _reindex_detail(omi))
+        if rollup:
+            step("rollup", lambda: f"{len(journal.rollup_journals(omi))} week(s) rolled up")
+        # The only fleet-propagating, irreversible step: opt-in, and last, so a
+        # failure above (report.aborted) keeps it from ever running.
+        if sync:
+            step("mesh-sync", lambda: _sync_detail(omi, node_id))
+
+        _persist(omi, report)
+        if report_note and not report.aborted:
+            with contextlib.suppress(Exception):
+                _write_report_note(omi, report)
+    return report

src/omind/mesh.py

@@ -79,6 +79,7 @@ class MeshError(Exception):
 GITIGNORE = """\
 .omi.lock
 .tmp-*
+*.scratch.md
 .obsidian/workspace.json
 .obsidian/workspace-mobile.json
 .obsidian/workspace

src/omind/paths.py

@@ -125,3 +125,13 @@ def transaction_dir(omi_dir: Path) -> Path:
     doing, is meaningless on another peer, and must never be mesh-synced.
     """
     return state_dir() / f"txn-{_omi_dir_digest(omi_dir)}"
+
+
+def maintain_lock_path(omi_dir: Path) -> Path:
+    """Single-instance mutex for ``omind maintain`` (one janitor per vault)."""
+    return state_dir() / f"maintain-{_omi_dir_digest(omi_dir)}.lock"
+
+
+def maintain_state_path(omi_dir: Path) -> Path:
+    """Last ``omind maintain`` run report — state-dir, never a vault note."""
+    return state_dir() / f"maintain-{_omi_dir_digest(omi_dir)}.json"

src/omind/server.py

@@ -364,7 +364,9 @@ def _resolve_agent(explicit: str | None) -> str:
             "confidence: high|medium|low, omit if unknown. conflicts_with: a "
             "[[wikilink]] to a memory this one DISAGREES with (use supersedes "
             "instead when this cleanly replaces the older fact). agent: your "
-            "self-declared identity (advisory attribution only)."
+            "self-declared identity (advisory attribution only). scratch: mark a "
+            "machine-local, auto-expiring note (never mesh-synced; archived after "
+            "7 idle days by `omind maintain`)."
         ),
     )
     def create_note(
@@ -382,6 +384,7 @@ def create_note(
         references: list[str] | None = None,
         agent: str = "",
         scope: str = "",
+        scratch: bool = False,
     ) -> dict[str, object]:
         # Scoped-write interlock (item #5): deny an out-of-scope write BEFORE it
         # lands. Raises (rendered as a tool error) in deny mode; returns a
@@ -406,7 +409,9 @@ def create_note(
             references=references or [],
             scope=note_scope,
         )
-        filename = store.create_note(fields)
+        # scratch=True marks this a machine-local, auto-expiring scratch note
+        # (item #5 part 2): never mesh-replicated, TTL-expired by `omind maintain`.
+        filename = store.create_note(fields, scratch=scratch)
         result: dict[str, object] = {"filename": filename, "agent": fields.agent}
         if scope_warning:
             result["scope_warning"] = scope_warning