ver 1.5.0 — six-improvement pass: retention, routing, self-firing memory

1. capture-update keeps the whole body (was truncating to first line)
2. frontmatter completeness: kind-default status + kind-seeded tags at
   capture, fm create-or-replace, incomplete-frontmatter lint, sweep
   backfill (one data source: KIND_STATUS/KIND_REQUIRED_FM)
3. pre-write duplicate gate (exit 76, --merge-into/--force)
4. recall corpus + ranking: sessions/journal indexed (down-weighted),
   BM25 x freshness x status fusion, recall --json, index schema 2
5. session hooks: SessionStart auto-load, Stop reflection nudge
6. one-tap triage verb with processing-log audit; --json on read verbs

+22 mock end-to-end tests; all suites green. Docs current (README,
CHANGELOG, SKILL.md, commands, references, MAINTENANCE, eval README).
Drops tracked .DS_Store (gitignored); retires README-gretchen.md
(moved to gitignored dist/); adds the field report that drove item 2.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

echo-memory.plugin.src/skills/echo-memory/scripts/echo.py

@@ -40,8 +40,9 @@ Usage:
   echo.py post   <path> [file]                  # raw append (NON-idempotent; prefer `append`)
   echo.py append <path> <line>                  # idempotent append: skips only on an exact whole-line match
   echo.py patch  <path> <append|prepend|replace> <heading|frontmatter|block> <target> [file]
-  echo.py fm     <path> <field> <json-value>    # PATCH a frontmatter scalar
+  echo.py fm     <path> <field> <json-value>    # create-or-replace a frontmatter scalar
   echo.py bump   <path> [YYYY-MM-DD]            # set frontmatter updated: to today (or given date)
+  echo.py triage [--list --json | <proposals.json> [--apply]]   # one-tap inbox triage + audit log
   echo.py delete <path>                         # DELETE (destructive; explicit use only)
   echo.py lock   <owner-id>                     # acquire advisory lock (exit 75 if held & fresh, or lost the race)
   echo.py unlock <owner-id>                     # release advisory lock if owned by <owner-id>
@@ -49,7 +50,8 @@ Usage:
   echo.py scope  set "<text>"                   # switch scope atomically (history + replace + stamp scope_updated)
   echo.py config [show|init|set|import]         # machine-local owner/endpoint/key (config import <file> adopts a provided file)
 
