echo/echo-memory.plugin.src/skills/echo-memory/scripts/echo_reflect.py

#!/usr/bin/env python3
"""echo_reflect.py — H5: automatic session-reflection capture.  [v1.0 — wired]

Today memory only accrues when the agent decides to write. This makes accrual the
default: at session end the model extracts durable items from the conversation and
emits a PROPOSAL set; this module dedups them against the entity index, previews them,
and applies the accepted ones via the normal `capture` path — so nothing is written
without a go-ahead (honors the "show before large/sensitive writes" safety rule).

DIVISION OF LABOR:
  * Extraction is MODEL-side (only the LLM has the conversation). The agent produces a
    JSON array of proposals matching PROPOSAL_SCHEMA.
  * This script is the deterministic spine: validate -> classify (new/update/inbox vs the
    entity index) -> preview -> apply on confirm. No NL understanding lives here, so it's
    testable offline.

PROPOSAL_SCHEMA (one per durable item the agent wants to remember):
  {
    "title":   "Bob Smith",                 # required
    "kind":    "person",                    # required unless "inbox": true
    "body":    "Principal at Acme; ...",      # markdown body (optional)
    "aliases": ["bob", "rs"],               # optional
    "tags":    ["client"],                  # optional; the kind is always seeded as a tag
    "sources": ["_agent/sessions/..."],     # optional backward links
    "date":    "YYYY-MM-DD",                # optional (meeting/decision kinds)
    "domain":  "business",                  # optional (area kind)
    "inbox":   false,                        # route to inbox when the home is unknown
    "confidence": 0.0-1.0                    # agent's own confidence; gates auto-suggest
  }

CLI (echo.py reflect): dry-run previews; --apply writes — matching bootstrap/migrate/sweep.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo                    # noqa: E402
import echo_index as idx_mod   # noqa: E402

MIN_CONFIDENCE = 0.6


def validate(proposals: list[dict]) -> tuple[list[dict], list[str]]:
    """Return (valid, errors). title always required; kind required unless inbox; a
    confidence below the floor is routed away (the agent should send it to the inbox)."""
    valid, errors = [], []
    for i, p in enumerate(proposals or []):
        if not str(p.get("title", "")).strip():
            errors.append(f"proposal[{i}]: missing title")
            continue
        is_inbox = bool(p.get("inbox"))
        if not is_inbox:
            kind = str(p.get("kind", "")).strip()
            if not kind:
                errors.append(f"proposal[{i}] '{p['title']}': missing kind (or set inbox:true)")
                continue
            if kind not in idx_mod.KIND_FOLDER:
                errors.append(f"proposal[{i}] '{p['title']}': unknown kind '{kind}'")
                continue
        if float(p.get("confidence", 1.0)) < MIN_CONFIDENCE:
            errors.append(f"proposal[{i}] '{p['title']}': confidence < {MIN_CONFIDENCE} — send to inbox instead")
            continue
        valid.append(p)
    return valid, errors


def classify(proposals: list[dict]) -> list[dict]:
    """Annotate each proposal with `_action` (create / update / inbox) and `_path`, by
    resolving its title against the entity index (alias-aware), so the preview shows
    create-vs-append and dedups against what's already in memory."""
    index = idx_mod.load()
    for p in proposals:
        if p.get("inbox") or not p.get("kind"):
            p["_action"], p["_path"] = "inbox", "inbox/captures/inbox.md"
            continue
        _, e = idx_mod.resolve(index, p["title"])
        if e and e.get("path"):
            p["_action"], p["_path"] = "update", e["path"]
        else:
            try:
                p["_path"] = idx_mod.derive_path(
                    p["kind"], idx_mod.slugify(p["title"]),
                    date=p.get("date"), domain=p.get("domain", "business"))
                p["_action"] = "create"
            except echo.EchoError as exc:
                p["_action"], p["_path"] = "error", str(exc)
    return proposals


def preview(proposals: list[dict]) -> str:
    """One confirmable line per proposal: action | kind | title -> target path."""
    rows = []
    for p in proposals:
        act = p.get("_action", "?")
        kind = "-" if act == "inbox" else p.get("kind", "?")
        rows.append(f"  {act:7} | {kind:9} | {p['title']}  ->  {p.get('_path', '?')}")
    return "\n".join(rows)


def apply_op(proposals: list[dict], confirm: bool = False) -> dict:
    """Core reflect (Phase 0): validate -> classify -> (apply via capture_op). Returns
    an envelope — rows carry the classified preview; with confirm, `results` carries
    each proposal's capture outcome. Never prints to stdout."""
    import echo_output
    valid, errors = validate(proposals)
    rows: list[dict] = []
    counts: dict[str, int] = {}
    if valid:
        classify(valid)
        counts = {a: sum(1 for p in valid if p.get("_action") == a)
                  for a in ("create", "update", "inbox", "error")}
        rows = [{"action": p.get("_action"), "kind": p.get("kind"),
                 "title": p["title"], "path": p.get("_path")} for p in valid]
    data = {"proposals": len(proposals or []), "valid": len(valid), "errors": errors,
            "counts": counts, "rows": rows, "dry_run": not confirm,
            "applied": 0, "gated": 0, "results": []}
    if not valid or not confirm:
        return echo_output.envelope("reflect", data)

    import echo_ops  # lazy: the apply path pulls in the capture/index/link stack
    applied = gated = 0
    results = []
    for p in valid:
        if p.get("_action") == "error":
            continue
        env = echo_ops.capture_op(
            p.get("kind"), p["title"], body_text=p.get("body") or "",
            aliases=p.get("aliases") or [], sources=p.get("sources") or [],
            tags=p.get("tags") or [],
            date=p.get("date"), domain=p.get("domain", "business"),
            inbox=bool(p.get("inbox")) or not p.get("kind"))
        act = env.get("action", "")
        if act == "duplicate-gate":
            gated += 1
        elif env.get("ok"):
            applied += 1
        results.append({"title": p["title"], "action": act,
                        "path": env.get("path"), "ok": bool(env.get("ok"))})
    data.update(applied=applied, gated=gated, results=results)
    return echo_output.envelope("reflect", data)


def apply(proposals: list[dict], confirm: bool = False) -> int:
    """CLI wrapper: same preview/summary text as pre-Phase-0. Without confirm it is a
    dry-run that writes nothing — the preview IS the confirmation step."""
    env = apply_op(proposals, confirm=confirm)
    for e in env["errors"]:
        print(f"skip: {e}", file=sys.stderr)
    if not env["valid"]:
        print("reflect: no valid proposals to apply.")
        return 0
    counts = env["counts"]
    print(f"reflect: {env['valid']} proposal(s) — "
          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} inbox"
          + (f", {counts['error']} error" if counts["error"] else ""))
    print("\n".join(f"  {r['action'] or '?':7} | {('-' if r['action'] == 'inbox' else r['kind'] or '?'):9} "
                    f"| {r['title']}  ->  {r['path'] or '?'}" for r in env["rows"]))
    if env["dry_run"]:
        print("\nreflect: dry-run — re-run with --apply to write these to memory.")
        return 0
    gated = env["gated"]
    print(f"reflect: applied {env['applied']}/{env['valid']} proposal(s)."
          + (f" {gated} stopped at the duplicate gate — re-propose with the existing "
             f"entity's title (or capture --merge-into <slug>)." if gated else ""))
    return 0