feat(convo): preserve authored timestamp from transcripts

docs/authored-at.md

@@ -0,0 +1,45 @@
+# Authored date (`authored_at`)
+
+Conversation transcripts carry a per-line ISO-8601 `timestamp` (both Claude Code and
+Codex JSONL). The miner records the most recent one per file as the drawer's
+**`authored_at`** — when the content was actually written.
+
+This is distinct from the ingest date:
+
+| Field | Meaning |
+|-------|---------|
+| `filed_at` / result `created_at` | When the drawer was **mined** (written to the palace). A bulk re-mine collapses these to a single instant. |
+| `authored_at` | When the underlying content was **written**, recovered from the transcript timestamps. Survives re-mining. |
+
+`authored_at` is surfaced in search results (and shown in the CLI `search` output), and is
+used as a deterministic tie-break in hybrid ranking: candidates with identical scores order
+with the more recently authored drawer first. Drawers without per-line timestamps (e.g.
+markdown) fall back to `filed_at`.
+
+## Backfilling existing memory
+
+New mines populate `authored_at` automatically. Drawers mined before this feature only have
+`filed_at`. Re-mining does **not** fix them — the scanner skips files already mined at the
+current `NORMALIZE_VERSION`. Two options:
+
+1. **In-place backfill (recommended — no re-embedding).** `scripts/backfill_authored_at.py`
+   reads each convos drawer's source transcript and updates only the `authored_at` metadata.
+   Idempotent and safe to re-run; embeddings are untouched.
+
+   ```bash
+   python scripts/backfill_authored_at.py \
+       --palace ~/.mempalace/palace \
+       --sessions ~/.claude --sessions ~/.codex          # dry run
+   python scripts/backfill_authored_at.py \
+       --palace ~/.mempalace/palace \
+       --sessions ~/.claude --sessions ~/.codex --apply  # write
+   ```
+
+   For the Docker MCP image, mount the volume and session dirs read-only — see the header of
+   `scripts/backfill_authored_at.py` for the exact `docker run` invocation.
+
+   > Back up first: `tar czf palace-backup.tgz -C <palace-dir> .` (or snapshot the
+   > `mempalace-data` volume).
+
+2. **Drop and recreate.** Delete the affected drawers and re-mine the transcripts; the fresh
+   mine stamps `authored_at`. Simpler, but re-embeds everything.

mempalace/convo_miner.py

@@ -10,6 +10,7 @@
 
 import os
 import sys
+import json
 import logging
 import stat
 from pathlib import Path
@@ -408,7 +409,42 @@ def scan_convos(convo_dir: str) -> list:
 # =============================================================================
 
 