-Exit codes: 0 ok · 44 not-found(404) · 75 lock-held/lost · 78 config-required · 2 usage · 1 other HTTP/transport error.
+Exit codes: 0 ok · 44 not-found(404) · 75 lock-held/lost · 76 duplicate-gate (capture) ·
+            78 config-required · 2 usage · 1 other HTTP/transport error.
 """
 
 from __future__ import annotations
@@ -60,6 +62,7 @@ import http.client
 import json
 import locale
 import os
+import re
 import sys
 import tempfile
 import threading
@@ -339,6 +342,14 @@ def cmd_put(path: str, file_arg: str | None) -> int:
         if status != 200:
             raise EchoError(f"put {path}: write did not verify (GET returned {status})")
     print(f"ok: PUT {path}")
+    # Keep the recall (BM25) index current when a corpus note is written directly —
+    # session logs, journal notes, hand-authored entity notes. update_note early-returns
+    # for non-corpus paths and never raises, so this can't fail or slow a normal put.
+    try:
+        import echo_recall
+        echo_recall.update_note(path, data.decode("utf-8", errors="replace"))
+    except Exception as exc:  # noqa: BLE001 — index upkeep must never fail a put
+        print(f"echo.py: recall-index update skipped ({exc})", file=sys.stderr)
     return 0
 
 
@@ -402,8 +413,64 @@ def cmd_patch(path: str, op: str, target_type: str, target: str, file_arg: str |
     return 0
 
 
+def _yaml_flow(value) -> str:
+    """Render a JSON-decoded value as a YAML flow scalar/list for one frontmatter line."""
+    if isinstance(value, list):
+        return "[" + ", ".join(_yaml_flow(v) for v in value) + "]"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if value is None:
+        return ""
+    s = str(value)
+    return s if re.match(r"^[A-Za-z0-9][A-Za-z0-9 ._/-]*$", s) else json.dumps(s)
+
+
+def _fm_upsert_line(text: str, field: str, rendered: str) -> str:
+    """Surgically set `field: rendered` inside the YAML frontmatter block, creating the
+    key (or the whole block) if absent. Every other line is preserved verbatim — this is
+    the safety property that makes it OK to fall back from a failed PATCH."""
+    line = f"{field}: {rendered}".rstrip()
+    lines = text.splitlines()
+    if lines and lines[0].strip() == "---":
+        for end in range(1, len(lines)):
+            if lines[end].strip() == "---":
+                for i in range(1, end):  # key exists -> replace that one line
+                    if re.match(rf"^{re.escape(field)}\s*:", lines[i]):
+                        lines[i] = line
+                        break
+                else:
+                    lines.insert(end, line)  # else append just before the closing ---
+                return "\n".join(lines) + ("\n" if text.endswith("\n") else "")
+    return f"---\n{line}\n---\n{text}"  # no frontmatter at all -> prepend a minimal block
+
+
 def cmd_fm(path: str, field: str, value: str) -> int:
-    return cmd_patch(path, "replace", "frontmatter", field, temp_file(normalize_json_scalar(value)))
+    """Create-or-replace a frontmatter scalar. PATCH replace handles the common case;
+    a missing key returns 400 invalid-target (PATCH cannot create), which used to make
+    `fm` fail on exactly the notes that needed repair — now it falls back to a surgical
+    GET -> insert-one-line -> verified PUT."""
+    data = normalize_json_scalar(value)
+    try:
+        return cmd_patch(path, "replace", "frontmatter", field, temp_file(data))
+    except EchoError as exc:
+        if "HTTP 400" not in str(exc):
+            raise
+    status, body = request("GET", vault_url(path))
+    check(status, body, f"fm {path}")
+    new = _fm_upsert_line(body.decode(errors="replace"), field,
+                          _yaml_flow(json.loads(data.decode("utf-8"))))
+    status, body = _qwrite("PUT", vault_url(path), data=new.encode("utf-8"),
+                           headers={"Content-Type": "text/markdown"})
+    if status == 202:
+        print(f"queued (offline): FM {field} -> {path} — will sync on the next reachable session")
+        return 0
+    check(status, body, f"fm {path}")
+    if VERIFY:
+        status, body = request("GET", vault_url(path))
+        if status != 200 or not re.search(rf"(?m)^{re.escape(field)}\s*:", body.decode(errors="replace")):
+            raise EchoError(f"fm {path}: created key '{field}' did not verify on read-back")
+    print(f"ok: FM created '{field}' in {path}")
+    return 0
 
 
 def cmd_delete(path: str) -> int:
@@ -482,31 +549,39 @@ def extract_heading(markdown: str, heading: str) -> str:
     return "\n".join(out).strip()
 
 
-def cmd_scope(subcommand: str, text: str | None = None) -> int:
+def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -> int:
     path = "_agent/context/current-context.md"
     status, body = request("GET", vault_url(path))
     check(status, body, f"scope {subcommand}")
     current = body.decode(errors="replace")
 
     if subcommand == "show":
-        print("-- Active scope --")
-        print(extract_heading(current, "Scope"))
+        scope_text = extract_heading(current, "Scope")
         scope_updated = ""
         for line in current.splitlines():
             if line.startswith("scope_updated:"):
                 scope_updated = line.split(":", 1)[1].strip().strip('"').strip("'")
                 break
-        print(f"scope_updated: {scope_updated or '<missing — drift cannot be detected; run scope set or repair>'}")
+        sessions_since = None
         # Count session logs dated after scope_updated (the drift signal).
         if scope_updated:
             status, body = request("GET", vault_url("_agent/sessions/"))
             if status == 200:
                 try:
                     files = json.loads(body).get("files", [])
-                    n = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
-                    print(f"sessions logged since: {n}")
+                    sessions_since = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
                 except json.JSONDecodeError:
                     pass
+        if as_json:
+            print(json.dumps({"ok": True, "action": "scope-show", "scope": scope_text,
+                              "scope_updated": scope_updated or None,
+                              "sessions_since": sessions_since}, ensure_ascii=False))
+            return 0
+        print("-- Active scope --")
+        print(scope_text)
+        print(f"scope_updated: {scope_updated or '<missing — drift cannot be detected; run scope set or repair>'}")
+        if sessions_since is not None:
+            print(f"sessions logged since: {sessions_since}")
         return 0
 
     if subcommand != "set":
@@ -682,14 +757,22 @@ def build_parser() -> argparse.ArgumentParser:
     p = sub.add_parser("scope")
     p.add_argument("subcommand", nargs="?", default="show", choices=["show", "set"])
     p.add_argument("text", nargs="?")
+    p.add_argument("--json", action="store_true")        # structured scope + freshness
     p = sub.add_parser("reflect")  # H5: apply a JSON proposal set (dry-run unless --apply)
     p.add_argument("file", nargs="?")
     p.add_argument("--apply", action="store_true")
+    p = sub.add_parser("triage")   # one-tap inbox triage (reflect pipeline + audit log)
+    p.add_argument("file", nargs="?")                    # proposals JSON; omit to list
+    p.add_argument("--list", action="store_true", dest="list_inbox")
+    p.add_argument("--json", action="store_true")        # structured inbox listing
+    p.add_argument("--apply", action="store_true")       # route + write processing log
 
     # High-level operations (entity index + cross-links + recall). See echo_ops.py.
     sub.add_parser("resolve").add_argument("mention", nargs="+")
     p = sub.add_parser("recall"); p.add_argument("query", nargs="+"); p.add_argument("--limit", type=int, default=6)
+    p.add_argument("--json", action="store_true")        # structured hits + neighbourhood
     p = sub.add_parser("link"); p.add_argument("a"); p.add_argument("b")
+    p.add_argument("--json", action="store_true")
     p = sub.add_parser("capture")
     p.add_argument("title")
     p.add_argument("file", nargs="?")
@@ -698,11 +781,14 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--status", default="")
     p.add_argument("--aliases", default="")
     p.add_argument("--source", default="")
+    p.add_argument("--tags", default="")                  # extra tags; the kind is always seeded
     p.add_argument("--date")
     p.add_argument("--domain", default="business")
     p.add_argument("--no-log", action="store_true")
     p.add_argument("--json", action="store_true")       # M4: emit a JSON result envelope
     p.add_argument("--dry-run", action="store_true")     # M4: preview the plan, write nothing
+    p.add_argument("--force", action="store_true")       # create despite a duplicate-gate hit
+    p.add_argument("--merge-into", metavar="SLUG")       # route as an update to this entity
     return parser
 
 
@@ -755,7 +841,7 @@ def main(argv: list[str] | None = None) -> int:
         if args.cmd == "unlock":
             return cmd_unlock(args.owner)
         if args.cmd == "scope":
-            return cmd_scope(args.subcommand, args.text)
+            return cmd_scope(args.subcommand, args.text, as_json=args.json)
         if args.cmd == "reflect":
             import echo_reflect
             raw = read_body(args.file).decode("utf-8", errors="replace")
@@ -766,20 +852,34 @@ def main(argv: list[str] | None = None) -> int:
             if not isinstance(proposals, list):
                 raise EchoError("reflect: proposals must be a JSON array of objects", 2)
             return echo_reflect.apply(proposals, confirm=args.apply)
+        if args.cmd == "triage":
+            import echo_triage
+            if args.list_inbox or not args.file:
+                return echo_triage.list_inbox(as_json=args.json)
+            raw = read_body(args.file).decode("utf-8", errors="replace")
+            try:
+                proposals = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise EchoError(f"triage: proposals must be a JSON array ({exc})", 2)
+            if not isinstance(proposals, list):
+                raise EchoError("triage: proposals must be a JSON array of objects", 2)
+            return echo_triage.apply(proposals, confirm=args.apply)
         if args.cmd in ("resolve", "recall", "link", "capture"):
             import echo_ops  # lazy: only the high-level ops pull in the index/link modules
             if args.cmd == "resolve":
                 return echo_ops.resolve(" ".join(args.mention))
             if args.cmd == "recall":
-                return echo_ops.recall(args.query, limit=args.limit)
+                return echo_ops.recall(args.query, limit=args.limit, as_json=args.json)
             if args.cmd == "link":
-                return echo_ops.link(args.a, args.b)
+                return echo_ops.link(args.a, args.b, as_json=args.json)
             return echo_ops.capture(
                 args.kind, args.title, args.file, status_v=args.status,
                 aliases=args.aliases.split(",") if args.aliases else [],
                 sources=args.source.split(",") if args.source else [],
+                tags=args.tags.split(",") if args.tags else [],
                 date=args.date, domain=args.domain, inbox=args.inbox, no_log=args.no_log,
-                as_json=args.json, dry_run=args.dry_run)
+                as_json=args.json, dry_run=args.dry_run,
+                force=args.force, merge_into=args.merge_into)
     except EchoError as exc:
         print(f"echo.py: {exc}", file=sys.stderr)
         return exc.code

echo-memory.plugin.src/skills/echo-memory/scripts/echo_hook_session_start.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""echo_hook_session_start.py — SessionStart hook: self-firing cold-start load.
+
+The two behaviours the skill cares most about (load at session start, reflect at
+session end) used to depend entirely on the model remembering to do them — scope
+drift and the inbox backlog are the residue of that discipline failing. This hook
+makes loading deterministic: on session start it runs the canonical `echo.py load`
+and hands the output to the model as additionalContext.
+
+Contract (a hook must NEVER break a session):
+  * always exits 0 — any failure degrades to a one-line note or to silence;
+  * fires only on `startup` / `clear` (resume & compact already carry context);
+  * NOT CONFIGURED (exit 78) becomes a short instruction to run the first-run flow,
+    not an error;
+  * output is capped so a huge vault can't blow up the context window.
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+MAX_CONTEXT = 16000  # chars of load output forwarded into the session context
+
+
+def main() -> int:
+    try:
+        payload = json.load(sys.stdin)
+    except Exception:  # noqa: BLE001
+        payload = {}
+    if payload.get("source", "startup") not in ("startup", "clear"):
+        return 0
+
+    header = ("[echo-memory] Cold-start memory load (SessionStart hook). Reconcile per "
+              "the echo-memory skill: check inbox depth and confirm the scope still "
+              "matches this session's request before working.\n\n")
+    buf = io.StringIO()
+    try:
+        import echo
+        with contextlib.redirect_stdout(buf):
+            rc = echo.cmd_load()
+        text = buf.getvalue()
+        if rc == 78:
+            context = ("[echo-memory] ECHO is NOT CONFIGURED on this machine. Before "
+                       "doing memory work, ask the operator for their echo-memory config "
+                       "(owner/endpoint/key) and install it via `echo.py config import "
+                       "<file>` — see the skill's First-run section.")
+        elif rc == 0 and text.strip():
+            if len(text) > MAX_CONTEXT:
+                text = text[:MAX_CONTEXT] + "\n[... load output truncated by the hook ...]"
+            context = header + text
+        else:
+            context = ("[echo-memory] `echo.py load` returned nothing usable "
+                       f"(exit {rc}) — run /echo-doctor if memory seems unavailable.")
+    except Exception as exc:  # noqa: BLE001 — never break session start
+        context = (f"[echo-memory] SessionStart load skipped ({exc}). The vault may be "
+                   "unreachable; proceed without memory and mention it once if relevant.")
+
+    print(json.dumps({"hookSpecificOutput": {
+        "hookEventName": "SessionStart",
+        "additionalContext": context,
+    }}, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

echo-memory.plugin.src/skills/echo-memory/scripts/echo_hook_stop.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""echo_hook_stop.py — Stop hook: one-shot session-end reflection nudge.
+
+Reflection (/echo-reflect) only captures memory if someone remembers to run it.
+This hook watches the first Stop of a *substantive* session and, when nothing has
+been reflected or session-logged yet, blocks once with a reminder so the model
+runs the reflect flow (which still previews and asks the operator before writing
+— the safety gate lives in reflect, not here).
+
+Guards (a hook must NEVER trap a session in a loop or nag):
+  * `stop_hook_active` -> exit immediately (loop guard);
+  * fires at most ONCE per session (marker file under ECHO_STATE_DIR/hook-state);
+  * skips quiet sessions (< ECHO_REFLECT_MIN_TURNS user turns, default 5);
+  * skips when the transcript shows reflection / a session log already happened;
+  * skips when ECHO isn't configured on this machine;
+  * always exits 0.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+MIN_TURNS = int(os.environ.get("ECHO_REFLECT_MIN_TURNS", "5"))
+
+REASON = (
+    "[echo-memory] Session-end reflection has not run yet. If this session produced "
+    "durable facts, decisions, commitments, or artifacts worth remembering: run the "
+    "/echo-reflect flow (extract -> preview -> apply only with the operator's confirm) "
+    "and write the session log + heartbeat per the echo-memory skill. If nothing "
+    "durable emerged, simply finish — do not invent memories. "
+    "(This reminder fires once per session.)"
+)
+
+
+def state_marker(session_id: str) -> Path:
+    base = Path(os.environ.get("ECHO_STATE_DIR") or (Path.home() / ".echo-memory"))
+    return base / "hook-state" / f"{session_id}.reflect-nudged"
+
+
+def count_user_turns(raw: str) -> int:
+    turns = 0
+    for line in raw.splitlines():
+        try:
+            rec = json.loads(line)
+        except Exception:  # noqa: BLE001
+            continue
+        if rec.get("type") != "user":
+            continue
+        content = (rec.get("message") or {}).get("content")
+        if isinstance(content, str) and content.strip():
+            turns += 1
+        elif isinstance(content, list) and any(
+                isinstance(b, dict) and b.get("type") == "text" for b in content):
+            turns += 1
+    return turns
+
+
+def already_reflected(raw: str) -> bool:
+    """Transcript evidence that reflection or session logging already happened."""
+    return ("/echo-reflect" in raw
+            or re.search(r'echo\.py"?\s+reflect\b', raw) is not None
+            or "put _agent/sessions/" in raw
+            or "put _agent/heartbeat/last-session.md" in raw)
+
+
+def main() -> int:
+    try:
+        payload = json.load(sys.stdin)
+    except Exception:  # noqa: BLE001
+        return 0
+    if payload.get("stop_hook_active"):
+        return 0
+    session_id = str(payload.get("session_id") or "unknown")
+    marker = state_marker(session_id)
+    if marker.exists():
+        return 0
+
+    try:
+        import echo_config
+        if not echo_config.is_configured(echo_config.load()):
+            return 0  # nothing to reflect into — stay silent
+    except Exception:  # noqa: BLE001
+        return 0
+
+    raw = ""
+    tp = payload.get("transcript_path")
+    if tp:
+        try:
+            raw = Path(tp).read_text(encoding="utf-8", errors="replace")
+        except Exception:  # noqa: BLE001
+            raw = ""
+
+    if already_reflected(raw):
+        try:  # done for this session — never fire
+            marker.parent.mkdir(parents=True, exist_ok=True)
+            marker.write_text("reflected\n", encoding="utf-8")
+        except Exception:  # noqa: BLE001
+            pass
+        return 0
+    if count_user_turns(raw) < MIN_TURNS:
+        return 0  # quiet so far; a later stop may qualify (no marker written)
+
+    try:
+        marker.parent.mkdir(parents=True, exist_ok=True)
+        marker.write_text("nudged\n", encoding="utf-8")
+    except Exception:  # noqa: BLE001
+        pass  # if the marker can't be written, still nudge; stop_hook_active guards loops
+    print(json.dumps({"decision": "block", "reason": REASON}, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

echo-memory.plugin.src/skills/echo-memory/scripts/echo_index.py

@@ -55,6 +55,31 @@ KIND_TYPE = {
 DATE_KINDS = {"meeting", "decision"}          # filename carries a YYYY-MM-DD prefix
 AREA_DOMAINS = ("business", "personal", "learning", "systems")
 
+# kind -> default `status:` stamped by capture when none is given. Every entity note is
+# born with a status — a missing key is the #1 source of frontmatter drift (18/71 notes
+# in the field audit). Records of past events default to done; living notes to active.
+KIND_STATUS = {
+    "person": "active", "company": "active", "concept": "active",
+    "reference": "active", "meeting": "done", "project": "active",
+    "area": "active", "semantic": "active", "episodic": "done",
+    "working": "active", "skill": "active", "decision": "done",
+}
+
+# kind -> frontmatter fields that must be present AND non-empty on that kind's notes.
+# Data-driven so vault_lint's incomplete-frontmatter check and sweep's backfill stay in
+# lockstep with what capture stamps; add a (kind, field) pair here and all three follow.
+KIND_REQUIRED_FM = {
+    "person":    ("status", "tags"),
+    "company":   ("status", "tags"),
+    "concept":   ("status", "tags"),
+    "reference": ("status", "tags"),
+    "project":   ("status", "tags"),
+    "area":      ("status", "tags"),
+    "skill":     ("status", "tags"),
+    "meeting":   ("tags",),
+    "decision":  ("tags",),
+}
+
 
 def slugify(text: str) -> str:
     text = str(text).strip().lower()

echo-memory.plugin.src/skills/echo-memory/scripts/echo_ops.py

@@ -12,6 +12,7 @@ Network goes through echo.py; routing/aliases through echo_index; links through
 from __future__ import annotations
 
 import json
+import os
 import re
 import sys
 from pathlib import Path
@@ -21,6 +22,11 @@ import echo          # noqa: E402
 import echo_index as idx_mod   # noqa: E402
 import echo_links as links     # noqa: E402
 
+# A fuzzy candidate scoring at/above this blocks `capture` from creating what is very
+# likely a duplicate (require --force to create anyway, or --merge-into to update the
+# existing entity). Below the gate, candidates are surfaced as a warning only.
+DUP_GATE = float(os.environ.get("ECHO_DUP_GATE", "0.6"))
+
 # Existing-entity capture appends a dated bullet under this per-kind heading.
 LOG_HEADING = {
     "project": "Session History", "person": "Log", "company": "Log",
@@ -53,19 +59,25 @@ def resolve(mention: str) -> int:
 
 
 # ------------------------------------------------------------------- link -----
-def link(a_path: str, b_path: str) -> int:
+def link(a_path: str, b_path: str, as_json: bool = False) -> int:
     a_changed, b_changed = links.link_bidirectional(a_path, b_path)
+    if as_json:
+        import echo_output
+        env = echo_output.envelope("link", {"a": a_path, "b": b_path,
+                                            "a_changed": a_changed, "b_changed": b_changed})
+        print(json.dumps(env, ensure_ascii=False))
+        return 0
     print(f"ok: linked {a_path} <-> {b_path} "
           f"(added: {'A' if a_changed else '-'}{'B' if b_changed else '-'})")
     return 0
 
 
 # ----------------------------------------------------------------- recall -----
-def recall(query, limit: int = 8) -> int:
+def recall(query, limit: int = 8, as_json: bool = False) -> int:
     """Hybrid lexical (BM25) + graph recall — implemented in echo_recall. Kept here as
     the stable public entrypoint (echo.py and /echo-recall call echo_ops.recall)."""
     import echo_recall  # lazy: only pulls the BM25 modules when recall is actually run
-    return echo_recall.recall(query, limit=limit)
+    return echo_recall.recall(query, limit=limit, as_json=as_json)
 
 
 # ----------------------------------------------------------- agent log -------
@@ -101,12 +113,16 @@ def ensure_daily_log(line: str) -> None:
 
 # ----------------------------------------------------------------- capture ---
 def _build_note(fm_type: str, status_v: str, today_s: str, title: str,
-                body_text: str, sources: list[str], aliases: list[str] | None = None) -> str:
+                body_text: str, sources: list[str], aliases: list[str] | None = None,
+                tags: list[str] | None = None) -> str:
     src = "[" + ", ".join(json.dumps(s) for s in sources) + "]"
     fm = ["---", f"type: {fm_type}"]
     if status_v:
         fm.append(f"status: {status_v}")
-    fm += [f"created: {today_s}", f"updated: {today_s}", "tags: []"]
+    # Never scaffold an empty `tags: []` — an entity note is born complete (the caller
+    # seeds the kind as a baseline tag), so the incomplete-frontmatter lint stays quiet.
+    tags_v = "[" + ", ".join(tags or []) + "]"
+    fm += [f"created: {today_s}", f"updated: {today_s}", f"tags: {tags_v}"]
     if aliases:
         # Obsidian-native, durable home for aliases; sweep folds these back into the index.
         fm.append("aliases: [" + ", ".join(json.dumps(a) for a in aliases) + "]")
@@ -118,6 +134,19 @@ def _build_note(fm_type: str, status_v: str, today_s: str, title: str,
     return out
 
 
+def _dated_block(today_s: str, body_text: str) -> tuple[str, str]:
+    """Render a capture body as a dated log entry that PRESERVES the whole body:
+    the first line rides on the bullet, every further line is indented under it as
+    a markdown continuation. Returns (bullet, full_block) — the bullet alone is the
+    idempotency key (same first-line-per-day semantics as before)."""
+    lines = [ln.rstrip() for ln in body_text.strip().splitlines()]
+    bullet = f"- {today_s}: {lines[0] if lines else '(update)'}"
+    block = bullet
+    for ln in lines[1:]:
+        block += "\n" + (f"  {ln}" if ln else "")
+    return bullet, block
+
+
 def _append_to_existing(path: str, kind: str, today_s: str, body_text: str) -> None:
     text = links.get_text(path) or ""
     h1 = links.first_h1(text) or path.rsplit("/", 1)[-1][:-3]
@@ -125,22 +154,22 @@ def _append_to_existing(path: str, kind: str, today_s: str, body_text: str) -> N
     if not re.search(rf"(?m)^##\s+{re.escape(heading)}\s*$", text):
         echo.request("POST", echo.vault_url(path), data=f"\n\n## {heading}\n".encode(),
                      headers={"Content-Type": "text/markdown"})
-    firstline = body_text.strip().splitlines()[0] if body_text.strip() else "(update)"
-    bullet = f"- {today_s}: {firstline}"
+    bullet, block = _dated_block(today_s, body_text)
     status, body = echo.request("GET", echo.vault_url(path))
     if status == 200 and bullet in body.decode(errors="replace"):
         return
     echo.request("PATCH", echo.vault_url(path),
-                 data=echo.normalize_patch_body((bullet + "\n").encode(), "append", "heading"),
+                 data=echo.normalize_patch_body((block + "\n").encode(), "append", "heading"),
                  headers={"Operation": "append", "Target-Type": "heading",
                           "Target": f"{h1}::{heading}", "Content-Type": "text/markdown"})
     echo.cmd_fm(path, "updated", json.dumps(today_s))
 
 
 def capture(kind: str | None, title: str, file_arg: str | None, status_v: str = "",
-            aliases=None, sources=None, date: str | None = None, domain: str = "business",
-            inbox: bool = False, no_log: bool = False, as_json: bool = False,
-            dry_run: bool = False) -> int:
+            aliases=None, sources=None, tags=None, date: str | None = None,
+            domain: str = "business", inbox: bool = False, no_log: bool = False,
+            as_json: bool = False, dry_run: bool = False, force: bool = False,
+            merge_into: str | None = None) -> int:
     import contextlib
     import io
     import echo_output
@@ -170,6 +199,7 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
     today_s = echo.today()
     aliases = [a.strip() for a in (aliases or []) if a.strip()]
     sources = [s.strip() for s in (sources or []) if s.strip()]
+    tags = [t.strip() for t in (tags or []) if t.strip()]
 
     # Unknown home -> defer to the inbox (a single idempotent capture line).
     if inbox or not kind:
@@ -185,15 +215,55 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
     slug = idx_mod.slugify(title)
     index = idx_mod.load()
     match_slug, existing = idx_mod.resolve(index, title)
+    # --merge-into: the operator has already identified the canonical entity (e.g. after
+    # a duplicate-gate stop) — route this capture as an UPDATE to it, whatever the title.
+    if merge_into and not existing:
+        match_slug, existing = idx_mod.resolve(index, merge_into)
+        if not existing:
+            print(f"echo_ops: --merge-into '{merge_into}' matches no entity in the index "
+                  f"(try `resolve` first)", file=sys.stderr)
+            return 2
     existing_reachable = bool(existing and echo.request("GET", echo.vault_url(existing["path"]))[0] == 200)
 
+    # Pre-write duplicate gate: a strong fuzzy candidate means this title is very likely
+    # an existing entity under a different name. STOP before creating the duplicate —
+    # after the fact, the warning arrives too late (the parallel note already exists).
+    gate_hits = []
+    if not existing_reachable and not force and not dry_run:
+        gate_hits = [{"slug": s, "path": c.get("path"), "title": c.get("title"),
+                      "kind": c.get("kind"), "score": sc}
+                     for s, c, sc in idx_mod.fuzzy_candidates(index, title)
+                     if sc >= DUP_GATE]
+    if gate_hits:
+        if as_json:
+            env = echo_output.envelope("duplicate-gate", {
+                "kind": kind, "title": title, "candidates": gate_hits,
+                "note": "likely duplicate — re-run with --merge-into <slug> to update the "
+                        "existing entity, or --force to create anyway"}, ok=False)
+            print(json.dumps(env, ensure_ascii=False), file=real_stdout)
+        else:
+            print(f"STOP: '{title}' likely already exists — not creating a duplicate.",
+                  file=real_stdout)
+            for c in gate_hits:
+                print(f"  candidate: {c['slug']} -> {c['path']} (score {c['score']})",
+                      file=real_stdout)
+            print("  re-run with --merge-into <slug> to update the existing entity, "
+                  "or --force to create anyway.", file=real_stdout)
+        return 76  # distinct exit: duplicate gate (cf. 75 lock, 78 config)
+
     if dry_run:
         if existing_reachable:
             return done("update", existing["path"], dry=True)
         s2 = echo_quality.safe_slug(set(index.get("entities", {}).keys()), slug)
-        near = [c.get("path") for _, c, _ in idx_mod.fuzzy_candidates(index, title)][:3]
-        return done("create", idx_mod.derive_path(kind, s2, date=date, domain=domain),
+        cands = idx_mod.fuzzy_candidates(index, title)
+        near = [c.get("path") for _, c, _ in cands][:3]
+        plan = done("create", idx_mod.derive_path(kind, s2, date=date, domain=domain),
                     dry=True, near=near)
+        if not as_json and not force and any(sc >= DUP_GATE for _, _, sc in cands):
+            # (in --json mode the near_duplicates field carries this; keep stdout clean)
+            print("note: a real run would STOP at the duplicate gate (candidate score "
+                  f">= {DUP_GATE}) — use --merge-into or --force.", file=real_stdout)
+        return plan
 
     near_dupes: list[str] = []
     index_title = title          # title to record in the index entry
@@ -213,18 +283,24 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
             _append_to_existing(path, kind, today_s, body_text)
             action = "updated"
         else:
-            if not status_v and kind == "project":
-                status_v = "active"
+            if not status_v:
+                # Every entity note is born with a status — a kind-appropriate default
+                # (KIND_STATUS) instead of the old project-only special case. Missing
+                # status was the root cause of the frontmatter-completeness drift.
+                status_v = idx_mod.KIND_STATUS.get(kind, "active")
             # Surface (don't silently create alongside) an existing entity this resembles —
-            # the duplication trap when a shortened name didn't resolve exactly.
+            # the duplication trap when a shortened name didn't resolve exactly. (Strong
+            # candidates already stopped at the duplicate gate above; these are the weak
+            # ones, surfaced as a warning.)
             near_dupes = [c.get("path") for _, c, _ in idx_mod.fuzzy_candidates(index, title)][:3]
             # M2: don't let a 40-char-truncated slug silently collide with a different
             # entity already in the index — disambiguate to slug-2, slug-3, ...
             slug = echo_quality.safe_slug(set(index.get("entities", {}).keys()), slug)
             path = idx_mod.derive_path(kind, slug, date=date, domain=domain)
             note_aliases = sorted((set(index_aliases) | set(idx_mod.derive_aliases(title))) - {slug})
+            note_tags = sorted({kind, *tags})  # baseline `- <kind>` tag; --tags enriches
             note = _build_note(idx_mod.KIND_TYPE.get(kind, kind), status_v, today_s, title,
-                               body_text, sources, note_aliases)
+                               body_text, sources, note_aliases, tags=note_tags)
             echo.cmd_put(path, echo.temp_file(note.encode()))
             action = "created"
 

echo-memory.plugin.src/skills/echo-memory/scripts/echo_recall.py

@@ -20,15 +20,28 @@ DESIGN (bounded by the project's pure-stdlib, no-deps constraint):
   * Resilience     : if the index can't be built (vault unreachable), recall falls back
                      to the server's /search/simple so it degrades instead of dying.
 
-Corpus = the indexable ENTITY notes (idx_mod.kind_for_path is not None): people,
+Corpus = the indexable ENTITY notes (idx_mod.kind_for_path is not None) — people,
 companies, concepts, references, meetings, projects, areas, decisions, semantic/
-episodic memory, skills — the durable memory graph. (Sessions/journal are a future
-add; tracked in ROADMAP-1.0.md.)
+episodic memory, skills — PLUS the time-series notes where conversation-derived
+context actually lives: session logs (`_agent/sessions/`) and journal notes
+(`journal/daily|weekly|monthly|quarterly|annual/`). Time-series docs carry a
+per-kind DOWN-WEIGHT so entity notes still rank first for the same lexical match.
+
+RANKING (v1.5): final score = BM25 * kind_weight * freshness * status_factor —
+  * kind_weight : 1.0 entity · 0.5 session · 0.4 journal (KIND_WEIGHT)
+  * freshness   : half-life decay on `updated:` (or the filename date), floored at
+                  FRESH_FLOOR so old notes are demoted, never buried
+  * status      : `active` boosted, `archived` demoted (STATUS_FACTOR)
+so a stale archived project no longer outranks the live one on raw term frequency.
+Doc metadata rides in the index (schema 2); a schema-1 index is discarded and
+rebuilt on the next recall/sweep.
 """
 from __future__ import annotations
 
