orchestrata_ultimate.session.hooks.session_auto_start_hook

Reference › orchestrata_ultimate › session › hooks › session_auto_start_hook

The Claude Code SessionStart hook that begins a session automatically (spec 006, user story 6). main reads the hook input (working directory and start source) and acts only in an Orchestrata project, recognised by a context.md file or a packages directory. When the source is anything other than a fresh startup (resume, clear, compact) it prints a short "Session Resumed" note and stops. On startup it builds a SessionController, detects an unclosed previous session (which begin auto-closes), generates a purpose from git context and objectives from the controller, begins the session and prints the session digest into the hook output. It then runs the project's scripts/run_ast_pipeline.py, reports success or the error tail, loads the resulting AST metrics, prints them with the top unmapped entries, updates the metrics section of context.md, saves a numbered snapshot and prints a diff against the previous snapshot. Failures in each step print a warning instead of raising, so the hook never blocks the start of a Claude session.

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_auto_start_hook.py

# mypy: ignore-errors
#!/usr/bin/env python3
"""SessionStart hook that automatically begins a new session.

This hook runs at Claude Code session start and:
1. Reads existing session context
2. Automatically starts a new session via SessionController.begin()
3. Outputs session info as context for Claude
4. Runs AST pipeline and surfaces porting metrics
5. Updates context.md with fresh AST data
6. Saves per-session AST snapshot for historical diff

Installation:
    Add to ~/.claude/settings.json:
    {
      "hooks": {
        "SessionStart": [
          {
            "hooks": [
              {
                "type": "command",
                "command": "python3 /path/to/session_auto_start_hook.py"
              }
            ]
          }
        ]
      }
    }
"""

from __future__ import annotations

import json
import logging
import re
import shutil
import subprocess  # noqa: S404
import sys
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING

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

if TYPE_CHECKING:
    from orchestrata_ultimate.session.models import SessionRecord

# 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))


logger = logging.getLogger(__name__)


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _read_hook_input() -> tuple[str, str]:
    """Read and parse hook input from stdin.

    Returns:
        Tuple of (cwd, source).
    """
    try:
        hook_input = sys.stdin.read()
        if hook_input.strip():
            data = json.loads(hook_input)
            cwd = data.get("cwd", ".")
            source = data.get("source", "startup")
            data.get("session_id", "unknown")
            return cwd, source
    except (json.JSONDecodeError, KeyError) as exc:
        logger.debug("hook input parse: %s", exc)
    return ".", "startup"


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _detect_auto_closed_session(
    controller: SessionController,
) -> SessionRecord | None:
    """Return the latest session if it was never properly closed, else None."""
    try:
        ctx = controller.ctx.read()
        latest = ctx.latest_session()
        if latest and latest.length is None:
            return latest
    except (OSError, ValueError, AttributeError) as e:
        logger.debug("Could not detect auto-closed session: %s", e)
    return None


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _build_session_output(
    controller: SessionController,
    session: SessionRecord,
    auto_closed_session: SessionRecord | None,
) -> list[str]:
    """Assemble output lines for a freshly started session.

    Args:
        controller: Active SessionController.
        session: The newly begun session record.
        auto_closed_session: Previous session that was auto-closed, or None.

    Returns:
        List of output lines (not yet joined).
    """
    output: list[str] = []

    if auto_closed_session:
        output.append("## Previous Session Auto-Closed")
        output.append("")
        output.append(
            f"**Session {auto_closed_session.seq:03d}** was not properly closed."
        )
        output.append("It has been auto-closed with status 'auto-closed'.")
        output.append("")
        output.append("---")
        output.append("")

    output.append("## New Session Started")
    output.append("")
    output.append(f"**Session**: {session.seq:03d}")
    output.append(f"**Started**: {session.when.strftime('%Y-%m-%d %H:%M')}")
    output.append(f"**Purpose**: {session.purpose}")
    output.append("")
    output.append("### Instructions")
    output.append("1. Review carryover tasks from previous session")
    output.append(
        "2. Read `incus/README.md` and verify `.env` is loaded before any infra work"
    )
    output.append("3. Set session objectives using `controller.log_work()`")
    output.append("4. Log technical decisions using `controller.log_choice()`")
    output.append(
        "5. End session with `controller.end(achievements, upcoming)` — this auto-generates CHANGELOG.md, creates git tag, and pushes it"
    )
    output.append("")
    output.append("### References")
    output.append("- `incus/README.md` — container setup, env var contract")
    output.append(
        "- `AGENTS.md` § Incus Infrastructure & Environment Variables — IP policy"
    )
    output.append("- `AGENTS.md` § Session Management — workflow details")
    output.append("")

    try:
        ctx = controller.ctx.read()
        pending = ctx.pending_work()
        if pending:
            output.append("### Carryover Tasks")
            for task in pending:
                prefix = f"[{task.priority}] " if task.priority else ""
                output.append(f"- {prefix}{task.text}")
            output.append("")
    except (OSError, ValueError, AttributeError) as e:
        logger.debug("Failed to load carryover tasks: %s", e)

    output.extend(_format_goal_injection(controller))

    return output


