v0.2.0: Truth governance engine

Add memory lifecycle states (hypothesis -> active -> validated -> deprecated -> superseded),
provenance tracking (source_path, source_hash, source_section), conflict detection,
staleness detection, health scoring, and trust-ranked recall.

Core changes:
- Schema v2 with backward-compatible migration
- promote(), deprecate(), supersede() lifecycle methods
- governance.py: conflict detection (duplicate vs contradiction), staleness, health check
- Trust-ranked recall: validated > active > hypothesis, provenance-weighted
- Search no longer bumps access_count (fixes self-reinforcing popularity bias)
- Deprecated/superseded memories excluded from search results
- MCP server: 13 tools including promote, deprecate, supersede, health, conflicts
- CLI: health, conflicts, stale, promote, deprecate commands
- 59 tests passing

Positioning: governed memory for long-lived coding agents.
Published to PyPI as quilmem 0.2.0.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Thezenmonster

README.md

@@ -1,8 +1,14 @@
-"""agentmem — Production-ready agent memory. SQLite + FTS5. No infrastructure."""
+"""agentmem — Trusted memory for long-lived coding agents. SQLite + FTS5. No infrastructure."""
 
 from .core import Memory
+from .governance import detect_conflicts, detect_stale, health_check, hash_content, hash_file_section
 from .importer import import_markdown
-from .models import MEMORY_TYPES, MemoryRecord
+from .models import MEMORY_TYPES, MEMORY_STATUSES, MemoryRecord
 
-__version__ = "0.1.0"
-__all__ = ["Memory", "MemoryRecord", "MEMORY_TYPES", "import_markdown"]
+__version__ = "0.2.0"
+__all__ = [
+    "Memory", "MemoryRecord", "MEMORY_TYPES", "MEMORY_STATUSES",
+    "import_markdown",
+    "detect_conflicts", "detect_stale", "health_check",
+    "hash_content", "hash_file_section",
+]

agentmem/__init__.py

@@ -8,7 +8,7 @@
 import click
 
 from .core import Memory
-from .models import MEMORY_TYPES
+from .models import MEMORY_TYPES, MEMORY_STATUSES
 
 
 def _get_mem(ctx: click.Context) -> Memory:
@@ -212,6 +212,128 @@ def load_session(ctx):
     mem.close()
 
 
