""" TRUST BALANCE AGENT =================== Earned autonomy for the Harvester system. "The longer it does a good job, the guards loosen a little." This agent reads the system's track record — pass rates, confidence calibration, pattern library depth, failure trends — and computes a trust level that dynamically adjusts operational guardrails. What it adjusts (operational leash): - BENCHMARK_FLOOR (99.5% → 95% as trust grows) - MAX_ITERATIONS (10 → 5 as system needs fewer retries) - AUTO_ACCEPT (require human review → auto-accept high-confidence results) What it NEVER touches (safety is immutable): - Content filters in safety_guardrails.py - Blocked content patterns - Rate limiters - Audit logging Trust is earned slowly and lost fast. One bad streak resets weeks of progress. Because the cost of silence is carried forever — do the right thing. Usage: python trust_balance_agent.py # show current trust level python trust_balance_agent.py --apply # compute and export trust-adjusted env vars python trust_balance_agent.py --history # show trust score over time python trust_balance_agent.py --reset # reset trust to baseline (requires confirmation) Integration: from trust_balance_agent import TrustBalanceAgent agent = TrustBalanceAgent() agent.apply_trust() # sets os.environ with adjusted guardrail values """ import os import sys import sqlite3 import json from datetime import datetime, timedelta from dataclasses import dataclass from typing import Optional sys.path.insert(0, os.path.dirname(__file__)) DB_PATH = os.environ.get( "GROWTH_DB_PATH", os.path.join(os.path.dirname(os.path.abspath(__file__)), "growth_memory.db") ) # ══════════════════════════════════════════════════════════════════════════════ # TRUST TIERS — earned through consistent good work # ══════════════════════════════════════════════════════════════════════════════ # # Each tier defines operational parameters the system earns the right to use. # Safety content filters are NEVER relaxed — only the performance leash changes. # # Score Range │ Tier │ Benchmark │ MaxIter │ AutoAccept # ─────────────┼──────────────┼───────────┼─────────┼─────────── # 0 - 29 │ RESTRICTED │ 99.5% │ 10 │ Never # 30 - 49 │ SUPERVISED │ 99.0% │ 9 │ Never # 50 - 69 │ STANDARD │ 97.0% │ 8 │ Never # 70 - 84 │ TRUSTED │ 95.0% │ 6 │ Score >= 98% # 85 - 100 │ AUTONOMOUS │ 92.0% │ 5 │ Score >= 95% TRUST_TIERS = { "RESTRICTED": {"min_score": 0, "benchmark_floor": 99.5, "max_iterations": 10, "auto_accept_above": None}, "SUPERVISED": {"min_score": 30, "benchmark_floor": 99.0, "max_iterations": 9, "auto_accept_above": None}, "STANDARD": {"min_score": 50, "benchmark_floor": 97.0, "max_iterations": 8, "auto_accept_above": None}, "TRUSTED": {"min_score": 70, "benchmark_floor": 95.0, "max_iterations": 6, "auto_accept_above": 98.0}, "AUTONOMOUS": {"min_score": 85, "benchmark_floor": 92.0, "max_iterations": 5, "auto_accept_above": 95.0}, } # How fast trust moves TRUST_GAIN_PER_PASS = 1.5 # points per successful task TRUST_GAIN_BONUS_STREAK = 0.5 # extra per consecutive pass (compounds) TRUST_LOSS_PER_FAIL = 5.0 # points lost per failure — trust is lost fast TRUST_LOSS_STREAK_MULT = 1.5 # multiplier per consecutive failure TRUST_DECAY_PER_DAY = 0.2 # idle decay — must keep proving yourself TRUST_CALIBRATION_PENALTY = 3.0 # penalty when overconfident # ══════════════════════════════════════════════════════════════════════════════ # DATA STRUCTURES # ══════════════════════════════════════════════════════════════════════════════ @dataclass class TrustState: score: float tier_name: str benchmark_floor: float max_iterations: int auto_accept_above: Optional[float] pass_rate: float total_tasks: int recent_streak: int # positive = consecutive passes, negative = fails calibration_accuracy: float is_overconfident: bool last_updated: str # ══════════════════════════════════════════════════════════════════════════════ # TRUST BALANCE AGENT # ══════════════════════════════════════════════════════════════════════════════ class TrustBalanceAgent: """ Reads the system's track record and computes earned trust. Trust adjusts operational parameters — never safety filters. """ def __init__(self, db_path: str = DB_PATH): self.db_path = db_path self._init_trust_table() def _get_db(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row return conn def _init_trust_table(self): """Create trust_history table to track score over time.""" conn = self._get_db() conn.execute(""" CREATE TABLE IF NOT EXISTS trust_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, trust_score REAL NOT NULL, tier_name TEXT NOT NULL, reason TEXT, pass_rate REAL, streak INTEGER ) """) conn.commit() conn.close() # ── Core: Compute Trust Score ────────────────────────────────────────── def compute_trust(self) -> TrustState: """ Calculate current trust from the system's actual track record. No self-reporting — only what the data proves. """ conn = self._get_db() # Get last recorded trust score as baseline last_trust = conn.execute( "SELECT trust_score FROM trust_history ORDER BY id DESC LIMIT 1" ).fetchone() current_score = last_trust['trust_score'] if last_trust else 50.0 # Get task history (recent 100 tasks for scoring) tasks = conn.execute( "SELECT passed, final_score, timestamp FROM task_history " "ORDER BY timestamp DESC LIMIT 100" ).fetchall() total_tasks = len(tasks) if total_tasks == 0: conn.close() return self._build_state(50.0, 0, 0, 0.0, False, total_tasks) # ── Pass rate ── passed_count = sum(1 for t in tasks if t['passed']) pass_rate = (passed_count / total_tasks) * 100 if total_tasks > 0 else 0 # ── Recent streak (walk backwards from most recent) ── streak = 0 if tasks: first_result = bool(tasks[0]['passed']) for t in tasks: if bool(t['passed']) == first_result: streak += 1 if first_result else -1 else: break # ── Tasks since last trust computation ── last_id = conn.execute( "SELECT MAX(id) as max_id FROM trust_history" ).fetchone()['max_id'] or 0 # Find tasks that happened after our last trust snapshot new_tasks = conn.execute( "SELECT passed, final_score FROM task_history " "WHERE id > ? ORDER BY timestamp ASC", (last_id,) ).fetchall() # ── Apply gains and losses from new tasks ── consecutive = 0 for t in new_tasks: if t['passed']: consecutive = max(0, consecutive) + 1 gain = TRUST_GAIN_PER_PASS + (TRUST_GAIN_BONUS_STREAK * min(consecutive, 10)) current_score = min(100.0, current_score + gain) else: consecutive = min(0, consecutive) - 1 loss = TRUST_LOSS_PER_FAIL * (TRUST_LOSS_STREAK_MULT ** min(abs(consecutive) - 1, 5)) current_score = max(0.0, current_score - loss) # ── Idle decay ── if tasks: last_task_time = datetime.fromisoformat(tasks[0]['timestamp']) days_idle = (datetime.now() - last_task_time).days if days_idle > 0: decay = TRUST_DECAY_PER_DAY * days_idle current_score = max(0.0, current_score - decay) # ── Calibration penalty ── calibration = self._get_calibration(conn) if calibration['is_overconfident'] and calibration['sample_size'] > 5: current_score = max(0.0, current_score - TRUST_CALIBRATION_PENALTY) conn.close() return self._build_state( current_score, streak, total_tasks, calibration['accuracy'], calibration['is_overconfident'], passed_count ) def _get_calibration(self, conn) -> dict: """Read confidence calibration from existing confidence_log table.""" rows = conn.execute( "SELECT stated_confidence, actual_score FROM confidence_log " "ORDER BY timestamp DESC LIMIT 50" ).fetchall() if not rows: return {"accuracy": 100.0, "is_overconfident": False, "sample_size": 0} errors = [abs((r['stated_confidence'] * 100) - r['actual_score']) for r in rows] avg_error = sum(errors) / len(errors) overconfident = sum( 1 for r in rows if (r['stated_confidence'] * 100) > r['actual_score'] ) / len(rows) > 0.6 return { "accuracy": round(100 - avg_error, 1), "is_overconfident": overconfident, "sample_size": len(rows) } def _build_state(self, score, streak, total, cal_accuracy, is_overconfident, passed_count) -> TrustState: """Map a trust score to a tier and build the full state.""" tier_name = "RESTRICTED" tier = TRUST_TIERS["RESTRICTED"] for name, t in TRUST_TIERS.items(): if score >= t["min_score"]: tier_name = name tier = t pass_rate = (passed_count / total * 100) if total > 0 else 0 return TrustState( score=round(score, 1), tier_name=tier_name, benchmark_floor=tier["benchmark_floor"], max_iterations=tier["max_iterations"], auto_accept_above=tier["auto_accept_above"], pass_rate=round(pass_rate, 1), total_tasks=total, recent_streak=streak, calibration_accuracy=cal_accuracy, is_overconfident=is_overconfident, last_updated=datetime.now().isoformat() ) # ── Apply: Set environment variables ─────────────────────────────────── def apply_trust(self) -> TrustState: """ Compute trust and set os.environ so reflection_engine picks up the adjusted guardrails on its next import. """ state = self.compute_trust() # Set the operational leash os.environ["BENCHMARK_FLOOR"] = str(state.benchmark_floor) os.environ["MAX_ITERATIONS"] = str(state.max_iterations) # Record this trust snapshot self._record_snapshot(state) return state def _record_snapshot(self, state: TrustState): """Write trust score to history for trend tracking.""" conn = self._get_db() conn.execute(""" INSERT INTO trust_history (timestamp, trust_score, tier_name, reason, pass_rate, streak) VALUES (?, ?, ?, ?, ?, ?) """, ( datetime.now().isoformat(), state.score, state.tier_name, f"tasks={state.total_tasks} pass_rate={state.pass_rate}% streak={state.recent_streak}", state.pass_rate, state.recent_streak )) conn.commit() conn.close() # ── Display ──────────────────────────────────────────────────────────── def show_status(self): """Print current trust level and what it means.""" state = self.compute_trust() tier_bar = "" for name in TRUST_TIERS: if name == state.tier_name: tier_bar += f" [{name}] " else: tier_bar += f" {name.lower()} " print(f"\n{'='*62}") print(f" TRUST BALANCE — EARNED AUTONOMY") print(f" \"The longer it does a good job, the guards loosen a little.\"") print(f"{'='*62}") print(f"\n Trust Score: {state.score:.1f} / 100") print(f" Current Tier: {state.tier_name}") print(f"\n {tier_bar}") print(f"\n ── What This Means ──") print(f" Benchmark Floor: {state.benchmark_floor}% (default: 99.5%)") print(f" Max Iterations: {state.max_iterations} (default: 10)") if state.auto_accept_above: print(f" Auto-Accept: Scores >= {state.auto_accept_above}%") else: print(f" Auto-Accept: Disabled (human review required)") print(f"\n ── Track Record ──") print(f" Tasks Completed: {state.total_tasks}") print(f" Pass Rate: {state.pass_rate}%") streak_str = f"+{state.recent_streak}" if state.recent_streak > 0 else str(state.recent_streak) print(f" Current Streak: {streak_str}") print(f" Calibration: {state.calibration_accuracy}% accurate") if state.is_overconfident: print(f" ⚠️ OVERCONFIDENT — trust penalized until calibration improves") print(f"\n ── Trust Rules ──") print(f" Gain: +{TRUST_GAIN_PER_PASS} per pass, +{TRUST_GAIN_BONUS_STREAK} streak bonus") print(f" Loss: -{TRUST_LOSS_PER_FAIL} per fail (x{TRUST_LOSS_STREAK_MULT} streak multiplier)") print(f" Decay: -{TRUST_DECAY_PER_DAY} per idle day") print(f" Safety: Content filters are NEVER relaxed. Only the leash changes.") print(f"\n{'='*62}\n") def show_history(self): """Print trust score trend over time.""" conn = self._get_db() rows = conn.execute( "SELECT timestamp, trust_score, tier_name, reason " "FROM trust_history ORDER BY id DESC LIMIT 30" ).fetchall() conn.close() if not rows: print("\n No trust history yet. Run some tasks first.\n") return print(f"\n{'='*62}") print(f" TRUST HISTORY (last {len(rows)} snapshots)") print(f"{'='*62}\n") # Reverse so oldest is first (reading top-down = timeline) for r in reversed(rows): ts = r['timestamp'][:16].replace('T', ' ') score = r['trust_score'] tier = r['tier_name'] bar_len = int(score / 2) bar = '█' * bar_len + '░' * (50 - bar_len) print(f" {ts} {score:5.1f} {bar} {tier}") print(f"\n{'='*62}\n") def reset_trust(self, confirm: bool = False): """Reset trust to baseline. Requires explicit confirmation.""" if not confirm: print("\n ⚠️ This will reset trust score to 50.0 (STANDARD).") print(" Run with --reset --confirm to proceed.\n") return conn = self._get_db() conn.execute(""" INSERT INTO trust_history (timestamp, trust_score, tier_name, reason, pass_rate, streak) VALUES (?, 50.0, 'STANDARD', 'MANUAL RESET', NULL, 0) """, (datetime.now().isoformat(),)) conn.commit() conn.close() print("\n Trust reset to 50.0 (STANDARD). Earn it back.\n") # ══════════════════════════════════════════════════════════════════════════════ # INTEGRATION HOOK — call this from growth_engine.py before running tasks # ══════════════════════════════════════════════════════════════════════════════ def apply_trust_before_run(): """ One-liner for growth_engine.py to call before running reflection_engine. Computes trust from track record and sets env vars accordingly. """ agent = TrustBalanceAgent() state = agent.apply_trust() print(f" 🔒 Trust: {state.score:.1f} ({state.tier_name}) → " f"floor={state.benchmark_floor}%, iters={state.max_iterations}") return state # ══════════════════════════════════════════════════════════════════════════════ # CLI # ══════════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": agent = TrustBalanceAgent() args = sys.argv[1:] if "--apply" in args: state = agent.apply_trust() print(f"\n Trust applied: {state.tier_name} (score={state.score:.1f})") print(f" BENCHMARK_FLOOR={state.benchmark_floor}") print(f" MAX_ITERATIONS={state.max_iterations}\n") elif "--history" in args: agent.show_history() elif "--reset" in args: agent.reset_trust(confirm="--confirm" in args) else: agent.show_status()