""" core/evolution/tier1_autonomous_updater.py Story 8.08: Tier1AutonomousUpdater — Epistemic Self-Updates Applies epistemic changes (KG entities, Qdrant scars, prompt templates, and GLOBAL_GENESIS_RULES.md rule additions) without requiring a PR. Tier 1 = NO code changes. None of the methods here create or modify .py files. All updates are audit-logged to data/observability/tier1_updates.jsonl. VERIFICATION_STAMP Story: 8.08 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 8/8 (BB1–BB4, WB1–WB4) Coverage: 100% """ from __future__ import annotations import json import os from dataclasses import dataclass, asdict from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional from core.evolution.meta_architect import ArchitectureAnalysis # --------------------------------------------------------------------------- # Result dataclass # --------------------------------------------------------------------------- @dataclass class Tier1Result: """Summary counts from a single Tier 1 autonomous update run.""" kg_entities_added: int scars_updated: int prompts_updated: int rules_updated: int # --------------------------------------------------------------------------- # Main class # --------------------------------------------------------------------------- class Tier1AutonomousUpdater: """ Applies Tier 1 (epistemic) self-updates derived from an ArchitectureAnalysis. Tier 1 updates are safe to apply without a PR because they never touch Python source files. The four update categories are: 1. KG entities — write new axiom/entity JSONL files under kg_base_path 2. Qdrant scars — upsert consolidated scar records into genesis_scars 3. Prompt templates — create/update .md or .txt files in prompts_dir 4. Rules additions — append new guardrail text to rules_file Every successful run writes one audit line to audit_log_path (JSONL). Parameters ---------- qdrant_client: Any object with an ``upsert(collection_name, points)`` method. When None, scar updates are skipped gracefully. kg_base_path: Root directory for KG entity files. Default: "KNOWLEDGE_GRAPH/entities" prompts_dir: Directory where prompt template files live. Default: "config/prompts" rules_file: Path to GLOBAL_GENESIS_RULES.md (or equivalent). Default: ".claude/rules/GLOBAL_GENESIS_RULES.md" audit_log_path: JSONL file for Tier 1 update audit entries. Default: "data/observability/tier1_updates.jsonl" """ def __init__( self, qdrant_client: Any = None, kg_base_path: Optional[str] = None, prompts_dir: Optional[str] = None, rules_file: Optional[str] = None, audit_log_path: Optional[str] = None, ) -> None: self.qdrant = qdrant_client self.kg_base = Path(kg_base_path or "KNOWLEDGE_GRAPH/entities") self.prompts_dir = Path(prompts_dir or "config/prompts") self.rules_file = Path(rules_file or ".claude/rules/GLOBAL_GENESIS_RULES.md") self.audit_log = Path(audit_log_path or "data/observability/tier1_updates.jsonl") # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def apply_tier1( self, analysis: ArchitectureAnalysis, epoch_id: str = "unknown", ) -> Tier1Result: """ Apply all Tier 1 updates derived from the given ArchitectureAnalysis. Parameters ---------- analysis: The result of a MetaArchitect.analyze() call with scope="epistemic". epoch_id: A string identifier for the current epoch (e.g. "epoch_042"). Used to name the KG entity file and tag the audit log. Returns ------- Tier1Result Counts of items updated in each category. """ kg_count = self._write_kg_entities(analysis, epoch_id) scar_count = self._update_scars(analysis) prompt_count = self._update_prompts(analysis) rule_count = self._append_rules(analysis) result = Tier1Result( kg_entities_added=kg_count, scars_updated=scar_count, prompts_updated=prompt_count, rules_updated=rule_count, ) self._write_audit(epoch_id, result) return result # ------------------------------------------------------------------ # Private helpers — KG entities # ------------------------------------------------------------------ def _write_kg_entities( self, analysis: ArchitectureAnalysis, epoch_id: str, ) -> int: """ Write one JSONL line per bottleneck/fix into {kg_base}/epoch_{epoch_id}_learnings.jsonl. Creates the directory if it does not exist. Returns the number of entities written. CRITICAL: Does NOT create or modify any .py files. """ if not analysis.bottlenecks and not analysis.recommended_fixes: return 0 self.kg_base.mkdir(parents=True, exist_ok=True) target = self.kg_base / f"epoch_{epoch_id}_learnings.jsonl" entities: list[dict] = [] # One entity per bottleneck for bottleneck in analysis.bottlenecks: entities.append({ "type": "bottleneck", "epoch_id": epoch_id, "timestamp": datetime.now(tz=timezone.utc).isoformat(), "description": bottleneck.description, "frequency": bottleneck.frequency, "affected_saga_ids": bottleneck.affected_saga_ids, "scar_ids": bottleneck.scar_ids, }) # One entity per fix proposal (only non-.py files honoured at Tier 1) for fix in analysis.recommended_fixes: if fix.target_file.endswith(".py"): # Tier 1 never references .py changes — skip silently continue entities.append({ "type": "fix_proposal", "epoch_id": epoch_id, "timestamp": datetime.now(tz=timezone.utc).isoformat(), "target_file": fix.target_file, "change_type": fix.change_type, "rationale": fix.rationale, }) if not entities: return 0 with target.open("a", encoding="utf-8") as fh: for entity in entities: fh.write(json.dumps(entity) + "\n") return len(entities) # ------------------------------------------------------------------ # Private helpers — Qdrant scars # ------------------------------------------------------------------ def _update_scars(self, analysis: ArchitectureAnalysis) -> int: """ Upsert consolidated scar records into the ``genesis_scars`` Qdrant collection — one point per bottleneck. When qdrant_client is None, returns 0 without raising. Returns the number of points upserted. """ if self.qdrant is None: return 0 if not analysis.bottlenecks: return 0 points = [] for i, bottleneck in enumerate(analysis.bottlenecks): # Build a minimal qdrant-style point dict point = { "id": f"t1_scar_{i}_{bottleneck.description[:24].replace(' ', '_')}", "payload": { "description": bottleneck.description, "frequency": bottleneck.frequency, "affected_saga_ids": bottleneck.affected_saga_ids, "scar_ids": bottleneck.scar_ids, "source": "tier1_autonomous_updater", "timestamp": datetime.now(tz=timezone.utc).isoformat(), }, # Qdrant requires a vector; supply a zero-vector placeholder. # Real production use would embed description via nomic-embed-text. "vector": [0.0] * 4, } points.append(point) try: self.qdrant.upsert( collection_name="genesis_scars", points=points, ) return len(points) except Exception: return 0 # ------------------------------------------------------------------ # Private helpers — prompt templates # ------------------------------------------------------------------ def _update_prompts(self, analysis: ArchitectureAnalysis) -> int: """ Create or update prompt template files in prompts_dir. Only creates .md or .txt files — NEVER .py files. One file per recommended fix whose target_file path ends with .md or .txt. Returns the number of prompt files created/updated. """ if not analysis.recommended_fixes: return 0 self.prompts_dir.mkdir(parents=True, exist_ok=True) count = 0 for fix in analysis.recommended_fixes: target = Path(fix.target_file) suffix = target.suffix.lower() # Tier 1 safety gate: only .md and .txt files allowed if suffix not in (".md", ".txt"): continue # Use only the filename part — write into prompts_dir dest = self.prompts_dir / target.name # Build prompt content block content_block = ( f"\n\n---\n" f"# Tier 1 Update — {datetime.now(tz=timezone.utc).date()}\n" f"## Rationale\n{fix.rationale}\n" f"## Change type\n{fix.change_type}\n" ) if dest.exists(): # Append to existing prompt file with dest.open("a", encoding="utf-8") as fh: fh.write(content_block) else: # Create new prompt file header = ( f"# Prompt Template: {target.stem}\n" f"# Auto-generated by Tier1AutonomousUpdater\n" f"# Created: {datetime.now(tz=timezone.utc).isoformat()}\n" ) with dest.open("w", encoding="utf-8") as fh: fh.write(header + content_block) count += 1 return count # ------------------------------------------------------------------ # Private helpers — rules file # ------------------------------------------------------------------ def _append_rules(self, analysis: ArchitectureAnalysis) -> int: """ Append a new guardrail rule to rules_file when the analysis identifies a missing guardrail (i.e. any bottleneck exists with scope='epistemic'). Rules are TEXT-ONLY additions — never modifies existing content. Returns 1 if a rule was appended, 0 otherwise. CRITICAL: Does NOT modify any .py files. """ if not analysis.bottlenecks: return 0 if analysis.scope != "epistemic": # Ontological issues are handled by PR — not Tier 1 return 0 # Build the rule text ts = datetime.now(tz=timezone.utc).isoformat() descriptions = "; ".join(b.description for b in analysis.bottlenecks[:3]) rule_text = ( f"\n\n---\n\n" f"## AUTO-APPENDED RULE (Tier 1 — {ts})\n\n" f"> Recurring epistemic bottleneck(s) detected:\n" f"> {descriptions}\n\n" f"**Guardrail**: When the above patterns recur, re-run MetaArchitect " f"analysis and escalate if frequency exceeds 3 within a 7-day window.\n" ) # Ensure parent directory exists self.rules_file.parent.mkdir(parents=True, exist_ok=True) # Append-only — never overwrite existing content with self.rules_file.open("a", encoding="utf-8") as fh: fh.write(rule_text) return 1 # ------------------------------------------------------------------ # Private helpers — audit log # ------------------------------------------------------------------ def _write_audit(self, epoch_id: str, result: Tier1Result) -> None: """ Append one JSON audit line to the audit_log JSONL file. Audit entry schema: timestamp ISO-8601 UTC epoch_id string kg_entities_added int scars_updated int prompts_updated int rules_updated int """ entry = { "timestamp": datetime.now(tz=timezone.utc).isoformat(), "epoch_id": epoch_id, "kg_entities_added": result.kg_entities_added, "scars_updated": result.scars_updated, "prompts_updated": result.prompts_updated, "rules_updated": result.rules_updated, } # Ensure parent directory exists self.audit_log.parent.mkdir(parents=True, exist_ok=True) with self.audit_log.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n")