orchestrata_ultimate.session.hooks.session_fact_hook

Reference › orchestrata_ultimate › session › hooks › session_fact_hook

The PostToolUse hook that records facts about tool use. extract_facts(tool_name, tool_input, tool_output, exit_code) returns a list of name/value facts: it dispatches Read calls, Bash calls (using the output and exit code) and Write or Edit calls to dedicated helpers, and returns an empty list for any other tool. get_session_id takes the identifier from the CLAUDE_SESSION_ID environment variable, else from a .claude_session_id file in the working directory, else generates a short identifier and saves it there. store_facts appends each fact as a JSON line (session id, name, value, timestamp and source) to .session_facts/<session id>.jsonl; this hook uses file storage, not the Redis or graph stores of FactEmitter. main reads the hook event from standard input, extracts the facts and stores them.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/hooks/session_fact_hook.py

"""Session Fact Hook module for Orchestrata."""

import json
import os
import sys
from collections.abc import Mapping
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

# !/usr/bin/env python3
"""PostToolUse hook that extracts facts from tool outputs.

This hook runs after each tool use and extracts facts that are relevant
to session governance. Facts are emitted to the fact storage system
(Redis/graph/file) for Prolog governance to query.

Usage as Claude Code hook:
    In settings.json or hooks config:
    {
        "hooks": {
            "PostToolUse": [
                {
                    "type": "command",
                    "command": "python3 /path/to/session_fact_hook.py"
                }
            ]
        }
    }
"""


def extract_facts(
    tool_name: str, tool_input: dict, tool_output: str, exit_code: int = 0
) -> list[dict]:
    """Extract facts from tool execution.

    Args:
        tool_name: Name of the tool (Read, Write, Bash, etc.)
        tool_input: Tool input parameters
        tool_output: Tool output/result
        exit_code: Exit code for Bash commands

    Returns:
        List of fact dicts with 'name' and 'value' keys
    """
    if tool_name == "Read":
        return _read_tool_facts(tool_input)
    if tool_name == "Bash":
        return _bash_tool_facts(tool_input, tool_output, exit_code)
    if tool_name in ("Write", "Edit"):
        return _write_tool_facts(tool_input)
    return []


def _read_tool_facts(tool_input: dict) -> list[dict]:
    """Extract facts from a Read tool use."""
    file_path = tool_input.get("file_path", "")
    facts = []
    # Context file read
    if "context.md" in file_path:
        facts.append({"name": "context_read", "value": True})
    # CLAUDE.md read (project understanding)
    if "CLAUDE.md" in file_path:
        facts.append({"name": "claude_md_read", "value": True})
    return facts


def _bash_tool_facts(tool_input: dict, tool_output: str, exit_code: int) -> list[dict]:
    """Extract facts from a Bash tool use."""
    command = tool_input.get("command", "")
    facts = []

    # Test execution
    if "pytest" in command or "python -m pytest" in command:
        facts.append({"name": "tests_run", "value": True})
        facts.append({"name": "tests_passing", "value": exit_code == 0})

    # Git status check
    if "git status" in command:
        clean = (
            "nothing to commit" in tool_output or "working tree clean" in tool_output
        )
        facts.append({"name": "git_clean", "value": clean})

    # Git commit
    if "git commit" in command and exit_code == 0:
        facts.append({"name": "changes_committed", "value": True})

    # Git push
    if "git push" in command and exit_code == 0:
        facts.append({"name": "changes_pushed", "value": True})

    return facts


def _write_tool_facts(tool_input: dict) -> list[dict]:
    """Extract facts from a Write or Edit tool use."""
    file_path = tool_input.get("file_path", "")
    facts = []
    # Context update
    if "context.md" in file_path:
        facts.append({"name": "work_logged", "value": True})
    # Handover written
    if "handover" in file_path.lower() or "HANDOVER" in file_path:
        facts.append({"name": "handover_written", "value": True})
    return facts


def get_session_id(environ: Mapping[str, str] | None = None) -> str:
    """Get current session ID from environment or generate one.

    Args:
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.
    """
    # Check for session ID in environment (set by SessionStart hook)
    env = os.environ if environ is None else environ
    session_id = env.get("CLAUDE_SESSION_ID")
    if session_id:
        return session_id

    # Check for session file
    session_file = Path(".claude_session_id")
    if session_file.exists():
        return session_file.read_text().strip()

    # Generate new session ID
    import uuid

    session_id = str(uuid.uuid4())[:8]
    session_file.write_text(session_id)
    return session_id


def store_facts(session_id: str, facts: list[dict]) -> None:
    """Store facts to file-based storage.

    Args:
        session_id: Current session ID
        facts: List of facts to store
    """
    if not facts:
        return

    # Use file-based storage (Redis/graph can be added later)
    facts_dir = Path(".session_facts")
    facts_dir.mkdir(exist_ok=True)

    facts_file = facts_dir / f"{session_id}.jsonl"
    timestamp = datetime.now(UTC).isoformat()

    with open(facts_file, "a") as f:
        for fact in facts:
            record = {
                "session_id": session_id,
                "name": fact["name"],
                "value": fact["value"],
                "recorded_at": timestamp,
                "source": "session_fact_hook",
            }
            f.write(json.dumps(record) + "\n")


def main(input_stream: Any = None) -> None:
    """Main entry point for PostToolUse hook.

    Args:
        input_stream: Optional file-like object to read JSON from
            (defaults to sys.stdin).
    """
    stream = input_stream or sys.stdin
    # Read hook input from stdin
    try:
        hook_input = json.loads(stream.read())
    except json.JSONDecodeError:
        # No input or invalid JSON
        print(json.dumps({"continue": True}))
        return

    # Extract tool information
    tool_name = hook_input.get("tool_name", "")
    tool_input = hook_input.get("tool_input", {})
    tool_output = hook_input.get("tool_output", "")
    exit_code = hook_input.get("exit_code", 0)

    # Get session ID
    session_id = get_session_id()

    # Extract facts from this tool use
    facts = extract_facts(tool_name, tool_input, tool_output, exit_code)

    # Store facts
    if facts:
        store_facts(session_id, facts)

        # Log extracted facts (visible in hook output)
        fact_names = [f["name"] for f in facts]
        print(
            json.dumps(
                {
                    "continue": True,
                    "message": f"Extracted facts: {', '.join(fact_names)}",
                }
            )
        )
    else:
        print(json.dumps({"continue": True}))


if __name__ == "__main__":
    main()