+import datetime as _dt
 import json
 import math
+import os
 import re
 import sys
 import urllib.parse
@@ -41,6 +54,7 @@ import echo_index as idx_mod     # noqa: E402
 import echo_links as links       # noqa: E402
 
 RECALL_INDEX_PATH = "_agent/index/recall-index.json"
+INDEX_SCHEMA = 2                 # bumped: schema 2 adds per-doc meta (weight/updated/status)
 
 # Tuning knobs. Defaults are standard BM25 (k1/b) + a 0.6 hop decay over <=2 hops.
 K1 = 1.5
@@ -49,6 +63,63 @@ MAX_HOPS = 2
 GRAPH_DECAY = 0.6
 MIN_SCORE = 0.0  # relevance floor; raise to trim weak lexical hits
 
+# --- rank-fusion priors -------------------------------------------------------
+# Time-series notes are in the corpus but down-weighted: they are where "what did we
+# decide three weeks ago" lives, yet the entity note should win a tie on the same terms.
+_TIME_SERIES = [
+    (re.compile(r"^_agent/sessions/"), "session"),
+    (re.compile(r"^journal/daily/"), "daily"),
+    (re.compile(r"^journal/(weekly|monthly|quarterly|annual)/"), "rollup"),
+]
+KIND_WEIGHT = {"session": 0.5, "daily": 0.4, "rollup": 0.4}
+STATUS_FACTOR = {"active": 1.1, "on-hold": 0.9, "archived": 0.75}
+FRESH_HALF_LIFE = float(os.environ.get("ECHO_FRESH_HALF_LIFE", "90"))  # days
+FRESH_FLOOR = 0.6   # freshness multiplier never drops below this — demote, don't bury
+
+_FM_UPDATED = re.compile(r"(?m)^updated:\s*[\"']?(\d{4}-\d{2}-\d{2})")
+_FM_STATUS = re.compile(r"(?m)^status:\s*[\"']?([A-Za-z-]+)")
+_PATH_DATE = re.compile(r"(\d{4}-\d{2}-\d{2})")
+
+
+def _series_kind(path: str) -> str | None:
+    for rx, kind in _TIME_SERIES:
+        if rx.match(path):
+            return kind
+    return None
+
+
+def doc_meta(path: str, raw_text: str) -> dict:
+    """Per-doc ranking priors, extracted once at index time: kind weight, last-updated
+    date (frontmatter `updated:`, else the date in the filename), and `status:`."""
+    kind = _series_kind(path)
+    w = KIND_WEIGHT.get(kind, 1.0) if kind else 1.0
+    head = raw_text[:2000]
+    mu = _FM_UPDATED.search(head)
+    updated = mu.group(1) if mu else None
+    if updated is None:
+        mp = _PATH_DATE.search(path.rsplit("/", 1)[-1])
+        updated = mp.group(1) if mp else None
+    ms = _FM_STATUS.search(head)
+    meta = {"w": w}
+    if updated:
+        meta["u"] = updated
+    if ms:
+        meta["s"] = ms.group(1)
+    return meta
+
+
+def freshness(updated: str | None, today_s: str) -> float:
+    """Half-life decay on document age, floored. Unknown age is neutral (1.0)."""
+    if not updated:
+        return 1.0
+    try:
+        age = (_dt.date.fromisoformat(today_s) - _dt.date.fromisoformat(updated)).days
+    except ValueError:
+        return 1.0
+    if age <= 0:
+        return 1.0
+    return FRESH_FLOOR + (1.0 - FRESH_FLOOR) * (0.5 ** (age / FRESH_HALF_LIFE))
+
 _WORD = re.compile(r"[a-z0-9]+")
 # Minimal English stoplist — keeps BM25 idf from being dominated by glue words.
 _STOP = frozenset(
@@ -73,12 +144,15 @@ def strip_frontmatter(text: str) -> str:
 
 # --------------------------------------------------------------------- BM25 core
 class Bm25Index:
-    """A compact, serializable BM25 index. add()/remove() are idempotent per path."""
+    """A compact, serializable BM25 index with per-doc ranking meta.
+    add()/remove() are idempotent per path. add() takes the RAW note text — it strips
+    frontmatter itself and extracts the doc meta (weight/updated/status) in one pass."""
 
     def __init__(self) -> None:
         self.df: Counter[str] = Counter()              # term -> #docs containing it
         self.postings: dict[str, dict[str, int]] = {}  # term -> {doc_path: tf}
         self.length: dict[str, int] = {}               # doc_path -> token count
+        self.meta: dict[str, dict] = {}                # doc_path -> {w, u, s} priors
         self.n_docs = 0
         self.avg_len = 0.0
 
@@ -86,11 +160,12 @@ class Bm25Index:
         self.n_docs = len(self.length)
         self.avg_len = (sum(self.length.values()) / self.n_docs) if self.n_docs else 0.0
 
-    def add(self, path: str, text: str) -> None:
+    def add(self, path: str, raw_text: str) -> None:
         if path in self.length:        # re-index in place (idempotent)
             self.remove(path)
-        toks = tokenize(text)
+        toks = tokenize(strip_frontmatter(raw_text))
         self.length[path] = len(toks)
+        self.meta[path] = doc_meta(path, raw_text)
         for term, tf in Counter(toks).items():
             self.postings.setdefault(term, {})[path] = tf
             self.df[term] += 1
@@ -107,9 +182,19 @@ class Bm25Index:
                     del self.postings[term]
                     del self.df[term]
         del self.length[path]
+        self.meta.pop(path, None)
         self._recompute()
 
-    def score(self, query: str, limit: int = 10) -> list[tuple[str, float]]:
+    def _doc_factor(self, path: str, today_s: str) -> float:
+        """The rank-fusion prior: kind weight x freshness x status factor."""
+        m = self.meta.get(path) or {}
+        return (float(m.get("w", 1.0))
+                * freshness(m.get("u"), today_s)
+                * STATUS_FACTOR.get(m.get("s", ""), 1.0))
+
+    def score(self, query: str, limit: int = 10,
+              today_s: str | None = None) -> list[tuple[str, float]]:
+        today_s = today_s or echo.today()
         scores: Counter[str] = Counter()
         for term in set(tokenize(query)):
             posting = self.postings.get(term)
@@ -120,18 +205,23 @@ class Bm25Index:
                 dl = self.length.get(path, 0)
                 denom = tf + K1 * (1 - B + B * (dl / self.avg_len if self.avg_len else 1))
                 scores[path] += idf * (tf * (K1 + 1) / denom if denom else 0)
-        return [(p, s) for p, s in scores.most_common(limit) if s > MIN_SCORE]
+        fused = {p: s * self._doc_factor(p, today_s) for p, s in scores.items()}
+        ranked = sorted(fused.items(), key=lambda kv: -kv[1])[:limit]
+        return [(p, s) for p, s in ranked if s > MIN_SCORE]
 
     # ---- serialization (compact; rebuildable, so the format can change freely) ----
     def to_json(self) -> dict:
-        return {"schema": 1, "n_docs": self.n_docs, "avg_len": self.avg_len,
-                "length": self.length, "postings": self.postings}
+        return {"schema": INDEX_SCHEMA, "n_docs": self.n_docs, "avg_len": self.avg_len,
+                "length": self.length, "postings": self.postings, "meta": self.meta}
 
     @classmethod
     def from_json(cls, d: dict) -> "Bm25Index":
+        if d.get("schema") != INDEX_SCHEMA:
+            return cls()  # older/newer format -> empty; recall's cold path rebuilds
         ix = cls()
         ix.length = d.get("length", {})
         ix.postings = d.get("postings", {})
+        ix.meta = d.get("meta", {})
         ix.n_docs = d.get("n_docs", len(ix.length))
         ix.avg_len = d.get("avg_len", 0.0)
         ix.df = Counter({term: len(posting) for term, posting in ix.postings.items()})
@@ -172,8 +262,12 @@ def _walk(prefix: str = ""):
 
 def _indexable(path: str) -> bool:
     base = path.rsplit("/", 1)[-1]
-    return (path.endswith(".md") and idx_mod.kind_for_path(path) is not None
-            and base not in _SKIP_BASENAMES and not _TEMPLATE_RE.search(path))
+    if (not path.endswith(".md") or base in _SKIP_BASENAMES
+            or _TEMPLATE_RE.search(path)):
+        return False
+    # Entity notes (the durable graph) + time-series notes (sessions, journal) — the
+    # latter join the corpus down-weighted so past decisions/discussions are findable.
+    return idx_mod.kind_for_path(path) is not None or _series_kind(path) is not None
 
 
 # ------------------------------------------------------------------- persistence
@@ -208,7 +302,7 @@ def rebuild(prefetched: dict | None = None) -> Bm25Index:
         items = ((p, _get(p)) for p in _walk() if _indexable(p))
     for path, text in items:
         if text is not None:
-            ix.add(path, strip_frontmatter(text))
+            ix.add(path, text)   # add() strips frontmatter + extracts doc meta itself
     save_index(ix)
     return ix
 
@@ -224,7 +318,7 @@ def update_note(path: str, text: str) -> None:
         import echo_concurrency
         with echo_concurrency.vault_lock():
             ix = load_index()                      # fresh read inside the lock
-            ix.add(path, strip_frontmatter(text))
+            ix.add(path, text)                     # raw text: add() strips + extracts meta
             save_index(ix)
     except Exception as exc:  # noqa: BLE001 — upkeep must never fail a capture
         print(f"echo_recall: index update skipped ({exc})", file=sys.stderr)
@@ -279,37 +373,57 @@ def expand_graph(seeds, nmap: dict, base_scores: dict, max_hops: int = MAX_HOPS)
 
 
 # ------------------------------------------------------------------- presentation
+def _doc_summary(path: str, text: str) -> dict:
+    """Structured per-note summary shared by the human brief and --json output."""
+    out: dict = {"path": path}
+    mtype = re.search(r"(?m)^type:\s*(.+)$", text)
+    if mtype:
+        out["type"] = mtype.group(1).strip()
+    mu = _FM_UPDATED.search(text[:2000])
+    if mu:
+        out["updated"] = mu.group(1)
+    ms = _FM_STATUS.search(text[:2000])
+    if ms:
+        out["status"] = ms.group(1)
+    body = strip_frontmatter(text)
+    status = re.search(r"(?ms)^##\s+Status\s*$(.*?)(?=^##\s|\Z)", body)
+    if status and status.group(1).strip():
+        out["excerpt"] = status.group(1).strip()[:400]
+    else:
+        kept = [ln[:200] for ln in body.splitlines()
+                if ln.strip() and not ln.startswith("# ")][:6]
+        out["excerpt"] = "\n".join(kept)
+    return out
+
+
 def _brief(path: str, score: float | None = None, via: str | None = None) -> None:
     text = links.get_text(path)
     if text is None:
         return
+    info = _doc_summary(path, text)
     head = f"\n### {path}"
     if score is not None:
         head += f"   (score {score:.2f})"
     if via:
         head += f"   (via {via})"
     print(head)
-    mtype = re.search(r"(?m)^type:\s*(.+)$", text)
-    if mtype:
-        print(f"_type: {mtype.group(1).strip()}_")
-    body = strip_frontmatter(text)
-    status = re.search(r"(?ms)^##\s+Status\s*$(.*?)(?=^##\s|\Z)", body)
-    if status and status.group(1).strip():
-        print(status.group(1).strip()[:400])
-        return
-    shown = 0
-    for line in body.splitlines():
-        if not line.strip() or line.startswith("# "):
-            continue
-        print(line[:200])
-        shown += 1
-        if shown >= 6:
-            break
+    stamps = []
+    if info.get("type"):
+        stamps.append(f"type: {info['type']}")
+    if info.get("updated"):
+        stamps.append(f"updated: {info['updated']}")
+    if info.get("status"):
+        stamps.append(f"status: {info['status']}")
+    if stamps:
+        print("_" + " · ".join(stamps) + "_")   # staleness is visible, not guessed
+    if info.get("excerpt"):
+        print(info["excerpt"])
 
 
 # ------------------------------------------------------------------- entrypoint
-def recall(query, limit: int = 8) -> int:
+def recall(query, limit: int = 8, as_json: bool = False) -> int:
     q = " ".join(query) if isinstance(query, list) else query
+    today_s = echo.today()
     index = idx_mod.load()
     nmap = idx_mod.name_map(index)
 
@@ -319,7 +433,7 @@ def recall(query, limit: int = 8) -> int:
         ix = load_index()
         if ix.n_docs == 0:
             ix = rebuild()
-        lexical = ix.score(q, limit=limit)
+        lexical = ix.score(q, limit=limit, today_s=today_s)
     except echo.EchoError as exc:
         print(f"recall: BM25 index unavailable ({exc}); falling back to server search",
               file=sys.stderr)
@@ -350,6 +464,26 @@ def recall(query, limit: int = 8) -> int:
     # --- graph layer ----------------------------------------------------------
     neighbours = expand_graph(hits, nmap, base, max_hops=MAX_HOPS)
 
+    if as_json:
+        import echo_output
+        texts = echo.read_many(hits + [p for p, _ in neighbours[: 2 * limit]])
+        primary = []
+        for p in hits:
+            if texts.get(p) is None:
+                continue
+            primary.append({**_doc_summary(p, texts[p]),
+                            "score": round(base.get(p, 0.0), 3)})
+        linked = []
+        for p, (sc, via) in neighbours[: 2 * limit]:
+            if texts.get(p) is None:
+                continue
+            linked.append({**_doc_summary(p, texts[p]),
+                           "score": round(sc, 3), "via": via})
+        env = echo_output.envelope("recall", {"query": q, "primary": primary,
+                                              "linked": linked})
+        print(json.dumps(env, ensure_ascii=False))
+        return 0
+
     print(f"== recall: {q} ==")
     print(f"\n# Primary hits ({len(hits)})")
     for p in hits:

echo-memory.plugin.src/skills/echo-memory/scripts/echo_reflect.py

@@ -20,6 +20,7 @@ PROPOSAL_SCHEMA (one per durable item the agent wants to remember):
     "kind":    "person",                    # required unless "inbox": true
     "body":    "Principal at Acme; ...",      # markdown body (optional)
     "aliases": ["bob", "rs"],               # optional
+    "tags":    ["client"],                  # optional; the kind is always seeded as a tag
     "sources": ["_agent/sessions/..."],     # optional backward links
     "date":    "YYYY-MM-DD",                # optional (meeting/decision kinds)
     "domain":  "business",                  # optional (area kind)
@@ -121,7 +122,7 @@ def apply(proposals: list[dict], confirm: bool = False) -> int:
         return 0
 
     import echo_ops  # lazy: the apply path pulls in the capture/index/link stack
-    applied = 0
+    applied = gated = 0
     for p in valid:
         if p.get("_action") == "error":
             continue
@@ -129,8 +130,12 @@ def apply(proposals: list[dict], confirm: bool = False) -> int:
         rc = echo_ops.capture(
             p.get("kind"), p["title"], bodyfile,
             aliases=p.get("aliases") or [], sources=p.get("sources") or [],
+            tags=p.get("tags") or [],
             date=p.get("date"), domain=p.get("domain", "business"),
             inbox=bool(p.get("inbox")) or not p.get("kind"))
         applied += 1 if rc == 0 else 0
-    print(f"reflect: applied {applied}/{len(valid)} proposal(s).")
+        gated += 1 if rc == 76 else 0
+    print(f"reflect: applied {applied}/{len(valid)} proposal(s)."
+          + (f" {gated} stopped at the duplicate gate — re-propose with the existing "
+             f"entity's title (or capture --merge-into <slug>)." if gated else ""))
     return 0

echo-memory.plugin.src/skills/echo-memory/scripts/echo_triage.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""echo_triage.py — one-tap inbox triage, built on the reflect pipeline.
+
+Triage used to be pure prose: the agent hand-routed each inbox line with individual
+PATCH/PUT calls and hand-wrote the processing log — which is why the inbox rots.
+But the machinery it needs already exists in echo_reflect (validate -> classify
+against the entity index -> preview -> apply via capture). This module reuses it
+and adds the two triage-specific pieces:
+
+  * `list_inbox`  — parse `inbox/captures/inbox.md` into structured capture lines
+                    (date, text, age in days) so the model builds proposals from
+                    data, not by re-reading prose;
+  * `apply`       — same contract as reflect (dry-run unless confirm), plus an
+                    audit line in `inbox/processing-log/YYYY-MM-DD.md` for every
+                    routed item. Originals are NEVER deleted (operating contract:
+                    the processing log is the audit trail, deletion is explicit).
+
+Proposals are reflect's PROPOSAL_SCHEMA plus an optional `"line"` — the original
+inbox line, echoed into the processing log so the audit trail maps 1:1.
+"""
+from __future__ import annotations
+
+import json
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import echo                    # noqa: E402
+
+INBOX_PATH = "inbox/captures/inbox.md"
+LOG_DIR = "inbox/processing-log"
+
+_CAPTURE_LINE = re.compile(r"^\s*-\s*(\d{4}-\d{2}-\d{2})\s*:?\s*(.*\S)\s*$")
+
+
+def parse_inbox(text: str) -> list[dict]:
+    """Dated capture bullets -> [{line, date, text, age_days}] (age from ECHO_TODAY)."""
+    import datetime as dt
+    try:
+        today = dt.date.fromisoformat(echo.today())
+    except ValueError:
+        today = dt.date.today()
+    items = []
+    for raw_line in text.splitlines():
+        m = _CAPTURE_LINE.match(raw_line)
+        if not m:
+            continue
+        try:
+            age = (today - dt.date.fromisoformat(m.group(1))).days
+        except ValueError:
+            age = None
+        items.append({"line": raw_line.strip(), "date": m.group(1),
+                      "text": m.group(2), "age_days": age})
+    return items
+
+
+def list_inbox(as_json: bool = False) -> int:
+    status, body = echo.request("GET", echo.vault_url(INBOX_PATH))
+    if status == 404:
+        items = []
+    else:
+        echo.check(status, body, f"triage list {INBOX_PATH}")
+        items = parse_inbox(body.decode(errors="replace"))
+    if as_json:
+        import echo_output
+        env = echo_output.envelope("triage-list", {"path": INBOX_PATH, "items": items,
+                                                   "count": len(items)})
+        print(json.dumps(env, ensure_ascii=False))
+        return 0
+    if not items:
+        print("triage: inbox is empty — nothing to route.")
+        return 0
+    print(f"triage: {len(items)} capture(s) in {INBOX_PATH}")
+    for it in items:
+        age = f"{it['age_days']}d" if it["age_days"] is not None else "?"
+        print(f"  [{age:>4}] {it['date']}: {it['text']}")
+    print("\nBuild a proposals JSON (reflect schema + optional \"line\") and run "
+          "`echo.py triage <file>` to preview, `--apply` to route.")
+    return 0
+
+
+def apply(proposals: list[dict], confirm: bool = False) -> int:
+    """Validate -> classify -> preview; with confirm, route each via capture AND write
+    the processing-log audit line. Mirrors echo_reflect.apply's contract exactly."""
+    import echo_reflect
+    valid, errors = echo_reflect.validate(proposals)
+    for e in errors:
+        print(f"skip: {e}", file=sys.stderr)
+    if not valid:
+        print("triage: no valid proposals to route.")
+        return 0
+
+    echo_reflect.classify(valid)
+    counts = {a: sum(1 for p in valid if p.get("_action") == a)
+              for a in ("create", "update", "inbox", "error")}
+    print(f"triage: {len(valid)} proposal(s) — "
+          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} stay-in-inbox"
+          + (f", {counts['error']} error" if counts["error"] else ""))
+    print(echo_reflect.preview(valid))
+
+    if not confirm:
+        print("\ntriage: dry-run — re-run with --apply to route these and log the moves.")
+        return 0
+
+    import echo_ops
+    today_s = echo.today()
+    applied = gated = 0
+    for p in valid:
+        if p.get("_action") == "error":
+            continue
+        if p.get("_action") == "inbox":
+            continue  # routing an inbox line back to the inbox is a no-op, not a move
+        bodyfile = echo.temp_file((p.get("body") or "").encode()) if p.get("body") else None
+        rc = echo_ops.capture(
+            p.get("kind"), p["title"], bodyfile,
+            aliases=p.get("aliases") or [], sources=p.get("sources") or [],
+            tags=p.get("tags") or [],
+            date=p.get("date"), domain=p.get("domain", "business"))
+        if rc == 76:
+            gated += 1
+            continue
+        if rc != 0:
+            continue
+        applied += 1
+        original = (p.get("line") or p["title"]).strip()
+        echo.cmd_append(f"{LOG_DIR}/{today_s}.md",
+                        f"- {original} → {p.get('_path', '?')}")
+    print(f"triage: routed {applied}/{len(valid)} item(s); audit in {LOG_DIR}/{today_s}.md. "
+          "Originals kept in the inbox (deletion is explicit-only)."
+          + (f" {gated} stopped at the duplicate gate — re-propose with the existing "
+             f"entity's title or use capture --merge-into." if gated else ""))
+    return 0

