feat(memory): routing-guard advisory and index-based entry locator

Add a structural detector that nudges the agent toward editing the
authoritative project file when a memory write looks like a rule or
value-assignment that belongs in a file (a recurring failure mode: a
correction rephrased as a "preference" and stored in memory while the
governing file stays stale). The guard never blocks — it only appends a
one-line advisory on add/replace — so false positives cost a sentence.

Also let replace/remove identify an entry by 0-based `index` (from
`list`) as a deterministic alternative to `old_text` substring matching;
out-of-range indices report the inventory so the retry is guided.

Slash completer now surfaces the matched alias as the menu label
(`/res` -> `/resume`) while keeping name matches ranked above alias-only
matches.

memory.md guidance updated for authoritative-files-first and the new
index-based locator.

Author: elkaix

src/pythinker_code/project_memory.py

@@ -275,11 +275,28 @@ def _match_one(entries: list[str], old_text: str) -> int | MemoryOpResult:
             )
         return matches[0]
 
-    async def replace(self, target: Target, old_text: str, new_content: str) -> MemoryOpResult:
+    def _locate(self, entries: list[str], old_text: str, index: int | None) -> int | MemoryOpResult:
+        """Resolve which entry to mutate. ``index`` (0-based, from `list`) is the
+        deterministic path — preferred when substring matching is uncertain;
+        ``old_text`` is the substring fallback. Out-of-range indices report the
+        inventory so the retry is guided, not a guess."""
+        if index is not None:
+            if not 0 <= index < len(entries):
+                return MemoryOpResult(
+                    False,
+                    f"No entry at index {index} ({len(entries)} stored). "
+                    f"Current entries:\n{self._inventory(entries)}",
+                )
+            return index
+        if old_text:
+            return self._match_one(entries, old_text)
+        return MemoryOpResult(False, "Provide old_text or index to identify the entry.")
+
+    async def replace(
+        self, target: Target, old_text: str, new_content: str, *, index: int | None = None
+    ) -> MemoryOpResult:
         old_text = old_text.strip()
         new_content = new_content.strip()
-        if not old_text:
-            return MemoryOpResult(False, "old_text cannot be empty.")
         if not new_content:
             return MemoryOpResult(False, "new_content cannot be empty. Use 'remove' to delete.")
         blocked = scan_memory_content(new_content)
@@ -293,7 +310,7 @@ async def replace(self, target: Target, old_text: str, new_content: str) -> Memo
                 return MemoryOpResult(
                     False, f"Memory read failed ({exc}); aborting write to avoid data loss."
                 )
-            idx = self._match_one(entries, old_text)
+            idx = self._locate(entries, old_text, index)
             if isinstance(idx, MemoryOpResult):
                 return idx
             limit = self._char_limit(target)
@@ -312,10 +329,10 @@ async def replace(self, target: Target, old_text: str, new_content: str) -> Memo
             await self._write_entries(target, candidate)
         return MemoryOpResult(True, "Entry replaced.")
 
-    async def remove(self, target: Target, old_text: str) -> MemoryOpResult:
+    async def remove(
+        self, target: Target, old_text: str, *, index: int | None = None
+    ) -> MemoryOpResult:
         old_text = old_text.strip()
-        if not old_text:
-            return MemoryOpResult(False, "old_text cannot be empty.")
         path = await self._path_for(target)
         async with self._async_lock, self._file_lock(path):
             try:
@@ -324,7 +341,7 @@ async def remove(self, target: Target, old_text: str) -> MemoryOpResult:
                 return MemoryOpResult(
                     False, f"Memory read failed ({exc}); aborting write to avoid data loss."
                 )
-            idx = self._match_one(entries, old_text)
+            idx = self._locate(entries, old_text, index)
             if isinstance(idx, MemoryOpResult):
                 return idx
             entries.pop(idx)

src/pythinker_code/tools/memory/__init__.py

@@ -6,7 +6,20 @@
 
 from pythinker_code.project_memory import ProjectMemoryStore
 from pythinker_code.soul.agent import Runtime
+from pythinker_code.tools.memory.routing_guard import routing_signals
 from pythinker_code.tools.utils import load_desc
+from pythinker_code.utils.logging import logger
+
+# Appended (never substituted) to a successful add/replace whose content looks
+# like it governs behavior defined in a project file. The tool is blind to
+# conversation context, so it can only nudge — enforcement lives in memory.md
+# guidance and the model's own awareness of the correction it just received.
+_ROUTING_ADVISORY = (
+    "\n\nNote: this looks like it may set a rule that lives in a project file "
+    "(a protocol, spec, or config you read). If so, edit that file directly — "
+    "memory does not change files you follow, so the change would be silently "
+    "lost. Memory is only for durable facts with no authoritative home."
+)
 
 
 class Params(BaseModel):