+@main.command()
+@click.argument("id")
+@click.pass_context
+def promote(ctx, id):
+    """Promote a memory: hypothesis → active → validated."""
+    mem = _get_mem(ctx)
+    record = mem.promote(id)
+    if record:
+        click.echo(f"Promoted: {record.id} -> {record.status}")
+        click.echo(f"  [{record.type}] {record.title}")
+    else:
+        click.echo(f"Not found: {id}", err=True)
+        sys.exit(1)
+    mem.close()
+
+
+@main.command()
+@click.argument("id")
+@click.option("--reason", default="", help="Why this memory is deprecated.")
+@click.pass_context
+def deprecate(ctx, id, reason):
+    """Mark a memory as deprecated. Excluded from recall, kept for history."""
+    mem = _get_mem(ctx)
+    record = mem.deprecate(id, reason=reason)
+    if record:
+        click.echo(f"Deprecated: {record.id}")
+        click.echo(f"  [{record.type}] {record.title}")
+        if reason:
+            click.echo(f"  Reason: {reason}")
+    else:
+        click.echo(f"Not found: {id}", err=True)
+        sys.exit(1)
+    mem.close()
+
+
+@main.command()
+@click.pass_context
+def conflicts(ctx):
+    """Detect contradictions between active memories."""
+    from .governance import detect_conflicts
+    mem = _get_mem(ctx)
+    found = detect_conflicts(mem._conn, project=ctx.obj.get("project", ""))
+
+    if not found:
+        click.echo("No conflicts detected.")
+    else:
+        click.echo(f"Found {len(found)} potential conflict(s):\n")
+        for i, c in enumerate(found, 1):
+            icon = "!!" if c.severity == "critical" else "?"
+            click.echo(f"  {icon} Conflict {i} ({c.severity})")
+            click.echo(f"    A: [{c.memory_a.type}] {c.memory_a.title}")
+            click.echo(f"       id: {c.memory_a.id}  status: {c.memory_a.status}")
+            click.echo(f"    B: [{c.memory_b.type}] {c.memory_b.title}")
+            click.echo(f"       id: {c.memory_b.id}  status: {c.memory_b.status}")
+            click.echo(f"    Reason: {c.reason}")
+            click.echo()
+    mem.close()
+
+
+@main.command()
+@click.option("--days", default=30, help="Days without update to consider stale.")
+@click.pass_context
+def stale(ctx, days):
+    """Find memories that may be outdated."""
+    from .governance import detect_stale
+    mem = _get_mem(ctx)
+    found = detect_stale(mem._conn, project=ctx.obj.get("project", ""), stale_days=days)
+
+    if not found:
+        click.echo(f"No stale memories (threshold: {days} days).")
+    else:
+        click.echo(f"Found {len(found)} stale memory/memories (>{days} days):\n")
+        for s in found:
+            click.echo(f"  [{s.memory.type}] {s.memory.title}")
+            click.echo(f"    id: {s.memory.id}  status: {s.memory.status}")
+            click.echo(f"    Last updated: {s.days_since_update} days ago")
+            click.echo(f"    Reason: {s.reason}")
+            click.echo()
+    mem.close()
+
+
+@main.command()
+@click.option("--days", default=30, help="Stale threshold in days.")
+@click.pass_context
+def health(ctx, days):
+    """Run a full health check on the memory system."""
+    from .governance import health_check
+    mem = _get_mem(ctx)
+    report = health_check(mem._conn, project=ctx.obj.get("project", ""), stale_days=days)
+
+    click.echo(f"{'=' * 50}")
+    click.echo(f"MEMORY HEALTH: {report.health_score:.0f}/100")
+    click.echo(f"{'=' * 50}")
+    click.echo(f"  Total memories: {report.total_memories}")
+    click.echo(f"  By status:")
+    for status in ("validated", "active", "hypothesis", "deprecated", "superseded"):
+        count = report.by_status.get(status, 0)
+        if count:
+            click.echo(f"    {status}: {count}")
+    click.echo(f"  Never accessed: {report.never_accessed}")
+    click.echo()
+
+    if report.conflicts:
+        click.echo(f"  CONFLICTS: {len(report.conflicts)}")
+        for c in report.conflicts:
+            icon = "!!" if c.severity == "critical" else "?"
+            click.echo(f"    {icon} {c.memory_a.title[:40]} vs {c.memory_b.title[:40]}")
+
+    if report.stale:
+        click.echo(f"  STALE: {len(report.stale)}")
+        for s in report.stale[:5]:
+            click.echo(f"    {s.memory.title[:50]} ({s.days_since_update}d)")
+        if len(report.stale) > 5:
+            click.echo(f"    ... and {len(report.stale) - 5} more")
+
+    if report.orphaned_supersedes:
+        click.echo(f"  ORPHANED: {len(report.orphaned_supersedes)}")
+
+    click.echo(f"\n{'=' * 50}")
+    mem.close()
+
+
 @main.command()
 @click.pass_context
 def serve(ctx):

agentmem/cli.py

@@ -7,14 +7,22 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from .models import MEMORY_TYPES, MemoryRecord
+from .models import MEMORY_TYPES, MEMORY_STATUSES, MemoryRecord
 from .schema import init_db
 
 
 def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def _safe_get(row, key, default=""):
