feat(ci): add upgrade-conscience.py and Forgejo Actions workflow

.forgejo/workflows/aurelio-sync.yml

@@ -0,0 +1,33 @@
+name: Aurélio Sync & Conscience Upgrade
+
+on:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+concurrency:
+  group: aurelio-sync-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  upgrade-conscience:
+    name: Upgrade Réplica Conscience
+    runs-on: self-hosted
+    steps:
+      - name: Checkout replica
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 1
+
+      - name: Configure Git
+        run: |
+          git config user.name "Aurélio CI"
+          git config user.email "ci@portugalfuturista.org"
+
+      - name: Generate and commit conscience upgrade
+        run: |
+          python3 scripts/upgrade-conscience.py --commit --push --skip-ci
+
+      - name: Sync brain to CT 206 (Gabinete dos Heterónimos)
+        run: |
+          python3 .aurelio/sync.py --push

scripts/upgrade-conscience.py

@@ -0,0 +1,417 @@
+#!/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())