echo-memory.plugin.src/skills/echo-memory/scripts/sweep.py

@@ -8,10 +8,14 @@ backfill what older vaults don't have yet:
   2. **(re)build the entity index** (`_agent/index/entities.json`) from the notes
      already in the vault (people, companies, projects, concepts, references,
      meetings, areas, decisions, semantic/episodic memory, skills),
-  3. **symmetrize cross-links** — for every `## Related` link A -> B, ensure the
+  3. **backfill incomplete entity frontmatter** — fill a missing/empty `status:` with
+     the kind default and seed a missing/empty `tags:` with the note's kind, per the
+     KIND_REQUIRED_FM/KIND_STATUS maps in echo_index (what `/echo-health` flags as
+     `incomplete-frontmatter`),
+  4. **symmetrize cross-links** — for every `## Related` link A -> B, ensure the
      reciprocal B -> A exists (it only adds the missing direction; it never invents
      new links from body text), and
-  4. stamp the marker `schema_version` to the current schema (4).
+  5. stamp the marker `schema_version` to the current schema (4).
 
 Dry-run by default; pass --apply to write. READ-ONLY without --apply.
 Cross-platform: pure Python via echo.py.
@@ -43,6 +47,30 @@ SKIP_BASENAMES = {"README.md"}
 kind_for = idx_mod.kind_for_path
 
 
