feat: Deterministic Persona Drift Detection (#1986) - V2

graph/config.py

@@ -475,6 +475,22 @@ class LangGraphConfig:
     # opts in via ``soul.self_edit_enabled: true``.
     soul_self_edit_enabled: bool = False
 
+    # Deterministic persona drift detection (#1986). A read-only curation pass
+    # (lineage of the dream/distill maintenance passes) periodically diffs the live
+    # ``SOUL.md`` against its EARLIEST recorded soul-history snapshot — net size
+    # delta, section churn (added/dropped markdown headings), and a ``difflib``
+    # retention ratio — and publishes a ``persona.drift_detected`` event on the bus
+    # when the drift score (``1 - retention``, 0..1) crosses ``threshold``. It never
+    # rewrites the persona: recovery already exists via the restore endpoint (ADR
+    # 0081), so this only SURFACES the signal. ``interval_hours`` sets the cadence
+    # (0 disables the pass); the pass is hosted by the checkpoint-prune maintenance
+    # loop, so it runs at most once per interval. Cheap (a diff of two small text
+    # files) and a true no-op until there's history to compare against, so it's on
+    # by default like the other curation passes.
+    soul_drift_enabled: bool = True
+    soul_drift_interval_hours: int = 24
+    soul_drift_threshold: float = 0.25
+
     # Knowledge store — sqlite + FTS5, see ``knowledge/store.py``.
     # The default path lives under ``/sandbox/`` to play well with the
     # bundled Docker volume; the store falls back to
@@ -1013,6 +1029,10 @@ def from_dict(
         auth = data.get("auth", {})
         runtime = data.get("runtime", {})
         operator = data.get("operator", {})
+        # `or {}` at each level: a present-but-empty `soul:` or `soul.drift:` block
+        # parses to None (top-level null is normalized above, a nested null is not).
+        soul = data.get("soul", {}) or {}
+        soul_drift = soul.get("drift", {}) or {}
         # Box runtime (Host layer, ADR 0047 D8) — `or {}` because a present-but-empty
         # section parses to None.
         network = data.get("network", {}) or {}
@@ -1080,7 +1100,14 @@ def from_dict(
             goal_eval_model=data.get("goal", {}).get("eval_model", cls.goal_eval_model),
             goal_verify_timeout=data.get("goal", {}).get("verify_timeout", cls.goal_verify_timeout),
             watches_enabled=data.get("watches", {}).get("enabled", cls.watches_enabled),
-            soul_self_edit_enabled=data.get("soul", {}).get("self_edit_enabled", cls.soul_self_edit_enabled),
+            soul_self_edit_enabled=soul.get("self_edit_enabled", cls.soul_self_edit_enabled),
+            soul_drift_enabled=bool(soul_drift.get("enabled", cls.soul_drift_enabled)),
+            soul_drift_interval_hours=int(
+                soul_drift.get("interval_hours", cls.soul_drift_interval_hours) or 0
+            ),
+            soul_drift_threshold=float(
+                soul_drift.get("threshold", cls.soul_drift_threshold) or 0.0
+            ),
             subagent_max_concurrency=subagents.get("max_concurrency", cls.subagent_max_concurrency),
             subagent_output_truncate=subagents.get("output_truncate", cls.subagent_output_truncate),
             knowledge_db_path=knowledge.get("db_path", cls.knowledge_db_path),

graph/config_io.py

@@ -26,6 +26,7 @@
 
 from __future__ import annotations
 
+import difflib
 import hashlib
 import logging
 import os
@@ -1025,6 +1026,90 @@ def read_soul_version(version_id: str) -> str | None:
         return None
 
 
+# ---------------------------------------------------------------------------
+# Persona drift detection (#1986)
+# ---------------------------------------------------------------------------
+# A deterministic, read-only curation pass: diff the live SOUL.md against a
+# baseline (the EARLIEST recorded soul-history snapshot, or an explicit pinned
+# version) and quantify how far the persona has drifted. The signals are pure
+# functions of the two texts — net size delta, section churn (added/dropped
+# markdown headings), and a difflib retention ratio — so identical inputs always
+# yield an identical score (no clock, no I/O, no model). Recovery is out of scope
+# (the restore endpoint already exists); this only surfaces the signal so a
+# background maintenance pass can publish ``persona.drift_detected``.
+
+# Markdown ATX heading (``#`` … ``######``) — the persona's section boundaries.
+_SOUL_SECTION_RE = re.compile(r"^#{1,6}[ \t]+(.+?)[ \t]*#*[ \t]*$", re.MULTILINE)
+
+
+def _soul_sections(text: str) -> list[str]:
+    """The markdown heading titles in a persona doc, in document order."""
+    return [m.group(1).strip() for m in _SOUL_SECTION_RE.finditer(text or "")]
+
+
+def compute_soul_drift(baseline: str, current: str) -> dict:
+    """Deterministic drift signals between a ``baseline`` persona and the ``current`` one.
+
+    Returns ``{score, retention, size_delta, baseline_size, current_size,
+    sections_added, sections_dropped, rationale}``. ``retention`` is difflib's
+    ``SequenceMatcher`` ratio (1.0 = identical); ``score`` is ``1 - retention``
+    (0.0 = no drift, 1.0 = total). ``sections_*`` compare the markdown heading
+    sets. Pure function of its two inputs — same inputs, same output."""
+    baseline = baseline or ""
+    current = current or ""
+    retention = difflib.SequenceMatcher(None, baseline, current, autojunk=False).ratio()
+    score = round(1.0 - retention, 4)
+    base_sections, cur_sections = _soul_sections(baseline), _soul_sections(current)
+    base_set, cur_set = set(base_sections), set(cur_sections)
+    sections_added = [s for s in cur_sections if s not in base_set]
+    sections_dropped = [s for s in base_sections if s not in cur_set]
+    size_delta = len(current) - len(baseline)
+    rationale = (
+        f"{round(retention * 100)}% of the baseline persona retained; "
+        f"size {size_delta:+d} chars ({len(baseline)}→{len(current)}); "
+        f"sections +{len(sections_added)}/-{len(sections_dropped)}"
+    )
+    return {
+        "score": score,
+        "retention": round(retention, 4),
+        "size_delta": size_delta,
+        "baseline_size": len(baseline),
+        "current_size": len(current),
+        "sections_added": sections_added,
+        "sections_dropped": sections_dropped,
+        "rationale": rationale,
+    }
+
+
+def detect_soul_drift(baseline_version_id: str | None = None) -> dict | None:
+    """Compare the live persona against a baseline soul-history snapshot.
+
+    The baseline is ``baseline_version_id`` when given (a pinned version), else the
+    EARLIEST recorded snapshot (the origin persona). Returns the
+    :func:`compute_soul_drift` report augmented with ``baseline_id`` /
+    ``baseline_saved_at``, or ``None`` when there is nothing to compare — no
+    recorded history, an unreadable/invalid baseline, or no live persona — so a
+    caller can cleanly skip publishing. Read-only and best-effort."""
+    current = read_soul()
+    if not current:
+        return None
+    if baseline_version_id:
+        baseline_id = baseline_version_id
+        baseline = read_soul_version(baseline_version_id)
+    else:
+        versions = list_soul_versions()  # newest first — the earliest is last
+        if not versions:
+            return None
+        baseline_id = versions[-1]["id"]
+        baseline = read_soul_version(baseline_id)
+    if baseline is None:
+        return None
+    report = compute_soul_drift(baseline, current)
+    report["baseline_id"] = baseline_id
+    report["baseline_saved_at"] = _soul_version_saved_at(baseline_id)
+    return report
+
+
 # ---------------------------------------------------------------------------
 # Gateway model discovery
 # ---------------------------------------------------------------------------

server/agent_init.py

@@ -843,6 +843,15 @@ async def _checkpoint_prune_loop() -> None:
                         log.info("[a2a-task-prune] removed %d orphaned push-config(s)", orphaned)
             except Exception:
                 log.exception("[a2a-task-prune] sweep failed")
+        # Persona drift curation (#1986) — a read-only diff of the live SOUL.md
+        # against its earliest recorded baseline snapshot; publishes
+        # persona.drift_detected when the deterministic drift score crosses the
+        # threshold. Self-gated to its own soul.drift.interval_hours cadence and
+        # best-effort, so it never blocks the prune sweep.
+        try:
+            await asyncio.to_thread(_maybe_run_soul_drift_pass, cfg)
+        except Exception:
+            log.exception("[soul-drift] pass failed")
         # Tick at the checkpoint interval if set, else hourly (so telemetry pruning
         # still runs when checkpoint pruning is off).
         await asyncio.sleep(max(1, interval_h or 1) * 3600)
@@ -906,6 +915,91 @@ async def _retire_thread(thread_id: str, *, harvest: bool | None = None, cascade
     return chunk_id
 
 
+# ── Persona drift curation (#1986) ───────────────────────────────────────────
+# A read-only curation pass (lineage of the dream/distill maintenance passes):
+# periodically diff the live SOUL.md against its earliest recorded soul-history
+# snapshot and, when the deterministic drift score crosses ``soul.drift.threshold``,
+# publish ``persona.drift_detected`` on the event bus. It never rewrites the
+# persona — recovery already exists via the restore endpoint (ADR 0081); this only
+# surfaces the signal. Hosted by the checkpoint-prune loop below (no new startup
+# wiring), but gated to its own ``soul.drift.interval_hours`` cadence so a fast
+# prune interval doesn't over-run it.
+_last_soul_drift_check: float = 0.0
+
+
+def _run_soul_drift_pass(cfg) -> dict | None:
+    """Run one persona-drift curation pass NOW (ignores the interval gate).
+
+    Returns the drift report when a comparison ran (whether or not it crossed the
+    threshold), else ``None`` (feature off, or nothing to compare against yet).
+    Publishes ``persona.drift_detected`` — carrying the score, the individual
+    signals, and a human-readable rationale — only when the score is at or above
+    ``soul.drift.threshold``. Best-effort: detection/publish failures are logged,
+    never raised into the hosting loop."""
+    if cfg is None or not getattr(cfg, "soul_drift_enabled", False):
+        return None
+    try:
+        from graph.config_io import detect_soul_drift
+
+        report = detect_soul_drift()
+    except Exception:
+        log.exception("[soul-drift] detection failed")
+        return None
+    if report is None:
+        return None  # no baseline snapshot / no live persona — nothing to compare
+
+    threshold = float(getattr(cfg, "soul_drift_threshold", 1.0) or 0.0)
+    if report["score"] >= threshold:
+        try:
+            _event_bus.publish(
+                "persona.drift_detected",
+                {
+                    "score": report["score"],
+                    "threshold": threshold,
+                    "baseline_id": report.get("baseline_id"),
+                    "baseline_saved_at": report.get("baseline_saved_at"),
+                    "signals": {
+                        "retention": report["retention"],
+                        "size_delta": report["size_delta"],
+                        "baseline_size": report["baseline_size"],
+                        "current_size": report["current_size"],
+                        "sections_added": report["sections_added"],
+                        "sections_dropped": report["sections_dropped"],
+                    },
+                    "rationale": report["rationale"],
+                },
+            )
+            log.info(
+                "[soul-drift] persona.drift_detected score=%.3f (threshold %.3f) — %s",
+                report["score"],
+                threshold,
+                report["rationale"],
+            )
+        except Exception:
+            log.exception("[soul-drift] event publish failed")
+    return report
+
+
+def _maybe_run_soul_drift_pass(cfg) -> dict | None:
+    """Gate :func:`_run_soul_drift_pass` to its own ``soul.drift.interval_hours``
+    cadence, then run it. Called every prune tick; a no-op until an interval has
+    elapsed since the last run (tracked on a monotonic module global so a config
+    reload re-paces without restarting the loop). ``interval_hours`` ``0`` disables."""
+    global _last_soul_drift_check
+    if cfg is None or not getattr(cfg, "soul_drift_enabled", False):
+        return None
+    interval_h = getattr(cfg, "soul_drift_interval_hours", 0) or 0
+    if interval_h <= 0:
+        return None
+    import time
+
+    now = time.monotonic()
+    if _last_soul_drift_check and (now - _last_soul_drift_check) < interval_h * 3600:
+        return None
+    _last_soul_drift_check = now
+    return _run_soul_drift_pass(cfg)
+
+
 # ── Opt-in plugin auto-update (#1720) ────────────────────────────────────────
 # Only plugins the operator lists in ``plugins.update_policy`` are ever touched;
 # a pinned-to-SHA plugin is never auto-updated. ``when: idle`` defers a plugin's