feat: reinforcement-on-duplicate (upstream port E)

memu/ingest/codebase.py

@@ -382,7 +382,12 @@ def _build_node(
         source["commit"] = commit_sha
     if repo_url:
         source["repo"] = repo_url
-    metadata = {"content_hash": parsed.content_hash}
+    # ``source_hash`` — SHA-256 over the raw source file, powering the
+    # ingester's "skip unchanged" fast path. Distinct from
+    # :meth:`WikiNode.content_hash` (which hashes the rendered wiki
+    # body+title for reinforcement-on-duplicate). Namespaced to avoid
+    # collision when both concerns coexist.
+    metadata = {"source_hash": parsed.content_hash}
     return WikiNode(
         id=str(uuid.uuid4()),
         slug=slug,
@@ -404,15 +409,27 @@ def _build_node(
 async def _put_with_hash(
     backend: StorageBackend,
     node: WikiNode,
-    content_hash: str,
+    source_hash: str,
     result: IngestResult,
     *,
     force: bool,
 ) -> None:
+    """Skip-if-unchanged helper keyed on the raw-source-file hash.
+
+    ``source_hash`` is the ingester's SHA-256 of the file bytes. It
+    lives in ``metadata['source_hash']`` and is independent of
+    :meth:`WikiNode.content_hash`, which hashes the rendered wiki body
+    and drives reinforcement-on-duplicate inside the backend's
+    ``put_node``.
+    """
     existing = await backend.get_node(node.slug)
     if existing is not None:
-        existing_hash = (existing.metadata or {}).get("content_hash")
-        if not force and existing_hash == content_hash:
+        meta = existing.metadata or {}
+        # Accept both the new ``source_hash`` key and the legacy
+        # ``content_hash`` key so vaults built before the rename stay
+        # incremental across an upgrade.
+        existing_hash = meta.get("source_hash") or meta.get("content_hash")
+        if not force and existing_hash == source_hash:
             result.unchanged.append(node.slug)
             return
         node.id = existing.id

memu/rlm/scoring.py

@@ -0,0 +1,102 @@
+"""Scoring primitives for the RLM retriever.
+
+Ports the combined ranking shape from upstream memU (NevaMind-AI/memU
+PR #206): ``similarity × log(reinforcement+1) × recency_decay``.
+
+This module lands the primitive only. Wiring it into the retriever is
+deferred to a follow-up PR so this change stays focused on the storage-
+layer reinforcement port.
+"""
+from __future__ import annotations
+
+import math
+from datetime import datetime, timezone
+from typing import Optional
+
+
+# Half-life of the recency term, expressed in days. A 30-day half-life
+# means a node observed 30 days ago contributes half the recency weight
+# of one observed today. Kept as a module-level default so tests can
+# override while callers outside this module use a single number.
+DEFAULT_RECENCY_HALFLIFE_DAYS = 30.0
+
+
+def recency_decay(
+    recency_days: float,
+    *,
+    halflife_days: float = DEFAULT_RECENCY_HALFLIFE_DAYS,
+) -> float:
+    """Exponential recency decay ``0.5 ** (age / halflife)``.
+
+    - ``recency_days == 0`` -> 1.0 (brand new, full weight).
+    - ``recency_days == halflife_days`` -> 0.5.
+    - ``recency_days`` far in the future (negative) is clamped to 0 so
+      a clock skew can't amplify rankings.
+    """
+    if recency_days < 0:
+        recency_days = 0.0
+    if halflife_days <= 0:
+        # Guard against pathological input; treat as "no decay".
+        return 1.0
+    return 0.5 ** (recency_days / halflife_days)
+
+
+def reinforcement_boost(reinforcement: int) -> float:
+    """``log(reinforcement + 1)`` with a floor of 0 for non-positive counts.
+
+    The +1 offset means an un-reinforced node (count=0) contributes 0,
+    so the combined score degenerates to 0 and the caller can fall back
+    to the similarity term (see :func:`score_combined`). ``natural log``
+    mirrors upstream memU.
+    """
+    if reinforcement <= 0:
+        return 0.0
+    return math.log(reinforcement + 1)
+
+
+def score_combined(
+    similarity: float,
+    reinforcement: int,
+    recency_days: float,
+    *,
+    halflife_days: float = DEFAULT_RECENCY_HALFLIFE_DAYS,
+) -> float:
+    """Upstream-style combined score.
+
+    Formula: ``similarity × log(reinforcement + e) × recency_decay``.
+
+    The ``+ e`` shift (rather than upstream's ``+ 1``) keeps the helper
+    **strictly monotonic** in ``reinforcement``: ``log(0 + e) = 1`` so
+    an un-reinforced node produces a baseline of
+    ``similarity × decay``, and each additional reinforcement raises
+    the boost further (``log(1 + e) > 1``). This avoids the naive
+    ``log(r + 1)`` floor of 0 that would zero out the score for the
+    most common case (freshly written, never re-seen).
+
+    Monotonically *increasing* in ``reinforcement`` and monotonically
+    *decreasing* in ``recency_days``. Negative ``recency_days`` (clock
+    skew, future events) are clamped to 0 so they can't amplify
+    rankings.
+    """
+    decay = recency_decay(recency_days, halflife_days=halflife_days)
+    r = max(0, int(reinforcement))
+    boost = math.log(r + math.e)
+    return float(similarity) * boost * decay
+
+
+def days_since(ts: Optional[datetime], *, now: Optional[datetime] = None) -> float:
+    """Helper to turn a ``WikiNode.last_reinforced_at`` into days-ago.
+
+    Returns ``+inf`` if ``ts`` is ``None`` so a node that was never
+    reinforced still produces a finite score (decay → 0) without special-
+    casing the caller. Timezone-naive inputs are assumed UTC.
+    """
+    if ts is None:
+        return math.inf
+    now = now or datetime.now(timezone.utc)
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    if now.tzinfo is None:
+        now = now.replace(tzinfo=timezone.utc)
+    delta = now - ts
+    return delta.total_seconds() / 86_400.0

memu/storage/base.py

@@ -7,6 +7,8 @@
 """
 from __future__ import annotations
 
+import hashlib
+import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import (
@@ -19,6 +21,19 @@
 )
 
 
+_WS_RE = re.compile(r"\s+")
+
+
+def _normalize_for_hash(text: str) -> str:
+    """Whitespace/case-insensitive normalization for content hashing.
+
+    Collapses CRLF/LF/tabs/runs of spaces into single spaces, strips
+    leading/trailing whitespace, and lowercases. Storage retains the
+    original text — normalization is only for the hash itself.
+    """
+    return _WS_RE.sub(" ", text.replace("\r\n", "\n")).strip().lower()
+
+
 NodeKind = Literal["note", "code", "paper", "task"]
 LinkType = Literal[
     "similar", "extends", "contradicts", "supersedes", "caused_by", "related"
@@ -65,6 +80,27 @@ class WikiNode:
     memory_type: str = "observation"
     source: Optional[dict[str, Any]] = None  # {path, symbol, commit} for code/paper nodes
     extra: dict[str, Any] = field(default_factory=dict)
+    reinforcement_count: int = 0
+    last_reinforced_at: Optional[datetime] = None
+
+    def content_hash(self) -> str:
+        """SHA-256 over normalized body+title.
+
+        Stable under whitespace-only edits (one extra newline, trailing
+        spaces, CRLF vs LF, tab vs space, leading/trailing blanks) and
+        case changes. Changes when title or the textual body changes.
+
+        Distinct from ``metadata['source_hash']`` used by the codebase
+        ingester (that one hashes the raw source file; this one hashes
+        the rendered wiki body+title and powers reinforcement-on-dup).
+        """
+        norm_title = _normalize_for_hash(self.title or "")
+        norm_body = _normalize_for_hash(self.body or "")
+        digest = hashlib.sha256()
+        digest.update(norm_title.encode("utf-8"))
+        digest.update(b"\x00")  # separator so title|body cannot collide with body|title
+        digest.update(norm_body.encode("utf-8"))
+        return digest.hexdigest()
 
     def to_frontmatter(self) -> dict[str, Any]:
         """Serializable frontmatter dict (ordered for readability)."""
@@ -100,6 +136,10 @@ def to_frontmatter(self) -> dict[str, Any]:
             fm["metadata"] = self.metadata
         if self.extra:
             fm["extra"] = self.extra
+        if self.reinforcement_count:
+            fm["reinforcement_count"] = self.reinforcement_count
+        if self.last_reinforced_at:
+            fm["last_reinforced_at"] = self.last_reinforced_at.isoformat()
         return fm
 
     def effective_time(self) -> Optional[datetime]:
@@ -163,7 +203,23 @@ class StorageBackend(Protocol):
     async def init(self) -> None: ...
 
     # --- node CRUD ---
-    async def put_node(self, node: WikiNode) -> WikiNode: ...
+    async def put_node(
+        self,
+        node: WikiNode,
+        *,
+        fencing_token: int | None = None,
+    ) -> WikiNode:
+        """Persist ``node``.
+
+        ``fencing_token`` is an optional guard produced by
+        :class:`memu.neighborhood_lock.NeighborhoodLock`. When provided,
+        the backend must verify the token matches the currently-held
+        lock on ``node.slug`` (inside the same transaction that mutates
+        the row) and raise :class:`memu.lane_lock.FencingTokenError` on
+        mismatch. When ``None`` (default) the backend skips the check
+        so existing callers stay unchanged.
+        """
+
     async def get_node(self, ref: str) -> Optional[WikiNode]:
         """Fetch by id or slug. Backends should accept either."""
 
@@ -173,6 +229,17 @@ async def list_nodes(
         self, kind: Optional[NodeKind] = None, limit: int = 100
     ) -> list[WikiNode]: ...
 
+    async def reinforce_node(self, ref: str) -> Optional[WikiNode]:
+        """Atomically bump ``reinforcement_count`` + ``last_reinforced_at``.
+
+        Returns the updated node, or ``None`` if no node matches ``ref``.
+        Never mutates title/body/tags/links/metadata — a pure salience
+        signal for duplicate-on-write or explicit "I saw this again"
+        callers. Also invoked internally by :meth:`put_node` when the
+        incoming node's ``content_hash()`` matches the stored node's.
+        """
+        ...
+
     # --- slug registry ---
     async def register_slug(self, slug: str, node_id: str, kind: NodeKind) -> None: ...
     async def resolve_slug(self, slug: str) -> Optional[str]: ...

memu/storage/markdown_backend.py

@@ -52,10 +52,71 @@ def _all_md(self) -> list[Path]:
         return [p for p in self.layout.root.rglob("*.md") if "_index" not in p.parts]
 
     # ----------------------------------------------------------------- CRUD
-    async def put_node(self, node: WikiNode) -> WikiNode:
-        node.updated_at = datetime.now(timezone.utc)
+    async def put_node(
+        self,
+        node: WikiNode,
+        *,
+        fencing_token: int | None = None,
+    ) -> WikiNode:
+        # TODO(neighborhood-lock): enforce fencing_token against
+        # ``.memu/locks/_fence/<slug>`` (see NEIGHBORHOOD_LOCK_DESIGN §5).
+        # The token is accepted-and-ignored for now so callers can
+        # thread it uniformly across tiers; SQLite is the tier that
+        # actively enforces it in this PR.
+        del fencing_token
+        now = datetime.now(timezone.utc)
+        # Reinforcement-on-duplicate: same slug + same content_hash() =>
+        # bump the counter instead of rewriting. Preserves the original
+        # ``created_at`` and body bytes. Different body resets the
+        # counter (a new generation of the memory).
+        existing = await self._find_by_slug(node.slug)
+        if existing is not None and existing.content_hash() == node.content_hash():
+            # Reinforcement preserves title+body (that's what equivalence
+            # means) but still refreshes side-channel fields the caller
+            # may have set on the incoming node (``happened_at``,
+            # ``metadata``, ``extra``, ``tags``, ``source``, ``agent_id``,
+            # etc.). Without this, callers who re-put a known-good node
+            # after enriching its metadata would silently lose the
+            # enrichment.
+            existing.reinforcement_count = (existing.reinforcement_count or 0) + 1
+            existing.last_reinforced_at = now
+            existing.updated_at = now
+            existing.tags = list(node.tags)
+            existing.metadata = dict(node.metadata or {})
+            existing.extra = dict(node.extra or {})
+            existing.source = node.source
+            existing.happened_at = node.happened_at
+            existing.agent_id = node.agent_id
+            existing.memory_type = node.memory_type
+            existing.salience = node.salience
+            existing.confidence = node.confidence
+            path = self._path_for(existing)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            fm = existing.to_frontmatter()
+            body = existing.body or f"# {existing.title}\n"
+            path.write_text(dump_frontmatter(fm, body), encoding="utf-8")
+            return existing
+        if existing is not None:
+            node.reinforcement_count = 0
+            node.last_reinforced_at = None
+        node.updated_at = now
         if node.created_at is None:
-            node.created_at = node.updated_at
+            node.created_at = now
+        path = self._path_for(node)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        fm = node.to_frontmatter()
+        body = node.body or f"# {node.title}\n"
+        path.write_text(dump_frontmatter(fm, body), encoding="utf-8")
+        return node
+
+    async def reinforce_node(self, ref: str) -> Optional[WikiNode]:
+        node = await self.get_node(ref)
+        if node is None:
+            return None
+        now = datetime.now(timezone.utc)
+        node.reinforcement_count = (node.reinforcement_count or 0) + 1
+        node.last_reinforced_at = now
+        node.updated_at = now
         path = self._path_for(node)
         path.parent.mkdir(parents=True, exist_ok=True)
         fm = node.to_frontmatter()
@@ -224,6 +285,8 @@ def _read(self, path: Path) -> Optional[WikiNode]:
             memory_type=fm.get("memory_type", "observation"),
             source=fm.get("source"),
             extra=dict(fm.get("extra") or {}),
+            reinforcement_count=int(fm.get("reinforcement_count") or 0),
+            last_reinforced_at=_parse_dt(fm.get("last_reinforced_at")),
         )