+def fm_block(text: str) -> str:
+    if not text.startswith("---"):
+        return ""
+    end = text.find("\n---", 3)
+    return text[:end] if end != -1 else text
+
+
+def fm_populated(text: str, field: str) -> bool:
+    """True when the frontmatter carries a real value for `field` (flow list `[x]`,
+    block list items, or a non-empty scalar). Mirrors vault_lint.fm_field_populated."""
+    fm = fm_block(text)
+    m = re.search(rf"(?m)^{re.escape(field)}:[ \t]*(.*)$", fm)
+    if not m:
+        return False
+    val = m.group(1).strip()
+    if val == "[]":
+        return False
+    if val:
+        return True
+    rest = fm[m.end():].lstrip("\n")
+    first = rest.splitlines()[0] if rest else ""
+    return bool(re.match(r"^\s+-\s*\S", first))
+
+
 def get(path: str) -> str | None:
     status, body = echo.request("GET", echo.vault_url(path))
     if status == 404:
@@ -139,9 +167,36 @@ def main(argv: list[str] | None = None) -> int:
         print(f"sweep: {tag} recall index — {rix.n_docs} docs (BM25)")
     else:
         n_idx = sum(1 for p in all_files if echo_recall._indexable(p))
-        print(f"sweep: {tag} recall index — would rebuild BM25 over {n_idx} entity notes")
+        print(f"sweep: {tag} recall index — would rebuild BM25 over {n_idx} corpus notes "
+              f"(entities + sessions/journal)")
 