@traces(
    spec="023",
    story="US-003",
    reqs=[
        "REQ-023-014",  # session start injects the goal into the conversation
        "REQ-023-015",  # first hook output includes statement + criteria status
    ],
)
def _format_goal_injection(controller: SessionController) -> list[str]:
    """US-003: inject the active goal into the session-start output."""
    try:
        goal = controller.get_goal()
    except (OSError, ValueError, AttributeError) as e:
        logger.debug("Failed to load goal context: %s", e)
        return []
    if goal is None:
        return []
    lines = ["### Active Goal (SPEC-023)", f"- **Statement**: {goal.statement}"]
    lines.append(f"- **Status**: {goal.status}")
    if goal.criteria:
        for criterion in goal.criteria:
            icon = "x" if criterion.status == "met" else " "
            lines.append(f"  - [{icon}] {criterion.description}")
    if goal.status == "blocked" and goal.blockers:
        lines.append(f"- **Blockers**: {', '.join(goal.blockers)}")
    elif goal.status == "active":
        remaining = [c.description for c in goal.criteria if c.status != "met"]
        if remaining:
            lines.append(
                f"- **Remaining**: {len(remaining)} criterion(s) — "
                "continue working toward the goal."
            )
    lines.append("")
    return lines


# ---------------------------------------------------------------------------
# AST context helpers (A + B + C)
# ---------------------------------------------------------------------------


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _load_ast_metrics(project_path: Path) -> dict | None:
    """Load and parse ast-deep-comparison.json."""
    path = project_path / "plans" / "ast-deep-comparison.json"
    if not path.exists():
        return None
    try:
        with path.open() as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        logger.warning("Failed to load AST metrics: %s", e)
        return None


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _format_ast_metrics(data: dict) -> list[str]:
    """Format AST summary as markdown table."""
    summary = data.get("summary", {})
    readiness = summary.get("readiness", {})
    total_pairs = summary.get("total_pairs", 0)
    unmapped_ult = summary.get("unmapped_ultimate", 0)
    unmapped_orc = summary.get("unmapped_orchestrata", 0)

    lines: list[str] = []
    lines.append("### Porting Progress")
    lines.append("")
    lines.append("| Metric | Count |")
    lines.append("|--------|-------|")
    lines.append(f"| Semantic pairs | {total_pairs} |")
    lines.extend(
        f"| {key.replace('_', ' ').title()} | {val} |" for key, val in readiness.items()
    )
    lines.append(f"| Unmapped (ultimate) | {unmapped_ult} |")
    lines.append(f"| Unmapped (orchestrata) | {unmapped_orc} |")
    lines.append("")
    return lines


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _format_top_unmapped(data: dict, limit: int = 10) -> list[str]:
    """List top unmapped class names."""
    unmapped = data.get("unmapped", {})
    ultimate_only = unmapped.get("ultimate_only", [])
    if not ultimate_only:
        return []
    names = ultimate_only[:limit]
    suffix = (
        f" and {len(ultimate_only) - limit} more" if len(ultimate_only) > limit else ""
    )
    lines: list[str] = []
    lines.append(f"**Top unmapped classes**: {', '.join(names)}{suffix}")
    lines.append("")
    return lines


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _save_ast_snapshot(project_path: Path, seq: int) -> Path | None:
    """Copy ast-deep-comparison.json to a per-session snapshot."""
    src = project_path / "plans" / "ast-deep-comparison.json"
    if not src.exists():
        return None
    dst = project_path / "plans" / f"ast-deep-comparison-{seq:03d}.json"
    try:
        shutil.copy2(src, dst)
        return dst
    except OSError as e:
        logger.warning("Failed to save AST snapshot: %s", e)
        return None


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _load_previous_snapshot(project_path: Path, current_seq: int) -> dict | None:
    """Find the most recent snapshot before current_seq."""
    plans_dir = project_path / "plans"
    if not plans_dir.exists():
        return None
    candidates = sorted(
        plans_dir.glob("ast-deep-comparison-*.json"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    for candidate in candidates:
        # Extract seq from filename
        match = re.search(r"-(\d{3})\.json$", candidate.name)
        if match:
            seq = int(match.group(1))
            if seq < current_seq:
                try:
                    with candidate.open() as f:
                        return json.load(f)
                except (json.JSONDecodeError, OSError) as e:
                    logger.debug("Failed to load snapshot %s: %s", candidate, e)
                    continue
    return None


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _format_ast_diff(current: dict, previous: dict | None) -> list[str]:
    """Compute delta between current and previous AST data."""
    if previous is None:
        return []
    cur_sum = current.get("summary", {})
    prev_sum = previous.get("summary", {})
    cur_read = cur_sum.get("readiness", {})
    prev_read = prev_sum.get("readiness", {})

    lines: list[str] = []
    lines.append("### Progress Since Last Session")
    lines.append("")

    # Readiness deltas
    all_keys = set(cur_read.keys()) | set(prev_read.keys())
    for key in sorted(all_keys):
        cur_v = cur_read.get(key, 0)
        prev_v = prev_read.get(key, 0)
        delta = cur_v - prev_v
        sign = f"(+{delta})" if delta > 0 else f"({delta})" if delta < 0 else "(=)"
        lines.append(f"- {key.replace('_', ' ').title()}: {cur_v} {sign}")

    # Unmapped delta
    cur_um = cur_sum.get("unmapped_ultimate", 0)
    prev_um = prev_sum.get("unmapped_ultimate", 0)
    um_delta = cur_um - prev_um
    um_sign = (
        f"(+{um_delta})" if um_delta > 0 else f"({um_delta})" if um_delta < 0 else "(=)"
    )
    lines.append(f"- Unmapped (ultimate): {cur_um} {um_sign}")

    lines.append("")
    return lines


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _update_context_md(project_path: Path, seq: int, data: dict | None) -> None:
    """Rewrite the ## Current State section of context.md with AST metrics."""
    context_path = project_path / "context.md"
    if data is None:
        return

    summary = data.get("summary", {})
    readiness = summary.get("readiness", {})
    total_pairs = summary.get("total_pairs", 0)
    unmapped_ult = summary.get("unmapped_ultimate", 0)
    unmapped_orc = summary.get("unmapped_orchestrata", 0)
    now = datetime.now().strftime("%Y-%m-%d %H:%M")

    new_section = f"""## Current State

- Active Session: {seq}
- Version: 0.2.1
- Branch: master
- Test State: unknown
- Status: OPEN
- AST Refresh: {now}

### Porting Metrics

| Metric | Count |
|--------|-------|
| Semantic pairs | {total_pairs} |
| Mostly complete | {readiness.get('mostly_complete', 0)} |
| Equal | {readiness.get('equal', 0)} |
| Unmapped (ultimate) | {unmapped_ult} |
| Unmapped (orchestrata) | {unmapped_orc} |

"""

    if context_path.exists():
        try:
            text = context_path.read_text()
        except OSError as e:
            logger.warning("Cannot read context.md: %s", e)
            return
    else:
        text = ""

    # Try to replace existing ## Current State block
    pattern = r"## Current State\n.*?\n(?=## |\Z)"
    if re.search(pattern, text, re.DOTALL):
        text = re.sub(pattern, new_section, text, count=1, flags=re.DOTALL)
    else:
        # Prepend if no Current State section found
        text = new_section + text

    try:
        context_path.write_text(text)
    except OSError as e:
        logger.warning("Cannot write context.md: %s", e)


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _generate_session_purpose(project_path: Path) -> str:
    """Generate a meaningful session purpose from git context.

    Uses branch name and recent commit messages to infer what work
    is in progress, rather than just a timestamp.
    """
    now = datetime.now()
    parts: list[str] = []

    # Try to get branch name
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            cwd=project_path,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0:
            branch = result.stdout.strip()
            if branch and branch != "HEAD":
                parts.append(f"[{branch}]")
    except FileNotFoundError, subprocess.SubprocessError:
        pass

    # Try to get most recent commit message as context
    try:
        result = subprocess.run(
            ["git", "log", "-1", "--format=%s"],
            cwd=project_path,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0 and result.stdout.strip():
            commit_msg = result.stdout.strip()
            # Truncate long messages
            if len(commit_msg) > 60:
                commit_msg = commit_msg[:57] + "..."
            parts.append(commit_msg)
    except FileNotFoundError, subprocess.SubprocessError:
        pass

    if parts:
        return f"{' '.join(parts)} — {now.strftime('%Y-%m-%d %H:%M')}"
    return f"Session {now.strftime('%Y-%m-%d %H:%M')}"


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def _generate_objectives(controller: SessionController) -> list[str]:  # type: ignore[name-defined]
    """Generate objectives from carryover tasks and context.

    Looks at pending work items and handoff data from the previous
    session to suggest what should be done next.
    """
    objectives: list[str] = []
    try:
        ctx = controller.ctx.read()
        # Add pending work items as objectives
        for item in ctx.pending_work()[:5]:
            prefix = f"[{item.priority}] " if item.priority else ""
            objectives.append(f"{prefix}{item.text}")

        # Add handoff active work if available
        latest = ctx.latest_session()
        if latest and latest.handoff:
            for work in latest.handoff.active_work[:3]:
                if work not in objectives:
                    objectives.append(work)
    except (OSError, ValueError, AttributeError) as e:
        logger.debug("Could not generate objectives: %s", e)

    return objectives


@traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
def main() -> None:
    """Start new session and output context."""

    cwd, source = _read_hook_input()
    project_path = Path(cwd)

    # Check if this is an Orchestrata project (has context.md or packages/)
    context_file = project_path / "context.md"
    packages_dir = project_path / "packages"

    if not context_file.exists() and not packages_dir.exists():
        # Not an Orchestrata project, skip
        return

    # Only auto-start on fresh startup, not resume/clear/compact
    if source != "startup":
        # For resume, just show current session info
        print("## Session Resumed")
        print(f"Source: {source}")
        return

    # Initialize controller
    try:
        controller = SessionController(project_path)
    except Exception as e:
        print("## Session Controller Error")
        print(f"Failed to initialize: {e}")
        return

    # Check for unclosed session and auto-close if needed
    auto_closed_session = _detect_auto_closed_session(controller)

    # Generate meaningful session purpose from git context
    purpose = _generate_session_purpose(project_path)
    objectives = _generate_objectives(controller)

    # Start new session (this will auto-close unclosed session)
    try:
        session = controller.begin(
            purpose=purpose,
            objectives=objectives,
        )
        output = _build_session_output(controller, session, auto_closed_session)
        print("\n".join(output))

    except Exception as e:
        print("## Session Start Failed")
        print(f"Error: {e}")
        print("")
        print("You may need to start the session manually.")
        return

    # --- Auto-run AST pipeline at session start ---
    print("")
    print("### AST Sync")
    try:
        result = subprocess.run(
            [sys.executable, str(project_path / "scripts" / "run_ast_pipeline.py")],
            cwd=project_path,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0:
            print("✅ AST pipeline complete (plans/ast-*.json refreshed)")
        else:
            print("⚠️ AST pipeline exited with errors:")
            print(result.stderr[-800:] if result.stderr else "(no stderr)")
    except Exception as e:
        print(f"⚠️ Could not run AST pipeline: {e}")

    # --- A + B + C: Load metrics, update context.md, snapshot, diff ---
    seq = session.seq
    ast_data = _load_ast_metrics(project_path)
    if ast_data:
        # A: Surface metrics
        print("")
        print("\n".join(_format_ast_metrics(ast_data)))
        print("\n".join(_format_top_unmapped(ast_data, limit=10)))

        # B: Update context.md
        _update_context_md(project_path, seq, ast_data)

        # C: Snapshot + diff
        snapshot_path = _save_ast_snapshot(project_path, seq)
        if snapshot_path:
            logger.debug("AST snapshot saved: %s", snapshot_path)
        previous = _load_previous_snapshot(project_path, seq)
        diff_lines = _format_ast_diff(ast_data, previous)
        if diff_lines:
            print("\n".join(diff_lines))
    else:
        print("")
        print("⚠️ No AST comparison data available (run `uv run ast-sync` manually)")


if __name__ == "__main__":
    main()