Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 3)
1ccf652 verified | """ | |
| aureon_code_integrator.py — The organism writes its own source, under review. | |
| When the Code Architect authors a change that targets existing aureon source | |
| (not a new skill in the library, but an edit to a .py file already in the | |
| tree), the change does not land directly. It lands *here*: | |
| 1. PROPOSE the change. We locate `old_text` inside `target_path`, | |
| build the would-be resulting file in memory, and run a | |
| static syntax check (ast.parse + compile) against it. | |
| A pending record is written to | |
| state/integrations_pending/<id>.json | |
| with the full before-snapshot and the new text. | |
| 2. CONFIRM the change by id. We take a backup of the current file | |
| under state/code_integrator_backups/<path>/<ts>.bak, | |
| write the new content, run one more compile on disk, | |
| and append an audit line to | |
| state/integrations_applied.jsonl | |
| with the unified diff. If the on-disk compile fails we | |
| restore from backup and record the failure. | |
| 3. REJECT the change by id. The pending record is moved to | |
| state/integrations_rejected.jsonl. No file is touched. | |
| 4. LIST pending edits. | |
| 5. HISTORY recent applied / rejected edits (audit log). | |
| The module never applies a change without an explicit `confirm_edit` | |
| call. It never writes outside the aureon-trading repository root. It | |
| refuses to touch files under `state/` (runtime scratch) or anywhere | |
| outside `aureon/`, `scripts/`, or repo-root `.py` files by default. | |
| Typical flow from outside: | |
| ci = get_code_integrator() | |
| prop = ci.propose_edit( | |
| target_path="aureon/core/aureon_lambda_engine.py", | |
| old_text="PHI = 1.618", | |
| new_text="PHI = 1.6180339887498948", | |
| rationale="tighten constant to 16 digits", | |
| ) | |
| # inspect prop["before_snapshot"], prop["after_preview"], prop["syntax_ok"] | |
| ci.confirm_edit(prop["pending_id"]) | |
| # the file is now modified and the change is in state/integrations_applied.jsonl | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import difflib | |
| import hashlib | |
| import json | |
| import logging | |
| import py_compile | |
| import shutil | |
| import threading | |
| import time | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional, Tuple | |
| logger = logging.getLogger("aureon.core.code_integrator") | |
| _REPO_ROOT = Path(__file__).resolve().parents[2] | |
| _STATE_ROOT = _REPO_ROOT / "state" | |
| _PENDING_DIR = _STATE_ROOT / "integrations_pending" | |
| _APPLIED_LOG = _STATE_ROOT / "integrations_applied.jsonl" | |
| _REJECTED_LOG = _STATE_ROOT / "integrations_rejected.jsonl" | |
| _BACKUP_ROOT = _STATE_ROOT / "code_integrator_backups" | |
| _IMPROVEMENT_LOG = _STATE_ROOT / "improvement_log.jsonl" | |
| # Paths the integrator is permitted to touch, relative to repo root. These | |
| # are prefixes — `aureon/` covers every module under aureon-trading/aureon/. | |
| _DEFAULT_ALLOWED_PREFIXES: Tuple[str, ...] = ( | |
| "aureon/", | |
| "scripts/", | |
| ) | |
| # Paths the integrator refuses to touch even if an allowed prefix also matches. | |
| _DEFAULT_DENIED_PREFIXES: Tuple[str, ...] = ( | |
| "state/", | |
| ".git/", | |
| "__pycache__/", | |
| ) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Helpers | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def _now() -> float: | |
| return time.time() | |
| def _relpath(p: Path) -> str: | |
| try: | |
| return str(p.resolve().relative_to(_REPO_ROOT)).replace("\\", "/") | |
| except Exception: | |
| return str(p) | |
| def _resolve_repo_path(target_path: str) -> Path: | |
| """Resolve `target_path` to an absolute path inside the repo, or raise.""" | |
| candidate = Path(target_path) | |
| if not candidate.is_absolute(): | |
| candidate = _REPO_ROOT / candidate | |
| candidate = candidate.resolve() | |
| try: | |
| candidate.relative_to(_REPO_ROOT) | |
| except ValueError: | |
| raise ValueError(f"path escapes repo root: {target_path}") | |
| return candidate | |
| def _check_allowed(target_path: str, | |
| allowed: Tuple[str, ...], | |
| denied: Tuple[str, ...]) -> None: | |
| rel = target_path.replace("\\", "/").lstrip("./") | |
| for bad in denied: | |
| if rel.startswith(bad): | |
| raise PermissionError(f"path under denied prefix {bad!r}: {rel}") | |
| # Root-level .py files are permitted even without a prefix match. | |
| if "/" not in rel and rel.endswith(".py"): | |
| return | |
| for good in allowed: | |
| if rel.startswith(good): | |
| return | |
| raise PermissionError( | |
| f"path {rel!r} is outside allowed prefixes {allowed!r}" | |
| ) | |
| def _sha256(text: str) -> str: | |
| return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() | |
| def _unified_diff(before: str, after: str, rel: str) -> str: | |
| diff = difflib.unified_diff( | |
| before.splitlines(keepends=True), | |
| after.splitlines(keepends=True), | |
| fromfile=f"a/{rel}", | |
| tofile=f"b/{rel}", | |
| n=3, | |
| ) | |
| return "".join(diff) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Dataclasses | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class PendingEdit: | |
| pending_id: str | |
| target_path: str # repo-relative | |
| rationale: str | |
| before_text: str | |
| after_text: str | |
| before_sha: str | |
| after_sha: str | |
| syntax_ok: bool | |
| syntax_error: str | |
| proposed_at: float | |
| match_count: int # how many times old_text appeared in before_text | |
| def preview(self, lines: int = 12) -> str: | |
| diff = _unified_diff(self.before_text, self.after_text, self.target_path) | |
| snippet = diff.splitlines()[:lines] | |
| return "\n".join(snippet) | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "pending_id": self.pending_id, | |
| "target_path": self.target_path, | |
| "rationale": self.rationale, | |
| "before_sha": self.before_sha, | |
| "after_sha": self.after_sha, | |
| "syntax_ok": self.syntax_ok, | |
| "syntax_error": self.syntax_error, | |
| "proposed_at": self.proposed_at, | |
| "match_count": self.match_count, | |
| } | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # The integrator | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class CodeIntegrator: | |
| """ | |
| Holds a pending queue of proposed source edits, syntax-checks them, | |
| and applies them only on explicit confirm. | |
| """ | |
| def __init__( | |
| self, | |
| allowed_prefixes: Tuple[str, ...] = _DEFAULT_ALLOWED_PREFIXES, | |
| denied_prefixes: Tuple[str, ...] = _DEFAULT_DENIED_PREFIXES, | |
| ) -> None: | |
| self.allowed_prefixes = tuple(allowed_prefixes) | |
| self.denied_prefixes = tuple(denied_prefixes) | |
| self._lock = threading.Lock() | |
| _PENDING_DIR.mkdir(parents=True, exist_ok=True) | |
| _BACKUP_ROOT.mkdir(parents=True, exist_ok=True) | |
| # ------------------------------------------------------------------ | |
| # Propose | |
| # ------------------------------------------------------------------ | |
| def propose_edit( | |
| self, | |
| target_path: str, | |
| old_text: str, | |
| new_text: str, | |
| rationale: str = "", | |
| ) -> Dict[str, Any]: | |
| """ | |
| Build a pending edit. Does NOT touch the file on disk. | |
| Returns: | |
| { | |
| "ok": bool, | |
| "pending_id": "...", | |
| "target_path": "aureon/...", | |
| "match_count": int, | |
| "before_sha": "...", | |
| "after_sha": "...", | |
| "syntax_ok": bool, | |
| "syntax_error": str, | |
| "preview": "<unified diff, trimmed>" | |
| } | |
| """ | |
| try: | |
| _check_allowed(target_path, self.allowed_prefixes, self.denied_prefixes) | |
| abs_path = _resolve_repo_path(target_path) | |
| except (PermissionError, ValueError) as e: | |
| return {"ok": False, "error": str(e)} | |
| if not abs_path.exists(): | |
| return {"ok": False, "error": f"file not found: {target_path}"} | |
| try: | |
| before_text = abs_path.read_text(encoding="utf-8") | |
| except Exception as e: | |
| return {"ok": False, "error": f"read failed: {e}"} | |
| if not old_text: | |
| return {"ok": False, "error": "old_text cannot be empty"} | |
| match_count = before_text.count(old_text) | |
| if match_count == 0: | |
| return { | |
| "ok": False, | |
| "error": "old_text not found in target file", | |
| "match_count": 0, | |
| } | |
| if match_count > 1: | |
| return { | |
| "ok": False, | |
| "error": ( | |
| f"old_text matches {match_count} times — refusing " | |
| "non-unique edit. Widen old_text to include surrounding " | |
| "context so the match becomes unique." | |
| ), | |
| "match_count": match_count, | |
| } | |
| # Build the would-be file. | |
| after_text = before_text.replace(old_text, new_text, 1) | |
| # Syntax check the *would-be* content. | |
| syntax_ok, syntax_err = self._syntax_check(after_text, target_path) | |
| pending_id = f"pe_{int(_now() * 1000)}_{uuid.uuid4().hex[:8]}" | |
| pending = PendingEdit( | |
| pending_id=pending_id, | |
| target_path=_relpath(abs_path), | |
| rationale=rationale or "", | |
| before_text=before_text, | |
| after_text=after_text, | |
| before_sha=_sha256(before_text), | |
| after_sha=_sha256(after_text), | |
| syntax_ok=syntax_ok, | |
| syntax_error=syntax_err, | |
| proposed_at=_now(), | |
| match_count=match_count, | |
| ) | |
| with self._lock: | |
| self._write_pending(pending) | |
| return { | |
| "ok": True, | |
| "pending_id": pending_id, | |
| "target_path": pending.target_path, | |
| "match_count": match_count, | |
| "before_sha": pending.before_sha, | |
| "after_sha": pending.after_sha, | |
| "syntax_ok": syntax_ok, | |
| "syntax_error": syntax_err, | |
| "preview": pending.preview(lines=24), | |
| } | |
| # ------------------------------------------------------------------ | |
| # Confirm / reject | |
| # ------------------------------------------------------------------ | |
| def confirm_edit(self, pending_id: str) -> Dict[str, Any]: | |
| """Apply the pending edit after backing up the current file.""" | |
| with self._lock: | |
| pending = self._load_pending(pending_id) | |
| if pending is None: | |
| return {"ok": False, "error": f"no such pending edit: {pending_id}"} | |
| if not pending.syntax_ok: | |
| return { | |
| "ok": False, | |
| "error": "refusing to apply: syntax check failed", | |
| "syntax_error": pending.syntax_error, | |
| } | |
| try: | |
| abs_path = _resolve_repo_path(pending.target_path) | |
| except Exception as e: | |
| return {"ok": False, "error": f"path resolve: {e}"} | |
| if not abs_path.exists(): | |
| return {"ok": False, "error": "target file disappeared"} | |
| # Guard: before_sha must still match current disk contents. | |
| try: | |
| current = abs_path.read_text(encoding="utf-8") | |
| except Exception as e: | |
| return {"ok": False, "error": f"re-read failed: {e}"} | |
| if _sha256(current) != pending.before_sha: | |
| return { | |
| "ok": False, | |
| "error": ( | |
| "target file changed since the edit was proposed — " | |
| "re-propose against the new file state." | |
| ), | |
| } | |
| # Backup. | |
| backup_path = self._backup(abs_path, current) | |
| # Capture metrics BEFORE the write so we can measure positive | |
| # improvement after the apply. Improvement is the organism's | |
| # signal that the edit actually did something. | |
| metrics_before = self._capture_metrics() | |
| # Write and compile. | |
| try: | |
| abs_path.write_text(pending.after_text, encoding="utf-8") | |
| except Exception as e: | |
| return {"ok": False, "error": f"write failed: {e}"} | |
| compile_ok, compile_err = self._disk_compile(abs_path) | |
| if not compile_ok: | |
| # Roll back. | |
| try: | |
| abs_path.write_text(current, encoding="utf-8") | |
| except Exception: | |
| pass | |
| return { | |
| "ok": False, | |
| "error": "applied file failed on-disk compile — rolled back", | |
| "compile_error": compile_err, | |
| "backup_path": str(backup_path), | |
| } | |
| metrics_after = self._capture_metrics() | |
| improvement = self._metrics_delta(metrics_before, metrics_after) | |
| # File-level deltas from the pending text itself — sensitive to | |
| # edits that activate dormant modules (new imports, new calls) | |
| # even when the tree-wide scan shows no change. | |
| file_delta = self._file_delta(pending.before_text, pending.after_text) | |
| improvement["file"] = file_delta | |
| # Upgrade verdict if file-level change is clearly positive. | |
| if improvement.get("verdict") == "no-op" and ( | |
| file_delta.get("lines_added", 0) > 0 | |
| or file_delta.get("imports_added", 0) > 0 | |
| or file_delta.get("calls_added", 0) > 0 | |
| ): | |
| improvement["verdict"] = "positive" | |
| improvement["verdict_source"] = "file_delta" | |
| # If this pending was part of a comparison, auto-reject the sibling. | |
| comparison_id, variant, sibling_id = self._read_comparison_meta(pending_id) | |
| # Audit. | |
| diff = _unified_diff(pending.before_text, pending.after_text, pending.target_path) | |
| record = { | |
| "applied_at": _now(), | |
| "pending_id": pending.pending_id, | |
| "target_path": pending.target_path, | |
| "rationale": pending.rationale, | |
| "before_sha": pending.before_sha, | |
| "after_sha": pending.after_sha, | |
| "backup_path": _relpath(backup_path), | |
| "diff": diff, | |
| "metrics_before": metrics_before, | |
| "metrics_after": metrics_after, | |
| "improvement": improvement, | |
| } | |
| if comparison_id: | |
| record["comparison_id"] = comparison_id | |
| record["variant"] = variant | |
| record["sibling_pending_id"] = sibling_id | |
| self._append_log(_APPLIED_LOG, record) | |
| # Dedicated improvement log — tails here to see organism progress. | |
| self._append_log(_IMPROVEMENT_LOG, { | |
| "applied_at": record["applied_at"], | |
| "pending_id": pending_id, | |
| "target_path": pending.target_path, | |
| "improvement": improvement, | |
| "comparison_id": comparison_id or None, | |
| "variant": variant or None, | |
| }) | |
| self._delete_pending(pending_id) | |
| sibling_rejection: Optional[Dict[str, Any]] = None | |
| if comparison_id and sibling_id: | |
| # Don't hold _lock recursively — sibling rejection doesn't | |
| # touch disk state beyond our own log files. | |
| sibling = self._load_pending(sibling_id) | |
| if sibling is not None: | |
| sib_record = { | |
| "rejected_at": _now(), | |
| "pending_id": sibling.pending_id, | |
| "target_path": sibling.target_path, | |
| "rationale": sibling.rationale, | |
| "reason": f"lost comparison {comparison_id} to variant={variant}", | |
| "comparison_id": comparison_id, | |
| "sibling_pending_id": pending_id, | |
| "before_sha": sibling.before_sha, | |
| "after_sha": sibling.after_sha, | |
| } | |
| self._append_log(_REJECTED_LOG, sib_record) | |
| self._delete_pending(sibling_id) | |
| sibling_rejection = { | |
| "sibling_pending_id": sibling_id, | |
| "reason": sib_record["reason"], | |
| } | |
| result = { | |
| "ok": True, | |
| "applied_at": record["applied_at"], | |
| "pending_id": pending_id, | |
| "target_path": pending.target_path, | |
| "backup_path": _relpath(backup_path), | |
| "diff": diff, | |
| "improvement": improvement, | |
| } | |
| if comparison_id: | |
| result["comparison_id"] = comparison_id | |
| result["variant"] = variant | |
| if sibling_rejection: | |
| result["sibling_rejected"] = sibling_rejection | |
| # Self-report: if no positive improvement, the loop is not doing | |
| # useful work — flag loudly so the reviewer sees it. | |
| verdict = improvement.get("verdict", "unknown") | |
| if verdict in ("no-op", "regression"): | |
| logger.warning( | |
| "edit applied but improvement verdict=%s on %s — loop may not be working", | |
| verdict, pending.target_path, | |
| ) | |
| else: | |
| logger.info( | |
| "edit applied: %s verdict=%s cls_delta=%s fn_delta=%s dp_delta=%s", | |
| pending.target_path, | |
| verdict, | |
| improvement.get("classes_delta"), | |
| improvement.get("functions_delta"), | |
| improvement.get("decision_points_delta"), | |
| ) | |
| # Continuity: hot-swap the module without the organism | |
| # perceiving a restart. Phi-calendar counts + singletons are | |
| # preserved across importlib.reload. Subjective time unbroken. | |
| try: | |
| from aureon.core.aureon_continuity_engine import get_continuity_engine | |
| ce = get_continuity_engine() | |
| swap = ce.apply_with_continuity( | |
| target_path=pending.target_path, | |
| pending_id=pending.pending_id, | |
| triggered_by="integrator.confirm", | |
| ) | |
| result["continuity"] = { | |
| "ok": swap.ok, | |
| "sandbox_ok": swap.sandbox_ok, | |
| "sandbox_detail": swap.sandbox_detail, | |
| "reload_ok": swap.reload_ok, | |
| "reload_error": swap.reload_error, | |
| "continuity_preserved": swap.continuity_preserved, | |
| "elapsed_s": swap.elapsed_s, | |
| } | |
| except Exception as e: | |
| # Continuity is best-effort. The file is already on disk; | |
| # fresh imports will pick it up regardless. Log softly. | |
| logger.debug("continuity engine unavailable: %s", e) | |
| return result | |
| def propose_comparison( | |
| self, | |
| target_path: str, | |
| old_text: str, | |
| aureon_new_text: str, | |
| human_new_text: str, | |
| rationale: str = "", | |
| ) -> Dict[str, Any]: | |
| """ | |
| Propose two competing edits for the same site so a reviewer can see | |
| whether Aureon's version beats a human baseline. | |
| Both variants are stored as independent pending edits that share | |
| a `comparison_id`. When one is confirmed, the other is auto-rejected | |
| with reason "lost comparison <id>" so the applied audit line can | |
| always be traced back to what the organism chose over what a human | |
| would have written. | |
| Returns: | |
| { | |
| "ok": bool, | |
| "comparison_id": "cmp_...", | |
| "aureon": { <propose_edit result> }, | |
| "human": { <propose_edit result> }, | |
| "side_by_side": "<aureon diff>\\n---\\n<human diff>" | |
| } | |
| """ | |
| comparison_id = f"cmp_{int(_now() * 1000)}_{uuid.uuid4().hex[:8]}" | |
| aureon_res = self.propose_edit( | |
| target_path=target_path, | |
| old_text=old_text, | |
| new_text=aureon_new_text, | |
| rationale=f"[comparison {comparison_id} variant=aureon] {rationale}", | |
| ) | |
| human_res = self.propose_edit( | |
| target_path=target_path, | |
| old_text=old_text, | |
| new_text=human_new_text, | |
| rationale=f"[comparison {comparison_id} variant=human] {rationale}", | |
| ) | |
| # Tag each pending record with its sibling + comparison id so the | |
| # audit trail can find the pair later. | |
| for res, variant, sibling in ( | |
| (aureon_res, "aureon", human_res), | |
| (human_res, "human", aureon_res), | |
| ): | |
| pid = res.get("pending_id") | |
| if not pid: | |
| continue | |
| try: | |
| fp = self._pending_file(pid) | |
| d = json.loads(fp.read_text(encoding="utf-8")) | |
| d["comparison_id"] = comparison_id | |
| d["variant"] = variant | |
| d["sibling_pending_id"] = sibling.get("pending_id") | |
| fp.write_text(json.dumps(d, indent=2), encoding="utf-8") | |
| except Exception as e: | |
| logger.debug("comparison tag failed: %s", e) | |
| # Side-by-side preview (trimmed). | |
| aureon_preview = aureon_res.get("preview", "") or "" | |
| human_preview = human_res.get("preview", "") or "" | |
| side_by_side = ( | |
| f"=== AUREON ({aureon_res.get('pending_id', '?')}) ===\n" | |
| f"syntax_ok={aureon_res.get('syntax_ok')} " | |
| f"err={aureon_res.get('syntax_error') or ''}\n" | |
| f"{aureon_preview}\n" | |
| f"\n=== HUMAN ({human_res.get('pending_id', '?')}) ===\n" | |
| f"syntax_ok={human_res.get('syntax_ok')} " | |
| f"err={human_res.get('syntax_error') or ''}\n" | |
| f"{human_preview}\n" | |
| ) | |
| return { | |
| "ok": bool(aureon_res.get("ok") or human_res.get("ok")), | |
| "comparison_id": comparison_id, | |
| "aureon": aureon_res, | |
| "human": human_res, | |
| "side_by_side": side_by_side, | |
| } | |
| def reject_edit(self, pending_id: str, reason: str = "") -> Dict[str, Any]: | |
| """Discard a pending edit; write a rejection audit line.""" | |
| with self._lock: | |
| pending = self._load_pending(pending_id) | |
| if pending is None: | |
| return {"ok": False, "error": f"no such pending edit: {pending_id}"} | |
| record = { | |
| "rejected_at": _now(), | |
| "pending_id": pending.pending_id, | |
| "target_path": pending.target_path, | |
| "rationale": pending.rationale, | |
| "reason": reason or "", | |
| "before_sha": pending.before_sha, | |
| "after_sha": pending.after_sha, | |
| } | |
| self._append_log(_REJECTED_LOG, record) | |
| self._delete_pending(pending_id) | |
| return {"ok": True, "pending_id": pending_id} | |
| # ------------------------------------------------------------------ | |
| # Listing / history | |
| # ------------------------------------------------------------------ | |
| def list_pending(self) -> List[Dict[str, Any]]: | |
| result: List[Dict[str, Any]] = [] | |
| if not _PENDING_DIR.exists(): | |
| return result | |
| for p in sorted(_PENDING_DIR.glob("*.json")): | |
| try: | |
| data = json.loads(p.read_text(encoding="utf-8")) | |
| # Strip bulk text from the listing; caller can fetch full | |
| # via inspect_pending(id). | |
| light = { | |
| k: v | |
| for k, v in data.items() | |
| if k not in ("before_text", "after_text") | |
| } | |
| result.append(light) | |
| except Exception: | |
| continue | |
| return result | |
| def inspect_pending(self, pending_id: str) -> Optional[Dict[str, Any]]: | |
| pending = self._load_pending(pending_id) | |
| if pending is None: | |
| return None | |
| d = pending.to_dict() | |
| d["preview"] = pending.preview(lines=40) | |
| return d | |
| def history(self, limit: int = 20, kind: str = "applied") -> List[Dict[str, Any]]: | |
| path = _APPLIED_LOG if kind == "applied" else _REJECTED_LOG | |
| if not path.exists(): | |
| return [] | |
| try: | |
| lines = path.read_text(encoding="utf-8").splitlines() | |
| except Exception: | |
| return [] | |
| recent = lines[-limit:] if limit > 0 else lines | |
| out: List[Dict[str, Any]] = [] | |
| for ln in recent: | |
| ln = ln.strip() | |
| if not ln: | |
| continue | |
| try: | |
| out.append(json.loads(ln)) | |
| except Exception: | |
| continue | |
| return out | |
| # ------------------------------------------------------------------ | |
| # Internals | |
| # ------------------------------------------------------------------ | |
| # ------------------------------------------------------------------ | |
| # Improvement capture — the "is this actually working?" signal | |
| # ------------------------------------------------------------------ | |
| def _capture_metrics(self) -> Dict[str, Any]: | |
| """ | |
| Snapshot measurable state that can change from an edit. Returns a | |
| compact dict. Safe to call on a live tree. | |
| """ | |
| metrics: Dict[str, Any] = { | |
| "taken_at": _now(), | |
| "modules_total": 0, | |
| "decision_points_total": 0, | |
| "decision_points_by_kind": {}, | |
| "classes_total": 0, | |
| "functions_total": 0, | |
| "loaded_aureon_modules": 0, | |
| } | |
| try: | |
| from aureon.core.aureon_self_introspection import get_self_introspection | |
| intro = get_self_introspection() | |
| snap = intro.scan(fast=False) | |
| by_kind: Dict[str, int] = {} | |
| for dp in snap.decision_points: | |
| by_kind[dp.kind] = by_kind.get(dp.kind, 0) + 1 | |
| cls_total = 0 | |
| fn_total = 0 | |
| for fp in snap.fingerprints: | |
| cls_total += len(fp.classes) | |
| fn_total += len(fp.functions) | |
| metrics.update({ | |
| "modules_total": snap.modules_total, | |
| "decision_points_total": len(snap.decision_points), | |
| "decision_points_by_kind": by_kind, | |
| "classes_total": cls_total, | |
| "functions_total": fn_total, | |
| "loaded_aureon_modules": len(intro.loaded_modules()), | |
| }) | |
| except Exception as e: | |
| metrics["error"] = f"capture failed: {e}" | |
| return metrics | |
| def _file_delta(before_text: str, after_text: str) -> Dict[str, Any]: | |
| """ | |
| File-level delta between the before and after text of a single | |
| pending edit. Counts lines, bytes, imports, and call expressions — | |
| so an edit that activates dormant modules (new imports, new calls) | |
| registers as positive even when the tree-wide scan shows no change. | |
| """ | |
| def _count_ast(text: str) -> Dict[str, int]: | |
| out = {"imports": 0, "calls": 0, "classes": 0, "functions": 0} | |
| try: | |
| tree = ast.parse(text) | |
| except Exception: | |
| return out | |
| for node in ast.walk(tree): | |
| if isinstance(node, (ast.Import, ast.ImportFrom)): | |
| # Count every imported name (import a, b → 2). | |
| out["imports"] += len(getattr(node, "names", []) or []) | |
| elif isinstance(node, ast.Call): | |
| out["calls"] += 1 | |
| elif isinstance(node, ast.ClassDef): | |
| out["classes"] += 1 | |
| elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): | |
| out["functions"] += 1 | |
| return out | |
| before_counts = _count_ast(before_text) | |
| after_counts = _count_ast(after_text) | |
| return { | |
| "lines_added": after_text.count("\n") - before_text.count("\n"), | |
| "bytes_added": len(after_text) - len(before_text), | |
| "imports_added": after_counts["imports"] - before_counts["imports"], | |
| "calls_added": after_counts["calls"] - before_counts["calls"], | |
| "classes_added": after_counts["classes"] - before_counts["classes"], | |
| "functions_added": after_counts["functions"] - before_counts["functions"], | |
| } | |
| def _metrics_delta(before: Dict[str, Any], after: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Compute a compact delta. Positive deltas on function/class counts | |
| indicate new code was added. Changes to decision-point kinds tell | |
| us whether gates/flags moved. | |
| """ | |
| def _num(d: Dict[str, Any], k: str) -> int: | |
| v = d.get(k, 0) | |
| try: | |
| return int(v) | |
| except Exception: | |
| return 0 | |
| kinds_before = dict(before.get("decision_points_by_kind") or {}) | |
| kinds_after = dict(after.get("decision_points_by_kind") or {}) | |
| all_kinds = set(kinds_before) | set(kinds_after) | |
| kinds_delta = { | |
| k: int(kinds_after.get(k, 0)) - int(kinds_before.get(k, 0)) | |
| for k in sorted(all_kinds) | |
| } | |
| delta = { | |
| "modules_total_delta": _num(after, "modules_total") - _num(before, "modules_total"), | |
| "decision_points_delta": ( | |
| _num(after, "decision_points_total") - _num(before, "decision_points_total") | |
| ), | |
| "decision_points_by_kind_delta": kinds_delta, | |
| "classes_delta": _num(after, "classes_total") - _num(before, "classes_total"), | |
| "functions_delta": _num(after, "functions_total") - _num(before, "functions_total"), | |
| "loaded_aureon_modules_delta": ( | |
| _num(after, "loaded_aureon_modules") - _num(before, "loaded_aureon_modules") | |
| ), | |
| } | |
| # Simple positive-improvement verdict: | |
| # - new code added (classes or functions up) → positive | |
| # - modules_total up (file added) → positive | |
| # - no change at all → flagged "no-op" | |
| # - only regressions → flagged "regression" | |
| positives = [ | |
| delta["classes_delta"], | |
| delta["functions_delta"], | |
| delta["modules_total_delta"], | |
| delta["loaded_aureon_modules_delta"], | |
| ] | |
| is_noop = all(v == 0 for v in positives) and delta["decision_points_delta"] == 0 | |
| has_positive = any(v > 0 for v in positives) or any(v > 0 for v in kinds_delta.values()) | |
| has_regression = any(v < 0 for v in positives) | |
| if is_noop: | |
| delta["verdict"] = "no-op" | |
| elif has_positive and not has_regression: | |
| delta["verdict"] = "positive" | |
| elif has_positive and has_regression: | |
| delta["verdict"] = "mixed" | |
| elif has_regression: | |
| delta["verdict"] = "regression" | |
| else: | |
| delta["verdict"] = "neutral" | |
| return delta | |
| def _read_comparison_meta(self, pending_id: str) -> Tuple[str, str, str]: | |
| """Return (comparison_id, variant, sibling_pending_id) or ("", "", "").""" | |
| fp = self._pending_file(pending_id) | |
| if not fp.exists(): | |
| return "", "", "" | |
| try: | |
| d = json.loads(fp.read_text(encoding="utf-8")) | |
| except Exception: | |
| return "", "", "" | |
| return ( | |
| str(d.get("comparison_id") or ""), | |
| str(d.get("variant") or ""), | |
| str(d.get("sibling_pending_id") or ""), | |
| ) | |
| def _syntax_check(self, text: str, target_path: str) -> Tuple[bool, str]: | |
| # We only full-parse .py files; everything else gets a pass. | |
| if not target_path.endswith(".py"): | |
| return True, "" | |
| try: | |
| ast.parse(text, filename=target_path) | |
| except SyntaxError as e: | |
| return False, f"SyntaxError at line {e.lineno}: {e.msg}" | |
| except Exception as e: | |
| return False, f"parse failed: {e}" | |
| return True, "" | |
| def _disk_compile(self, path: Path) -> Tuple[bool, str]: | |
| if path.suffix != ".py": | |
| return True, "" | |
| try: | |
| py_compile.compile(str(path), doraise=True) | |
| except py_compile.PyCompileError as e: | |
| return False, str(e) | |
| except Exception as e: | |
| return False, f"py_compile error: {e}" | |
| return True, "" | |
| def _pending_file(self, pending_id: str) -> Path: | |
| return _PENDING_DIR / f"{pending_id}.json" | |
| def _write_pending(self, pending: PendingEdit) -> None: | |
| data = { | |
| "pending_id": pending.pending_id, | |
| "target_path": pending.target_path, | |
| "rationale": pending.rationale, | |
| "before_text": pending.before_text, | |
| "after_text": pending.after_text, | |
| "before_sha": pending.before_sha, | |
| "after_sha": pending.after_sha, | |
| "syntax_ok": pending.syntax_ok, | |
| "syntax_error": pending.syntax_error, | |
| "proposed_at": pending.proposed_at, | |
| "match_count": pending.match_count, | |
| } | |
| self._pending_file(pending.pending_id).write_text( | |
| json.dumps(data, indent=2), encoding="utf-8" | |
| ) | |
| def _load_pending(self, pending_id: str) -> Optional[PendingEdit]: | |
| fp = self._pending_file(pending_id) | |
| if not fp.exists(): | |
| return None | |
| try: | |
| d = json.loads(fp.read_text(encoding="utf-8")) | |
| except Exception: | |
| return None | |
| return PendingEdit( | |
| pending_id=d["pending_id"], | |
| target_path=d["target_path"], | |
| rationale=d.get("rationale", ""), | |
| before_text=d["before_text"], | |
| after_text=d["after_text"], | |
| before_sha=d["before_sha"], | |
| after_sha=d["after_sha"], | |
| syntax_ok=bool(d.get("syntax_ok", False)), | |
| syntax_error=d.get("syntax_error", ""), | |
| proposed_at=float(d.get("proposed_at", 0.0)), | |
| match_count=int(d.get("match_count", 0)), | |
| ) | |
| def _delete_pending(self, pending_id: str) -> None: | |
| try: | |
| self._pending_file(pending_id).unlink(missing_ok=True) | |
| except Exception: | |
| pass | |
| def _backup(self, abs_path: Path, content: str) -> Path: | |
| ts = time.strftime("%Y%m%d_%H%M%S") | |
| rel = _relpath(abs_path) | |
| bk_dir = _BACKUP_ROOT / rel | |
| bk_dir.mkdir(parents=True, exist_ok=True) | |
| bk_path = bk_dir / f"{ts}.bak" | |
| try: | |
| bk_path.write_text(content, encoding="utf-8") | |
| except Exception as e: | |
| logger.warning("backup write failed (%s): %s", bk_path, e) | |
| return bk_path | |
| def _append_log(self, path: Path, record: Dict[str, Any]) -> None: | |
| try: | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| with path.open("a", encoding="utf-8") as f: | |
| f.write(json.dumps(record, default=str) + "\n") | |
| except Exception as e: | |
| logger.warning("log append failed (%s): %s", path, e) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Singleton | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| _instance: Optional[CodeIntegrator] = None | |
| _instance_lock = threading.Lock() | |
| def get_code_integrator() -> CodeIntegrator: | |
| global _instance | |
| with _instance_lock: | |
| if _instance is None: | |
| _instance = CodeIntegrator() | |
| return _instance | |
| if __name__ == "__main__": | |
| import argparse | |
| parser = argparse.ArgumentParser(description="Aureon code integrator CLI.") | |
| sub = parser.add_subparsers(dest="cmd", required=True) | |
| p_list = sub.add_parser("list", help="List pending edits") | |
| p_hist = sub.add_parser("history", help="Show applied/rejected history") | |
| p_hist.add_argument("--kind", choices=("applied", "rejected"), default="applied") | |
| p_hist.add_argument("--limit", type=int, default=10) | |
| p_show = sub.add_parser("show", help="Show full pending edit with diff preview") | |
| p_show.add_argument("pending_id") | |
| p_confirm = sub.add_parser("confirm", help="Confirm a pending edit by id") | |
| p_confirm.add_argument("pending_id") | |
| p_reject = sub.add_parser("reject", help="Reject a pending edit by id") | |
| p_reject.add_argument("pending_id") | |
| p_reject.add_argument("--reason", default="") | |
| p_propose = sub.add_parser("propose", help="Propose an edit from file contents on stdin") | |
| p_propose.add_argument("--target", required=True) | |
| p_propose.add_argument("--old", required=True, help="Path to a file containing old_text") | |
| p_propose.add_argument("--new", required=True, help="Path to a file containing new_text") | |
| p_propose.add_argument("--rationale", default="") | |
| args = parser.parse_args() | |
| ci = get_code_integrator() | |
| if args.cmd == "list": | |
| print(json.dumps(ci.list_pending(), indent=2, default=str)) | |
| elif args.cmd == "history": | |
| print(json.dumps(ci.history(limit=args.limit, kind=args.kind), indent=2, default=str)) | |
| elif args.cmd == "show": | |
| d = ci.inspect_pending(args.pending_id) | |
| print(json.dumps(d, indent=2, default=str) if d else "not found") | |
| elif args.cmd == "confirm": | |
| print(json.dumps(ci.confirm_edit(args.pending_id), indent=2, default=str)) | |
| elif args.cmd == "reject": | |
| print(json.dumps(ci.reject_edit(args.pending_id, reason=args.reason), indent=2, default=str)) | |
| elif args.cmd == "propose": | |
| old_text = Path(args.old).read_text(encoding="utf-8") | |
| new_text = Path(args.new).read_text(encoding="utf-8") | |
| result = ci.propose_edit( | |
| target_path=args.target, | |
| old_text=old_text, | |
| new_text=new_text, | |
| rationale=args.rationale, | |
| ) | |
| print(json.dumps(result, indent=2, default=str)) | |