@@ -16,6 +29,11 @@ class Params(BaseModel):
     old_text: str | None = Field(
         default=None, description="Unique substring identifying the entry to replace/remove."
     )
+    index: int | None = Field(
+        default=None,
+        description="0-based index of the entry to replace/remove (from `list`). "
+        "Deterministic alternative to old_text — prefer it when a substring match is uncertain.",
+    )
 
 
 class Memory(CallableTool2[Params]):
@@ -38,9 +56,14 @@ async def __call__(self, params: Params) -> ToolReturnValue:
             return ToolError(
                 message="content is required for add/replace.", brief="memory: no content"
             )
-        if params.action in ("replace", "remove") and not (params.old_text or "").strip():
+        if (
+            params.action in ("replace", "remove")
+            and not (params.old_text or "").strip()
+            and params.index is None
+        ):
             return ToolError(
-                message="old_text is required for replace/remove.", brief="memory: no old_text"
+                message="old_text or index is required for replace/remove.",
+                brief="memory: no locator",
             )
 
         if params.action == "list":
@@ -51,10 +74,12 @@ async def __call__(self, params: Params) -> ToolReturnValue:
             result = await self._store.add(params.target, params.content or "")
         elif params.action == "replace":
             result = await self._store.replace(
-                params.target, params.old_text or "", params.content or ""
+                params.target, params.old_text or "", params.content or "", index=params.index
             )
         else:
-            result = await self._store.remove(params.target, params.old_text or "")
+            result = await self._store.remove(
+                params.target, params.old_text or "", index=params.index
+            )
 
         if not result.ok:
             message = result.message
@@ -72,4 +97,18 @@ async def __call__(self, params: Params) -> ToolReturnValue:
         rearm = getattr(self._runtime, "rearm_injection", None)
         if rearm is not None:
             rearm("project_memory")
-        return ToolOk(output=result.message, message=result.message, brief=params.action)
+        message = result.message
+        if params.action in ("add", "replace"):
+            signals = routing_signals(params.content or "")
+            if signals:
+                logger.bind(
+                    event="memory_guard_advisory",
+                    action=params.action,
+                    target=params.target,
+                    signals=signals,
+                ).debug(
+                    "memory write tripped routing advisory: {preview}",
+                    preview=(params.content or "")[:60],
+                )
+                message += _ROUTING_ADVISORY
+        return ToolOk(output=message, message=message, brief=params.action)

src/pythinker_code/tools/memory/memory.md

@@ -5,20 +5,33 @@ WRITE FACTS, NOT INSTRUCTIONS:
 - "Project uses pytest with xdist" ✓ — "Run tests with pytest -n 4" ✗
 - "User prefers concise answers" ✓ — "Always respond concisely" ✗
 
+AUTHORITATIVE FILES FIRST:
+If a fact would change how you already behave according to a file you read (AGENTS.md, a
+protocol/spec/config), EDIT that file — do not store it here. Memory does not change files
+you follow, so the change would be silently lost. Rephrasing a correction as a preference
+("set the limit to 100" → "user prefers a 100-word limit") does NOT turn a file edit into a
+memory fact. When the user corrects a rule, find and edit the file that governs it first.
+
 TWO TARGETS:
 - `memory`: project facts — conventions, architecture notes, gotchas, key file locations.
 - `user`: how the user likes to work in this repo — preferences, style, do/don't.
 
 DO NOT store: task progress, session outcomes, "fixed bug X / merged PR Y /
 Phase N done", PR/issue numbers, commit SHAs, file counts, or anything stale
 within a week. Those are transient and belong in the session journal, not memory.
-NEVER store secrets, tokens, or credentials.
+Store only non-obvious facts — deviations, gotchas, conventions — not baseline
+behavior any agent would assume. Store rules the user actually stated, never ones
+you inferred from conversation fragments. NEVER store secrets, tokens, or credentials.
 
 ACTIONS:
 - `add`: append a new entry (requires `content`).