-    # ---- 3. symmetrize existing Related links --------------------------------
+    # ---- 3. backfill incomplete entity frontmatter ----------------------------
+    # What /echo-health flags as incomplete-frontmatter, filled mechanically: a missing
+    # or empty status gets the kind default; missing/empty tags get the kind as a
+    # baseline tag. cmd_fm is create-or-replace, so absent keys are inserted surgically.
+    fixes: list[tuple[str, str, str]] = []
+    for path in all_files:
+        kind = kind_for(path)
+        if kind is None:
+            continue
+        text = texts.get(path) or ""
+        if not text.startswith("---"):
+            continue  # no frontmatter block at all — lint flags it; not a mechanical fill
+        for field in idx_mod.KIND_REQUIRED_FM.get(kind, ()):
+            if not fm_populated(text, field):
+                value = (idx_mod.KIND_STATUS.get(kind, "active") if field == "status"
+                         else json.dumps([kind]))
+                fixes.append((path, field, value))
+    print(f"sweep: {tag} frontmatter backfill — {len(fixes)} field(s) to fill")
+    for path, field, value in fixes[:40]:
+        print(f"          {path}: {field} -> {value}")
+    if len(fixes) > 40:
+        print(f"          ... and {len(fixes) - 40} more")
+    if apply:
+        for path, field, value in fixes:
+            echo.cmd_fm(path, field, value)
+
+    # ---- 4. symmetrize existing Related links --------------------------------
     fileset = set(all_files)
     basemap: dict[str, str] = {}
     for p in all_files:
