orchestrata_ultimate.session.hooks.session_pre_close_hook

Reference › orchestrata_ultimate › session › hooks › session_pre_close_hook

The PreToolUse hook that gates closing a session (spec 006, user story 6). It prints a JSON verdict with continue true or false. main reads the hook input from standard input and takes the working directory from it. In a directory that is not an Orchestrata project (no context.md and no implementation directory) it allows the close. Otherwise it builds a SessionController and resumes the active session; if that fails it allows the close with a warning message. validate_checklist checks that work is logged, artifacts are recorded, decisions are documented and a handoff is set. run_tests runs pytest and check_git_clean checks the working tree, and each failure is added to the list. With any failure the hook answers continue: false, with the failures and a numbered list of instructions for completing the checklist; otherwise it answers continue: true with a confirmation message. The same rules are enforced in code by validate_and_end of the controller.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

# mypy: ignore-errors
#!/usr/bin/env python3
"""PreToolUse hook that validates session checklist before close.

This hook runs before session close operations and:
1. Validates all checklist items are complete
2. Blocks close if any required item fails
3. Returns JSON with continue: false if validation fails

Installation:
    Add to ~/.claude/settings.json:
    {
      "hooks": {
        "PreToolUse": [
          {
            "matcher": ".*session.*close.*|.*end.*session.*",
            "hooks": [
              {
                "type": "command",
                "command": "python3 /path/to/session_pre_close_hook.py"
              }
            ]
          }
        ]
      }
    }
"""

import json
import subprocess
import sys
from collections.abc import Mapping
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.session import SessionController

# Add package src to path for imports
_impl_src = Path(__file__).resolve().parent.parent.parent
if str(_impl_src) not in sys.path:
    sys.path.insert(0, str(_impl_src))


@traces(
    spec="006",
    story="FR-011",
    reqs=[
        "REQ-006-FR11-001",
        "REQ-006-FR11-002",
        "REQ-006-FR11-003",
        "REQ-006-FR11-004",
        "REQ-006-FR11-005",
    ],
)
def run_tests(project_path: Path, env: Mapping[str, str] | None = None) -> bool:
    """Run pytest and return True if tests pass.

    Args:
        project_path: Project root to run tests in.
        env: Optional explicit environment mapping for the ``uv`` subprocess
            (for testing, e.g. simulating a missing ``uv``/PATH); when
            omitted, the subprocess inherits the real process environment.
    """
    try:
        result = subprocess.run(
            ["uv", "run", "pytest", "tests/", "-q"],
            cwd=project_path,
            capture_output=True,
            timeout=300,
            env=env,
        )
        return result.returncode == 0
    except subprocess.TimeoutExpired, FileNotFoundError:
        return False


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-002"])
def check_git_clean(project_path: Path, env: Mapping[str, str] | None = None) -> bool:
    """Check if git working directory is clean.

    Args:
        project_path: Project root to check.
        env: Optional explicit environment mapping for the ``git`` subprocess
            (for testing, e.g. simulating a missing ``git``/PATH); when
            omitted, the subprocess inherits the real process environment.
    """
    try:
        result = subprocess.run(
            ["git", "status", "--porcelain"],
            cwd=project_path,
            capture_output=True,
            text=True,
            env=env,
        )
        # Clean if output is empty (only untracked files allowed)
        output = result.stdout.strip()
        if not output:
            return True
        # Check if only untracked files (lines starting with ??)
        lines = output.split("\n")
        return all(line.startswith("??") for line in lines if line)
    except FileNotFoundError:
        return True  # No git, consider clean


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-002"])
def validate_checklist(controller: SessionController) -> list[str]:
    """Validate session checklist and return list of failures."""
    failures = []
    session = controller.active_session

    if session is None:
        failures.append("No active session")
        return failures

    # Check 1: Work logged (log not empty)
    if not session.log:
        failures.append("No work logged (session.log is empty)")

    # Check 2: Artifacts recorded (artifacts not empty)
    if not session.artifacts:
        failures.append("No artifacts recorded (session.artifacts is empty)")

    # Check 3: Technical decisions documented (choices not empty)
    if not session.choices:
        failures.append("No technical decisions documented (session.choices is empty)")

    # Check 4: Handoff set (handoff is not None)
    if session.handoff is None:
        failures.append("Handoff not set (call set_handoff() before closing)")

    return failures


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-002"])
def main() -> None:
    """Validate checklist and output result."""
    # Read hook input from stdin
    try:
        hook_input = sys.stdin.read()
        if hook_input.strip():
            data = json.loads(hook_input)
            cwd = data.get("cwd", ".")
        else:
            cwd = "."
    except json.JSONDecodeError, KeyError:
        cwd = "."

    project_path = Path(cwd)

    # Check if this is an Orchestrata project
    context_file = project_path / "context.md"
    impl_dir = project_path / "implementation"

    if not context_file.exists() and not impl_dir.exists():
        # Not an Orchestrata project, allow close
        result = {"continue": True}
        print(json.dumps(result))
        return

    # Initialize controller
    try:
        controller = SessionController(project_path)
        # Try to resume active session
        controller.resume()
    except Exception as e:
        # Cannot initialize, allow close but warn
        result = {
            "continue": True,
            "message": f"Warning: Could not validate session: {e}",
        }
        print(json.dumps(result))
        return

    # Validate checklist
    failures = validate_checklist(controller)

    # Check tests pass
    if not run_tests(project_path):
        failures.append("Tests do not pass (uv run pytest tests/ failed)")

    # Check git clean
    if not check_git_clean(project_path):
        failures.append("Git working directory is dirty (uncommitted changes)")

    if failures:
        result = {
            "continue": False,
            "message": "Session checklist validation failed",
            "failures": failures,
            "instructions": [
                "Complete all checklist items before closing (see AGENTS.md § Mandatory Startup Checklist):",
                "1. Log work: controller.log_work('description', ['details'])",
                "2. Log artifacts: controller.log_artifact('path', line_count, 'summary')",
                "3. Log decisions: controller.log_choice('choice', 'reason')",
                "4. Set handoff: controller.set_handoff(decisions=[...], ...)",
                "5. Ensure tests pass: uv run pytest tests/",
                "6. Commit changes: git add . && git commit -m 'message'",
                "7. End session: controller.end(achievements, upcoming) — auto-generates CHANGELOG.md + git tag + push",
            ],
        }
    else:
        result = {
            "continue": True,
            "message": "Session checklist validated successfully",
        }

    print(json.dumps(result))


if __name__ == "__main__":
    main()