-- `replace`: update an entry (`old_text` = a unique substring of the target entry; `content` = new text).
-- `remove`: delete an entry (`old_text` = a unique substring of the target entry).
-- `list`: show current entries with their sizes and how much room is free (read-only).
+- `replace`: update an entry — identify it by `index` or `old_text`; `content` = new text.
+- `remove`: delete an entry — identify it by `index` or `old_text`.
+- `list`: show current entries, each with its `[index]`, size, and how much room is free (read-only).
+
+IDENTIFYING AN ENTRY: prefer `index` (the `[N]` shown by `list`) — it is exact. Use `old_text`
+(a unique substring) only when you are sure of the literal stored text. If a remove/replace
+fails to match, run `list` and retry with the `index` rather than guessing the substring again.
 
 WHEN A WRITE IS REJECTED FOR SPACE:
 The error reports the exact free budget and lists existing entries. Do NOT retry the

src/pythinker_code/tools/memory/routing_guard.py

@@ -0,0 +1,61 @@
+"""Structural detector for memory writes that probably belong in a file.
+
+A recurring failure mode: the user *corrects* a rule that lives in an
+authoritative project file (a protocol/spec/config the agent reads), and the
+agent — to satisfy the "write facts, not instructions" rule — rephrases the
+correction as a "preference" and stores it in memory. The file stays stale, the
+correction is silently lost, and memory bloats with directives.
+
+The tool is blind to conversation context, so it cannot *know* whether a
+governing file exists. It can only spot the structural shape of a correction and
+*advise*. This module is that detector: pure, deterministic, no project
+knowledge, no I/O. The tool layer appends a one-line nudge when a signal trips;
+it never blocks, so false positives cost nothing but a sentence of output.
+"""
+
+from __future__ import annotations
+
+import re
+
+# A token that looks like a config/doc filename. High precision: if the entry
+# names such a file, that file — not memory — is the source of truth.
+_FILE_REF = re.compile(
+    r"\b[\w./-]+\.(?:md|markdown|ya?ml|json|toml|ini|cfg|conf|txt|env)\b",
+    re.IGNORECASE,
+)
+
+# Value-assignment grammar: a directive verb steering a number ("set ... to 100",
+# "raise ... 12"). Bounded gap so it stays a local phrase, not a whole paragraph.
+_DIRECTIVE_VERB = re.compile(
+    r"\b(?:set|change|update|increase|decrease|raise|lower|bump|cap|adjust)\b[^.\n]{0,40}\d+",
+    re.IGNORECASE,
+)
+
+# A limit/threshold word adjacent to a number, in either order. This is what
+# catches the rephrased-as-preference evasion ("limit of 100").
+_LIMIT_NUM = re.compile(
+    r"\b(?:limit|max(?:imum)?|min(?:imum)?|threshold|cap|quota)\b[^.\n]{0,20}\d+"
+    r"|\d+[^.\n]{0,20}\b(?:limit|max(?:imum)?|min(?:imum)?|threshold|cap|quota)\b",
+    re.IGNORECASE,
+)
+
+# Prescriptive imperatives — the grammar of a rule rather than a fact.
+_IMPERATIVE = re.compile(r"\b(?:always|never|must)\b", re.IGNORECASE)
+
+
+def routing_signals(content: str) -> list[str]:
+    """Return the structural signals tripped by ``content``, in stable order.
+
+    ``"file_ref"`` — the entry names a config/doc file (that file is the likely
+    source of truth). ``"directive"`` — the entry reads like a rule or a
+    value-assignment rather than a standing fact. Empty list means nothing
+    tripped. Order is deterministic (``file_ref`` before ``directive``) and each
+    signal appears at most once, so callers can log it verbatim.
+    """
+    text = content or ""
+    signals: list[str] = []
+    if _FILE_REF.search(text):
+        signals.append("file_ref")
+    if _DIRECTIVE_VERB.search(text) or _LIMIT_NUM.search(text) or _IMPERATIVE.search(text):
+        signals.append("directive")
+    return signals

src/pythinker_code/ui/shell/prompt.py