-def _file_chunks_locked(collection, source_file, chunks, wing, room, agent, extract_mode):
+def _extract_authored_at(filepath):
+    """Most-recent message timestamp in a transcript, used as the drawer's authored date.
+
+    Both Claude Code and Codex JSONL transcripts carry a top-level ISO-8601
+    ``timestamp`` on each line. We take the max so ``authored_at`` reflects when the
+    content was actually written, independent of when it was mined (``filed_at``).
+    This restores chronology: a session from days ago keeps its real date even when
+    re-mined today, instead of every drawer collapsing to ingest time. Returns None
+    for formats without per-line timestamps (e.g. plain ``.md``).
+    """
+    path = Path(filepath)
+    if path.suffix != ".jsonl":
+        return None
+    latest = None
+    try:
+        with path.open(encoding="utf-8", errors="ignore") as fh:
+            for line in fh:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    ts = json.loads(line).get("timestamp")
+                except (ValueError, TypeError, AttributeError):
+                    continue
+                # ISO-8601 timestamps are strings; guard against a non-string
+                # ``timestamp`` so a malformed line can't raise TypeError on compare.
+                if isinstance(ts, str) and (latest is None or ts > latest):
+                    latest = ts
+    except OSError:
+        return None
+    return latest
+
+
+def _file_chunks_locked(
+    collection, source_file, chunks, wing, room, agent, extract_mode, authored_at=None
+):
     """Lock the source file, purge stale drawers, and upsert fresh chunks.
 
     Combines the per-file serialization that prevents concurrent agents from
@@ -463,6 +499,7 @@ def _file_chunks_locked(collection, source_file, chunks, wing, room, agent, extr
                         "chunk_index": chunk["chunk_index"],
                         "added_by": agent,
                         "filed_at": filed_at,
+                        "authored_at": authored_at if authored_at is not None else filed_at,
                         "ingest_mode": "convos",
                         "extract_mode": extract_mode,
                         "normalize_version": NORMALIZE_VERSION,
@@ -726,7 +763,14 @@ def _mine_convos_impl(
         # Lock + purge stale + file fresh chunks. Lock serializes concurrent
         # agents; purge removes pre-v2 drawers so the schema bump applies.
         drawers_added, room_delta, skipped = _file_chunks_locked(
-            collection, source_file, chunks, wing, room, agent, extract_mode
+            collection,
+            source_file,
+            chunks,
+            wing,
+            room,
+            agent,
+            extract_mode,
+            authored_at=_extract_authored_at(filepath),
         )
         if skipped:
             files_skipped += 1

mempalace/layers.py

@@ -306,6 +306,9 @@ def search(self, query: str, wing: str = None, room: str = None, n_results: int
             lines.append(f"      {snippet}")
             if source:
                 lines.append(f"      src: {source}")
+            authored = (meta.get("authored_at") or "")[:10]
+            if authored:
+                lines.append(f"      authored: {authored}")
 
         return "\n".join(lines)
 

mempalace/searcher.py

@@ -221,7 +221,18 @@ def _hybrid_rank(
         r["bm25_score"] = round(raw, 3)
         scored.append((vector_weight * vec_sim + bm25_weight * norm, r))
 
-    scored.sort(key=lambda pair: pair[0], reverse=True)
+    # Break exact score ties toward the more recently authored drawer so equal-score
+    # candidates rank chronologically instead of in arbitrary backend order. ISO-8601
+    # ``authored_at`` strings sort chronologically; missing dates sort oldest.
+    # authored_at lives at the top level on the search_memories path and nested under
+    # "metadata" on the candidate-union path; check both so the tie-break works for each.
+    scored.sort(
+        key=lambda pair: (
+            pair[0],
+            pair[1].get("authored_at") or pair[1].get("metadata", {}).get("authored_at") or "",
+        ),
+        reverse=True,
+    )
     results[:] = [r for _, r in scored]
     return results
 
@@ -681,6 +692,7 @@ def _metadata_filter_sql(row_id_expr: str) -> tuple[str, list[str]]:
                 "source_file": Path(full_source).name if full_source else "?",
                 "source_path": full_source,
                 "created_at": meta.get("filed_at", "unknown"),
+                "authored_at": meta.get("authored_at", meta.get("filed_at", "unknown")),
                 # No vector distance available in BM25-only mode.
                 "similarity": None,
                 "distance": None,
@@ -783,6 +795,7 @@ def _merge_bm25_union_candidates(
                 "source_file": Path(full_source).name if full_source else "?",
                 "source_path": full_source,
                 "created_at": meta.get("filed_at", "unknown"),
+                "authored_at": meta.get("authored_at", meta.get("filed_at", "unknown")),
                 "similarity": None,
                 "distance": None,
                 "effective_distance": None,
@@ -1198,6 +1211,7 @@ def search_memories(
             "source_file": Path(source).name if source else "?",
             "source_path": source,
             "created_at": meta.get("filed_at", "unknown"),
+            "authored_at": meta.get("authored_at", meta.get("filed_at", "unknown")),
             "similarity": round(_distance_to_similarity(effective_dist, metric), 3),
             "distance": round(dist, 4),
             "effective_distance": round(effective_dist, 4),

scripts/backfill_authored_at.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Backfill ``authored_at`` onto existing conversation drawers.
+
+New mines stamp ``authored_at`` automatically (see ``convo_miner._extract_authored_at``),
+but drawers mined before that change only have ``filed_at`` (ingest time). Re-mining does
+NOT fix them: the scanner skips files already mined at the current ``NORMALIZE_VERSION``.
+
+This migration updates the affected drawers IN PLACE — metadata only, embeddings are left
+untouched, so there is no re-embedding cost. It is idempotent (drawers already correct are
+skipped) and safe to re-run. It only touches ``ingest_mode == "convos"`` drawers; markdown
+drawers have no per-line timestamps and keep their ``filed_at`` fallback.
+
+Drawers whose source transcript is no longer on disk are left as-is (they keep falling back
+to ``filed_at``), so point ``--sessions`` at the directories that still hold your ``.jsonl``
+transcripts (e.g. ``~/.claude`` and ``~/.codex``).
+
+Usage (dry-run prints what would change; pass --apply to write):
+
+    python scripts/backfill_authored_at.py \
+        --palace ~/.mempalace/palace \
+        --sessions ~/.claude --sessions ~/.codex [--apply]
+
+In Docker (the MCP image), mount the volume and your session dirs read-only:
+
+    docker run --rm \
+      -v mempalace-data:/data \
+      -v ~/.claude:/sessions/claude:ro -v ~/.codex:/sessions/codex:ro \
+      -v "$PWD/scripts/backfill_authored_at.py:/tmp/backfill.py:ro" \
+      --entrypoint /app/.venv/bin/python mempalace:local \
+      /tmp/backfill.py --palace /data/.mempalace/palace \
+        --sessions /sessions/claude --sessions /sessions/codex --apply
+"""
+
+import argparse
+import glob
+import os
+
+import chromadb
+
+from mempalace.convo_miner import _extract_authored_at
+
+COLLECTION = "mempalace_drawers"
+PAGE = 2000
+BATCH = 1000
+
+
+def _index_sessions(session_dirs):
+    """Map ``basename.jsonl -> realpath`` for every transcript under the given dirs."""
+    index = {}
+    for root in session_dirs:
+        for f in glob.glob(os.path.join(os.path.expanduser(root), "**", "*.jsonl"), recursive=True):
+            index.setdefault(os.path.basename(f), f)
+    return index
+
+
+def backfill_authored_at(collection, session_dirs, apply=False):
+    """Stamp ``authored_at`` on convos drawers from their source transcript timestamps.
+
+    Returns a stats dict: ``scanned``, ``updated``, ``resolved_files``, ``unresolved_files``.
+    """
+    index = _index_sessions(session_dirs)
+    cache = {}
+    unresolved = set()
+    pending_ids, pending_metas = [], []
+    scanned = updated = 0
+
+    def flush():
+        nonlocal pending_ids, pending_metas, updated
+        if pending_ids and apply:
+            collection.update(ids=pending_ids, metadatas=pending_metas)
+        updated += len(pending_ids)
+        pending_ids, pending_metas = [], []
+
+    offset = 0
+    while True:
+        res = collection.get(
+            where={"ingest_mode": "convos"}, include=["metadatas"], limit=PAGE, offset=offset
+        )
+        ids = res["ids"]
+        if not ids:
+            break
+        for drawer_id, meta in zip(ids, res["metadatas"]):
+            scanned += 1
+            basename = os.path.basename(meta.get("source_file") or "")
+            if basename in cache:
+                authored = cache[basename]
+            else:
+                path = index.get(basename)
+                authored = _extract_authored_at(path) if path else None
+                cache[basename] = authored
+                if path is None and basename:
+                    unresolved.add(basename)
+            if authored and meta.get("authored_at") != authored:
+                new_meta = dict(meta)
+                new_meta["authored_at"] = authored
+                pending_ids.append(drawer_id)
+                pending_metas.append(new_meta)
+                if len(pending_ids) >= BATCH:
+                    flush()
+        offset += len(ids)
+    flush()
+    return {
+        "scanned": scanned,
+        "updated": updated,
+        "resolved_files": sum(1 for v in cache.values() if v),
+        "unresolved_files": len(unresolved),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--palace", required=True, help="Path to the ChromaDB palace dir")
+    parser.add_argument(
+        "--sessions",
+        action="append",
+        default=[],
+        required=True,
+        help="Directory holding .jsonl transcripts (repeatable)",
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Write changes (default is a dry run that only reports counts)",
+    )
+    args = parser.parse_args()
+
+    client = chromadb.PersistentClient(path=os.path.expanduser(args.palace))
+    collection = client.get_collection(COLLECTION)
+    stats = backfill_authored_at(collection, args.sessions, apply=args.apply)
+    mode = "APPLIED" if args.apply else "DRY-RUN (use --apply to write)"
+    print(
+        f"{mode}: scanned={stats['scanned']} updated={stats['updated']} "
+        f"resolved_files={stats['resolved_files']} unresolved_files={stats['unresolved_files']}"
+    )
+
+
+if __name__ == "__main__":
+    main()