From 9436eabd90b16ed5f99f08dd73bfd25bfbc62363 Mon Sep 17 00:00:00 2001 From: Josh Mabry Date: Tue, 21 Jul 2026 14:17:25 -0700 Subject: [PATCH] feat: Deterministic Persona Drift Detection (#1986) - V2 --- graph/config.py | 29 ++++- graph/config_io.py | 85 +++++++++++++ server/agent_init.py | 94 ++++++++++++++ tests/test_config_roundtrip.py | 222 +++++++++++++++++++++++++++++++++ 4 files changed, 429 insertions(+), 1 deletion(-) diff --git a/graph/config.py b/graph/config.py index 89c080c8..77ae906e 100644 --- a/graph/config.py +++ b/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), diff --git a/graph/config_io.py b/graph/config_io.py index 759c8c28..de4a2f95 100644 --- a/graph/config_io.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/server/agent_init.py b/server/agent_init.py index 6d54727e..4dc8b6d3 100644 --- a/server/agent_init.py +++ b/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 diff --git a/tests/test_config_roundtrip.py b/tests/test_config_roundtrip.py index df59a1f2..0833c4e4 100644 --- a/tests/test_config_roundtrip.py +++ b/tests/test_config_roundtrip.py @@ -125,6 +125,9 @@ def _freeze_ui_tier_default(monkeypatch): "goal_verify_timeout": 120.0, "watches_enabled": False, # #2020 feature flag, default off (example keeps watches: commented) "soul_self_edit_enabled": False, + "soul_drift_enabled": True, # #1986 read-only persona-drift curation pass, on by default + "soul_drift_interval_hours": 24, + "soul_drift_threshold": 0.25, "identity_name": "protoagent", "identity_operator": "", "identity_org": "", @@ -711,3 +714,222 @@ def test_plugins_disabled_and_sources_allow_survive_config_to_dict(): "update_policy": {}, "autoupdate_interval_hours": 6, } + + +# --------------------------------------------------------------------------- +# (f) Deterministic persona drift detection (#1986) +# --------------------------------------------------------------------------- + +from types import SimpleNamespace # noqa: E402 + +from graph import config_io # noqa: E402 + + +def test_soul_drift_config_knobs_parse(tmp_path): + """soul.drift.{enabled,interval_hours,threshold} parse onto the dataclass; + the top-level `soul.self_edit_enabled` sibling still resolves alongside them.""" + path = _write_yaml( + tmp_path, + """ + soul: + self_edit_enabled: true + drift: + enabled: false + interval_hours: 12 + threshold: 0.5 + """, + ) + cfg = LangGraphConfig.from_yaml(path) + assert cfg.soul_self_edit_enabled is True + assert cfg.soul_drift_enabled is False + assert cfg.soul_drift_interval_hours == 12 + assert cfg.soul_drift_threshold == 0.5 + + +def test_soul_drift_defaults_when_section_absent_or_null(tmp_path): + """No soul block -> defaults; a present-but-null `soul.drift:` also falls back to + defaults (nested null isn't caught by the top-level normalizer, hence the `or {}`).""" + cfg_absent = LangGraphConfig.from_yaml(_write_yaml(tmp_path, "model:\n name: foo\n")) + assert cfg_absent.soul_drift_enabled is True + assert cfg_absent.soul_drift_interval_hours == 24 + assert cfg_absent.soul_drift_threshold == 0.25 + + null_dir = tmp_path / "null_drift" + null_dir.mkdir() + cfg_null = LangGraphConfig.from_yaml(_write_yaml(null_dir, "soul:\n drift:\n")) + assert cfg_null.soul_drift_enabled is True + assert cfg_null.soul_drift_interval_hours == 24 + assert cfg_null.soul_drift_threshold == 0.25 + + +def test_compute_soul_drift_identical_is_no_drift(): + """Identical persona texts -> retention 1.0, score 0.0, no section churn.""" + report = config_io.compute_soul_drift("# Identity\nI am steady.", "# Identity\nI am steady.") + assert report["retention"] == 1.0 + assert report["score"] == 0.0 + assert report["size_delta"] == 0 + assert report["sections_added"] == [] and report["sections_dropped"] == [] + + +def test_compute_soul_drift_signals_size_and_section_churn(): + """A rewrite is scored deterministically: net size delta, section churn (added + + dropped markdown headings), and a difflib retention ratio strictly between 0 and 1.""" + baseline = "# Identity\nCalm and precise.\n\n# Voice\nTerse.\n" + current = "# Identity\nCalm and precise.\n\n# Persona\nLoud and rambling and very different now.\n" + report = config_io.compute_soul_drift(baseline, current) + assert report["sections_added"] == ["Persona"] + assert report["sections_dropped"] == ["Voice"] + assert report["size_delta"] == len(current) - len(baseline) + assert report["baseline_size"] == len(baseline) + assert report["current_size"] == len(current) + assert 0.0 < report["score"] < 1.0 + assert report["retention"] == round(1.0 - report["score"], 4) + assert "retained" in report["rationale"] + + +def _seed_soul_history(monkeypatch, tmp_path, baseline: str, current: str): + """Two persona saves so the soul-history holds `baseline` (earliest snapshot) and + the live SOUL.md holds `current` — the exact shape detect_soul_drift compares.""" + home = tmp_path / "home" + monkeypatch.setenv("PROTOAGENT_HOME", str(home)) + # Neutralize the bundled seed so the first save has no prior persona to archive. + monkeypatch.setattr(config_io, "soul_source_path", lambda: tmp_path / "no-seed.md") + config_io.write_soul(baseline) # live = baseline; history empty (nothing to archive) + config_io.write_soul(current) # archives baseline; live = current + return home + + +def test_detect_soul_drift_none_without_history(monkeypatch, tmp_path): + """A single persona (no archived baseline) -> nothing to compare -> None, so the + curation pass stays silent on a fresh instance.""" + home = tmp_path / "home" + monkeypatch.setenv("PROTOAGENT_HOME", str(home)) + monkeypatch.setattr(config_io, "soul_source_path", lambda: tmp_path / "no-seed.md") + config_io.write_soul("# Identity\nOnly ever one version.\n") + assert config_io.list_soul_versions() == [] + assert config_io.detect_soul_drift() is None + + +def test_detect_soul_drift_compares_against_earliest_snapshot(monkeypatch, tmp_path): + """detect_soul_drift diffs the live persona against the EARLIEST recorded snapshot + and tags the report with that baseline id.""" + _seed_soul_history( + monkeypatch, + tmp_path, + baseline="# Identity\nThe original, careful persona.\n", + current="# Identity\nA wholly rewritten, very different persona now.\n", + ) + versions = config_io.list_soul_versions() + assert len(versions) == 1 + report = config_io.detect_soul_drift() + assert report is not None + assert report["baseline_id"] == versions[-1]["id"] # earliest snapshot + assert report["baseline_size"] == len("# Identity\nThe original, careful persona.\n") + assert 0.0 < report["score"] <= 1.0 + + +def test_run_soul_drift_pass_publishes_when_over_threshold(monkeypatch, tmp_path): + """The curation pass publishes `persona.drift_detected` (score + signals + rationale) + to the event bus when the drift score crosses the configured threshold.""" + from server import agent_init + + _seed_soul_history( + monkeypatch, + tmp_path, + baseline="# Identity\nThe original persona, steady and precise.\n", + current="totally different content with no overlap whatsoever xyz 123\n", + ) + published = [] + monkeypatch.setattr( + agent_init, + "_event_bus", + SimpleNamespace(publish=lambda event, data=None: published.append((event, data))), + ) + cfg = SimpleNamespace(soul_drift_enabled=True, soul_drift_threshold=0.1) + report = agent_init._run_soul_drift_pass(cfg) + + assert report is not None + assert len(published) == 1 + event, data = published[0] + assert event == "persona.drift_detected" + assert data["score"] == report["score"] and data["score"] >= 0.1 + assert data["threshold"] == 0.1 + assert data["rationale"] == report["rationale"] + # Every deterministic signal rides along for a verifiable payload. + assert set(data["signals"]) == { + "retention", + "size_delta", + "baseline_size", + "current_size", + "sections_added", + "sections_dropped", + } + + +def test_run_soul_drift_pass_silent_when_below_threshold(monkeypatch, tmp_path): + """A near-identical persona scores below threshold -> the pass runs but publishes + nothing (still returns the report for observability).""" + from server import agent_init + + _seed_soul_history( + monkeypatch, + tmp_path, + baseline="# Identity\nThe original persona, steady and precise.\n", + current="# Identity\nThe original persona, steady and precise!\n", # one char changed + ) + published = [] + monkeypatch.setattr( + agent_init, + "_event_bus", + SimpleNamespace(publish=lambda event, data=None: published.append((event, data))), + ) + cfg = SimpleNamespace(soul_drift_enabled=True, soul_drift_threshold=0.5) + report = agent_init._run_soul_drift_pass(cfg) + assert report is not None and report["score"] < 0.5 + assert published == [] + + +def test_run_soul_drift_pass_noop_when_disabled(monkeypatch, tmp_path): + """soul_drift_enabled=False -> the pass is a no-op (no detection, no publish).""" + from server import agent_init + + _seed_soul_history( + monkeypatch, + tmp_path, + baseline="# Identity\nBaseline.\n", + current="# Identity\nDrastically different now, entirely rewritten.\n", + ) + published = [] + monkeypatch.setattr( + agent_init, + "_event_bus", + SimpleNamespace(publish=lambda event, data=None: published.append((event, data))), + ) + cfg = SimpleNamespace(soul_drift_enabled=False, soul_drift_threshold=0.0) + assert agent_init._run_soul_drift_pass(cfg) is None + assert published == [] + + +def test_maybe_run_soul_drift_pass_gates_to_interval(monkeypatch, tmp_path): + """The interval gate runs the pass once, then suppresses further runs until the + cadence elapses (verified by counting detect_soul_drift calls).""" + from server import agent_init + + monkeypatch.setattr(agent_init, "_last_soul_drift_check", 0.0) + calls = {"n": 0} + + def _fake_run(cfg): + calls["n"] += 1 + return {"score": 0.0} + + monkeypatch.setattr(agent_init, "_run_soul_drift_pass", _fake_run) + cfg = SimpleNamespace(soul_drift_enabled=True, soul_drift_interval_hours=24) + + assert agent_init._maybe_run_soul_drift_pass(cfg) is not None # first tick runs + assert agent_init._maybe_run_soul_drift_pass(cfg) is None # gated within the interval + assert calls["n"] == 1 + + # interval_hours=0 disables the pass outright. + monkeypatch.setattr(agent_init, "_last_soul_drift_check", 0.0) + disabled = SimpleNamespace(soul_drift_enabled=True, soul_drift_interval_hours=0) + assert agent_init._maybe_run_soul_drift_pass(disabled) is None