@@ -401,14 +401,15 @@ def get_completions(
         typed_lower = typed.lower()
         seen: set[str] = set()
 
-        def emit(cmd: SlashCommand[Any]) -> Iterable[Completion]:
+        def emit(cmd: SlashCommand[Any], label: str | None = None) -> Iterable[Completion]:
             if cmd.name in seen:
                 return
             seen.add(cmd.name)
+            shown = label or cmd.name
             yield Completion(
-                text=f"/{cmd.name}",
+                text=f"/{shown}",
                 start_position=-len(token),
-                display=f"/{cmd.name}",
+                display=f"/{shown}",
                 display_meta=self._display_meta(cmd),
             )
 
@@ -417,35 +418,39 @@ def emit(cmd: SlashCommand[Any]) -> Iterable[Completion]:
                 yield from emit(cmd)
             return
 
-        def match_tier(cmd: SlashCommand[Any]) -> int | None:
-            """Lower tier = stronger match. Name matches rank above alias matches
-            so typing toward a command name (e.g. ``/report`` → ``/reports``)
-            wins over a command that only matches via an exact alias."""
+        def match_tier(cmd: SlashCommand[Any]) -> tuple[int, str] | None:
+            """Return ``(tier, label)`` or ``None``. Lower tier = stronger match.
+            Name matches rank above alias matches so typing toward a command name
+            (e.g. ``/report`` → ``/reports``) wins over a command that only matches
+            via an exact alias. For alias-only matches the label is the matched
+            alias, so the menu surfaces what the user typed toward (e.g. ``/res``
+            → ``/resume``) rather than the differently-named command (``/sessions``)."""
             name_lower = cmd.name.lower()
             if name_lower == typed_lower:
-                return 0
+                return (0, cmd.name)
             if name_lower.startswith(typed_lower):
-                return 1
-            alias_prefix = False
+                return (1, cmd.name)
+            alias_prefix: str | None = None
             for alias in cmd.aliases:
                 alias_lower = alias.lower()
                 if alias_lower == typed_lower:
-                    return 2
-                if alias_lower.startswith(typed_lower):
-                    alias_prefix = True
-            return 3 if alias_prefix else None
+                    return (2, alias)
+                if alias_prefix is None and alias_lower.startswith(typed_lower):
+                    alias_prefix = alias
+            return (3, alias_prefix) if alias_prefix is not None else None
 
         # Rank by (match tier, command-name length, name): the closest, shortest
         # command name surfaces first within each tier.
-        matched: list[tuple[int, int, str, SlashCommand[Any]]] = []
+        matched: list[tuple[int, int, str, str, SlashCommand[Any]]] = []
         for cmd in self._available_commands:
-            tier = match_tier(cmd)
-            if tier is not None:
-                matched.append((tier, len(cmd.name), cmd.name, cmd))
+            result = match_tier(cmd)
+            if result is not None:
+                tier, label = result
+                matched.append((tier, len(cmd.name), cmd.name, label, cmd))
         matched.sort(key=lambda item: (item[0], item[1], item[2]))
 
-        for _, _, _, cmd in matched:
-            yield from emit(cmd)
+        for _, _, _, label, cmd in matched:
+            yield from emit(cmd, label)
 
     def _disabled_during_task(self, cmd: SlashCommand[Any]) -> bool:
         """True when a running turn blocks this shell-level command."""

tests/core/test_project_memory.py

@@ -239,6 +239,36 @@ async def test_remove_deletes_matching_entry(tmp_path, monkeypatch):
     assert not r.ok and "No entry matched" in r.message
 
 
+async def test_remove_by_index(tmp_path, monkeypatch):
+    """An entry can be deleted by its 0-based index (from `list`) without
+    guessing a substring — the fix for stale/imprecise old_text matches."""
+    store = _store(tmp_path, monkeypatch)
+    await store.add("memory", "uses pytest")
+    await store.add("memory", "uses ruff")
+
+    r = await store.remove("memory", "", index=0)
+    assert r.ok and await store.read_entries("memory") == ["uses ruff"]
+
+
+async def test_replace_by_index(tmp_path, monkeypatch):
+    store = _store(tmp_path, monkeypatch)
+    await store.add("memory", "uses pytest")
+    await store.add("memory", "uses ruff")
+
+    r = await store.replace("memory", "", "uses ruff + biome", index=1)
+    assert r.ok and await store.read_entries("memory") == ["uses pytest", "uses ruff + biome"]
+
+
+async def test_index_out_of_range_errors_with_inventory(tmp_path, monkeypatch):
+    store = _store(tmp_path, monkeypatch)
+    await store.add("memory", "uses pytest")
+
+    r = await store.remove("memory", "", index=5)
+    assert not r.ok
+    assert "index 5" in r.message
+    assert "uses pytest" in r.message  # inventory guides the retry
+
+
 async def test_snapshot_builds_block_with_priority_and_budget(tmp_path, monkeypatch):
     monkeypatch.setenv("PYTHINKER_SHARE_DIR", str(tmp_path / "share"))
     from pythinker_code.project_memory import ProjectMemoryStore