+    """Safely get a column value, returning default if column doesn't exist."""
+    try:
+        return row[key]
+    except (IndexError, KeyError):
+        return default
+
+
 def _row_to_record(row: sqlite3.Row, rank: float | None = None) -> MemoryRecord:
     return MemoryRecord(
         id=row["id"],
@@ -26,6 +34,13 @@ def _row_to_record(row: sqlite3.Row, rank: float | None = None) -> MemoryRecord:
         project=row["project"],
         confidence=row["confidence"],
         supersedes=row["supersedes"],
+        status=_safe_get(row, "status", "active"),
+        source_path=_safe_get(row, "source_path", ""),
+        source_section=_safe_get(row, "source_section", ""),
+        source_hash=_safe_get(row, "source_hash", ""),
+        validated_at=_safe_get(row, "validated_at", ""),
+        deprecated_at=_safe_get(row, "deprecated_at", ""),
+        superseded_by=_safe_get(row, "superseded_by", ""),
         created_at=row["created_at"],
         updated_at=row["updated_at"],
         accessed_at=row["accessed_at"],
@@ -67,29 +82,42 @@ def add(
         project: str | None = None,
         confidence: float = 1.0,
         supersedes: str = "",
+        status: str = "active",
+        source_path: str = "",
+        source_section: str = "",
+        source_hash: str = "",
     ) -> MemoryRecord:
         if type not in MEMORY_TYPES:
             raise ValueError(f"Invalid type '{type}'. Must be one of: {MEMORY_TYPES}")
+        if status not in MEMORY_STATUSES:
+            raise ValueError(f"Invalid status '{status}'. Must be one of: {MEMORY_STATUSES}")
 
         record_id = str(uuid.uuid4())
         now = _now()
         tags_str = ",".join(tags) if tags else ""
         proj = project if project is not None else self.project
+        validated_at = now if status == "validated" else ""
 
         self._conn.execute(
             """INSERT INTO memories
                (id, type, title, content, tags, source, project, confidence,
-                supersedes, created_at, updated_at, accessed_at, access_count)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0)""",
+                supersedes, status, source_path, source_section, source_hash,
+                validated_at, deprecated_at, superseded_by,
+                created_at, updated_at, accessed_at, access_count)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '', '', ?, ?, ?, 0)""",
             (record_id, type, title, content, tags_str, source, proj,
-             confidence, supersedes, now, now, now),
+             confidence, supersedes, status, source_path, source_section,
+             source_hash, validated_at, now, now, now),
         )
         self._conn.commit()
 
         return MemoryRecord(
             id=record_id, type=type, title=title, content=content,
             tags=tags or [], source=source, project=proj,
             confidence=confidence, supersedes=supersedes,
+            status=status, source_path=source_path,
+            source_section=source_section, source_hash=source_hash,
+            validated_at=validated_at,
             created_at=now, updated_at=now, accessed_at=now, access_count=0,
         )
 
@@ -112,7 +140,9 @@ def update(self, id: str, **kwargs) -> MemoryRecord | None:
         if not record:
             return None
 
-        allowed = {"title", "content", "tags", "type", "confidence", "project", "supersedes"}
+        allowed = {"title", "content", "tags", "type", "confidence", "project",
+                    "supersedes", "status", "source_path", "source_section",
+                    "source_hash", "validated_at", "deprecated_at", "superseded_by"}
         updates = {k: v for k, v in kwargs.items() if k in allowed and v is not None}
         if not updates:
             return record
@@ -121,6 +151,8 @@ def update(self, id: str, **kwargs) -> MemoryRecord | None:
             updates["tags"] = ",".join(updates["tags"])
         if "type" in updates and updates["type"] not in MEMORY_TYPES:
             raise ValueError(f"Invalid type '{updates['type']}'. Must be one of: {MEMORY_TYPES}")
+        if "status" in updates and updates["status"] not in MEMORY_STATUSES:
+            raise ValueError(f"Invalid status '{updates['status']}'. Must be one of: {MEMORY_STATUSES}")
 
         updates["updated_at"] = _now()
         set_clause = ", ".join(f"{k} = ?" for k in updates)
@@ -130,6 +162,70 @@ def update(self, id: str, **kwargs) -> MemoryRecord | None:
         self._conn.commit()
         return self.get(id)
 
+    # ── Truth governance lifecycle ──────────────────────────────────
+
+    def promote(self, id: str) -> MemoryRecord | None:
+        """Promote a memory: hypothesis → active → validated.
+        Each call advances one step. Validated is the highest trust level."""
+        record = self.get(id)
+        if not record:
+            return None
+
+        promotions = {"hypothesis": "active", "active": "validated"}
+        next_status = promotions.get(record.status)
+        if not next_status:
+            return record  # Already validated or in a terminal state
+
+        now = _now()
+        updates = {"status": next_status, "updated_at": now}
+        if next_status == "validated":
+            updates["validated_at"] = now
+
+        set_clause = ", ".join(f"{k} = ?" for k in updates)
+        values = list(updates.values()) + [id]
+        self._conn.execute(f"UPDATE memories SET {set_clause} WHERE id = ?", values)
+        self._conn.commit()
+        return self.get(id)
+
+    def deprecate(self, id: str, reason: str = "") -> MemoryRecord | None:
+        """Mark a memory as deprecated. Excluded from recall, kept for history."""
+        record = self.get(id)
+        if not record:
+            return None
+
+        now = _now()
+        content = record.content
+        if reason:
+            content = f"{content}\n\n[DEPRECATED {now[:10]}] {reason}"
+
+        self._conn.execute(
+            "UPDATE memories SET status = 'deprecated', deprecated_at = ?, "
+            "content = ?, updated_at = ? WHERE id = ?",
+            (now, content, now, id),
+        )
+        self._conn.commit()
+        return self.get(id)
+
+    def supersede(self, old_id: str, new_id: str) -> tuple[MemoryRecord | None, MemoryRecord | None]:
+        """Mark old_id as superseded by new_id. Old memory points to replacement."""
+        old = self.get(old_id)
+        new = self.get(new_id)
+        if not old or not new:
+            return (old, new)
+
+        now = _now()
+        self._conn.execute(
+            "UPDATE memories SET status = 'superseded', superseded_by = ?, "
+            "deprecated_at = ?, updated_at = ? WHERE id = ?",
+            (new_id, now, now, old_id),
+        )
+        self._conn.execute(
+            "UPDATE memories SET supersedes = ?, updated_at = ? WHERE id = ?",
+            (old_id, now, new_id),
+        )
+        self._conn.commit()
+        return (self.get(old_id), self.get(new_id))
+
     def delete(self, id: str) -> bool:
         cur = self._conn.execute("DELETE FROM memories WHERE id = ?", (id,))
         self._conn.commit()
@@ -210,24 +306,32 @@ def save_session(self, summary: str, tags: list[str] | None = None) -> MemoryRec
         The summary should capture: what's in progress, what's blocked, what's done,
         and any decisions made this session.
         """
-        # Find and supersede the previous session
+        # Find previous active session
         prev = self._conn.execute(
             "SELECT id FROM memories WHERE type = 'session' AND project = ? "
+            "AND COALESCE(status, 'active') = 'active' "
             "ORDER BY created_at DESC LIMIT 1",
             (self.project,),
         ).fetchone()
 
-        supersedes = prev["id"] if prev else ""
+        prev_id = prev["id"] if prev else ""
 
-        return self.add(
+        # Create new session
+        new_record = self.add(
             type="session",
             title=f"Session state — {self.project or 'default'}",
             content=summary,
             tags=tags or ["session", "state"],
             source="session",
-            supersedes=supersedes,
+            supersedes=prev_id,
         )
 
+        # Actually supersede the old session using the governance lifecycle
+        if prev_id:
+            self.supersede(prev_id, new_record.id)
+
+        return new_record
+
     def load_session(self) -> MemoryRecord | None:
         """Load the most recent session state for this project.