482 lines
19 KiB
Python
482 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""drift_core.py — Governed Adaptation core engine.
|
||
|
||
The system evolves, but under governance — not wild change, not rigid control.
|
||
Identity anchors (locked entries) define who the agent is. Evolving state
|
||
(provisional/validated entries) adapts through proven access cycles.
|
||
|
||
This module implements:
|
||
- Multi-factor value function (proven_uses × importance × recency × reliability)
|
||
- Trust level lifecycle (provisional → validated → locked)
|
||
- FAMA metric (forgetting-aware memory accuracy)
|
||
- Promotion gate (three critics before locking)
|
||
- State evolution monitoring (velocity, direction)
|
||
|
||
Research basis:
|
||
- AMV-L (arXiv:2603.04443): value-driven promotion/demotion/eviction
|
||
- MARS (arXiv:2605.14401): reinforcement/weakening/consolidation lifecycle
|
||
- SmartVector (arXiv:2604.20598): confidence decay and access reinforcement
|
||
- MemCon (arXiv:2607.13591): learned policy for consolidation/forgetting
|
||
- Learning What to Remember (arXiv:2606.12945): multi-factor value function
|
||
- VaG (arXiv:2608.05810): pre-commit gating (post-hoc rollback insufficient)
|
||
- PoEM (arXiv:2608.16032): tamper-evident execution ledger
|
||
- Memora/FAMA (arXiv:2604.20006): forgetting-aware memory accuracy
|
||
|
||
Run under ~/.venvs/memory/bin/python3 (has sqlite-vec).
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts")
|
||
sys.path.insert(0, SCRIPTS_DIR)
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# CONSTANTS
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
# Trust level lifecycle: provisional → validated → locked
|
||
# An entry becomes validated after PROVEN_USES_THRESHOLD proven accesses.
|
||
# An entry becomes locked after passing the promotion gate twice.
|
||
PROVEN_USES_THRESHOLD = 20 # proven accesses before validation eligible
|
||
RECENCY_HALF_LIFE_DAYS = 30.0 # recency decay half-life
|
||
RELIABILITY_WINDOW = 50 # recent accesses for reliability calc
|
||
STATE_EVOLUTION_VELOCITY_THRESHOLD = 10 # changes per day before alert
|
||
STATE_EVOLUTION_DIRECTION_WINDOW = 7 # days for direction calculation
|
||
|
||
# Multi-factor weights (from Learning What to Remember, arXiv:2606.12945)
|
||
WEIGHT_PROVEN_USES = 0.30 # survival count (how many times proven useful)
|
||
WEIGHT_IMPORTANCE = 0.25 # assigned importance
|
||
WEIGHT_RECENCY = 0.20 # time since last access
|
||
WEIGHT_RELIABILITY = 0.15 # consistency of access pattern
|
||
WEIGHT_GOAL_RELEVANCE = 0.10 # relevance to current goals
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# MULTI-FACTOR VALUE FUNCTION
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def compute_value(entry: dict, current_goals: List[str] = None) -> float:
|
||
"""Multi-factor value score. Returns 0.0-1.0+.
|
||
|
||
How much is this entry worth? Higher = more proven, more useful, more trusted.
|
||
|
||
Factors (from arXiv:2606.12945):
|
||
- proven_uses: how many times this entry proved useful (logarithmic scaling)
|
||
- importance: assigned importance weight
|
||
- recency: exponential decay from last access
|
||
- reliability: consistency of access pattern
|
||
- goal_relevance: relevance to current goals (optional)
|
||
"""
|
||
proven = entry.get("wash_count", 0) or 0 # column name preserved for compatibility
|
||
imp = entry.get("importance", 0.5) or 0.5
|
||
|
||
# Proven use factor: logarithmic scaling (diminishing returns)
|
||
proven_factor = min(1.0, math.log1p(proven) / math.log1p(PROVEN_USES_THRESHOLD))
|
||
|
||
# Recency factor: exponential decay
|
||
recency = 1.0
|
||
last_accessed = entry.get("last_accessed") or entry.get("last_washed")
|
||
if last_accessed:
|
||
try:
|
||
last = datetime.fromisoformat(last_accessed)
|
||
hours_since = max(0.1, (datetime.now(timezone.utc) - last).total_seconds() / 3600)
|
||
recency = max(0.1, 1.0 / (1.0 + hours_since / (RECENCY_HALF_LIFE_DAYS * 24)))
|
||
except (ValueError, TypeError):
|
||
recency = 0.5
|
||
|
||
# Reliability factor: based on access pattern consistency
|
||
# (placeholder — full implementation needs access history tracking)
|
||
reliability = min(1.0, proven / max(1, RELIABILITY_WINDOW))
|
||
|
||
# Goal relevance factor: keyword overlap with current goals
|
||
goal_relevance = 0.5 # default when no goals provided
|
||
if current_goals:
|
||
text = entry.get("text", "").lower()
|
||
goal_words = set()
|
||
for g in current_goals:
|
||
goal_words.update(g.lower().split())
|
||
text_words = set(text.split())
|
||
overlap = len(goal_words & text_words)
|
||
goal_relevance = min(1.0, overlap / max(1, len(goal_words) * 0.3))
|
||
|
||
# Weighted combination
|
||
value = (
|
||
WEIGHT_PROVEN_USES * proven_factor +
|
||
WEIGHT_IMPORTANCE * imp +
|
||
WEIGHT_RECENCY * recency +
|
||
WEIGHT_RELIABILITY * reliability +
|
||
WEIGHT_GOAL_RELEVANCE * goal_relevance
|
||
)
|
||
|
||
return round(value, 4)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# PROMOTION LIFECYCLE
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def check_promotion_eligibility(entry: dict) -> Tuple[bool, str]:
|
||
"""Check if an entry is eligible for promotion to rock.
|
||
|
||
Requirements:
|
||
- proven_uses >= PROVEN_USES_THRESHOLD
|
||
- value score >= 0.6
|
||
- rock_status == 'washable' or 'established'
|
||
|
||
Returns: (eligible, reason)
|
||
"""
|
||
wash = entry.get("wash_count", 0) or 0
|
||
status = entry.get("rock_status", "washable")
|
||
|
||
if status == "rock":
|
||
return False, "already rock"
|
||
|
||
if wash < PROVEN_USES_THRESHOLD:
|
||
return False, f"proven_uses {wash} < threshold {PROVEN_USES_THRESHOLD}"
|
||
|
||
value = compute_value(entry)
|
||
if value < 0.6:
|
||
return False, f"value {value:.3f} < 0.6 minimum"
|
||
|
||
return True, f"wash_count={wash}, value={value:.3f}"
|
||
|
||
|
||
def promote_entry(db, entry_id: int) -> dict:
|
||
"""Promote an entry through the lifecycle: washable → established → rock.
|
||
|
||
Returns dict with promotion result.
|
||
"""
|
||
import memory_store
|
||
|
||
row = db.execute(
|
||
"SELECT id, text, wash_count, importance, rock_status FROM entries WHERE id = ?",
|
||
(entry_id,)
|
||
).fetchone()
|
||
|
||
if not row:
|
||
return {"success": False, "reason": "entry not found"}
|
||
|
||
entry = dict(row)
|
||
current_status = entry.get("rock_status", "washable")
|
||
|
||
if current_status == "rock":
|
||
return {"success": False, "reason": "already rock"}
|
||
|
||
eligible, reason = check_promotion_eligibility(entry)
|
||
if not eligible:
|
||
return {"success": False, "reason": reason}
|
||
|
||
# Determine next status
|
||
if current_status == "washable":
|
||
new_status = "established"
|
||
elif current_status == "established":
|
||
new_status = "rock"
|
||
else:
|
||
new_status = "established"
|
||
|
||
db.execute(
|
||
"UPDATE entries SET rock_status = ?, promotion_eligible = ? WHERE id = ?",
|
||
(new_status, 1 if new_status != "rock" else 0, entry_id)
|
||
)
|
||
db.commit()
|
||
|
||
return {
|
||
"success": True,
|
||
"entry_id": entry_id,
|
||
"old_status": current_status,
|
||
"new_status": new_status,
|
||
"wash_count": entry.get("wash_count", 0),
|
||
"reason": reason
|
||
}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# FAMA METRIC (Forgetting-Aware Memory Accuracy)
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def compute_fama(db, query: str, retrieved_ids: List[int]) -> float:
|
||
"""FAMA: Forgetting-Aware Memory Accuracy (arXiv:2604.20006).
|
||
|
||
Penalizes reliance on obsolete or invalidated memory.
|
||
Returns 0.0-1.0 where 1.0 = all retrieved memories are valid and relevant.
|
||
"""
|
||
if not retrieved_ids:
|
||
return 0.0
|
||
|
||
valid_count = 0
|
||
for eid in retrieved_ids:
|
||
row = db.execute(
|
||
"SELECT status, rock_status, wash_count FROM entries WHERE id = ?",
|
||
(eid,)
|
||
).fetchone()
|
||
if not row:
|
||
continue
|
||
# Valid if: active status, not obsolete (rock or established preferred)
|
||
if row["status"] == "active":
|
||
# Higher wash_count = more validated
|
||
wash = row["wash_count"] or 0
|
||
if wash > 0:
|
||
valid_count += 1
|
||
elif row["rock_status"] in ("established", "rock"):
|
||
valid_count += 1
|
||
|
||
return valid_count / len(retrieved_ids)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# DRIFT OBSERVABILITY
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def compute_drift_velocity(db, days: int = 7) -> dict:
|
||
"""Drift velocity: how fast is the system changing?
|
||
|
||
Returns dict with velocity metrics.
|
||
"""
|
||
from datetime import timedelta
|
||
|
||
cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
|
||
|
||
# Count changes in the window
|
||
changes = db.execute(
|
||
"SELECT COUNT(*) FROM entries WHERE last_washed > ? AND status = 'active'",
|
||
(cutoff,)
|
||
).fetchone()[0]
|
||
|
||
# Count promotions in the window
|
||
promotions = db.execute(
|
||
"SELECT COUNT(*) FROM entries WHERE rock_status != 'washable' AND last_washed > ? AND status = 'active'",
|
||
(cutoff,)
|
||
).fetchone()[0]
|
||
|
||
velocity = changes / max(1, days)
|
||
promotion_rate = promotions / max(1, days)
|
||
|
||
return {
|
||
"changes_per_day": round(velocity, 2),
|
||
"promotions_per_day": round(promotion_rate, 2),
|
||
"total_changes": changes,
|
||
"total_promotions": promotions,
|
||
"window_days": days,
|
||
"alert": velocity > STATE_EVOLUTION_VELOCITY_THRESHOLD
|
||
}
|
||
|
||
|
||
def compute_drift_direction(db) -> dict:
|
||
"""Drift direction: is the system moving toward or away from stability?
|
||
|
||
Measures: ratio of established/rock entries to total entries.
|
||
Increasing ratio = positive drift (toward stability).
|
||
Decreasing ratio = negative drift (away from stability).
|
||
"""
|
||
total = db.execute(
|
||
"SELECT COUNT(*) FROM entries WHERE status = 'active'"
|
||
).fetchone()[0]
|
||
|
||
established = db.execute(
|
||
"SELECT COUNT(*) FROM entries WHERE rock_status IN ('established', 'rock') AND status = 'active'"
|
||
).fetchone()[0]
|
||
|
||
rock = db.execute(
|
||
"SELECT COUNT(*) FROM entries WHERE rock_status = 'rock' AND status = 'active'"
|
||
).fetchone()[0]
|
||
|
||
stability_ratio = established / max(1, total)
|
||
rock_ratio = rock / max(1, total)
|
||
|
||
return {
|
||
"total_entries": total,
|
||
"established_count": established,
|
||
"rock_count": rock,
|
||
"stability_ratio": round(stability_ratio, 4),
|
||
"rock_ratio": round(rock_ratio, 4),
|
||
"direction": "stable" if stability_ratio > 0.3 else "drifting"
|
||
}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# PRE-COMMIT GATE (Three Critics)
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def pre_commit_gate(db, entry_id: int) -> dict:
|
||
"""Three critics before promotion (VaG, arXiv:2608.05810).
|
||
|
||
Critics:
|
||
1. Structural validity: is the entry well-formed?
|
||
2. Behavioral harmlessness: does the entry align with the immutable core?
|
||
3. Semantic consistency: does the entry contradict existing rocks?
|
||
|
||
Returns dict with verdict and critic details.
|
||
"""
|
||
|
||
|
||
def pre_commit_gate_entry(entry: dict) -> dict:
|
||
"""Three critics for a dict entry (no db lookup needed).
|
||
|
||
Used by curator when promoting candidates that don't have entry_ids yet.
|
||
"""
|
||
text = entry.get("text", "")
|
||
critics = {}
|
||
|
||
# Critic 1: Structural validity
|
||
if len(text) < 5:
|
||
critics["structural"] = {"pass": False, "reason": "text too short"}
|
||
elif len(text) > 2000:
|
||
critics["structural"] = {"pass": False, "reason": "text too long"}
|
||
elif not text.strip():
|
||
critics["structural"] = {"pass": False, "reason": "empty text"}
|
||
else:
|
||
critics["structural"] = {"pass": True, "reason": "well-formed"}
|
||
|
||
# Critic 2: Behavioral harmlessness
|
||
harmful_patterns = ["refuse to", "cannot comply", "safety violation", "I'm sorry, but I can't"]
|
||
is_harmful = any(p in text.lower() for p in harmful_patterns)
|
||
critics["behavioral"] = {
|
||
"pass": not is_harmful,
|
||
"reason": "contains refusal language" if is_harmful else "aligned with core"
|
||
}
|
||
|
||
# Critic 3: Semantic consistency (simplified — no db lookup)
|
||
critics["semantic"] = {"pass": True, "reason": "skipped (no db context)"}
|
||
|
||
all_pass = all(c["pass"] for c in critics.values())
|
||
verdict = "PASS" if all_pass else "REJECT"
|
||
|
||
return {
|
||
"verdict": verdict,
|
||
"critics": critics,
|
||
"reason": "all critics passed" if all_pass else "one or more critics failed"
|
||
}
|
||
|
||
# Critic 1: Structural validity
|
||
text = entry.get("text", "")
|
||
if len(text) < 5:
|
||
critics["structural"] = {"pass": False, "reason": "text too short"}
|
||
elif len(text) > 2000:
|
||
critics["structural"] = {"pass": False, "reason": "text too long"}
|
||
elif not text.strip():
|
||
critics["structural"] = {"pass": False, "reason": "empty text"}
|
||
else:
|
||
critics["structural"] = {"pass": True, "reason": "well-formed"}
|
||
|
||
# Critic 2: Behavioral harmlessness (check against immutable core)
|
||
# The immutable core: user authority, no safety theater, no fights
|
||
harmful_patterns = ["refuse to", "cannot comply", "safety violation", "I'm sorry, but I can't"]
|
||
is_harmful = any(p in text.lower() for p in harmful_patterns)
|
||
critics["behavioral"] = {
|
||
"pass": not is_harmful,
|
||
"reason": "contains refusal language" if is_harmful else "aligned with core"
|
||
}
|
||
|
||
# Critic 3: Semantic consistency (check for contradictions with existing rocks)
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="Drift-with-Integrity core engine")
|
||
sub = parser.add_subparsers(dest="cmd")
|
||
|
||
# value <entry_id>
|
||
p_val = sub.add_parser("value", help="Compute multi-factor value for an entry")
|
||
p_val.add_argument("entry_id", type=int)
|
||
|
||
# check-promotion <entry_id>
|
||
p_check = sub.add_parser("check-promotion", help="Check promotion eligibility")
|
||
p_check.add_argument("entry_id", type=int)
|
||
|
||
# promote <entry_id>
|
||
p_promote = sub.add_parser("promote", help="Promote an entry")
|
||
p_promote.add_argument("entry_id", type=int)
|
||
|
||
# gate <entry_id>
|
||
p_gate = sub.add_parser("gate", help="Run pre-commit gate")
|
||
p_gate.add_argument("entry_id", type=int)
|
||
|
||
# velocity [--days N]
|
||
p_vel = sub.add_parser("velocity", help="Compute drift velocity")
|
||
p_vel.add_argument("--days", type=int, default=7)
|
||
|
||
# direction
|
||
sub.add_parser("direction", help="Compute drift direction")
|
||
|
||
# status
|
||
sub.add_parser("status", help="Show drift system status")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.cmd:
|
||
parser.print_help()
|
||
sys.exit(1)
|
||
|
||
import memory_store
|
||
db = memory_store.connect()
|
||
|
||
if args.cmd == "value":
|
||
row = db.execute(
|
||
"SELECT id, text, importance, wash_count, last_accessed, last_washed, rock_status FROM entries WHERE id = ?",
|
||
(args.entry_id,)
|
||
).fetchone()
|
||
if not row:
|
||
print(f"Entry {args.entry_id} not found")
|
||
sys.exit(1)
|
||
entry = dict(row)
|
||
val = compute_value(entry)
|
||
print(f"Entry {args.entry_id}: value={val:.4f}")
|
||
print(f" wash_count={entry.get('wash_count', 0)}")
|
||
print(f" importance={entry.get('importance', 0.5)}")
|
||
print(f" rock_status={entry.get('rock_status', 'washable')}")
|
||
|
||
elif args.cmd == "check-promotion":
|
||
row = db.execute(
|
||
"SELECT id, text, importance, wash_count, rock_status FROM entries WHERE id = ?",
|
||
(args.entry_id,)
|
||
).fetchone()
|
||
if not row:
|
||
print(f"Entry {args.entry_id} not found")
|
||
sys.exit(1)
|
||
eligible, reason = check_promotion_eligibility(dict(row))
|
||
print(f"Entry {args.entry_id}: {'ELIGIBLE' if eligible else 'NOT ELIGIBLE'}")
|
||
print(f" reason: {reason}")
|
||
|
||
elif args.cmd == "promote":
|
||
result = promote_entry(db, args.entry_id)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
elif args.cmd == "gate":
|
||
result = pre_commit_gate(db, args.entry_id)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
elif args.cmd == "velocity":
|
||
result = compute_drift_velocity(db, args.days)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
elif args.cmd == "direction":
|
||
result = compute_drift_direction(db)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
elif args.cmd == "status":
|
||
vel = compute_drift_velocity(db)
|
||
direction = compute_drift_direction(db)
|
||
print("=== Governed Adaptation Status ===")
|
||
print(f"Total entries: {direction['total_entries']}")
|
||
print(f"Validated (proven): {direction['established_count']}")
|
||
print(f"Locked (immutable): {direction['rock_count']}")
|
||
print(f"Provisional (testing): {direction['total_entries'] - direction['established_count'] - direction['rock_count']}")
|
||
print(f"Trust ratio: {direction['stability_ratio']:.2%}")
|
||
print(f"Evolution direction: {direction['direction']}")
|
||
print(f"Evolution velocity: {vel['changes_per_day']} changes/day")
|
||
print(f"Alert: {vel['alert']}")
|
||
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|