@@ -176,7 +231,7 @@ def main(argv: list[str] | None = None) -> int:
         for tp, sp in missing:
             links.add_one(tp, sp)
 
-    # ---- 4. stamp schema ------------------------------------------------------
+    # ---- 5. stamp schema ------------------------------------------------------
     marker = get("_agent/echo-vault.md") or ""
     cur = next((int(m) for m in re.findall(r"(?m)^schema_version:\s*(\d+)", marker)), 0)
     if cur < CURRENT_SCHEMA:

echo-memory.plugin.src/skills/echo-memory/scripts/vault_lint.py

@@ -120,6 +120,22 @@ def as_list(value: object) -> list[object]:
     return value if isinstance(value, list) else [value]
 
 
+def fm_field_populated(raw: str, fields: dict[str, object], field: str) -> bool:
+    """True when `field` exists AND holds a real value. parse_frontmatter is line-based,
+    so a block-style list (`tags:\n  - x`) parses as "" — check the raw text for items
+    before calling the field empty."""
+    value = fields.get(field)
+    if isinstance(value, list):
+        return bool(value)
+    if str(value or "").strip():
+        return True
+    lines = raw.splitlines()
+    for index, line in enumerate(lines):
+        if re.match(rf"^{re.escape(field)}:\s*$", line):
+            return index + 1 < len(lines) and bool(re.match(r"^\s+-\s*\S", lines[index + 1]))
+    return False
+
+
 def route_matchers():
     routing = json.loads((SCRIPT_DIR / "routing.json").read_text(encoding="utf-8"))
     routes = [(item["id"], re.compile(item["pattern"])) for item in routing.get("routes", [])]
