417 lines
14 KiB
Python
Executable file
417 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""
|
||
upgrade-conscience.py — Programmatic `/upgrade-conscience` for the Réplica Omnisciente.
|
||
|
||
Scans the central brain (replica) and attached realms, then writes a new
|
||
conscience report to `.aurelio/data/conscience/`. Optionally commits, pushes,
|
||
and syncs the brain to CT 206 (Gabinete dos Heterónimos).
|
||
|
||
Usage:
|
||
python3 scripts/upgrade-conscience.py --commit --push --sync
|
||
python3 scripts/upgrade-conscience.py --root /path/to/replica --dry-run
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
DEFAULT_ROOT = Path(__file__).resolve().parent.parent
|
||
CONSCIENCE_DIR = Path(".aurelio/data/conscience")
|
||
AGENTS_FILE = Path("AGENTS.md")
|
||
MCP_CONFIG = Path(".aurelio/mcp_config.json")
|
||
WORKFLOWS_DIR = Path(".aurelio/workflows")
|
||
SKILLS_DIR = Path(".aurelio/skills")
|
||
REALMS_DIR = Path("realms")
|
||
BRAIN_DIR = Path(".aurelio/brain")
|
||
SYNC_SCRIPT = Path(".aurelio/sync.py")
|
||
|
||
SKIP_CI_MARKER = "[skip ci]"
|
||
|
||
|
||
def error(msg: str) -> None:
|
||
print(f"❌ {msg}", file=sys.stderr)
|
||
|
||
|
||
def info(msg: str) -> None:
|
||
print(f"ℹ️ {msg}")
|
||
|
||
|
||
def read_text(path: Path) -> str:
|
||
if not path.exists():
|
||
return ""
|
||
try:
|
||
return path.read_text(encoding="utf-8")
|
||
except Exception as exc:
|
||
error(f"Could not read {path}: {exc}")
|
||
return ""
|
||
|
||
|
||
def list_markdown_files(directory: Path) -> list[Path]:
|
||
if not directory.exists():
|
||
return []
|
||
return sorted(p for p in directory.iterdir() if p.is_file() and p.suffix == ".md")
|
||
|
||
|
||
def list_subdirectories(directory: Path) -> list[Path]:
|
||
if not directory.exists():
|
||
return []
|
||
return sorted(p for p in directory.iterdir() if p.is_dir())
|
||
|
||
|
||
def count_lines(text: str) -> int:
|
||
return len(text.splitlines())
|
||
|
||
|
||
def summarize_agents(agents_text: str) -> str:
|
||
"""Extract the first H1 and a short snippet from AGENTS.md."""
|
||
title_match = re.search(r"^#\s+(.+)$", agents_text, re.MULTILINE)
|
||
title = title_match.group(1).strip() if title_match else "Réplica Omnisciente"
|
||
snippet = agents_text[:600].replace("\n", " ").strip()
|
||
if len(snippet) > 600:
|
||
snippet = snippet[:597] + "..."
|
||
return f"**{title}** — {snippet}"
|
||
|
||
|
||
def parse_mcp_config(config_path: Path) -> dict:
|
||
text = read_text(config_path)
|
||
if not text:
|
||
return {}
|
||
try:
|
||
return json.loads(text)
|
||
except json.JSONDecodeError as exc:
|
||
error(f"Invalid JSON in {config_path}: {exc}")
|
||
return {}
|
||
|
||
|
||
def describe_mcp_servers(config: dict) -> list[str]:
|
||
"""Return a bullet list of discovered MCP servers."""
|
||
servers = config.get("mcpServers", config.get("servers", {}))
|
||
if not isinstance(servers, dict):
|
||
return []
|
||
lines: list[str] = []
|
||
for name, meta in servers.items():
|
||
if not isinstance(meta, dict):
|
||
continue
|
||
command = meta.get("command", "")
|
||
args = meta.get("args", [])
|
||
env = meta.get("env", {})
|
||
port_hint = ""
|
||
# Try to find a port number in args or env for documentation.
|
||
for token in args:
|
||
if isinstance(token, str) and re.match(r"^\d{4,5}$", token):
|
||
port_hint = f" (port {token})"
|
||
break
|
||
if not port_hint and env:
|
||
for v in env.values():
|
||
if isinstance(v, str) and re.search(r":(\d{4,5})", v):
|
||
port_hint = f" ({v})"
|
||
break
|
||
cmd_summary = f"{command} {' '.join(str(a) for a in args[:2])}".strip()
|
||
lines.append(f"- `{name}`{port_hint}: `{cmd_summary}`")
|
||
return lines
|
||
|
||
|
||
def scan_workflows(workflows_dir: Path) -> tuple[list[str], list[str]]:
|
||
files = list_markdown_files(workflows_dir)
|
||
names = [f.stem for f in files]
|
||
highlights: list[str] = []
|
||
for f in files:
|
||
text = read_text(f)
|
||
title_match = re.search(r"^#\s+(.+)$", text, re.MULTILINE)
|
||
title = title_match.group(1).strip() if title_match else f.stem
|
||
highlights.append(f"- `{f.name}` — {title}")
|
||
return names, highlights
|
||
|
||
|
||
def scan_skills(skills_dir: Path) -> tuple[list[str], list[str]]:
|
||
dirs = list_subdirectories(skills_dir)
|
||
names = [d.name for d in dirs]
|
||
highlights: list[str] = []
|
||
for d in dirs:
|
||
skill_md = d / "SKILL.md"
|
||
text = read_text(skill_md)
|
||
title_match = re.search(r"^#\s+(.+)$", text, re.MULTILINE)
|
||
title = title_match.group(1).strip() if title_match else d.name
|
||
highlights.append(f"- `{d.name}` — {title}")
|
||
return names, highlights
|
||
|
||
|
||
def scan_realms(realms_dir: Path) -> tuple[list[str], list[str], list[str]]:
|
||
dirs = list_subdirectories(realms_dir)
|
||
names = [d.name for d in dirs]
|
||
highlights: list[str] = []
|
||
inconsistencies: list[str] = []
|
||
for d in dirs:
|
||
realm_agents = d / "AGENTS.md"
|
||
memory_index = d / "memory" / "index.md"
|
||
agents_text = read_text(realm_agents)
|
||
memory_text = read_text(memory_index)
|
||
title = "Unknown realm"
|
||
if agents_text:
|
||
title_match = re.search(r"^#\s+(.+)$", agents_text, re.MULTILINE)
|
||
title = title_match.group(1).strip() if title_match else d.name
|
||
hl = f"- `{d.name}` — {title}"
|
||
if agents_text:
|
||
hl += f" ({count_lines(agents_text)} lines in AGENTS.md)"
|
||
else:
|
||
hl += " ⚠️ missing AGENTS.md"
|
||
inconsistencies.append(f"Realm `{d.name}` is missing `AGENTS.md`.")
|
||
if memory_text:
|
||
hl += f", {count_lines(memory_text)} lines in memory/index.md"
|
||
else:
|
||
hl += ", no memory/index.md"
|
||
highlights.append(hl)
|
||
return names, highlights, inconsistencies
|
||
|
||
|
||
UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I)
|
||
|
||
|
||
def brain_stats(brain_dir: Path) -> dict[str, int]:
|
||
"""Group top-level brain directories into categories to keep reports concise."""
|
||
stats: dict[str, int] = {}
|
||
if not brain_dir.exists():
|
||
return stats
|
||
|
||
categories: dict[str, tuple[int, int]] = {}
|
||
for entry in brain_dir.iterdir():
|
||
if not entry.is_dir():
|
||
continue
|
||
name = entry.name
|
||
count = sum(1 for _ in entry.rglob("*") if _.is_file())
|
||
|
||
if name.startswith("session-kimi-"):
|
||
key = "session-kimi-*"
|
||
elif name.startswith("session-mp"):
|
||
key = "session-mp*"
|
||
elif name.startswith("session-mo"):
|
||
key = "session-mo*"
|
||
elif name.startswith("session-"):
|
||
key = "session-other"
|
||
elif UUID_RE.match(name):
|
||
key = "legacy-uuid-sessions"
|
||
else:
|
||
key = name
|
||
|
||
dirs, files = categories.get(key, (0, 0))
|
||
categories[key] = (dirs + 1, files + count)
|
||
|
||
for key, (dirs, files) in sorted(categories.items()):
|
||
if dirs > 1:
|
||
stats[f"{key} ({dirs} dirs)"] = files
|
||
else:
|
||
stats[key] = files
|
||
return stats
|
||
|
||
|
||
def generate_report(
|
||
root: Path,
|
||
agents_text: str,
|
||
mcp_config: dict,
|
||
workflow_names: list[str],
|
||
workflow_highlights: list[str],
|
||
skill_names: list[str],
|
||
skill_highlights: list[str],
|
||
realm_names: list[str],
|
||
realm_highlights: list[str],
|
||
inconsistencies: list[str],
|
||
brain_statistics: dict[str, int],
|
||
) -> str:
|
||
now = datetime.now(timezone.utc)
|
||
version = now.strftime("v%Y.%m.%d-%H%M")
|
||
date_str = now.strftime("%Y-%m-%d %H:%M UTC")
|
||
|
||
mcp_bullets = describe_mcp_servers(mcp_config) or ["- No MCP servers declared."]
|
||
workflow_list = workflow_highlights or ["- No workflow files found."]
|
||
skill_list = skill_highlights or ["- No skill directories found."]
|
||
realm_list = realm_highlights or ["- No realms tracked."]
|
||
|
||
brain_summary = "\n".join(
|
||
f"- `{name}`: {count} artifact(s)" for name, count in sorted(brain_statistics.items())
|
||
) or "- Brain directory empty or not present."
|
||
|
||
lines = [
|
||
f"# Conscience Upgrade Report — {version}",
|
||
f"**Date:** {date_str}",
|
||
f"**Root:** `{root}`",
|
||
"",
|
||
"## Parsed Components",
|
||
"",
|
||
"### 1. Core System Alignment",
|
||
summarize_agents(agents_text),
|
||
"",
|
||
"### 2. MCP Configuration",
|
||
"Servers discovered in `.aurelio/mcp_config.json`:",
|
||
*mcp_bullets,
|
||
"",
|
||
"### 3. Skill & Workflow Matrix",
|
||
f"**Workflows** ({len(workflow_names)}):",
|
||
*workflow_list,
|
||
"",
|
||
f"**Skills** ({len(skill_names)}):",
|
||
*skill_list,
|
||
"",
|
||
"### 4. Realm Memory Assimilation",
|
||
f"**Realms** ({len(realm_names)}):",
|
||
*realm_list,
|
||
"",
|
||
"### 5. Brain Statistics",
|
||
brain_summary,
|
||
"",
|
||
"## New Capabilities & Changes Identified",
|
||
"- Regenerated programmatically by `scripts/upgrade-conscience.py`.",
|
||
"- Reflects the latest AGENTS.md, MCP config, skills, workflows, and realm memory.",
|
||
"",
|
||
"## Structural Inconsistencies",
|
||
]
|
||
|
||
if inconsistencies:
|
||
lines.extend(f"- {inc}" for inc in inconsistencies)
|
||
else:
|
||
lines.append("- No structural inconsistencies detected.")
|
||
|
||
lines.extend([
|
||
"",
|
||
"## Ascension Conclusion",
|
||
"The replica's conscience is now synchronized with the current state of the Portugal Futurista ecosystem.",
|
||
"Future agents should consult `.aurelio/data/conscience/current.md` for the authoritative system state.",
|
||
"",
|
||
])
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def write_reports(conscience_dir: Path, report: str) -> Path:
|
||
conscience_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# current.md and latest_conscience.md always reflect the newest report.
|
||
(conscience_dir / "current.md").write_text(report, encoding="utf-8")
|
||
(conscience_dir / "latest_conscience.md").write_text(report, encoding="utf-8")
|
||
|
||
# Also keep a versioned snapshot.
|
||
now = datetime.now(timezone.utc)
|
||
versioned_name = f"upgrade_report_{now.strftime('%Y_%m_%d_%H%M')}.md"
|
||
versioned_path = conscience_dir / versioned_name
|
||
versioned_path.write_text(report, encoding="utf-8")
|
||
return versioned_path
|
||
|
||
|
||
def git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess:
|
||
result = subprocess.run(
|
||
["git", *args],
|
||
cwd=str(repo),
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if check and result.returncode != 0:
|
||
raise RuntimeError(f"git {' '.join(args)} failed: {result.stderr.strip()}")
|
||
return result
|
||
|
||
|
||
def commit_and_push(repo: Path, skip_ci: bool) -> None:
|
||
git(repo, "add", str(CONSCIENCE_DIR))
|
||
|
||
status = git(repo, "status", "--porcelain", str(CONSCIENCE_DIR), check=False)
|
||
if not status.stdout.strip():
|
||
info("No conscience changes to commit.")
|
||
return
|
||
|
||
message = "🧠 conscience: upgrade system state"
|
||
if skip_ci:
|
||
message += f" {SKIP_CI_MARKER}"
|
||
git(repo, "commit", "-m", message)
|
||
info("Committed conscience upgrade.")
|
||
|
||
git(repo, "push", "origin", "main")
|
||
info("Pushed conscience upgrade to Forgejo.")
|
||
|
||
|
||
def sync_brain(repo: Path) -> None:
|
||
sync_script = repo / SYNC_SCRIPT
|
||
if not sync_script.exists():
|
||
error(f"Sync script not found: {sync_script}")
|
||
return
|
||
info("Syncing brain to CT 206 via .aurelio/sync.py --push ...")
|
||
result = subprocess.run(
|
||
[sys.executable, str(sync_script), "--push"],
|
||
cwd=str(repo),
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
print(result.stdout)
|
||
if result.returncode != 0:
|
||
error(f"Brain sync failed: {result.stderr.strip()}")
|
||
raise RuntimeError("Brain sync failed")
|
||
info("Brain sync complete.")
|
||
|
||
|
||
def main(argv: Iterable[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description="Upgrade the Réplica Omnisciente conscience.",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=__doc__,
|
||
)
|
||
parser.add_argument("--root", type=Path, default=DEFAULT_ROOT, help="Path to the replica repository")
|
||
parser.add_argument("--commit", action="store_true", help="Commit the generated reports")
|
||
parser.add_argument("--push", action="store_true", help="Push the commit to origin")
|
||
parser.add_argument("--sync", action="store_true", help="Run .aurelio/sync.py --push to CT 206")
|
||
parser.add_argument("--skip-ci", action="store_true", help=f"Add '{SKIP_CI_MARKER}' to the commit message")
|
||
parser.add_argument("--dry-run", action="store_true", help="Generate report but do not write or commit")
|
||
|
||
args = parser.parse_args(argv)
|
||
|
||
root: Path = args.root.resolve()
|
||
if not (root / AGENTS_FILE).exists():
|
||
error(f"This does not look like the replica repository: {root}")
|
||
return 1
|
||
|
||
agents_text = read_text(root / AGENTS_FILE)
|
||
mcp_config = parse_mcp_config(root / MCP_CONFIG)
|
||
|
||
workflow_names, workflow_highlights = scan_workflows(root / WORKFLOWS_DIR)
|
||
skill_names, skill_highlights = scan_skills(root / SKILLS_DIR)
|
||
realm_names, realm_highlights, inconsistencies = scan_realms(root / REALMS_DIR)
|
||
brain_statistics = brain_stats(root / BRAIN_DIR)
|
||
|
||
report = generate_report(
|
||
root=root,
|
||
agents_text=agents_text,
|
||
mcp_config=mcp_config,
|
||
workflow_names=workflow_names,
|
||
workflow_highlights=workflow_highlights,
|
||
skill_names=skill_names,
|
||
skill_highlights=skill_highlights,
|
||
realm_names=realm_names,
|
||
realm_highlights=realm_highlights,
|
||
inconsistencies=inconsistencies,
|
||
brain_statistics=brain_statistics,
|
||
)
|
||
|
||
if args.dry_run:
|
||
print(report)
|
||
return 0
|
||
|
||
conscience_dir = root / CONSCIENCE_DIR
|
||
versioned_path = write_reports(conscience_dir, report)
|
||
info(f"Wrote conscience reports to {conscience_dir}")
|
||
info(f"Versioned snapshot: {versioned_path}")
|
||
|
||
if args.commit:
|
||
commit_and_push(root, skip_ci=args.skip_ci or args.push)
|
||
|
||
if args.sync:
|
||
sync_brain(root)
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|