ATOCore/deploy/hooks/capture_stop.py

#!/usr/bin/env python3
"""Claude Code Stop hook: capture interaction to AtoCore.

Reads the Stop hook JSON from stdin, extracts the last user prompt
from the transcript JSONL, and POSTs to the AtoCore /interactions
endpoint with reinforcement enabled (no extraction).

Fail-open: always exits 0, logs errors to stderr only.

Environment variables:
    ATOCORE_URL     Base URL of the AtoCore instance (default: http://dalidou:8100)
    ATOCORE_CAPTURE_DISABLED  Set to "1" to disable capture (kill switch)

Usage in ~/.claude/settings.json:
    "Stop": [{
        "matcher": "",
        "hooks": [{
            "type": "command",
            "command": "python /path/to/capture_stop.py",
            "timeout": 15
        }]
    }]
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request

ATOCORE_URL = os.environ.get("ATOCORE_URL", "http://dalidou:8100")
TIMEOUT_SECONDS = 10

# Minimum prompt length to bother capturing. Single-word acks,
# slash commands, and empty lines aren't useful interactions.
MIN_PROMPT_LENGTH = 15

# Maximum response length to capture. Truncate very long assistant
# responses to keep the interactions table manageable.
MAX_RESPONSE_LENGTH = 50_000


def main() -> None:
    """Entry point. Always exits 0."""
    try:
        _capture()
    except Exception as exc:
        print(f"capture_stop: {exc}", file=sys.stderr)


def _capture() -> None:
    if os.environ.get("ATOCORE_CAPTURE_DISABLED") == "1":
        return

    raw = sys.stdin.read()
    if not raw.strip():
        return

    hook_data = json.loads(raw)

    session_id = hook_data.get("session_id", "")
    assistant_message = hook_data.get("last_assistant_message", "")
    transcript_path = hook_data.get("transcript_path", "")
    cwd = hook_data.get("cwd", "")

    prompt = _extract_last_user_prompt(transcript_path)
    if not prompt or len(prompt.strip()) < MIN_PROMPT_LENGTH:
        return

    response = assistant_message or ""
    if len(response) > MAX_RESPONSE_LENGTH:
        response = response[:MAX_RESPONSE_LENGTH] + "\n\n[truncated]"

    project = _infer_project(cwd)

    payload = {
        "prompt": prompt,
        "response": response,
        "client": "claude-code",
        "session_id": session_id,
        "project": project,
        "reinforce": True,
    }

    body = json.dumps(payload, ensure_ascii=True).encode("utf-8")
    req = urllib.request.Request(
        f"{ATOCORE_URL}/interactions",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    resp = urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS)
    result = json.loads(resp.read().decode("utf-8"))
    print(
        f"capture_stop: recorded interaction {result.get('id', '?')} "
        f"(project={project or 'none'}, prompt_chars={len(prompt)}, "
        f"response_chars={len(response)})",
        file=sys.stderr,
    )


def _extract_last_user_prompt(transcript_path: str) -> str:
    """Read the JSONL transcript and return the last real user prompt.

    Skips meta messages (isMeta=True) and system/command messages
    (content starting with '<').
    """
    if not transcript_path:
        return ""

    # Normalize path for the current OS
    path = os.path.normpath(transcript_path)
    if not os.path.isfile(path):
        return ""

    last_prompt = ""
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue

                if entry.get("type") != "user":
                    continue
                if entry.get("isMeta", False):
                    continue

                msg = entry.get("message", {})
                if not isinstance(msg, dict):
                    continue

                content = msg.get("content", "")

                if isinstance(content, str):
                    text = content.strip()
                elif isinstance(content, list):
                    # Content blocks: extract text blocks
                    parts = []
                    for block in content:
                        if isinstance(block, str):
                            parts.append(block)
                        elif isinstance(block, dict) and block.get("type") == "text":
                            parts.append(block.get("text", ""))
                    text = "\n".join(parts).strip()
                else:
                    continue

                # Skip system/command XML and very short messages
                if text.startswith("<") or len(text) < MIN_PROMPT_LENGTH:
                    continue

                last_prompt = text
    except OSError:
        pass

    return last_prompt


# Project inference from working directory.
# Maps known repo paths to AtoCore project IDs. The user can extend
# this table or replace it with a registry lookup later.
_PROJECT_PATH_MAP: dict[str, str] = {
    # Add mappings as needed, e.g.:
    # "C:\\Users\\antoi\\gigabit": "p04-gigabit",
    # "C:\\Users\\antoi\\interferometer": "p05-interferometer",
}


def _infer_project(cwd: str) -> str:
    """Try to map the working directory to an AtoCore project."""
    if not cwd:
        return ""
    norm = os.path.normpath(cwd).lower()
    for path_prefix, project_id in _PROJECT_PATH_MAP.items():
        if norm.startswith(os.path.normpath(path_prefix).lower()):
            return project_id
    return ""


if __name__ == "__main__":
    main()