""" core/evolution/meta_architect.py Story 8.03: MetaArchitect — Scar-Driven Structural Analysis Reads Qdrant L3 scars and Postgres L4 sagas, identifies structural bottlenecks, and determines if a code-level fix or prompt-level fix is warranted. VERIFICATION_STAMP Story: 8.03 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, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class Bottleneck: """Represents a recurring failure pattern identified across scars and sagas.""" description: str frequency: int affected_saga_ids: list[str] = field(default_factory=list) scar_ids: list[str] = field(default_factory=list) @dataclass class FixProposal: """A proposed structural fix for an identified bottleneck.""" target_file: str change_type: str rationale: str @dataclass class ArchitectureAnalysis: """ Complete result of a MetaArchitect analysis run. scope: "epistemic" — prompt-level fix sufficient, no PR needed. "ontological" — code-level fix required; must raise a PR. """ bottlenecks: list[Bottleneck] = field(default_factory=list) recommended_fixes: list[FixProposal] = field(default_factory=list) scope: str = "epistemic" # "epistemic" | "ontological" # --------------------------------------------------------------------------- # MetaArchitect # --------------------------------------------------------------------------- # Default log path — can be overridden via constructor for testability _DEFAULT_LOG_PATH = Path("/mnt/e/genesis-system/data/observability/meta_architect_log.jsonl") # Cosine-similarity threshold for clustering scars _CLUSTER_THRESHOLD = 0.85 class MetaArchitect: """ Scar-driven structural analyser. Queries: • Qdrant L3 — semantic scar vectors (past failure memories) • Postgres L4 — swarm_sagas with status='PARTIAL_FAIL' Then clusters recurring failure patterns, determines whether they are prompt-fixable (epistemic) or require a code PR (ontological), and writes the analysis to a JSONL log. All external I/O is dependency-injected so every method is fully mockable in tests — zero real network calls are ever required. """ def __init__( self, qdrant_client: Any = None, pg_connection: Any = None, log_path: Optional[Path] = None, ) -> None: self.qdrant_client = qdrant_client self.pg_connection = pg_connection self.log_path: Path = log_path or _DEFAULT_LOG_PATH # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def analyze(self, lookback_days: int = 7) -> ArchitectureAnalysis: """ Run a full scar-driven architecture analysis. Parameters ---------- lookback_days: How many days back to scan for failed sagas and scars. Returns ------- ArchitectureAnalysis Populated with bottlenecks, fix proposals, and a scope label. """ scars = self._query_scars(lookback_days) sagas = self._query_sagas(lookback_days) clusters = self._cluster_scars(scars) bottlenecks = self._identify_bottlenecks(clusters, sagas) scope = self._determine_scope(bottlenecks) proposals = self._generate_proposals(bottlenecks, scope) analysis = ArchitectureAnalysis( bottlenecks=bottlenecks, recommended_fixes=proposals, scope=scope, ) self._write_log(analysis, lookback_days) return analysis # ------------------------------------------------------------------ # Private helpers — query layer # ------------------------------------------------------------------ def _query_scars(self, lookback_days: int) -> list[dict]: """ Retrieve recent scar records from Qdrant L3. The Qdrant client is expected to expose a ``scroll`` method that returns a list of point dicts, each with at least: • "id" — scar identifier • "payload" — dict with "description" and "vector" keys When ``qdrant_client`` is None, returns an empty list. """ if self.qdrant_client is None: return [] try: results = self.qdrant_client.scroll( collection_name="scars", limit=1000, with_payload=True, with_vectors=True, ) scars = [] for point in results: scars.append({ "id": str(point.get("id", "")), "description": point.get("payload", {}).get("description", ""), "vector": point.get("vector", []), }) return scars except Exception: return [] def _query_sagas(self, lookback_days: int) -> list[dict]: """ Retrieve PARTIAL_FAIL sagas from Postgres L4. The pg_connection is expected to behave like a psycopg2 connection with a ``cursor()`` context manager. SQL executed: SELECT id, description, error_trace FROM swarm_sagas WHERE status = 'PARTIAL_FAIL' AND created_at > NOW() - INTERVAL '{lookback_days} days' When ``pg_connection`` is None, returns an empty list. """ if self.pg_connection is None: return [] try: sql = ( "SELECT id, description, error_trace " "FROM swarm_sagas " "WHERE status = 'PARTIAL_FAIL' " f"AND created_at > NOW() - INTERVAL '{lookback_days} days'" ) cursor = self.pg_connection.cursor() cursor.execute(sql) rows = cursor.fetchall() sagas = [] for row in rows: sagas.append({ "id": str(row[0]), "description": row[1] or "", "error_trace": row[2] or "", }) return sagas except Exception: return [] # ------------------------------------------------------------------ # Private helpers — analysis layer # ------------------------------------------------------------------ def _cluster_scars(self, scars: list[dict]) -> list[list[dict]]: """ Group scars by semantic similarity (cosine ≥ 0.85). Uses cosine similarity between scar vectors when available. Falls back to exact description-text matching when vectors are absent (e.g. during testing with mock data that has no real embeddings). Returns a list of clusters, where each cluster is a list of scar dicts that are semantically similar to one another. """ if not scars: return [] # Determine whether scars carry real embedding vectors has_vectors = any( isinstance(s.get("vector"), (list, tuple)) and len(s.get("vector", [])) > 0 for s in scars ) if has_vectors: return self._cluster_by_cosine(scars) else: return self._cluster_by_text(scars) def _cluster_by_cosine(self, scars: list[dict]) -> list[list[dict]]: """Greedy cosine-based clustering (O(n²) — acceptable for ≤1 000 scars).""" assigned = [False] * len(scars) clusters: list[list[dict]] = [] for i, scar_i in enumerate(scars): if assigned[i]: continue cluster = [scar_i] assigned[i] = True vec_i = scar_i.get("vector", []) for j in range(i + 1, len(scars)): if assigned[j]: continue vec_j = scars[j].get("vector", []) if self._cosine_similarity(vec_i, vec_j) >= _CLUSTER_THRESHOLD: cluster.append(scars[j]) assigned[j] = True clusters.append(cluster) return clusters def _cluster_by_text(self, scars: list[dict]) -> list[list[dict]]: """ Fallback: cluster by identical normalised description text. One cluster per unique description; singletons get their own cluster. """ buckets: dict[str, list[dict]] = {} for scar in scars: key = scar.get("description", "").strip().lower() or scar.get("id", "") buckets.setdefault(key, []).append(scar) return list(buckets.values()) @staticmethod def _cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float: """ Compute cosine similarity between two equal-length float vectors. Returns 0.0 on zero-length or mismatched inputs. """ if not vec_a or not vec_b or len(vec_a) != len(vec_b): return 0.0 dot = sum(a * b for a, b in zip(vec_a, vec_b)) mag_a = sum(a * a for a in vec_a) ** 0.5 mag_b = sum(b * b for b in vec_b) ** 0.5 if mag_a == 0.0 or mag_b == 0.0: return 0.0 return dot / (mag_a * mag_b) def _identify_bottlenecks( self, clusters: list[list[dict]], sagas: list[dict], ) -> list[Bottleneck]: """ Map scar clusters to Bottleneck instances, cross-referencing sagas. A cluster becomes a Bottleneck when: • It contains ≥1 scar, AND • Either the cluster has frequency > 1 OR at least one saga description matches the cluster's description theme. All clusters are surfaced as bottlenecks (frequency ≥ 1) so that even singleton failures are visible for review. """ bottlenecks: list[Bottleneck] = [] for cluster in clusters: if not cluster: continue scar_ids = [s["id"] for s in cluster] # Representative description from the most common text in cluster description = cluster[0].get("description", "unknown failure pattern") # Find sagas whose description overlaps with this cluster theme desc_lower = description.lower() affected_saga_ids: list[str] = [] for saga in sagas: saga_desc = (saga.get("description") or saga.get("error_trace") or "").lower() # Overlap: any word ≥5 chars from description appears in saga text words = [w for w in desc_lower.split() if len(w) >= 5] if words and any(w in saga_desc for w in words): affected_saga_ids.append(saga["id"]) elif not words: # Short/generic description — attach all sagas to surface them affected_saga_ids.append(saga["id"]) bottlenecks.append( Bottleneck( description=description, frequency=len(cluster), affected_saga_ids=list(dict.fromkeys(affected_saga_ids)), # dedupe scar_ids=scar_ids, ) ) return bottlenecks def _determine_scope(self, bottlenecks: list[Bottleneck]) -> str: """ Classify the required fix scope. "epistemic" — all bottlenecks are prompt-fixable: - description mentions "prompt", "instruction", "context", or - no .py file appears in any FixProposal target (pre-proposal stage) "ontological" — at least one bottleneck's description references code artefacts (.py files) OR the word "code", "module", "function", "class", "import", or "refactor". This is intentionally conservative: when in doubt, escalate to "ontological" so that structural regressions are not silently swept under prompt changes. """ if not bottlenecks: return "epistemic" ontological_keywords = { ".py", "module", "function", "class", "import", "refactor", "code", "implementation", } # Prompt-fixable keywords shift it epistemic epistemic_keywords = {"prompt", "instruction", "context", "system prompt"} for bottleneck in bottlenecks: desc_lower = bottleneck.description.lower() tokens = set(desc_lower.split()) # Check for .py suffix (appears as part of a word like "router.py") has_py_file = any(".py" in token for token in tokens) has_ontological = has_py_file or any(kw in desc_lower for kw in ontological_keywords if kw != ".py") has_epistemic = any(kw in desc_lower for kw in epistemic_keywords) if has_ontological and not has_epistemic: return "ontological" return "epistemic" def _generate_proposals( self, bottlenecks: list[Bottleneck], scope: str, ) -> list[FixProposal]: """ Generate FixProposal instances for each bottleneck. For "epistemic" scope → proposals target prompt/config files. For "ontological" scope → proposals target Python source files. """ proposals: list[FixProposal] = [] for bottleneck in bottlenecks: desc_lower = bottleneck.description.lower() if scope == "ontological": # Try to extract a .py filename from the description target_file = self._extract_py_file(desc_lower) if not target_file: # Generic structural fix target_file = "core/evolution/meta_architect.py" proposals.append( FixProposal( target_file=target_file, change_type="refactor", rationale=( f"Recurring bottleneck (×{bottleneck.frequency}): " f"{bottleneck.description}" ), ) ) else: # Epistemic — prompt-level fix proposals.append( FixProposal( target_file="prompts/system_prompt.md", change_type="prompt_update", rationale=( f"Prompt-fixable bottleneck (×{bottleneck.frequency}): " f"{bottleneck.description}" ), ) ) return proposals @staticmethod def _extract_py_file(text: str) -> str: """ Extract the first *.py filename token from free-form text. Returns empty string if none found. """ for token in text.split(): clean = token.strip(".,;:'\"()[]{}") if clean.endswith(".py"): return clean return "" # ------------------------------------------------------------------ # Logging # ------------------------------------------------------------------ def _write_log(self, analysis: ArchitectureAnalysis, lookback_days: int) -> None: """ Append one JSON line to the meta_architect log. Log entry schema: timestamp ISO-8601 UTC lookback_days scope bottleneck_count fix_count bottlenecks list of dicts recommended_fixes list of dicts """ entry = { "timestamp": datetime.now(tz=timezone.utc).isoformat(), "lookback_days": lookback_days, "scope": analysis.scope, "bottleneck_count": len(analysis.bottlenecks), "fix_count": len(analysis.recommended_fixes), "bottlenecks": [ { "description": b.description, "frequency": b.frequency, "affected_saga_ids": b.affected_saga_ids, "scar_ids": b.scar_ids, } for b in analysis.bottlenecks ], "recommended_fixes": [ { "target_file": f.target_file, "change_type": f.change_type, "rationale": f.rationale, } for f in analysis.recommended_fixes ], } # Ensure parent directory exists self.log_path.parent.mkdir(parents=True, exist_ok=True) with self.log_path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n")