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

#!/usr/bin/env python3
"""echo_hook_stop.py — Stop hook: one-shot session-end reflection nudge.

Reflection (/echo-reflect) only captures memory if someone remembers to run it.
This hook watches the first Stop of a *substantive* session and, when nothing has
been reflected or session-logged yet, blocks once with a reminder so the model
runs the reflect flow (which still previews and asks the operator before writing
— the safety gate lives in reflect, not here).

Guards (a hook must NEVER trap a session in a loop or nag):
  * `stop_hook_active` -> exit immediately (loop guard);
  * fires at most ONCE per session (marker file under ECHO_STATE_DIR/hook-state);
  * skips quiet sessions (< ECHO_REFLECT_MIN_TURNS user turns, default 5);
  * skips when the transcript shows reflection / a session log already happened;
  * skips when ECHO isn't configured on this machine;
  * always exits 0.
"""
from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

MIN_TURNS = int(os.environ.get("ECHO_REFLECT_MIN_TURNS", "5"))

REASON = (
    "[echo-memory] Session-end reflection has not run yet. If this session produced "
    "durable facts, decisions, commitments, or artifacts worth remembering: run the "
    "/echo-reflect flow (extract -> preview -> apply only with the operator's confirm) "
    "and write the session log + heartbeat per the echo-memory skill. If nothing "
    "durable emerged, simply finish — do not invent memories. "
    "(This reminder fires once per session.)"
)


def state_marker(session_id: str) -> Path:
    base = Path(os.environ.get("ECHO_STATE_DIR") or (Path.home() / ".echo-memory"))
    return base / "hook-state" / f"{session_id}.reflect-nudged"


def count_user_turns(raw: str) -> int:
    turns = 0
    for line in raw.splitlines():
        try:
            rec = json.loads(line)
        except Exception:  # noqa: BLE001
            continue
        if rec.get("type") != "user":
            continue
        content = (rec.get("message") or {}).get("content")
        if isinstance(content, str) and content.strip():
            turns += 1
        elif isinstance(content, list) and any(
                isinstance(b, dict) and b.get("type") == "text" for b in content):
            turns += 1
    return turns


def already_reflected(raw: str) -> bool:
    """Transcript evidence that reflection or session logging already happened."""
    return ("/echo-reflect" in raw
            or re.search(r'echo\.py"?\s+reflect\b', raw) is not None
            or "put _agent/sessions/" in raw
            or "put _agent/heartbeat/last-session.md" in raw)


def main() -> int:
    try:
        payload = json.load(sys.stdin)
    except Exception:  # noqa: BLE001
        return 0
    if payload.get("stop_hook_active"):
        return 0
    session_id = str(payload.get("session_id") or "unknown")
    marker = state_marker(session_id)
    if marker.exists():
        return 0

    try:
        import echo_config
        if not echo_config.is_configured(echo_config.load()):
            return 0  # nothing to reflect into — stay silent
    except Exception:  # noqa: BLE001
        return 0

    raw = ""
    tp = payload.get("transcript_path")
    if tp:
        try:
            raw = Path(tp).read_text(encoding="utf-8", errors="replace")
        except Exception:  # noqa: BLE001
            raw = ""

    if already_reflected(raw):
        try:  # done for this session — never fire
            marker.parent.mkdir(parents=True, exist_ok=True)
            marker.write_text("reflected\n", encoding="utf-8")
        except Exception:  # noqa: BLE001
            pass
        return 0
    if count_user_turns(raw) < MIN_TURNS:
        return 0  # quiet so far; a later stop may qualify (no marker written)

    try:
        marker.parent.mkdir(parents=True, exist_ok=True)
        marker.write_text("nudged\n", encoding="utf-8")
    except Exception:  # noqa: BLE001
        pass  # if the marker can't be written, still nudge; stop_hook_active guards loops
    print(json.dumps({"decision": "block", "reason": REASON}, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())