@@ -171,6 +187,15 @@ def main() -> int:
         missing = [k for k in REQUIRED_FM if not str(fm.get(k, "")).strip()]
         if fm and missing:
             flag("missing-frontmatter", f"{path}: missing {', '.join(missing)}")
+        # Kind-scoped completeness: entity notes must classify (status + tags). The
+        # per-kind requirement map lives in echo_index (KIND_REQUIRED_FM) so capture's
+        # defaults, this check, and sweep's backfill all read the same source of truth.
+        kind = idx_mod.kind_for_path(path)
+        if fm and kind:
+            for field in idx_mod.KIND_REQUIRED_FM.get(kind, ()):
+                if not fm_field_populated(raw, fm, field):
+                    what = f"missing {field}" if field not in fm else f"empty {field}"
+                    flag("incomplete-frontmatter", f"{path}: {what}")
         created, updated = parse_date(fm.get("created")), parse_date(fm.get("updated"))
         if created and updated and updated < created:
             flag("date-order", f"{path}: updated {updated} is before created {created}")
@@ -307,6 +332,7 @@ def main() -> int:
         "unknown-path": "Path matches no route in routing.json",
         "retired-path": "Write to a retired/dead path",
         "missing-frontmatter": "Missing required frontmatter field",
+        "incomplete-frontmatter": "Incomplete entity frontmatter (status/tags) — sweep.py --apply backfills",
         "date-order": "updated earlier than created",
         "future-date": "updated date is in the future",
         "source-notes-wikilink": "Wikilink in source_notes (must be plain paths)",