agentic-ai-engineering/pipeline/expertise.py

#!/usr/bin/env python3
"""
Expertise File Manager — per-cron compounding memory.

Each recurring cron job gets its own expertise file at
  ~/.hermes/expertise/<job-name>.yaml

The file is loaded at job start and updated at job end, making every run
smarter than the last.

Usage:
  # Load expertise into context (inline at top of cron prompt):
  python -m pipeline.expertise --load --job triage

  # Append a new insight after work is done:
  python -m pipeline.expertise --append --job triage --domain github-issues \
    --insight "Issues with label 'bug' most often miss reproduction steps"

  # Read all expertise entries (JSON for tool consumption):
  python -m pipeline.expertise --read --job triage
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from datetime import datetime, timezone
from typing import Any, Optional

try:
    import yaml
except ModuleNotFoundError:
    class yaml:
        @staticmethod
        def safe_load(f):
            text = f.read()
            if not text.strip():
                return None
            try:
                return json.loads(text)
            except json.JSONDecodeError as exc:
                # ponytail: no silent YAML-to-empty fallback; install PyYAML for real YAML files.
                raise RuntimeError("PyYAML is required to read non-JSON expertise files") from exc

        @staticmethod
        def dump(data, f, **_kwargs):
            json.dump(data, f, indent=2, ensure_ascii=False)


# ── Paths ─────────────────────────────────────────────────────────────────────


def expertise_dir() -> str:
    """Return ~/.hermes/expertise/ directory, creating it if needed."""
    base = os.path.join(os.path.expanduser("~"), ".hermes", "expertise")
    os.makedirs(base, exist_ok=True)
    return base


def expertise_path(job_name: str) -> str:
    return os.path.join(expertise_dir(), f"{job_name}.yaml")


# ── Data Model ────────────────────────────────────────────────────────────────


ENTRY_SCHEMA = {
    "domain": str,
    "insight": str,
    "date": str,
    "source": str,
    "verified": bool,
}


def default_expertise() -> dict:
    return {
        "updatable": True,
        "version": 1,
        "entries": [],
    }


# ── Load / Save ───────────────────────────────────────────────────────────────


def load(job_name: str) -> dict:
    path = expertise_path(job_name)
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f) or default_expertise()
            data.setdefault("updatable", True)
            data.setdefault("version", 1)
            data.setdefault("entries", [])
            return data
    return default_expertise()


def save(job_name: str, data: dict) -> None:
    path = expertise_path(job_name)
    with open(path, "w", encoding="utf-8") as f:
        yaml.dump(
            data,
            f,
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )


# ── Operations ────────────────────────────────────────────────────────────────


def append_entry(
    job_name: str,
    domain: str,
    insight: str,
    source: Optional[str] = None,
    verified: bool = False,
) -> dict:
    """Append a new insight entry to the job's expertise file."""
    data = load(job_name)
    if not data.get("updatable", True):
        print(f"Expertise file for '{job_name}' is not updatable.", file=sys.stderr)
        return data

    entry = {
        "domain": domain,
        "insight": insight,
        "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "source": source or f"cron:{job_name}",
        "verified": verified,
    }
    data["entries"].append(entry)
    save(job_name, data)
    return data


def render_context(job_name: str) -> str:
    """Render expertise entries as a context string for agent prompts."""
    data = load(job_name)
    if not data["entries"]:
        return ""
    lines = [f"# Expertise ({job_name})"]
    for i, e in enumerate(data["entries"], 1):
        tag = "✓" if e.get("verified") else "·"
        lines.append(f"{tag} [{e['domain']}] {e['insight']} ({e['date']})")
    return "\n".join(lines)


def purge_domain(job_name: str, domain: str) -> dict:
    """Remove all entries for a specific domain."""
    data = load(job_name)
    before = len(data["entries"])
    data["entries"] = [e for e in data["entries"] if e.get("domain") != domain]
    removed = before - len(data["entries"])
    if removed:
        save(job_name, data)
        print(f"Removed {removed} entries for domain '{domain}'.")
    else:
        print(f"No entries found for domain '{domain}'.")
    return data


# ── CLI ───────────────────────────────────────────────────────────────────────


def main():
    parser = argparse.ArgumentParser(description="Expertise file manager")
    parser.add_argument("--job", required=True, help="Job name (corresponds to expertise filename)")
    parser.add_argument("--load", action="store_true", help="Print expertise as context (for injection at cron start)")
    parser.add_argument("--read", action="store_true", help="Dump entire expertise file as JSON")
    parser.add_argument("--append", action="store_true", help="Append a new insight entry")
    parser.add_argument("--domain", default="general", help="Domain tag for the insight")
    parser.add_argument("--insight", help="The insight text to store")
    parser.add_argument("--source", help="Source identifier (default: cron:<job>)")
    parser.add_argument("--verified", action="store_true", default=False, help="Mark insight as verified")
    parser.add_argument("--purge", help="Remove all entries for a domain")

    args = parser.parse_args()

    if args.load:
        print(render_context(args.job))
        return

    if args.read:
        data = load(args.job)
        print(json.dumps(data, indent=2))
        return

    if args.purge:
        purge_domain(args.job, args.purge)
        return

    if args.append:
        if not args.insight:
            print("ERROR: --insight is required with --append", file=sys.stderr)
            sys.exit(1)
        append_entry(
            args.job,
            domain=args.domain,
            insight=args.insight,
            source=args.source,
            verified=args.verified,
        )
        print(f"Appended insight to {args.job}.yaml")
        return

    parser.print_help()


if __name__ == "__main__":
    main()