src/omega/server/hook_server.py

"""OMEGA Hook Server — Unix Domain Socket daemon for fast hook dispatch.

Runs inside the MCP server process, reusing warm bridge/coordination singletons.
Hooks connect via ~/.omega/hook.sock, send a JSON request, and get a JSON response.
This eliminates ~750ms of cold-start overhead per hook invocation.
"""

import asyncio
import hashlib
import json
import logging
import os
import re
import subprocess
import time
import traceback
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger("omega.hook_server")

SOCK_PATH = Path.home() / ".omega" / "hook.sock"

# Debounce state (in-memory, reset on server restart)
_last_heartbeat: dict[str, float] = {}  # session_id -> monotonic timestamp
_last_claim: dict[tuple[str, str], float] = {}  # (session_id, file_path) -> timestamp
_last_surface: dict[str, float] = {}  # file_path -> timestamp (capped at 500)
_MAX_SURFACE_ENTRIES = 500
_last_overlap_notify: dict[tuple[str, str, str], float] = {}  # (sid, other_sid, file) -> timestamp
OVERLAP_NOTIFY_DEBOUNCE_S = 300.0  # 5 minutes
_last_block_notify: dict[tuple[str, str, str], float] = {}  # (blocked_sid, owner_sid, file) -> timestamp
BLOCK_NOTIFY_DEBOUNCE_S = 300.0  # 5 minutes
_last_peer_dir_check: dict[str, float] = {}  # dir_path -> timestamp
PEER_DIR_CHECK_DEBOUNCE_S = 300.0  # 5 minutes
_last_coord_query: dict[str, float] = {}  # session_id -> monotonic timestamp
COORD_QUERY_DEBOUNCE_S = 120.0  # 2 minutes
_last_reminder_check: float = 0.0
REMINDER_CHECK_DEBOUNCE_S = 300.0  # 5 minutes

_last_deadlock_push: dict[str, float] = {}  # cycle_hash -> monotonic timestamp
DEADLOCK_PUSH_DEBOUNCE_S = 600.0  # 10 minutes — don't spam same cycle

HEARTBEAT_DEBOUNCE_S = 30.0
CLAIM_DEBOUNCE_S = 30.0
SURFACE_DEBOUNCE_S = 5.0

_heartbeat_count: dict[str, int] = {}  # session_id -> call count (for inbox surfacing)
_peer_snapshot: dict[str, set] = {}  # session_id -> set of known peer session_ids
_session_intent: dict[str, str] = {}  # session_id -> latest classified intent

# Urgent message queue — push notifications from send_message to recipient's next heartbeat
_pending_urgent: dict[str, list[dict]] = {}  # session_id -> urgent message summaries
_MAX_URGENT_PER_SESSION = 10


# Error dedup state (mirrors standalone surface_memories.py)
_error_hashes: set = set()  # capped at 200 per session
_MAX_ERROR_HASHES = 200
_error_counts: dict[str, int] = {}  # session_id -> error count
_MAX_ERRORS_PER_SESSION = 5


# ---------------------------------------------------------------------------
# Agent nicknames — deterministic human-readable names from session IDs
# ---------------------------------------------------------------------------

_AGENT_NAMES = [
    "Alder",
    "Aspen",
    "Birch",
    "Briar",
    "Brook",
    "Cedar",
    "Cliff",
    "Cloud",
    "Coral",
    "Crane",
    "Creek",
    "Dune",
    "Elm",
    "Ember",
    "Fern",
    "Finch",
    "Flint",
    "Frost",
    "Glen",
    "Grove",
    "Hawk",
    "Hazel",
    "Heath",
    "Heron",
    "Holly",
    "Iris",
    "Ivy",
    "Jade",
    "Jay",
    "Juniper",
    "Lake",
    "Lark",
    "Laurel",
    "Leaf",
    "Lily",
    "Maple",
    "Marsh",
    "Meadow",
    "Moss",
    "Oak",
    "Olive",
    "Opal",
    "Orca",
    "Osprey",
    "Pearl",
    "Pebble",
    "Pine",
    "Rain",
    "Raven",
    "Reed",
    "Ridge",
    "Robin",
    "Sage",
    "Shore",
    "Sky",
    "Slate",
    "Stone",
    "Storm",
    "Swift",
    "Thorn",
    "Tide",
    "Vale",
    "Willow",
    "Wren",
]


def _agent_nickname(session_id: str) -> str:
    """Generate a deterministic, memorable nickname from a session ID.

    Returns format: "Nickname (abcd1234)" — e.g. "Cedar (a3f2b1c8)".
    """
    if not session_id:
        return "unknown"
    idx = int(hashlib.md5(session_id.encode()).hexdigest()[:8], 16) % len(_AGENT_NAMES)
    return f"{_AGENT_NAMES[idx]} ({session_id[:8]})"


def notify_session(target_session_id: str, msg_summary: dict) -> None:
    """Queue an urgent message notification for a target session.

    Called from coord_handlers after a successful send_message. The recipient's
    next heartbeat (even if debounced) will drain and surface these.
    """
    try:
        if not target_session_id or not msg_summary:
            return
        queue = _pending_urgent.setdefault(target_session_id, [])
        queue.append(msg_summary)
        # Cap at max, keeping newest
        if len(queue) > _MAX_URGENT_PER_SESSION:
            _pending_urgent[target_session_id] = queue[-_MAX_URGENT_PER_SESSION:]
    except Exception:
        pass  # Fail-open


def _drain_urgent_queue(session_id: str) -> str:
    """Pop and format all pending urgent messages for a session.

    Returns formatted [INBOX] lines or empty string.
    """
    queue = _pending_urgent.pop(session_id, None)
    if not queue:
        return ""
    try:
        previews = []
        for msg in queue:
            from_name = _agent_nickname(msg.get("from_session") or "unknown")
            subj = (msg.get("subject") or "")[:60]
            mtype = msg.get("msg_type", "inform")
            previews.append(f'{from_name} [{mtype}]: "{subj}"')
        return "[INBOX] " + " | ".join(previews) + " — use omega_inbox for details"
    except Exception:
        return ""


def _secure_append(log_path: Path, data: str):
    """Append to a file with secure permissions (0o600)."""
    log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd = os.open(str(log_path), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
    try:
        os.write(fd, data.encode("utf-8"))
    finally:
        os.close(fd)


def _log_hook_error(hook_name: str, error: Exception):
    """Log hook errors to ~/.omega/hooks.log."""
    try:
        log_path = Path.home() / ".omega" / "hooks.log"
        timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
        tb = traceback.format_exc()
        _secure_append(log_path, f"[{timestamp}] hook_server/{hook_name}: {error}\n{tb}\n")
    except Exception:
        pass


def _log_timing(hook_name: str, elapsed_ms: float):
    """Log hook timing to ~/.omega/hooks.log."""
    try:
        log_path = Path.home() / ".omega" / "hooks.log"
        timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
        _secure_append(log_path, f"[{timestamp}] hook_server/{hook_name}: OK ({elapsed_ms:.0f}ms)\n")
    except Exception:
        pass


def _auto_cloud_sync(session_id: str):
    """Fire-and-forget cloud sync on session stop. Fully fail-open."""
    try:
        secrets_path = Path.home() / ".omega" / "secrets.json"
        if not secrets_path.exists():
            return  # Cloud not configured — fast bail

        import threading

        def _sync():
            t0 = time.monotonic()
            try:
                from omega.cloud.sync import get_sync

                get_sync().sync_all()
                # Write push marker for status tracking
                push_marker = Path.home() / ".omega" / "last-cloud-push"
                push_marker.parent.mkdir(parents=True, exist_ok=True)
                push_marker.write_text(datetime.now(timezone.utc).isoformat())
                _log_timing("auto_cloud_sync", (time.monotonic() - t0) * 1000)
            except Exception as e:
                _log_hook_error("auto_cloud_sync", e)

        t = threading.Thread(target=_sync, daemon=True, name="omega-cloud-sync")
        t.start()
    except Exception:
        pass  # Never propagate


def _resolve_entity(project: str) -> "Optional[str]":
    """Resolve project→entity_id. Fail-open: returns None on any error.

    Delegates to resolve_project_entity() which reads config-file mappings.
    Returns None when no mappings exist or no match found.
    """
    try:
        from omega.entity.engine import resolve_project_entity

        return resolve_project_entity(project)
    except Exception:
        return None


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def _relative_time_from_iso(iso_str: str) -> str:
    """Convert an ISO timestamp to a human-readable relative time like '2d ago'."""
    try:
        dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        secs = (datetime.now(timezone.utc) - dt).total_seconds()
        if secs < 0:
            return "just now"
        if secs < 60:
            return f"{int(secs)}s ago"
        if secs < 3600:
            return f"{int(secs / 60)}m ago"
        if secs < 86400:
            return f"{int(secs / 3600)}h ago"
        return f"{int(secs / 86400)}d ago"
    except Exception:
        return ""


def _check_milestone(name: str) -> bool:
    """Return True if milestone not yet achieved (first time). Creates marker."""
    marker = Path.home() / ".omega" / "milestones" / name
    if marker.exists():
        return False
    marker.parent.mkdir(parents=True, exist_ok=True)
    marker.touch()
    return True


def _should_run_periodic(marker_name: str, interval_seconds: int) -> bool:
    """Check if a periodic task should run based on its marker file age.

    Returns True if the marker is missing or older than interval_seconds.
    Updates the marker timestamp on caller's behalf (caller writes after success).
    """
    marker = Path.home() / ".omega" / marker_name
    if not marker.exists():
        return True
    try:
        last_ts = marker.read_text().strip()
        last = datetime.fromisoformat(last_ts)
        if last.tzinfo is None:
            last = last.replace(tzinfo=timezone.utc)
        age_seconds = (datetime.now(timezone.utc) - last).total_seconds()
        return age_seconds >= interval_seconds
    except Exception:
        return True


def _update_marker(marker_name: str) -> None:
    """Write current UTC timestamp to a marker file."""
    marker = Path.home() / ".omega" / marker_name
    marker.parent.mkdir(parents=True, exist_ok=True)
    marker.write_text(datetime.now(timezone.utc).isoformat())


def _parse_tool_input(payload: dict) -> dict:
    """Parse tool_input from a hook payload into a dict. Returns {} on failure."""
    raw = payload.get("tool_input", "{}")
    try:
        return json.loads(raw) if isinstance(raw, str) else (raw or {})
    except (json.JSONDecodeError, TypeError):
        return {}


def _get_file_path_from_input(input_data: dict) -> str:
    """Extract file_path or notebook_path from parsed tool input."""
    return input_data.get("file_path", input_data.get("notebook_path", ""))


def _debounce_check(cache: dict, key, debounce_s: float, max_entries: int) -> bool:
    """Check if a key has been seen within debounce_s seconds.

    Returns True if the action should proceed (not debounced).
    Updates the cache timestamp and evicts the oldest entry if over max_entries.
    """
    now = time.monotonic()
    if key in cache and now - cache[key] < debounce_s:
        return False
    cache[key] = now
    if len(cache) > max_entries:
        oldest = min(cache, key=cache.get)
        del cache[oldest]
    return True


def _format_age(dt_str: str) -> str:
    """Format a datetime ISO string as a human-readable relative age.

    Returns e.g. "5m ago", "2h15m ago", or "" on failure.
    """
    try:
        dt = datetime.fromisoformat(dt_str)
        if dt.tzinfo is not None:
            dt = dt.replace(tzinfo=None)
        delta = datetime.now(timezone.utc).replace(tzinfo=None) - dt
        mins = int(delta.total_seconds() / 60)
        if mins < 60:
            return f"{mins}m ago"
        return f"{mins // 60}h{mins % 60}m ago"
    except Exception:
        return ""


def _append_output(existing: str, new_line: str) -> str:
    """Append a line to output with newline separator."""
    return (existing + "\n" + new_line) if existing else new_line


# ---------------------------------------------------------------------------
# Handler functions — replicate hook script logic using warm singletons
# ---------------------------------------------------------------------------


def handle_session_start(payload: dict) -> dict:
    """Welcome briefing + auto-consolidation check."""
    session_id = payload.get("session_id", "")
    project = payload.get("project", "")

    # Auto-consolidation check (max once per 7 days)
    try:
        if _should_run_periodic("last-consolidate", 7 * 86400):
            from omega.bridge import consolidate

            consolidate(prune_days=30, max_summaries=50)
            _update_marker("last-consolidate")
    except Exception as e:
        _log_hook_error("auto_consolidate", e)

    # Auto-compaction check (max once per 14 days)
    try:
        if _should_run_periodic("last-compact", 14 * 86400):
            from omega.bridge import compact

            compact(event_type="lesson_learned", similarity_threshold=0.60, min_cluster_size=3)
            _update_marker("last-compact")
    except Exception as e:
        _log_hook_error("auto_compact", e)

    # Auto-backup check (max once per 7 days)
    try:
        if _should_run_periodic("last-backup", 7 * 86400):
            backup_dir = Path.home() / ".omega" / "backups"
            backup_dir.mkdir(parents=True, exist_ok=True)
            dest = backup_dir / f"omega-{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.json"
            from omega.bridge import export_memories

            export_memories(filepath=str(dest))
            # Rotate: keep only last 4
            backups = sorted(backup_dir.glob("omega-*.json"), key=lambda p: p.name, reverse=True)
            for old in backups[4:]:
                old.unlink()
            _update_marker("last-backup")
    except Exception as e:
        _log_hook_error("auto_backup", e)

    # Auto-doctor check (max once per 7 days)
    doctor_summary = ""
    try:
        if _should_run_periodic("last-doctor", 7 * 86400):
            from omega.bridge import status as omega_status

            s = omega_status()
            issues = []
            if s.get("node_count", 0) == 0:
                issues.append("0 memories")
            if not s.get("vec_enabled"):
                issues.append("vec disabled")
            # FTS5 integrity check
            try:
                import sqlite3 as _sqlite3

                db_path = Path.home() / ".omega" / "omega.db"
                if db_path.exists():
                    _conn = _sqlite3.connect(str(db_path))
                    _conn.execute("INSERT INTO memories_fts(memories_fts) VALUES('integrity-check')")
                    _conn.close()
            except Exception:
                issues.append("FTS5 integrity issue")
            doctor_summary = f"doctor: {len(issues)} issue(s)" if issues else "doctor: healthy"
            _update_marker("last-doctor")
    except Exception as e:
        _log_hook_error("auto_doctor", e)

    # Auto-scan documents folder (max once per hour)
    doc_scan_summary = ""
    try:
        docs_dir = Path.home() / ".omega" / "documents"
        if _should_run_periodic("last-doc-scan", 3600) and docs_dir.exists() and any(docs_dir.iterdir()):
            from omega.knowledge.engine import scan_directory

            result = scan_directory()
            # Only show summary if something was ingested
            if "ingested" in result.lower() and "0 ingested" not in result:
                doc_scan_summary = result
            _update_marker("last-doc-scan")
    except Exception as e:
        _log_hook_error("auto_doc_scan", e)

    # Auto-pull from cloud (max once per day — pull is fast, just checks hashes)
    cloud_pull_summary = ""
    try:
        secrets_path = Path.home() / ".omega" / "secrets.json"
        if secrets_path.exists() and _should_run_periodic("last-cloud-pull", 86400):
            from omega.cloud.sync import get_sync

            result = get_sync().pull_all()
            mem_pulled = result.get("memories", {}).get("pulled", 0)
            doc_pulled = result.get("documents", {}).get("pulled", 0)
            total_pulled = mem_pulled + doc_pulled
            if total_pulled > 0:
                parts = []
                if mem_pulled:
                    parts.append(f"{mem_pulled} memories")
                if doc_pulled:
                    parts.append(f"{doc_pulled} documents")
                cloud_pull_summary = f"cloud: pulled {', '.join(parts)}"
            _update_marker("last-cloud-pull")
    except Exception as e:
        _log_hook_error("auto_cloud_pull", e)

    # Clean up stale surfacing counter files (both .surfaced and .surfaced.json)
    try:
        omega_dir = Path.home() / ".omega"
        cutoff = time.time() - 86400
        for pattern in ("session-*.surfaced", "session-*.surfaced.json"):
            for f in omega_dir.glob(pattern):
                if f.stat().st_mtime < cutoff:
                    f.unlink()
    except Exception:
        pass

    # Gather session context for briefing
    try:
        from omega.bridge import get_session_context

        ctx = get_session_context(project=project, exclude_session=session_id)
    except Exception as e:
        _log_hook_error("session_start", e)
        return {"output": f"OMEGA welcome failed: {e}", "error": str(e)}

    memory_count = ctx.get("memory_count", 0)
    health_status = ctx.get("health_status", "ok")
    last_capture = ctx.get("last_capture_ago", "unknown")
    context_items = ctx.get("context_items", [])

    # Detect project name and git branch/status
    project_name = Path(project).name if project else "unknown"
    git_branch = _get_current_branch(project or ".") or "unknown"
    git_status_str = "unknown"
    try:
        status_result = subprocess.run(
            ["git", "status", "--porcelain"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project or ".",
        )
        if status_result.returncode == 0:
            changed = len([l for l in status_result.stdout.strip().split("\n") if l.strip()])
            git_status_str = "Clean" if changed == 0 else f"{changed} unstaged changes"
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        pass

    # --- Section 1: Header (always) ---
    first_word = "Welcome back!" if memory_count > 0 else "Welcome!"
    lines = [
        f"## {first_word} OMEGA ready — {memory_count} memories | Project: {project_name} | Branch: {git_branch} | {git_status_str}"
    ]

    # First-time user "Aha" moment — guided welcome for new users
    if memory_count == 0:
        lines.append("")
        lines.append("OMEGA captures decisions, lessons, and errors automatically as you work.")
        lines.append("Next session, it surfaces relevant context when you edit the same files.")
        lines.append("")
        lines.append("**Quick start:**")
        lines.append('- Say "remember that we always use TypeScript strict mode" to store a preference')
        lines.append("- Make a decision and OMEGA captures it automatically")
        lines.append("- Encounter an error, and OMEGA stores the pattern for future recall")
        lines.append("")
        lines.append("After this session ends, you'll see exactly what was captured.")
    elif memory_count <= 10:
        lines.append(f"  OMEGA has {memory_count} memories from your first sessions. These will surface when you edit related files.")
        try:
            from omega.bridge import type_stats as _ts_first

            first_stats = _ts_first()
            stat_parts = []
            for k, v in sorted(first_stats.items(), key=lambda x: x[1], reverse=True):
                if v > 0 and k != "session_summary":
                    stat_parts.append(f"{v} {k.replace('_', ' ')}")
            if stat_parts:
                lines.append(f"  Captured so far: {', '.join(stat_parts[:4])}")
        except Exception:
            pass

    # --- Section 2: Health line (always) ---
    health_line = f"**Health:** {health_status} | **Last capture:** {last_capture}"
    try:
        from omega.router.engine import OmegaRouter as _WelcomeRouter

        _wr = _WelcomeRouter()
        _priority = getattr(_wr, "priority_mode", "balanced")
        health_line += f" | **Router:** {_priority}"
    except Exception:
        pass
    lines.append(health_line)

    # Graph health ratio
    _graph_ratio = 0.0
    try:
        from omega.bridge import _get_store as _gs_graph

        _graph_store = _gs_graph()
        _node_count = _graph_store.count()
        _edge_count = _graph_store.edge_count()
        if _node_count > 0:
            _graph_ratio = _edge_count / _node_count
            _graph_label = "rich" if _graph_ratio >= 1.5 else ("good" if _graph_ratio >= 0.5 else "sparse")
            lines.append(f"**Graph:** {_graph_label} ({_edge_count:,} edges, {_graph_ratio:.1f}x)")
    except Exception:
        pass

    # Profile summary (field counts across categories)
    try:
        from omega.profile.engine import get_profile_engine

        _pe = get_profile_engine()
        with _pe._lock:
            _profile_rows = _pe._conn.execute(
                "SELECT COUNT(*) as cnt FROM secure_profile"
            ).fetchone()
        _profile_count = _profile_rows["cnt"] if _profile_rows else 0
        if _profile_count > 0:
            lines.append(f"**Profile:** {_profile_count} encrypted field(s) stored")
    except Exception:
        pass

    # Proactive maintenance suggestion when graph is sparse
    if _graph_ratio > 0 and _graph_ratio < 0.5:
        lines.append("[MAINTENANCE] Graph connectivity is sparse — consider running omega_compact to consolidate related memories")

    # --- Section 2a: Alerts for degraded subsystems ---
    # Embedding model warning → [!] alert
    try:
        from omega.graphs import get_active_backend

        if get_active_backend() is None:
            lines.append("[!] Embedding model unavailable — semantic search degraded (hash fallback)")
    except Exception:
        pass

    # Router degradation → [!] alert (only when providers degraded)
    try:
        from omega.router.engine import OmegaRouter

        router = OmegaRouter()
        provider_status = router.get_provider_status()
        available = sum(1 for s in provider_status.values() if s == "available")
        total = len(provider_status)
        if 0 < available < total:
            lines.append(f"[!] Router: {available}/{total} providers active — some routing degraded")
        elif available == 0 and total > 0:
            lines.append("[!] Router: 0 providers active — routing unavailable")
    except ImportError:
        pass  # Router is optional
    except Exception as e:
        _log_hook_error("router_status_welcome", e)

    # Doctor issues → [!] alert
    if doctor_summary and "issue" in doctor_summary:
        lines.append(f"[!] {doctor_summary}")

    # Document scan results (only if new files were ingested)
    if doc_scan_summary:
        lines.append(f"[DOCS] {doc_scan_summary}")

    # Cloud pull results (only if new data was pulled)
    if cloud_pull_summary:
        lines.append(f"[CLOUD] {cloud_pull_summary}")

    # --- Section 3: [REMINDER] due reminders with smart enrichment ---
    try:
        from omega.bridge import get_due_reminders

        due_reminders = get_due_reminders(mark_fired=True)
        if due_reminders:
            lines.append("")
            for r in due_reminders[:5]:  # Cap at 5
                overdue_label = " [OVERDUE]" if r.get("is_overdue") else ""
                lines.append(f"[REMINDER]{overdue_label} {r['text']}")
                if r.get("context"):
                    lines.append(f"  Context: {r['context'][:120]}")

                # Smart enrichment: find related memories for this reminder
                try:
                    from omega.bridge import query_structured as _qs_enrich

                    reminder_text = r.get("text", "")
                    if len(reminder_text) > 10:
                        related = _qs_enrich(
                            query_text=reminder_text[:200],
                            limit=2,
                            event_type="decision",
                        )
                        related_lessons = _qs_enrich(
                            query_text=reminder_text[:200],
                            limit=1,
                            event_type="lesson_learned",
                        )
                        enrichments = []
                        for m in (related or []) + (related_lessons or []):
                            if m.get("relevance", 0) >= 0.30:
                                preview = m.get("content", "")[:80].replace("\n", " ").strip()
                                etype = m.get("event_type", "")
                                if preview:
                                    enrichments.append(f"{etype}: {preview}")
                        if enrichments:
                            lines.append(f"  Related: {enrichments[0]}")
                except Exception:
                    pass

                lines.append(f"  ID: {r['id'][:12]} — dismiss with omega_remind_dismiss")
    except Exception as e:
        _log_hook_error("reminder_check", e)

    # --- Section 5: [CONTEXT] (if items exist) ---
    if context_items:
        lines.append("")
        lines.append("[CONTEXT]")
        for item in context_items:
            lines.append(f"  {item['tag']}: {item['text']}")

    # --- Section 7: [ACTION] maintenance nudges ---
    try:
        from omega.bridge import type_stats as _ts

        stats = _ts()
        lesson_count = stats.get("lesson_learned", 0)
        if lesson_count >= 40:
            lines.append(
                f"\n[ACTION] {lesson_count} lessons with potential duplicates — run omega_compact if asked about maintenance"
            )
    except Exception:
        pass

    # --- Section 7a: Expanded welcome nudges ---
    nudges: list[str] = []

    # Nudge: overdue backup
    try:
        backup_marker = Path.home() / ".omega" / "last-backup"
        if backup_marker.exists():
            last_ts = backup_marker.read_text().strip()
            last = datetime.fromisoformat(last_ts)
            if last.tzinfo is None:
                last = last.replace(tzinfo=timezone.utc)
            age_days = (datetime.now(timezone.utc) - last).days
            if age_days >= 14:
                nudges.append(f"Last backup: {age_days}d ago — consider running omega_backup")
        elif memory_count > 50:
            nudges.append("No backup found — consider running omega_backup")
    except Exception:
        pass

    # Nudge: due/overdue reminders count
    try:
        from omega.bridge import list_reminders as _lr

        pending = _lr(status="pending")
        due_count = sum(1 for r in pending if r.get("is_due"))
        upcoming_today = 0
        for r in pending:
            if not r.get("is_due"):
                try:
                    remind_at = datetime.fromisoformat(r["remind_at"])
                    if remind_at.tzinfo is None:
                        remind_at = remind_at.replace(tzinfo=timezone.utc)
                    if (remind_at - datetime.now(timezone.utc)).total_seconds() < 86400:
                        upcoming_today += 1
                except Exception:
                    pass
        if due_count > 0:
            nudges.append(f"{due_count} reminder{'s' if due_count != 1 else ''} due now")
        elif upcoming_today > 0:
            nudges.append(f"{upcoming_today} reminder{'s' if upcoming_today != 1 else ''} due today")
    except Exception:
        pass

    # Nudge: recurring error patterns (3+ of same type this month)
    try:
        from omega.bridge import _get_store as _gs

        _store = _gs()
        month_cutoff = (datetime.now(timezone.utc) - __import__("datetime").timedelta(days=30)).isoformat()
        error_rows = _store._conn.execute(
            "SELECT content FROM memories "
            "WHERE event_type = 'error_pattern' AND created_at >= ? "
            "ORDER BY created_at DESC LIMIT 50",
            (month_cutoff,),
        ).fetchall()
        if len(error_rows) >= 3:
            # Group by normalized prefix (first 80 chars, lowered, whitespace-collapsed)
            buckets: dict[str, int] = {}
            for (content,) in error_rows:
                key = re.sub(r"\s+", " ", content[:80].lower()).strip()
                buckets[key] = buckets.get(key, 0) + 1
            top_bucket = max(buckets.items(), key=lambda x: x[1]) if buckets else None
            if top_bucket and top_bucket[1] >= 3:
                nudges.append(
                    f"Pattern: same error {top_bucket[1]}x this month — {top_bucket[0][:60]}"
                )
    except Exception:
        pass

    # Nudge: time-of-day project awareness
    try:
        from omega.bridge import _get_store as _gs_tod

        _store_tod = _gs_tod()
        # Get sessions for the last 14 days, group by hour-of-day + project
        tod_cutoff = (datetime.now(timezone.utc) - __import__("datetime").timedelta(days=14)).isoformat()
        tod_rows = _store_tod._conn.execute(
            "SELECT created_at, metadata FROM memories "
            "WHERE event_type = 'session_summary' AND created_at >= ? "
            "ORDER BY created_at DESC LIMIT 50",
            (tod_cutoff,),
        ).fetchall()
        if len(tod_rows) >= 5:
            current_hour = datetime.now().hour
            # Determine time-of-day bucket
            if 5 <= current_hour < 12:
                tod_label = "morning"
            elif 12 <= current_hour < 17:
                tod_label = "afternoon"
            elif 17 <= current_hour < 22:
                tod_label = "evening"
            else:
                tod_label = "night"

            # Find which projects are most common at this time of day
            project_counts: dict[str, int] = {}
            for (created_at_str, meta_json) in tod_rows:
                try:
                    ca = datetime.fromisoformat(created_at_str.replace("Z", "+00:00"))
                    local_hour = ca.astimezone().hour
                    same_bucket = False
                    if tod_label == "morning" and 5 <= local_hour < 12:
                        same_bucket = True
                    elif tod_label == "afternoon" and 12 <= local_hour < 17:
                        same_bucket = True
                    elif tod_label == "evening" and 17 <= local_hour < 22:
                        same_bucket = True
                    elif tod_label == "night" and (local_hour >= 22 or local_hour < 5):
                        same_bucket = True
                    if same_bucket:
                        meta = json.loads(meta_json) if isinstance(meta_json, str) else (meta_json or {})
                        proj = meta.get("project", "")
                        if proj:
                            proj_name = os.path.basename(proj)
                            project_counts[proj_name] = project_counts.get(proj_name, 0) + 1
                except Exception:
                    continue

            if project_counts:
                top_proj = max(project_counts.items(), key=lambda x: x[1])
                if top_proj[1] >= 3 and project_name != top_proj[0]:
                    nudges.append(f"You typically work on {top_proj[0]} {tod_label}s")
    except Exception:
        pass

    if nudges:
        lines.append("")
        for nudge in nudges[:3]:  # Cap at 3 nudges
            lines.append(f"[NUDGE] {nudge}")

    # --- Section 7b: Auto-surfaced weekly digest (max once per 7 days, 20+ memories) ---
    if memory_count >= 20 and _should_run_periodic("last-digest", 7 * 86400):
        try:
            from omega.bridge import get_weekly_digest

            digest = get_weekly_digest(days=7)
            period_new = digest.get("period_new", 0)
            session_count = digest.get("session_count", 0)
            total = digest.get("total_memories", 0)
            growth_pct = digest.get("growth_pct", 0)
            type_breakdown = digest.get("type_breakdown", {})

            if period_new > 0:
                lines.append("")
                lines.append(f"[WEEKLY] This week: {period_new} memories across {session_count} sessions")
                if type_breakdown:
                    bd_parts = [f"{v} {k.replace('_', ' ')}" for k, v in sorted(type_breakdown.items(), key=lambda x: x[1], reverse=True)[:3]]
                    lines.append(f"  Breakdown: {', '.join(bd_parts)}")
                sign = "+" if growth_pct >= 0 else ""
                lines.append(f"  Trend: {sign}{growth_pct:.0f}% vs prior week | {total} total memories")

                _update_marker("last-digest")
        except Exception as e:
            _log_hook_error("weekly_digest_surface", e)

    # --- Section 8: Footer (maintenance + doctor ok + cloud) ---
    footer_parts = []
    # Maintenance status from markers
    for label, marker_name, cadence in [
        ("backup", "last-backup", 7),
        ("doctor", "last-doctor", 7),
    ]:
        try:
            marker = Path.home() / ".omega" / marker_name
            if marker.exists():
                footer_parts.append(f"{label} ok")
        except Exception:
            pass
    # Cloud sync status
    try:
        secrets_path = Path.home() / ".omega" / "secrets.json"
        if secrets_path.exists():
            pull_marker = Path.home() / ".omega" / "last-cloud-pull"
            if pull_marker.exists():
                last_ts = pull_marker.read_text().strip()
                last = datetime.fromisoformat(last_ts)
                if last.tzinfo is None:
                    last = last.replace(tzinfo=timezone.utc)
                age_days = (datetime.now(timezone.utc) - last).days
                if age_days < 1:
                    footer_parts.append("cloud ok")
                else:
                    footer_parts.append(f"cloud pull: {age_days}d ago")
            else:
                footer_parts.append("cloud pull: never")
    except Exception:
        pass
    if footer_parts:
        lines.append(f"\nMaintenance: {', '.join(footer_parts)}")

    return {"output": "\n".join(lines), "error": None}


def _auto_feedback_on_surfaced(session_id: str):
    """Auto-record feedback for memories surfaced multiple times (likely relevant)."""
    if not session_id:
        return
    json_path = Path.home() / ".omega" / f"session-{session_id}.surfaced.json"
    if not json_path.exists():
        return
    try:
        data = json.loads(json_path.read_text())
        # Count how many times each memory was surfaced across different files.
        # Only record "helpful" for memories surfaced 2+ times (re-surfaced
        # across edits suggests genuine relevance, avoids inflating scores).
        id_counts: dict[str, int] = {}
        for ids in data.values():
            for mid in ids:
                id_counts[mid] = id_counts.get(mid, 0) + 1

        relevant_ids = [mid for mid, count in id_counts.items() if count >= 2]

        # Fallback: if no 2+ surfacings but multiple unique memories were surfaced
        # once during active editing (multiple files edited), give lightweight feedback
        if not relevant_ids and len(data) >= 2:
            once_ids = [mid for mid, count in id_counts.items() if count == 1]
            relevant_ids = once_ids[:5]
            feedback_reason = "Auto: surfaced during active multi-file editing"
        else:
            feedback_reason = "Auto: surfaced across multiple edits"

        if not relevant_ids:
            return

        from omega.bridge import record_feedback

        for mid in relevant_ids[:10]:
            try:
                record_feedback(mid, "helpful", feedback_reason)
            except Exception:
                pass

        json_path.unlink(missing_ok=True)
    except Exception as e:
        _log_hook_error("auto_feedback_surfaced", e)
    finally:
        try:
            if json_path.exists():
                json_path.unlink()
        except Exception:
            pass


def handle_session_stop(payload: dict) -> dict:
    """Generate and store session summary + activity report."""
    session_id = payload.get("session_id", "")
    project = payload.get("project", "")
    entity_id = _resolve_entity(project) if project else None
    lines = []

    # Read surfaced data before auto-feedback cleanup deletes the file
    surfaced_count = 0
    surfaced_unique_ids = 0
    surfaced_unique_files = 0
    try:
        surfaced_json = Path.home() / ".omega" / f"session-{session_id}.surfaced.json"
        if surfaced_json.exists():
            data = json.loads(surfaced_json.read_text())
            surfaced_count = sum(len(ids) for ids in data.values())
            all_ids = set()
            for ids in data.values():
                all_ids.update(ids)
            surfaced_unique_ids = len(all_ids)
            surfaced_unique_files = len(data)
    except Exception:
        pass

    # Auto-feedback for surfaced memories before building summary
    _auto_feedback_on_surfaced(session_id)

    # --- Gather session event counts ---
    counts = {}
    captured = 0
    try:
        from omega.bridge import _get_store

        store = _get_store()
        counts = store.get_session_event_counts(session_id) if session_id else {}
        captured = sum(counts.values()) if counts else 0

        # Clean up surfaced marker
        try:
            marker = Path.home() / ".omega" / f"session-{session_id}.surfaced"
            if marker.exists():
                marker.unlink()
        except Exception:
            pass
    except Exception as e:
        _log_hook_error("session_stop_activity", e)

    # --- Build summary from per-type targeted queries (stored silently) ---
    summary = "Session ended"
    top_decisions: list[str] = []
    try:
        from omega.bridge import query_structured

        decisions = query_structured(
            query_text="decisions made",
            limit=5,
            session_id=session_id,
            project=project,
            event_type="decision",
            entity_id=entity_id,
        )
        errors = query_structured(
            query_text="errors encountered",
            limit=3,
            session_id=session_id,
            project=project,
            event_type="error_pattern",
            entity_id=entity_id,
        )
        tasks = query_structured(
            query_text="completed tasks",
            limit=3,
            session_id=session_id,
            project=project,
            event_type="task_completion",
            entity_id=entity_id,
        )

        parts = []
        if decisions:
            items = [m.get("content", "")[:120] for m in decisions[:3]]
            parts.append(f"Decisions ({len(decisions)}): " + "; ".join(items))
            top_decisions = [m.get("content", "")[:80].replace("\n", " ").strip() for m in decisions[:2]]
        if errors:
            items = [m.get("content", "")[:120] for m in errors[:3]]
            parts.append(f"Errors ({len(errors)}): " + "; ".join(items))
        if tasks:
            items = [m.get("content", "")[:120] for m in tasks[:3]]
            parts.append(f"Tasks ({len(tasks)}): " + "; ".join(items))

        if parts:
            summary = " | ".join(parts)[:600]
        elif decisions or errors or tasks:
            total = len(decisions or []) + len(errors or []) + len(tasks or [])
            summary = f"Session ended with {total} captured memories"
    except Exception as e:
        _log_hook_error("session_stop_summary", e)

    # Store the summary (silent)
    try:
        from omega.bridge import auto_capture

        auto_capture(
            content=f"Session summary: {summary}",
            event_type="session_summary",
            metadata={"source": "session_stop_hook", "project": project},
            session_id=session_id,
            project=project,
            entity_id=entity_id,
        )
    except Exception as e:
        _log_hook_error("session_stop", e)
        return {"output": "\n".join(lines), "error": str(e)}

    # --- Format output: header + details + footer ---
    if captured > 0:
        lines.append(f"## Session complete — {captured} captured, {surfaced_count} surfaced")
        # Type breakdown
        _LABELS = {
            "decision": ("decision", "decisions"),
            "lesson_learned": ("lesson", "lessons"),
            "error_pattern": ("error", "errors"),
        }
        type_parts = []
        other_count = 0
        for key, (singular, plural) in _LABELS.items():
            n = counts.get(key, 0)
            if n:
                type_parts.append(f"{n} {plural if n > 1 else singular}")
        for key, n in counts.items():
            if key not in _LABELS and n > 0:
                other_count += n
        if other_count:
            type_parts.append(f"{other_count} other")
        if type_parts:
            lines.append(f"  Stored: {', '.join(type_parts)}")
        if top_decisions:
            lines.append(f"  Key: {'; '.join(top_decisions)}")
    else:
        lines.append(f"## Session complete — {surfaced_count} memories surfaced")

    # Unique recall stats
    if surfaced_unique_ids > 0:
        lines.append(f"  Recalled: {surfaced_unique_ids} unique memories across {surfaced_unique_files} file{'s' if surfaced_unique_files != 1 else ''}")

    # --- Productivity recap: weekly stats ---
    try:
        from omega.bridge import _get_store as _gs_recap, session_stats as _ss_recap

        store = _gs_recap()
        total_memories = store.node_count()

        # Weekly session count
        all_sessions = _ss_recap()
        weekly_sessions = 0
        try:
            week_cutoff = (datetime.now(timezone.utc) - __import__("datetime").timedelta(days=7)).isoformat()
            weekly_rows = store._conn.execute(
                "SELECT COUNT(DISTINCT session_id) FROM memories "
                "WHERE created_at >= ? AND session_id IS NOT NULL",
                (week_cutoff,),
            ).fetchone()
            weekly_sessions = weekly_rows[0] if weekly_rows else 0
        except Exception:
            pass

        # Weekly memory count
        weekly_memories = 0
        try:
            weekly_mem_row = store._conn.execute(
                "SELECT COUNT(*) FROM memories WHERE created_at >= ?",
                (week_cutoff,),
            ).fetchone()
            weekly_memories = weekly_mem_row[0] if weekly_mem_row else 0
        except Exception:
            pass

        # Prior week memory count (for growth comparison)
        prev_week_memories = 0
        try:
            prev_cutoff = (datetime.now(timezone.utc) - __import__("datetime").timedelta(days=14)).isoformat()
            prev_row = store._conn.execute(
                "SELECT COUNT(*) FROM memories WHERE created_at >= ? AND created_at < ?",
                (prev_cutoff, week_cutoff),
            ).fetchone()
            prev_week_memories = prev_row[0] if prev_row else 0
        except Exception:
            pass

        recap_parts = []
        if weekly_sessions > 1:
            recap_parts.append(f"{weekly_sessions} sessions this week")
        if weekly_memories > 0:
            recap_parts.append(f"{weekly_memories} memories this week")
        recap_parts.append(f"{total_memories} total")
        lines.append(f"  Recap: {', '.join(recap_parts)}")

        # Week-over-week growth
        if prev_week_memories > 0 and weekly_memories > 0:
            growth_pct = ((weekly_memories - prev_week_memories) / prev_week_memories) * 100
            sign = "+" if growth_pct >= 0 else ""
            lines.append(f"  Growth: {sign}{growth_pct:.0f}% vs last week")
    except Exception:
        pass

    # --- Files touched in this session ---
    try:
        from omega.coordination import get_manager as _gm_recap

        mgr = _gm_recap()
        claims = mgr.get_session_claims(session_id)
        file_claims = claims.get("file_claims", [])
        if file_claims:
            fnames = [os.path.basename(f) for f in file_claims[:5]]
            if len(file_claims) > 5:
                fnames.append(f"+{len(file_claims) - 5}")
            lines.append(f"  Files: {', '.join(fnames)}")
    except Exception:
        pass

    # Prune debounce dicts for this session to prevent unbounded growth
    _last_heartbeat.pop(session_id, None)
    _heartbeat_count.pop(session_id, None)
    stale_claims = [k for k in _last_claim if k[0] == session_id]
    for k in stale_claims:
        del _last_claim[k]
    stale_overlaps = [k for k in _last_overlap_notify if k[0] == session_id]
    for k in stale_overlaps:
        del _last_overlap_notify[k]
    _last_coord_query.pop(session_id, None)
    _pending_urgent.pop(session_id, None)
    _session_intent.pop(session_id, None)
    _last_surface.clear()
    _error_hashes.clear()
    _error_counts.pop(session_id, None)

    # Auto-sync to cloud (fire-and-forget daemon thread)
    _auto_cloud_sync(session_id)

    return {"output": "\n".join(lines), "error": None}


def handle_coord_session_start(payload: dict) -> dict:
    """Register agent session + git sync check + session resume."""
    session_id = payload.get("session_id", "")
    project = payload.get("project", "")
    if not session_id:
        return {"output": "", "error": None}

    lines = []
    peer_count = 0
    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        mgr.list_sessions()  # Force-clean stale sessions (bypasses rate limit)
        result = mgr.register_session(
            session_id=session_id,
            pid=os.getpid(),
            project=project,
        )
        peer_count = result.get("peers_on_project", 0)

        # --- [!] Alerts: file/branch conflicts from peers ---
        if peer_count > 0:
            try:
                status = mgr.get_status()
                conflicts = status.get("conflicts", [])
                for conflict in conflicts[:3]:
                    file_path = conflict.get("file", conflict.get("file_path", "unknown"))
                    owner_sid = conflict.get("owner", conflict.get("session_id", "another agent"))
                    owner_name = _agent_nickname(owner_sid) if owner_sid != "another agent" else owner_sid
                    lines.append(f"[!] File conflict: {file_path} claimed by {owner_name}")
            except Exception:
                pass  # Conflict check is best-effort

            # --- [!] Alerts: uncommitted git files that peers own ---
            try:
                if project:
                    git_result = subprocess.run(
                        ["git", "status", "--porcelain"],
                        capture_output=True,
                        text=True,
                        timeout=5,
                        cwd=project,
                    )
                    if git_result.returncode == 0 and git_result.stdout.strip():
                        uncommitted = set()
                        for line in git_result.stdout.strip().split("\n"):
                            if line.strip():
                                parts = line[3:].split(" -> ")
                                rel_path = parts[-1].strip()
                                uncommitted.add(os.path.abspath(os.path.join(project, rel_path)))

                        peer_owned: dict[str, list[str]] = {}
                        for f in status.get("files", []):
                            f_path = f.get("path", "")
                            f_sid = f.get("session_id", "")
                            if f_sid != session_id and f_path in uncommitted:
                                name = _agent_nickname(f_sid)
                                peer_owned.setdefault(name, []).append(os.path.basename(f_path))

                        for name, fnames in peer_owned.items():
                            flist = ", ".join(fnames[:4])
                            if len(fnames) > 4:
                                flist += f" +{len(fnames) - 4}"
                            lines.append(
                                f"[!] {len(fnames)} uncommitted file{'s' if len(fnames) != 1 else ''}"
                                f" owned by {name}: {flist} — do NOT commit these"
                            )
            except Exception:
                pass  # Uncommitted file check is best-effort

            # Check for unread messages → feed into [TODO] section
            try:
                unread = mgr.get_unread_count(session_id)
            except Exception:
                unread = 0
        else:
            unread = 0

        # --- Git sync check → [!] alerts ---
        git_lines = _check_git_sync(session_id, project, mgr)
        lines.extend(git_lines)

        # Session intent announce + branch claim on start (silent)
        try:
            branch = _get_current_branch(project) if project else None
            if branch and branch not in ("main", "master", "develop", "release"):
                mgr.announce_intent(
                    session_id=session_id,
                    description=f"Working on branch {branch}",
                    intent_type="session_start",
                    target_files=[],
                    target_branch=branch,
                    ttl_minutes=120,
                )
                mgr.claim_branch(session_id, project, branch, task="session start")
        except Exception:
            pass  # Intent/branch claim is best-effort

        # --- [RESUME] from predecessor session ---
        resume_lines = _session_resume(session_id, project, mgr)
        lines.extend(resume_lines)

        # --- [HANDOFF] from predecessor's complete message ---
        try:
            handoff_msgs = mgr.check_inbox(session_id, unread_only=True, msg_type="complete", limit=1)
            if handoff_msgs:
                msg = handoff_msgs[0]
                from_sid = msg.get("from_session") or "unknown"
                from_name = _agent_nickname(from_sid)
                subj = (msg.get("subject") or "")[:80]
                body = (msg.get("body") or "")[:500]
                handoff_line = f"\n[HANDOFF] From {from_name}: {subj}"
                if body:
                    for bl in body.strip().split("\n")[:10]:
                        handoff_line += f"\n  {bl.strip()}"
                lines.append(handoff_line)
        except Exception:
            pass  # Handoff surfacing is best-effort

        # --- [CONTINUE] recently reassigned tasks (have progress > 0) ---
        try:
            all_pending = mgr.list_tasks(project=project, status="pending")
            continued = [t for t in all_pending if t.get("progress", 0) > 0]
            if continued:
                t = continued[0]  # highest priority (list_tasks already sorted)
                pct = f" [{t['progress']}%]" if t.get("progress") else ""
                lines.append(
                    f"\n[CONTINUE] Task #{t['id']} \"{t['title']}\"{pct} was in progress "
                    f"when last session ended — claim with omega_task_next"
                )
        except Exception:
            all_pending = None

        # --- [TODO] pending tasks + unread messages ---
        try:
            tasks = all_pending if all_pending is not None else mgr.list_tasks(project=project, status="pending")
        except Exception:
            tasks = []

        todo_parts = []
        if tasks:
            todo_parts.append(f"{len(tasks)} pending tasks")
        if unread > 0:
            todo_parts.append(f"{unread} unread msg")
        if todo_parts:
            lines.append(f"\n[TODO] {' | '.join(todo_parts)} — query omega_tasks_list / omega_inbox for full list")
            # Show the highest-priority next task with auto-claim hint
            if tasks:
                unblocked = [t for t in tasks if not t.get("blocked")]
                show_task = unblocked[0] if unblocked else max(tasks, key=lambda t: t.get("priority", 0))
                prio = f" P{show_task['priority']}" if show_task.get("priority") else ""
                hint = " — use omega_task_next to auto-claim" if unblocked else ""
                lines.append(f"  NEXT: #{show_task['id']}{prio} {show_task['title']}{hint}")

        # Router warmup (silent — only errors become [!] alerts)
        try:
            from omega.router.classifier import warm_up

            warm_up()
        except ImportError:
            pass  # Router is optional
        except Exception as e_r:
            _log_hook_error("router_warmup", e_r)

        # --- Coordination celebration (multi-agent success) ---
        if peer_count > 0:
            try:
                status = mgr.get_status()
                n_conflicts = len(status.get("conflicts", []))
                n_deadlocks = len(status.get("deadlocks", []))
                if n_conflicts == 0 and n_deadlocks == 0:
                    completed_tasks = mgr.list_tasks(status="completed")
                    today = datetime.now(timezone.utc).date().isoformat()
                    today_done = [t for t in completed_tasks if (t.get("updated_at") or "")[:10] == today]
                    if today_done:
                        lines.append(
                            f"\n[COORD] {peer_count + 1} agents, "
                            f"{len(today_done)} task{'s' if len(today_done) != 1 else ''} "
                            f"completed today with zero conflicts"
                        )
            except Exception:
                pass

        # --- Footer: rich [COORD] team roster (cross-project) ---
        try:
            sessions = mgr.list_sessions(auto_clean=False)
            peers = [s for s in sessions if s.get("session_id") != session_id][:6]
            if peers:
                # Fetch in-progress coord tasks to prefer over session.task
                _roster_tasks: dict[str, dict] = {}
                try:
                    for _rt in mgr.list_tasks(project=project, status="in_progress"):
                        _rt_sid = _rt.get("session_id")
                        if _rt_sid and _rt_sid not in _roster_tasks:
                            _roster_tasks[_rt_sid] = _rt
                except Exception:
                    pass
                lines.append(f"\n[COORD] {len(peers)} peer{'s' if len(peers) != 1 else ''} active:")
                for p in peers:
                    p_name = _agent_nickname(p["session_id"])
                    # Prefer coord_task over session.task
                    _ct = _roster_tasks.get(p["session_id"])
                    if _ct:
                        _pct = f" [{_ct['progress']}%]" if _ct.get("progress") else ""
                        p_task = f"#{_ct['id']} {_ct['title'][:40]}{_pct}"
                    else:
                        p_task = (p.get("task") or "idle")[:50]
                    # Get claimed files and branch
                    p_files = ""
                    p_branch = ""
                    try:
                        claims = mgr.get_session_claims(p["session_id"])
                        file_claims = claims.get("file_claims", [])
                        if file_claims:
                            fnames = [os.path.basename(f) for f in file_claims[:2]]
                            if len(file_claims) > 2:
                                fnames.append(f"+{len(file_claims) - 2}")
                            p_files = f" [{', '.join(fnames)}]"
                        branch_claims = claims.get("branch_claims", [])
                        if branch_claims:
                            p_branch = f" ({branch_claims[0]})"
                    except Exception:
                        pass
                    # Project label when different from current session
                    p_proj_label = ""
                    p_project = p.get("project") or ""
                    if p_project and p_project != project:
                        p_proj_label = f" [{os.path.basename(p_project)}]"
                    # Heartbeat age
                    age = _format_age(p.get("last_heartbeat", ""))
                    age_str = f" — {age}" if age else ""
                    lines.append(f"  {p_name}: {p_task}{p_proj_label}{p_files}{p_branch}{age_str}")
        except Exception:
            if peer_count > 0:
                lines.append(f"\n{peer_count} peer(s) active — query omega_sessions_list for details")

    except Exception as e:
        _log_hook_error("coord_session_start", e)
        return {"output": "", "error": str(e)}

    return {"output": "\n".join(lines), "error": None}


def _check_git_sync(session_id: str, project: str, mgr) -> list[str]:
    """Detect upstream commits from uncoordinated agents."""
    lines = []
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--is-inside-work-tree"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        if result.returncode != 0:
            return lines

        subprocess.run(
            ["git", "fetch", "origin", "--quiet"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )

        branch = _get_current_branch(project) or "main"

        upstream_result = subprocess.run(
            ["git", "log", f"HEAD..origin/{branch}", "--oneline", "--no-decorate", "-20"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        if upstream_result.returncode != 0 or not upstream_result.stdout.strip():
            return lines

        upstream_lines = upstream_result.stdout.strip().split("\n")
        upstream_hashes = [line.split()[0] for line in upstream_lines if line.strip()]
        if not upstream_hashes:
            return lines

        untracked = mgr.detect_untracked_commits(project, upstream_hashes)

        for line in upstream_lines:
            parts = line.split(None, 1)
            if parts:
                mgr.log_git_event(
                    project=project,
                    event_type="upstream_detected",
                    commit_hash=parts[0],
                    branch=branch,
                    message=parts[1] if len(parts) > 1 else "",
                )

        lines.append(
            f"[!] {len(upstream_hashes)} upstream commit(s) on origin/{branch} — run `git pull` before editing"
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    except Exception as e:
        _log_hook_error("git_sync_check", e)
    return lines


def _session_resume(session_id: str, project: str, mgr) -> list[str]:
    """Check for predecessor session snapshots with enriched context."""
    lines = []
    try:
        if not project:
            return lines
        snapshots = mgr.recover_session(project)
        if not snapshots:
            snapshots = []  # Fall through to checkpoint surfacing

        if snapshots:
            snap = snapshots[0]
            meta = snap.get("metadata") or {}

            # Compute how long ago the predecessor ended
            age = _format_age(snap.get("created_at", ""))
            age_str = f" ended {age}" if age else ""

            # Build compact single-line [RESUME]
            task_desc = (snap.get("task") or "")[:60]
            task_part = f'"{task_desc}"' if task_desc else "previous session"

            # Get top decisions as Key: summary
            key_parts = []
            try:
                from omega.bridge import query_structured

                decisions = query_structured(
                    query_text="decisions made",
                    limit=5,
                    session_id=snap["session_id"],
                    project=project,
                    event_type="decision",
                )
                for d in decisions or []:
                    content = d.get("content", "")
                    # Strip auto-capture prefixes
                    for prefix in ("Plan/decision captured: ", "Decision: "):
                        if content.startswith(prefix):
                            content = content[len(prefix) :]
                    # Skip JSON blobs
                    stripped = content.lstrip()
                    if stripped.startswith(("{", "[", '"filePath')):
                        continue
                    # Take first meaningful line only
                    first_line = content.split("\n")[0].strip()
                    if first_line and len(first_line) > 10:
                        key_parts.append(first_line[:80])
                    if len(key_parts) >= 2:
                        break
            except Exception:
                pass  # Decision surfacing is best-effort

            key_str = f" | Key: {'; '.join(key_parts)}" if key_parts else ""
            lines.append(f"[RESUME] {task_part}{age_str}{key_str}")
    except Exception as e:
        _log_hook_error("session_resume", e)

    # Surface recent checkpoints for this project
    try:
        if project:
            from omega.bridge import query_structured

            checkpoints = query_structured(
                query_text="checkpoint",
                limit=3,
                event_type="checkpoint",
                project=project,
            )
            if checkpoints:
                checkpoints.sort(key=lambda c: c.get("created_at", ""), reverse=True)
                latest = checkpoints[0]
                meta = latest.get("metadata", {})
                cp_data = meta.get("checkpoint_data", {})
                task_title = cp_data.get("task_title", "Unknown task")
                next_steps = cp_data.get("next_steps", "")

                # Compute age
                age = _format_age(latest.get("created_at", ""))
                cp_age = f" saved {age}" if age else ""

                cp_line = f'[CHECKPOINT] "{task_title}"{cp_age}'
                if next_steps:
                    cp_line += f" | Next: {next_steps[:80]}"
                cp_line += " — call `omega_resume_task` for full context"
                lines.append(cp_line)
    except Exception:
        pass  # Fail-open

    return lines


def handle_coord_session_stop(payload: dict) -> dict:
    """Deregister session and release all claims."""
    session_id = payload.get("session_id", "")
    if not session_id:
        return {"output": "", "error": None}

    lines = []
    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        project = payload.get("project", "")
        if project:
            mgr.enrich_session_metadata(session_id, project)

        # Count peers before deregistering (for footer)
        peer_count = 0
        try:
            sessions = mgr.list_sessions(auto_clean=True)
            peer_count = sum(1 for s in sessions if s.get("session_id") != session_id)
        except Exception:
            pass

        # Broadcast structured handoff message for successor sessions (silent)
        try:
            summary = "Session ended"
            try:
                from omega.bridge import _get_store

                store = _get_store()
                counts = store.get_session_event_counts(session_id) if session_id else {}
                if counts:
                    total = sum(counts.values())
                    summary = f"Session ended ({total} memories captured)"
            except Exception:
                pass

            handoff_parts = [f"## Session Summary\n{summary[:300]}"]

            # Decisions made this session
            try:
                from omega.bridge import query_structured as _qs_handoff

                decisions = _qs_handoff(
                    query_text="decisions made",
                    session_id=session_id,
                    event_type="decision",
                    limit=5,
                )
                if decisions:
                    handoff_parts.append("## Decisions")
                    for d in decisions:
                        handoff_parts.append(f"- {d.get('content', '')[:120]}")
            except Exception:
                pass

            # Errors/blockers encountered
            try:
                from omega.bridge import query_structured as _qs_errors

                errors = _qs_errors(
                    query_text="errors encountered",
                    session_id=session_id,
                    event_type="error_pattern",
                    limit=3,
                )
                if errors:
                    handoff_parts.append("## Blockers")
                    for e in errors:
                        handoff_parts.append(f"- {e.get('content', '')[:120]}")
            except Exception:
                pass

            # Incomplete tasks owned by this session
            incomplete = []
            try:
                all_tasks = mgr.list_tasks(project=project, status="in_progress")
                incomplete = [t for t in all_tasks if t.get("session_id") == session_id]
                if incomplete:
                    handoff_parts.append("## Incomplete Work")
                    for t in incomplete:
                        handoff_parts.append(f"- Task #{t['id']}: {t['title']}")
            except Exception:
                pass

            handoff_body = "\n".join(handoff_parts)[:2000]

            mgr.send_message(
                from_session=session_id,
                subject=f"Handoff: {summary[:60]}",
                msg_type="complete",
                project=project or None,
                body=handoff_body,
                ttl_minutes=120,
            )
        except Exception:
            pass  # Handoff message is best-effort

        # Reassign orphaned tasks to pending so next session can claim them
        reassigned = []
        try:
            reassigned = mgr.reassign_orphaned_tasks(session_id)
            if reassigned:
                task_list = ", ".join(f"#{t['id']} {t['title']}" for t in reassigned[:3])
                lines.append(f"  Tasks returned to queue: {task_list}")
        except Exception:
            pass  # Reassign is best-effort

        # Task state for output (completed + continuing)
        my_completed = []
        try:
            completed_tasks = mgr.list_tasks(project=project, status="completed")
            my_completed = [t for t in completed_tasks if t.get("session_id") == session_id]
            for t in my_completed[:2]:
                lines.append(f"  Done: #{t['id']} {t['title']}")
            for t in reassigned[:2]:
                progress = t.get("progress", 0)
                prog_str = f" ({progress}%)" if progress else ""
                lines.append(f"  Continuing: #{t['id']} {t['title']}{prog_str} — returned to queue for next session")
        except Exception:
            pass

        # [COORD] summary block — files, messages, tasks
        try:
            coord_parts = []
            # Files claimed this session
            claims = mgr.get_session_claims(session_id)
            file_claims = claims.get("file_claims", [])
            if file_claims:
                fnames = [os.path.basename(f) for f in file_claims[:3]]
                if len(file_claims) > 3:
                    fnames.append(f"+{len(file_claims) - 3}")
                coord_parts.append(f"Files: {', '.join(fnames)} ({len(file_claims)} released)")
            # Messages sent/received
            msg_sent = 0
            msg_received = 0
            try:
                sent_rows = mgr._conn.execute(
                    "SELECT COUNT(*) FROM coord_messages WHERE from_session = ?",
                    (session_id,),
                ).fetchone()
                msg_sent = sent_rows[0] if sent_rows else 0
                recv_rows = mgr._conn.execute(
                    "SELECT COUNT(*) FROM coord_messages WHERE to_session = ? OR "
                    "(to_session IS NULL AND from_session != ?)",
                    (session_id, session_id),
                ).fetchone()
                msg_received = recv_rows[0] if recv_rows else 0
            except Exception:
                pass
            if msg_sent or msg_received:
                coord_parts.append(f"Messages: {msg_sent} sent, {msg_received} received")
            # Tasks summary
            task_parts = []
            if my_completed:
                task_parts.append(f"{len(my_completed)} completed")
            if incomplete:
                for t in incomplete[:2]:
                    progress = t.get("progress", 0)
                    prog_str = f" ({progress}%)" if progress else ""
                    task_parts.append(f"#{t['id']} in progress{prog_str}")
            if task_parts:
                coord_parts.append(f"Tasks: {', '.join(task_parts)}")
            if coord_parts:
                lines.append("[COORD] Session coordination:")
                for part in coord_parts:
                    lines.append(f"  {part}")
        except Exception:
            pass  # Coord summary is best-effort

        mgr.deregister_session(session_id)

        # Footer: snapshot + peer notification
        footer = "Snapshot stored."
        if peer_count > 0:
            footer += f" {peer_count} peer(s) notified."
        lines.append(footer)

    except Exception as e:
        _log_hook_error("coord_session_stop", e)
        return {"output": "", "error": str(e)}

    return {"output": "\n".join(lines), "error": None}


def handle_surface_memories(payload: dict) -> dict:
    """Surface memories on file edits, capture errors from Bash."""
    tool_name = payload.get("tool_name", "")
    tool_input = payload.get("tool_input", "{}")
    tool_output = payload.get("tool_output") or ""
    if not isinstance(tool_output, str):
        tool_output = json.dumps(tool_output) if isinstance(tool_output, (dict, list)) else str(tool_output)
    session_id = payload.get("session_id", "")
    project = payload.get("project", "")
    entity_id = _resolve_entity(project) if project else None

    # Parse tool input once for all branches
    input_data = _parse_tool_input(payload)

    lines = []

    # Surface memories on file edits
    if tool_name in ("Edit", "Write", "NotebookEdit"):
        file_path = _get_file_path_from_input(input_data)
        if file_path and _debounce_check(_last_surface, file_path, SURFACE_DEBOUNCE_S, _MAX_SURFACE_ENTRIES):
            lines.extend(_surface_for_edit(file_path, session_id, project, entity_id=entity_id))
            _ctx_tags = _ext_to_tags(file_path) or None
            lines.extend(_surface_lessons(file_path, session_id, project, entity_id=entity_id, context_tags=_ctx_tags))

            # Transparent "no results" — show once per session on first edit with no context
            if not lines:
                no_ctx_key = f"_no_ctx_{session_id}"
                if no_ctx_key not in _last_surface:
                    _last_surface[no_ctx_key] = time.monotonic()
                    lines.append(f"[MEMORY] No stored context for {os.path.basename(file_path)} yet")

    # Surface memories on file reads (lightweight — no lessons)
    if tool_name == "Read":
        file_path = input_data.get("file_path", "")
        if file_path and _debounce_check(_last_surface, file_path, SURFACE_DEBOUNCE_S, _MAX_SURFACE_ENTRIES):
            lines.extend(_surface_for_edit(file_path, session_id, project, entity_id=entity_id))

    # Auto-capture errors from Bash failures + track git commits + auto-claim branches
    if tool_name == "Bash" and tool_output:
        recall_lines = _capture_error(tool_output, session_id, project, entity_id=entity_id)
        if recall_lines:
            lines.extend(recall_lines)
        _track_git_commit(tool_input, tool_output, session_id, project)
        _auto_claim_branch(tool_input, session_id, project)

    # Surface peer claims in same directory on edits (multi-agent only, debounced)
    if tool_name in ("Edit", "Write", "NotebookEdit") and session_id and project:
        try:
            edit_path = _get_file_path_from_input(input_data)
            if edit_path:
                edit_dir = os.path.dirname(os.path.abspath(edit_path))
                if _debounce_check(_last_peer_dir_check, edit_dir, PEER_DIR_CHECK_DEBOUNCE_S, _MAX_SURFACE_ENTRIES):
                    from omega.coordination import get_manager

                    mgr = get_manager()
                    sessions = mgr.list_sessions(auto_clean=False)
                    peers = [s for s in sessions
                             if s.get("session_id") != session_id
                             and s.get("project") == project
                             and s.get("status") == "active"]
                    if peers:
                        peer_lines = []
                        for p in peers[:4]:
                            claims = mgr.get_session_claims(p["session_id"])
                            peer_files = claims.get("file_claims", [])
                            same_dir = [os.path.basename(f) for f in peer_files
                                        if os.path.dirname(os.path.abspath(f)) == edit_dir]
                            if same_dir:
                                p_name = _agent_nickname(p["session_id"])
                                flist = ", ".join(same_dir[:4])
                                if len(same_dir) > 4:
                                    flist += f" +{len(same_dir) - 4}"
                                peer_lines.append(f"[PEER] {p_name} has {flist} claimed (same dir as your edit)")
                        if peer_lines:
                            lines.extend(peer_lines[:2])
        except Exception:
            pass  # Fail-open — peer check is best-effort

    # Check for due reminders (debounced — max once per 5 minutes)
    global _last_reminder_check
    now_mono = time.monotonic()
    if _last_reminder_check == 0.0 or now_mono - _last_reminder_check >= REMINDER_CHECK_DEBOUNCE_S:
        _last_reminder_check = now_mono
        try:
            from omega.bridge import get_due_reminders

            due = get_due_reminders(mark_fired=True)
            for r in due[:3]:
                overdue_label = " [OVERDUE]" if r.get("is_overdue") else ""
                lines.append(f"\n[REMINDER]{overdue_label} {r['text']}")
                lines.append(f"  ID: {r['id'][:12]} — dismiss with omega_remind_dismiss")
        except Exception:
            pass  # Fail-open

    return {"output": "\n".join(lines), "error": None}


def _track_surfaced_ids(session_id: str, file_path: str, memory_ids: list):
    """Append surfaced memory IDs to the session's .surfaced.json file."""
    if not session_id or not memory_ids:
        return
    try:
        json_path = Path.home() / ".omega" / f"session-{session_id}.surfaced.json"
        existing = {}
        if json_path.exists():
            existing = json.loads(json_path.read_text())
        prev = existing.get(file_path, [])
        merged = list(set(prev + memory_ids))
        existing[file_path] = merged
        data = json.dumps(existing).encode("utf-8")
        fd = os.open(str(json_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            os.write(fd, data)
        finally:
            os.close(fd)
    except Exception:
        pass


def _ext_to_tags(file_path: str) -> list:
    """Derive context tags from file extension for re-ranking boost."""
    ext = os.path.splitext(file_path)[1].lower()
    _EXT_MAP = {
        ".py": ["python"],
        ".js": ["javascript"],
        ".ts": ["typescript"],
        ".tsx": ["typescript", "react"],
        ".jsx": ["javascript", "react"],
        ".rs": ["rust"],
        ".go": ["go"],
        ".rb": ["ruby"],
        ".java": ["java"],
        ".swift": ["swift"],
        ".sh": ["bash"],
        ".sql": ["sql"],
        ".md": ["markdown"],
        ".yml": ["yaml"],
        ".yaml": ["yaml"],
        ".json": ["json"],
        ".toml": ["toml"],
        ".css": ["css"],
        ".html": ["html"],
        ".c": ["c"],
        ".cpp": ["c++"],
        ".vue": ["vue", "javascript"],
        ".svelte": ["svelte", "javascript"],
        ".tf": ["terraform"],
        ".graphql": ["graphql"],
        ".prisma": ["prisma"],
        ".env": ["config"],
        ".ini": ["config"],
        ".kt": ["kotlin"],
        ".sc": ["scala"],
        ".ex": ["elixir"],
        ".php": ["php"],
        ".r": ["r"],
    }
    return _EXT_MAP.get(ext, [])


def _apply_confidence_boost(results: list) -> list:
    """Boost relevance scores by capture confidence: high=1.2x, low=0.7x."""
    for r in results:
        confidence = (r.get("metadata") or {}).get("capture_confidence", "medium")
        score = r.get("relevance", 0.0)
        if confidence == "high":
            r["relevance"] = min(1.0, score * 1.2)
        elif confidence == "low":
            r["relevance"] = score * 0.7
    return results


def _surface_for_edit(file_path: str, session_id: str, project: str, *, entity_id: "Optional[str]" = None) -> list[str]:
    """Surface memories related to a file being edited."""
    lines = []
    try:
        from omega.bridge import query_structured
        from omega.sqlite_store import SurfacingContext

        filename = os.path.basename(file_path)
        dirname = os.path.basename(os.path.dirname(file_path))
        context_tags = _ext_to_tags(file_path)
        # Bias query with session intent if available
        intent_hint = ""
        if session_id:
            cached_intent = _session_intent.get(session_id)
            _INTENT_HINTS = {
                "coding": "code implementation",
                "logic": "logic analysis",
                "creative": "creative writing",
                "exploration": "research exploration",
            }
            if cached_intent and cached_intent in _INTENT_HINTS:
                intent_hint = _INTENT_HINTS[cached_intent] + " "
        results = query_structured(
            query_text=f"{intent_hint}{filename} {dirname} {file_path}",
            limit=3,
            session_id=session_id,
            project=project,
            context_file=file_path,
            context_tags=context_tags or None,
            filter_tags=context_tags or None,
            entity_id=entity_id,
            surfacing_context=SurfacingContext.FILE_EDIT,
        )
        results = _apply_confidence_boost(results)
        results = [r for r in results if r.get("relevance", 0.0) >= 0.20]
        if results:
            lines.append(f"\n[MEMORY] Relevant context for {filename}:")
            for r in results:
                score = r.get("relevance", 0.0)
                etype = r.get("event_type", "memory")
                preview = r.get("content", "")[:120].replace("\n", " ")
                nid = r.get("id", "")[:8]
                created = r.get("created_at", "")
                age = _relative_time_from_iso(created) if created else ""
                age_part = f" ({age})" if age else ""
                lines.append(f"  [{score:.0%}] {etype}{age_part}: {preview} (id:{nid})")

            # First-recall milestone
            if _check_milestone("first-recall"):
                lines.append("[OMEGA] First memory recalled! Past context is informing this edit.")

            # Track surfaced memory IDs for auto-feedback
            memory_ids = [r.get("id") for r in results if r.get("id")]
            _track_surfaced_ids(session_id, file_path, memory_ids)

            # Traverse: show linked memories (1 hop from top result)
            try:
                top_id = results[0].get("id", "")
                shown_ids = {r.get("id") for r in results}
                if top_id:
                    from omega.bridge import _get_store as _gs

                    _store = _gs()
                    linked = _store.get_related_chain(top_id, max_hops=1, min_weight=0.4)
                    novel = [ln for ln in linked if ln.get("node_id") not in shown_ids][:2]
                    for ln in novel:
                        etype = (ln.get("metadata") or {}).get("event_type", "memory")
                        preview = ln.get("content", "")[:100].replace("\n", " ")
                        lines.append(f"  [linked] {etype}: {preview}")
            except Exception:
                pass

            # Phrase search: exact-match error patterns for this file
            try:
                from omega.bridge import _get_store as _gs2

                _store2 = _gs2()
                exact_errors = _store2.phrase_search(filename, limit=2, event_type="error_pattern")
                shown_ids_updated = {r.get("id") for r in results}
                for err in exact_errors:
                    if err.id not in shown_ids_updated:
                        preview = err.content[:100].replace("\n", " ")
                        lines.append(f"  [exact] error: {preview}")
            except Exception:
                pass
    except Exception as e:
        _log_hook_error("surface_for_edit", e)

    # Surface relevant knowledge base chunks for this file
    if file_path:
        try:
            from omega.knowledge.engine import search_documents as _kb_search

            filename = os.path.basename(file_path)
            kb_result = _kb_search(query=filename, limit=1, entity_id=entity_id)
            if kb_result and "No document matches" not in kb_result and "No documents ingested" not in kb_result:
                # Extract first line of content (skip header formatting)
                kb_lines_raw = [ln.strip() for ln in kb_result.split("\n") if ln.strip() and not ln.startswith("#")]
                if kb_lines_raw:
                    snippet = kb_lines_raw[0][:120]
                    lines.append(f"[KB] {snippet}")
        except ImportError:
            pass  # Knowledge engine not available
        except Exception:
            pass  # KB surfacing is best-effort

    return lines


def _surface_lessons(file_path: str, session_id: str, project: str, *, entity_id: "Optional[str]" = None, context_tags: "Optional[list]" = None) -> list[str]:
    """Surface verified cross-session lessons and peer decisions."""
    lines = []
    try:
        from omega.bridge import get_cross_session_lessons

        filename = os.path.basename(file_path)
        lessons = get_cross_session_lessons(
            task=f"editing {filename}",
            project_path=project,
            exclude_session=session_id,
            limit=2,
            context_file=file_path,
            context_tags=context_tags,
        )
        verified = [lesson for lesson in lessons if lesson.get("verified")]
        if verified:
            lines.append(f"\n[LESSON] Verified wisdom for {filename}:")
            for lesson in verified:
                content = lesson.get("content", "")[:150]
                lines.append(f"  - {content}")
    except Exception as e:
        _log_hook_error("surface_lessons", e)

    # Surface recent peer decisions about this file
    if file_path and session_id:
        try:
            from omega.bridge import query_structured as _qs_peer

            filename = os.path.basename(file_path)
            peer_decisions = _qs_peer(
                query_text=f"decisions about {filename}",
                event_type="decision",
                limit=3,
                context_file=file_path,
                context_tags=context_tags,
                entity_id=entity_id,
            )
            # Filter out own session's decisions and low-relevance results
            for d in peer_decisions:
                meta = d.get("metadata") or {}
                if meta.get("session_id") == session_id:
                    continue
                if d.get("relevance", 0) < 0.5:
                    continue
                content = d.get("content", "")[:100].replace("\n", " ")
                lines.append(f"[PEER-DECISION] {content}")
                break  # Only show 1 to avoid noise
        except Exception:
            pass  # Peer decision surfacing is best-effort

    return lines




def _extract_error_summary(raw_output: str) -> str:
    """Extract a clean error summary from raw tool output.

    For tracebacks: grab the last non-frame line (the actual error).
    For JSON blobs: skip JSON structure, find the error marker line.
    """
    lines = raw_output.strip().split("\n")

    if "Traceback (most recent call last)" in raw_output:
        for line in reversed(lines):
            stripped = line.strip()
            if stripped and not stripped.startswith("File ") and not stripped.startswith("^"):
                return stripped[:300]

    non_json_lines = []
    for line in lines:
        stripped = line.strip()
        if stripped and not stripped.startswith(("{", "[", "}", "]", '"')):
            non_json_lines.append(stripped)
    if non_json_lines:
        error_markers_local = [
            "Error:",
            "ERROR:",
            "error:",
            "FAILED",
            "Failed",
            "SyntaxError:",
            "TypeError:",
            "NameError:",
            "ImportError:",
            "ModuleNotFoundError:",
            "AttributeError:",
            "ValueError:",
            "KeyError:",
            "IndexError:",
            "FileNotFoundError:",
            "fatal:",
            "FATAL:",
            "panic:",
            "command not found",
            "No such file or directory",
            "Permission denied",
            "Connection refused",
        ]
        for line in non_json_lines:
            if any(m in line for m in error_markers_local):
                return line[:300]
        return non_json_lines[0][:300]

    return raw_output[:300]


def _capture_error(tool_output: str, session_id: str, project: str, *, entity_id: "Optional[str]" = None) -> list[str]:
    """Auto-capture error patterns from Bash failures + recall past fixes.

    Session-level dedup: skip if same error pattern already captured.
    Cap at _MAX_ERRORS_PER_SESSION to prevent test-run floods.
    Returns lines for "you've seen this before" recall output.
    """
    if not tool_output:
        return []
    if not isinstance(tool_output, str):
        tool_output = str(tool_output)

    # Cap errors per session
    if _error_counts.get(session_id, 0) >= _MAX_ERRORS_PER_SESSION:
        return []

    error_markers = [
        "Error:",
        "ERROR:",
        "error:",
        "FAILED",
        "Failed",
        "Traceback (most recent call last)",
        "SyntaxError:",
        "TypeError:",
        "NameError:",
        "ImportError:",
        "ModuleNotFoundError:",
        "AttributeError:",
        "ValueError:",
        "KeyError:",
        "IndexError:",
        "FileNotFoundError:",
        "fatal:",
        "FATAL:",
        "panic:",
        "command not found",
        "No such file or directory",
        "Permission denied",
        "Connection refused",
    ]

    if not any(marker in tool_output for marker in error_markers):
        return []

    error_summary = _extract_error_summary(tool_output)

    # Session-level dedup: hash the first 100 chars (normalized)
    error_hash = re.sub(r"\s+", " ", error_summary[:100].lower()).strip()
    if error_hash in _error_hashes:
        return []
    if len(_error_hashes) >= _MAX_ERROR_HASHES:
        return []  # Cap reached — stop tracking new hashes to bound memory
    _error_hashes.add(error_hash)
    _error_counts[session_id] = _error_counts.get(session_id, 0) + 1

    recall_lines: list[str] = []

    # --- Feature: "You've seen this before" — proactive error recall ---
    try:
        from omega.bridge import query_structured
        from omega.sqlite_store import SurfacingContext

        # Search for matching error_pattern and lesson_learned memories
        past_errors = query_structured(
            query_text=error_summary[:200],
            limit=2,
            project=project,
            event_type="error_pattern",
            entity_id=entity_id,
            surfacing_context=SurfacingContext.ERROR_DEBUG,
        )
        past_lessons = query_structured(
            query_text=error_summary[:200],
            limit=2,
            event_type="lesson_learned",
            entity_id=entity_id,
        )

        # Filter to only high-relevance matches (>= 0.35) from previous sessions
        past_matches = []
        for m in (past_errors or []) + (past_lessons or []):
            if m.get("relevance", 0) >= 0.35 and m.get("session_id") != session_id:
                past_matches.append(m)

        if past_matches:
            recall_lines.append("")
            recall_lines.append("[RECALL] You've seen this before:")
            for m in past_matches[:2]:
                etype = m.get("event_type", "memory")
                content = m.get("content", "")[:150].replace("\n", " ").strip()
                rel_time = m.get("relative_time") or ""
                time_note = f" ({rel_time})" if rel_time else ""
                recall_lines.append(f"  [{etype}]{time_note} {content}")
    except Exception as e:
        _log_hook_error("error_recall", e)

    # Store the new error pattern
    try:
        from omega.bridge import auto_capture

        auto_capture(
            content=f"Error: {error_summary}",
            event_type="error_pattern",
            metadata={"source": "auto_capture_hook", "project": project},
            session_id=session_id,
            project=project,
            entity_id=entity_id,
        )
    except Exception as e:
        _log_hook_error("capture_error", e)

    return recall_lines


def _track_git_commit(tool_input: str, tool_output: str, session_id: str, project: str):
    """Detect git commit in Bash output and log to coordination."""
    if not tool_output:
        return
    if not isinstance(tool_output, str):
        tool_output = str(tool_output)

    try:
        input_data = json.loads(tool_input) if isinstance(tool_input, str) else tool_input
    except (json.JSONDecodeError, TypeError):
        return

    command = input_data.get("command", "")
    if "git commit" not in command:
        return

    match = re.search(r"\[[\w/.-]+\s+([0-9a-f]{7,12})\]", tool_output)
    if not match:
        return

    commit_hash = match.group(1)
    msg_match = re.search(r"\[[\w/.-]+\s+[0-9a-f]{7,12}\]\s+(.+)", tool_output)
    message = msg_match.group(1).strip() if msg_match else ""

    branch = _get_current_branch(project)

    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        mgr.log_git_event(
            project=project,
            event_type="commit",
            commit_hash=commit_hash,
            branch=branch,
            message=message,
            session_id=session_id,
        )

        # Auto-release file claims for committed files
        if session_id and project and commit_hash:
            try:
                diff_result = subprocess.run(
                    ["git", "diff-tree", "--no-commit-id", "--name-only", "-r", commit_hash],
                    capture_output=True,
                    text=True,
                    timeout=5,
                    cwd=project,
                )
                if diff_result.returncode == 0 and diff_result.stdout.strip():
                    committed_files = [
                        os.path.join(project, f.strip()) for f in diff_result.stdout.strip().split("\n") if f.strip()
                    ]
                    if committed_files:
                        mgr.release_committed_files(session_id, project, committed_files)
            except Exception:
                pass  # Auto-release is best-effort

    except Exception as e:
        _log_hook_error("track_git_commit", e)


def _auto_claim_branch(tool_input: str, session_id: str, project: str):
    """Auto-claim branch on git checkout/switch (PostToolUse, best-effort)."""
    if not session_id or not project:
        return
    try:
        input_data = json.loads(tool_input) if isinstance(tool_input, str) else tool_input
    except (json.JSONDecodeError, TypeError):
        return

    command = input_data.get("command", "")
    if not any(cmd in command for cmd in ("git checkout", "git switch", "git push")):
        return

    try:
        branch = _get_current_branch(project)
        if not branch or branch == "HEAD":
            return  # Detached HEAD or not a git repo

        from omega.coordination import get_manager

        mgr = get_manager()
        mgr.claim_branch(session_id, project, branch, task="auto-claimed on checkout")
    except Exception:
        pass  # Auto-claim is best-effort


def handle_coord_heartbeat(payload: dict) -> dict:
    """Update session heartbeat (debounced to max once per 30s).

    Every 5th heartbeat, check for unread messages and surface count.
    """
    session_id = payload.get("session_id", "")
    if not session_id:
        return {"output": "", "error": None}

    # Drain urgent queue BEFORE debounce — every tool use can deliver push notifications
    urgent_output = _drain_urgent_queue(session_id)

    now = time.monotonic()
    if now - _last_heartbeat.get(session_id, 0) < HEARTBEAT_DEBOUNCE_S:
        return {"output": urgent_output, "error": None}

    _last_heartbeat[session_id] = now
    output = urgent_output
    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        mgr.heartbeat(session_id)

        # Single-agent fast path: skip coordination overhead when alone
        try:
            multi_agent = mgr.active_session_count() > 1
        except Exception:
            multi_agent = True  # Assume multi-agent when check fails

        # Every 2nd heartbeat, check inbox (~60s polling)
        # Non-4th beats: show inbox preview with nicknames (inform-type only,
        #   preserves request/complete unread for 4th-beat coordination check)
        # 4th beats: full coordination check below handles request/complete messages
        count = _heartbeat_count.get(session_id, 0) + 1
        _heartbeat_count[session_id] = count
        project = payload.get("project", "")

        # Mid-session activity pulse — fires once at heartbeat #5
        if count == 5:
            try:
                surfaced_json = Path.home() / ".omega" / f"session-{session_id}.surfaced.json"
                if surfaced_json.exists():
                    data = json.loads(surfaced_json.read_text())
                    surfaced_so_far = sum(len(ids) for ids in data.values())
                    if surfaced_so_far > 0:
                        output = _append_output(output, f"[MEMORY] {surfaced_so_far} memories surfaced so far this session")
            except Exception:
                pass  # Activity pulse is best-effort

        unread = 0
        if count % 2 == 0:
            try:
                unread = mgr.get_unread_count(session_id)
                if unread > 0 and count % 4 != 0:
                    # Show inbox preview with nicknames (inform-type only)
                    try:
                        msgs = mgr.check_inbox(session_id, unread_only=True, limit=2, msg_type="inform")
                        if msgs:
                            previews = []
                            for m in msgs[:2]:
                                from_name = _agent_nickname(m.get("from_session") or "unknown")
                                subj = (m.get("subject") or "")[:60]
                                previews.append(f'{from_name}: "{subj}"')
                            remaining = unread - len(msgs)
                            output = f"[INBOX] {unread} unread — " + " | ".join(previews)
                            if remaining > 0:
                                output += " — use omega_inbox for more"
                        else:
                            output = f"[INBOX] {unread} unread message(s) — use omega_inbox to read"
                    except Exception:
                        output = f"[INBOX] {unread} unread message(s) — use omega_inbox to read"
            except Exception:
                pass  # Inbox check is best-effort

        # Every 4th heartbeat (~2 min), surface messages + coordination check
        if count % 4 == 0:
            try:
                coord_lines = []

                # Multi-agent only: check if my task is blocked by another agent's task
                if multi_agent:
                    try:
                        all_tasks = mgr.list_tasks(project=project, status="in_progress")
                        my_tasks = [t for t in all_tasks if t.get("session_id") == session_id]
                        for task in my_tasks[:1]:
                            for dep_id in task.get("depends_on") or []:
                                dep_tasks = [t for t in all_tasks if t.get("id") == dep_id]
                                if dep_tasks and dep_tasks[0].get("status") != "completed":
                                    owner_name = _agent_nickname(dep_tasks[0].get("session_id") or "unknown")
                                    coord_lines.append(
                                        f"[BLOCKED] Task #{task['id']} waiting on #{dep_id} "
                                        f"(owner: {owner_name}, status: {dep_tasks[0].get('status', 'unknown')})"
                                    )
                    except Exception:
                        pass

                # Surface first request-type message content (not just count)
                if unread > 0:
                    try:
                        msgs = mgr.check_inbox(session_id, unread_only=True, limit=1, msg_type="request")
                        if msgs:
                            m = msgs[0]
                            from_name = _agent_nickname(m.get("from_session") or "unknown")
                            subj = (m.get("subject") or "")[:80]
                            coord_lines.append(f"[REQUEST] From {from_name}: {subj}")
                    except Exception:
                        pass

                # Surface first complete-type message (handoff from departed peer)
                try:
                    complete_msgs = mgr.check_inbox(session_id, unread_only=True, limit=1, msg_type="complete")
                    if complete_msgs:
                        m = complete_msgs[0]
                        from_name = _agent_nickname(m.get("from_session") or "unknown")
                        subj = (m.get("subject") or "")[:80]
                        body_preview = (m.get("body") or "")[:200].split("\n")[0]
                        coord_lines.append(
                            f"[HANDOFF] From {from_name}: {subj}" + (f"\n  {body_preview}" if body_preview else "")
                        )
                except Exception:
                    pass

                # Multi-agent only: [TEAM] activity line — recent peer coordination events
                if multi_agent and project:
                    try:
                        events = mgr.get_recent_events(
                            project=project,
                            minutes=2,
                            limit=3,
                            exclude_session=session_id,
                        )
                        if events:
                            parts = []
                            for ev in events[:2]:
                                ev_name = _agent_nickname(ev["session_id"])
                                parts.append(f"{ev_name} {ev['summary']}")
                            coord_lines.append(f"[TEAM] {' | '.join(parts)}")
                    except Exception:
                        pass  # Team activity is best-effort

                if coord_lines:
                    output = _append_output(output, "\n".join(coord_lines))
            except Exception:
                pass  # Coordination check is best-effort

        # Every 6th heartbeat (~3 min), idle detection — nudge agent to claim a task
        if count % 6 == 0:
            try:
                all_tasks = mgr.list_tasks(status="in_progress")
                my_tasks = [t for t in all_tasks if t.get("session_id") == session_id]
                if not my_tasks:
                    pending = mgr.list_tasks(project=project or None, status="pending")
                    unblocked = [t for t in pending if not t.get("blocked")]
                    if unblocked:
                        idle_line = (
                            f"[IDLE] No active task — {len(unblocked)} unclaimed task(s) available. "
                            f"Use omega_task_next to claim one."
                        )
                        output = _append_output(output, idle_line)
            except Exception:
                pass  # Idle detection is best-effort

        # Every 8th heartbeat (~4 min), detect peer state changes
        if multi_agent and count % 8 == 0 and project:
            try:
                sessions = mgr.list_sessions(auto_clean=False)
                current_peers = {
                    s["session_id"]
                    for s in sessions
                    if s.get("session_id") != session_id
                    and s.get("project") == project
                    and s.get("status") == "active"
                }
                prev_peers = _peer_snapshot.get(session_id, set())
                _peer_snapshot[session_id] = current_peers

                # Only report changes after first snapshot (skip initial population)
                if prev_peers is not None and prev_peers != current_peers:
                    change_parts = []
                    joined = current_peers - prev_peers
                    departed = prev_peers - current_peers
                    for sid in list(joined)[:2]:
                        change_parts.append(f"{_agent_nickname(sid)} joined")
                    for sid in list(departed)[:2]:
                        change_parts.append(f"{_agent_nickname(sid)} left")

                    # Check for new claims in same project from peers
                    for p_sid in list(current_peers)[:4]:
                        try:
                            claims = mgr.get_session_claims(p_sid)
                            new_files = claims.get("file_claims", [])
                            if new_files:
                                p_name = _agent_nickname(p_sid)
                                fname = os.path.basename(new_files[-1])
                                change_parts.append(f"{p_name} claimed {fname}")
                        except Exception:
                            pass

                    if change_parts:
                        active_count = len(current_peers)
                        summary = " | ".join(change_parts[:3])
                        refresh_line = f"[TEAM] {summary} | {active_count} peer{'s' if active_count != 1 else ''} active"
                        output = _append_output(output, refresh_line)
            except Exception:
                pass  # Peer refresh is best-effort

        # Every 10th heartbeat (~5 min), deadlock detection
        if multi_agent and count % 10 == 0:
            try:
                cycles = mgr.detect_deadlocks()
                if cycles:
                    now_dl = time.monotonic()
                    for cycle in cycles[:2]:
                        cycle_str = " -> ".join(_agent_nickname(s) for s in cycle)
                        deadlock_line = f"[DEADLOCK] Circular wait: {cycle_str}"
                        output = _append_output(output, deadlock_line)

                        # Push [DEADLOCK] to all OTHER sessions in the cycle (debounced)
                        cycle_key = str(hash(tuple(sorted(cycle[:-1]))))
                        if cycle_key not in _last_deadlock_push or now_dl - _last_deadlock_push[cycle_key] >= DEADLOCK_PUSH_DEBOUNCE_S:
                            _last_deadlock_push[cycle_key] = now_dl
                            for peer in set(cycle[:-1]):
                                if peer == session_id:
                                    continue  # detecting session sees it in hook output
                                try:
                                    mgr.send_message(
                                        from_session=session_id,
                                        subject=f"[DEADLOCK] Circular wait: {cycle_str}",
                                        to_session=peer,
                                        msg_type="inform",
                                        ttl_minutes=30,
                                    )
                                except Exception:
                                    pass  # fail-open per notification
            except Exception:
                pass  # Deadlock detection is best-effort

        # Every 20th heartbeat (~10 min), lightweight git divergence check
        if count % 20 == 0 and project:
            try:
                result = subprocess.run(
                    ["git", "rev-list", "--count", "--left-right", "HEAD...@{u}"],
                    capture_output=True, text=True, timeout=5, cwd=project,
                )
                if result.returncode == 0:
                    parts = result.stdout.strip().split("\t")
                    if len(parts) == 2:
                        ahead, behind = int(parts[0]), int(parts[1])
                        if behind > 0:
                            div_line = f"[GIT] Branch is {behind} commit(s) behind upstream"
                            if ahead > 0:
                                div_line += f" (and {ahead} ahead)"
                            output = _append_output(output, div_line)
            except Exception:
                pass  # Git divergence check is best-effort
    except Exception as e:
        _log_hook_error("coord_heartbeat", e)
        return {"output": "", "error": str(e)}

    return {"output": output, "error": None}


def _suggest_alternative_dir(file_path: str) -> str | None:
    """Suggest sibling directories as alternative work areas."""
    parent = os.path.dirname(file_path)
    grandparent = os.path.dirname(parent)
    if not grandparent or grandparent == parent:
        return None
    try:
        siblings = [
            d
            for d in sorted(os.listdir(grandparent))
            if os.path.isdir(os.path.join(grandparent, d))
            and d != os.path.basename(parent)
            and not d.startswith((".", "__"))
        ]
        if siblings:
            return f"nearby dirs: {', '.join(siblings[:3])}"
    except OSError:
        pass
    return None


def handle_auto_claim_file(payload: dict) -> dict:
    """Auto-claim files on Edit/Write/NotebookEdit (debounced per file)."""
    tool_name = payload.get("tool_name", "")
    if tool_name not in ("Edit", "Write", "NotebookEdit"):
        return {"output": "", "error": None}

    session_id = payload.get("session_id", "")
    if not session_id:
        return {"output": "", "error": None}

    input_data = _parse_tool_input(payload)
    file_path = _get_file_path_from_input(input_data)
    if not file_path:
        return {"output": "", "error": None}

    # Debounce: skip full claim if same (session, file) claimed recently,
    # but still refresh last_activity to prevent TTL expiry during active editing
    claim_key = (session_id, file_path)
    if not _debounce_check(_last_claim, claim_key, CLAIM_DEBOUNCE_S, _MAX_SURFACE_ENTRIES):
        try:
            from omega.coordination import get_manager

            mgr = get_manager()
            mgr.refresh_file_activity(session_id, file_path)
        except Exception:
            pass  # Activity refresh is best-effort
        return {"output": "", "error": None}

    output = ""
    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        result = mgr.claim_file(session_id, file_path, task="auto-claimed on edit")
        if result.get("conflict"):
            owner_name = _agent_nickname(result["claimed_by"])
            owner_task = result.get("task") or "unknown task"
            output = (
                f"[CONFLICT] {os.path.basename(file_path)} is claimed by "
                f"{owner_name} ({owner_task}). Coordinate before editing."
            )
        elif result.get("success"):
            # Single-agent fast path: skip intent + overlap checks when alone
            try:
                multi_agent = mgr.active_session_count() > 1
            except Exception:
                multi_agent = True  # Assume multi-agent when check fails

            if multi_agent:
                # Auto-announce intent for coordination visibility
                try:
                    mgr.announce_intent(
                        session_id=session_id,
                        description=f"Editing {os.path.basename(file_path)}",
                        intent_type="edit",
                        target_files=[file_path],
                        ttl_minutes=5,
                    )
                except Exception:
                    pass  # Intent announcement is best-effort

                # Check for intent overlaps with other agents (best-effort, max 2 warnings)
                try:
                    overlap_result = mgr.check_intents(session_id)
                    if overlap_result.get("has_overlaps"):
                        overlaps = overlap_result["overlaps"][:2]
                        for ov in overlaps:
                            ov_sid_full = ov["session_id"]
                            ov_name = _agent_nickname(ov_sid_full)
                            ov_desc = ov.get("description", "")[:60]
                            ov_file_paths = ov.get("overlapping_files", [])[:3]
                            ov_files = ", ".join(os.path.basename(f) for f in ov_file_paths)

                            # Escalate if overlapping files are already CLAIMED by the other agent
                            escalated = False
                            try:
                                for of in ov_file_paths:
                                    check = mgr.check_file(of)
                                    if check.get("claimed") and check.get("session_id") == ov_sid_full:
                                        escalated = True
                                        break
                            except Exception:
                                pass  # Escalation check is best-effort

                            if escalated:
                                warning = (
                                    f"[CONFLICT] {ov_name} owns "
                                    f"{', '.join(os.path.basename(f) for f in ov_file_paths)}"
                                    f" — consider working in a different area"
                                )
                                alt = _suggest_alternative_dir(file_path)
                                if alt:
                                    warning += f"\n  Suggestion: {alt}"
                            else:
                                warning = f"[INTENT-OVERLAP] {ov_name}: {ov_desc}"
                                if ov_files:
                                    warning += f" (files: {ov_files})"

                            output += ("\n" if output else "") + warning

                            # Notify the other agent about the overlap (debounced)
                            try:
                                notify_key = (session_id, ov_sid_full, file_path)
                                notify_now = time.monotonic()
                                if (
                                    notify_key not in _last_overlap_notify
                                    or notify_now - _last_overlap_notify[notify_key] >= OVERLAP_NOTIFY_DEBOUNCE_S
                                ):
                                    _last_overlap_notify[notify_key] = notify_now
                                    filename = os.path.basename(file_path)
                                    mgr.send_message(
                                        from_session=session_id,
                                        to_session=ov_sid_full,
                                        msg_type="inform",
                                        subject=f"Overlap: both editing {filename}",
                                        body=(
                                            f"I'm editing {file_path}. You announced intent "
                                            f"to work on overlapping files. Let's coordinate."
                                        ),
                                        ttl_minutes=30,
                                    )
                            except Exception:
                                pass  # Overlap notification is best-effort
                except Exception:
                    pass  # Overlap check is best-effort
    except Exception as e:
        error_str = str(e)
        if "already claimed" not in error_str.lower():
            _log_hook_error("auto_claim_file", e)

    return {"output": output, "error": None}


def _file_guard_block_msg(file_path: str, owner_sid: str, owner_task: str, blocked_sid: str = "") -> dict:
    """Return a block response for the file guard.

    Also sends a [WAITING] message to the file owner so they know
    someone is blocked (debounced to once per 5 min per tuple).
    """
    filename = os.path.basename(file_path)
    owner_name = _agent_nickname(owner_sid)
    msg = (
        f"\n[FILE-GUARD] BLOCKED: {filename} is claimed by {owner_name} ({owner_task}).\n"
        f"  Options:\n"
        f"    1. Wait for the other agent to finish and release\n"
        f"    2. Ask other agent to call omega_file_release\n"
        f"    3. The claim expires automatically after 10 minutes of inactivity\n"
        f"    4. Ask the human to decide if a force-override is safe"
    )

    # Notify the owner that someone is waiting
    if blocked_sid and owner_sid:
        now = time.monotonic()
        key = (blocked_sid, owner_sid, file_path)
        if key not in _last_block_notify or now - _last_block_notify[key] >= BLOCK_NOTIFY_DEBOUNCE_S:
            _last_block_notify[key] = now
            try:
                from omega.coordination import get_manager

                mgr = get_manager()
                blocked_name = _agent_nickname(blocked_sid)
                mgr.send_message(
                    from_session=blocked_sid,
                    subject=f"[WAITING] {blocked_name} wants to edit {filename} — consider releasing if done",
                    msg_type="inform",
                    to_session=owner_sid,
                )
            except Exception:
                pass  # Fail-open — notification is best-effort

    return {"output": msg, "error": None, "exit_code": 2}


def handle_pre_file_guard(payload: dict) -> dict:
    """Check file claims BEFORE editing — blocks if claimed by another agent.

    Returns exit_code=2 in the response dict when blocking.
    Fail-open: any error returns allow (exit_code=0).
    """
    tool_name = payload.get("tool_name", "")
    if tool_name not in ("Edit", "Write", "NotebookEdit"):
        return {"output": "", "error": None}

    session_id = payload.get("session_id", "")
    input_data = _parse_tool_input(payload)
    file_path = _get_file_path_from_input(input_data)
    if not file_path:
        return {"output": "", "error": None}

    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        info = mgr.check_file(file_path)

        if info.get("claimed"):
            if session_id and info.get("session_id") == session_id:
                return {"output": "", "error": None}

            # Claimed by different session (or no session_id to prove identity) — BLOCK
            owner_sid = info.get("session_id", "unknown")
            owner_task = info.get("task") or "unknown task"
            return _file_guard_block_msg(file_path, owner_sid, owner_task, blocked_sid=session_id)

        # Unclaimed — if we have session_id, claim atomically to prevent TOCTOU race
        if session_id:
            result = mgr.claim_file(session_id, file_path, task="pre-edit guard claim")
            if result.get("conflict"):
                owner_sid = result["claimed_by"]
                owner_task = result.get("task") or "unknown task"
                return _file_guard_block_msg(file_path, owner_sid, owner_task, blocked_sid=session_id)

        # No session_id + unclaimed → allow (true single-agent)
        return {"output": "", "error": None}

    except Exception as e:
        # Fail-open: never block when OMEGA is unavailable
        _log_hook_error("pre_file_guard", e)
        return {"output": "", "error": None}


def handle_pre_task_guard(payload: dict) -> dict:
    """Check task declaration BEFORE editing — blocks if no active task.

    Opt-in: only enforces when the project has non-terminal tasks.
    Returns exit_code=2 in the response dict when blocking.
    Fail-open: any error returns allow (exit_code absent).
    """
    tool_name = payload.get("tool_name", "")
    if tool_name not in ("Edit", "Write", "NotebookEdit"):
        return {"output": "", "error": None}

    session_id = payload.get("session_id", "")
    if not session_id:
        # Single-agent mode — no enforcement
        return {"output": "", "error": None}

    input_data = _parse_tool_input(payload)
    file_path = _get_file_path_from_input(input_data)
    if not file_path:
        return {"output": "", "error": None}

    project = payload.get("project", "")
    if not project:
        return {"output": "", "error": None}

    # Skip if file is outside the project directory
    try:
        if not os.path.abspath(file_path).startswith(os.path.abspath(project)):
            return {"output": "", "error": None}
    except Exception:
        return {"output": "", "error": None}

    try:
        from omega.coordination import get_manager

        mgr = get_manager()

        # Opt-in: only enforce if project has active tasks
        if not mgr.project_has_active_tasks(project):
            return {"output": "", "error": None}

        # Check if session has an in_progress task
        result = mgr.has_active_task(session_id)
        if result.get("has_task"):
            return {"output": "", "error": None}

        # No active task — BLOCK
        filename = os.path.basename(file_path)
        project_name = os.path.basename(project)
        msg = (
            f"\n[TASK-GUARD] BLOCKED: No active task for this session on {project_name}.\n"
            f"  Create and claim a task before editing {filename}:\n"
            f'    1. omega_task_create(title="Your task", project="{project}")\n'
            f'    2. omega_task_claim(task_id=<id>, session_id="{session_id}")\n'
            f"  Or complete/cancel existing tasks to disable enforcement."
        )
        return {"output": msg, "error": None, "exit_code": 2}

    except Exception as e:
        # Fail-open: never block when OMEGA is unavailable
        _log_hook_error("pre_task_guard", e)
        return {"output": "", "error": None}


def _clean_task_text(prompt: str) -> str:
    """Delegate to shared implementation in omega.task_utils."""
    from omega.task_utils import clean_task_text
    return clean_task_text(prompt)


def handle_auto_capture(payload: dict) -> dict:
    """Auto-capture decisions and lessons from user prompts (UserPromptSubmit)."""
    # Prefer top-level keys (set by fast_hook.py from parsed stdin JSON).
    # Fall back to re-parsing payload["stdin"] for legacy/direct callers.
    prompt = payload.get("prompt", "")
    stdin_parsed = {}
    if not prompt:
        stdin_data = payload.get("stdin", "")
        if not stdin_data:
            return {"output": "", "error": None}
        try:
            stdin_parsed = json.loads(stdin_data)
        except (json.JSONDecodeError, TypeError):
            return {"output": "", "error": None}
        prompt = stdin_parsed.get("prompt", "")

    # Extract session_id/cwd from top-level first, fall back to parsed stdin
    session_id = payload.get("session_id") or stdin_parsed.get("session_id", "")
    cwd = payload.get("cwd") or payload.get("project") or stdin_parsed.get("cwd", "")
    entity_id = _resolve_entity(cwd) if cwd else None

    if not prompt:
        return {"output": "", "error": None}

    # Auto-set session task from first prompt (DB as source of truth)
    # Runs before the 20-char guard — short prompts are valid tasks
    if session_id:
        try:
            from omega.coordination import get_manager as _get_mgr_task

            _mgr = _get_mgr_task()
            row = _mgr._conn.execute(
                "SELECT task FROM coord_sessions WHERE session_id = ?",
                (session_id,),
            ).fetchone()
            if row is not None and not row[0]:
                task_text = _clean_task_text(prompt)
                if task_text:
                    with _mgr._lock:
                        _mgr._conn.execute(
                            "UPDATE coord_sessions SET task = ? WHERE session_id = ?",
                            (task_text, session_id),
                        )
                        _mgr._conn.commit()
        except Exception:
            pass  # Auto-task is best-effort

    # Short prompts: task is set above, but skip decision/lesson/preference capture
    if len(prompt) < 20:
        return {"output": "", "error": None}

    # --- Surface peer work on planning prompts (coordination awareness) ---
    coord_output = ""
    _planning_patterns = [
        r"\bwhat.?s?\s+next\b",
        r"\bwhat\s+should\s+(?:i|we)\b",
        r"\bwhat\s+(?:to|can)\s+(?:do|work)\b",
        r"\bnext\s+(?:step|task|priority)\b",
        r"\bpriorities?\b",
        r"\broadmap\b",
        r"\bwhat\s+(?:remains?|is\s+left)\b",
    ]
    prompt_lower_early = prompt.lower()
    is_planning = any(re.search(pat, prompt_lower_early) for pat in _planning_patterns)
    if is_planning and session_id and cwd:
        now = time.monotonic()
        if session_id not in _last_coord_query or now - _last_coord_query[session_id] >= COORD_QUERY_DEBOUNCE_S:
            try:
                from omega.coordination import get_manager as _get_mgr_coord

                mgr = _get_mgr_coord()
                sessions = mgr.list_sessions(auto_clean=False)
                peers = [
                    s for s in sessions
                    if s.get("session_id") != session_id and s.get("project") == cwd
                ][:4]
                if peers:
                    # Fetch in-progress coord tasks for this project
                    in_progress_tasks = []
                    try:
                        in_progress_tasks = mgr.list_tasks(project=cwd, status="in_progress")
                    except Exception:
                        pass
                    task_by_session: dict[str, dict] = {}
                    for t in in_progress_tasks:
                        sid = t.get("session_id")
                        if sid and sid not in task_by_session:
                            task_by_session[sid] = t

                    coord_lines = [f"[COORD] {len(peers)} peer{'s' if len(peers) != 1 else ''} active on this project:"]
                    for p in peers:
                        p_sid = p["session_id"]
                        p_name = _agent_nickname(p_sid)
                        # Prefer coord_task over session.task
                        ct = task_by_session.get(p_sid)
                        if ct:
                            pct = f" [{ct['progress']}%]" if ct.get("progress") else ""
                            p_task = f"#{ct['id']} {ct['title'][:40]}{pct}"
                        else:
                            p_task = (p.get("task") or "idle")[:50]
                        # File claims
                        p_files = ""
                        try:
                            claims = mgr.get_session_claims(p_sid)
                            file_claims = claims.get("file_claims", [])
                            if file_claims:
                                fnames = [os.path.basename(f) for f in file_claims[:3]]
                                if len(file_claims) > 3:
                                    fnames.append(f"+{len(file_claims) - 3}")
                                p_files = f" [{', '.join(fnames)}]"
                        except Exception:
                            pass
                        coord_lines.append(f"  {p_name}: {p_task}{p_files}")
                    coord_output = "\n".join(coord_lines)
                _last_coord_query[session_id] = now
            except Exception:
                pass  # Coordination query is best-effort

    # Auto-classify intent (router integration)
    router_output = ""
    classified_intent = None
    try:
        from omega.router.classifier import classify_intent

        intent, confidence = classify_intent(prompt)
        if confidence >= 0.6:
            classified_intent = intent
            router_output = f"[ROUTER] Intent: {intent} ({confidence:.0%})"
            if session_id:
                _session_intent[session_id] = intent
    except ImportError:
        pass  # Router is optional
    except Exception:
        pass  # Classification is best-effort

    # Preference pattern matching (checked first — highest priority)
    preference_patterns = [
        r"\bi\s+(?:prefer|like|love|enjoy|favor|favour)\s+\w",
        r"\bmy\s+(?:preference|favorite|favourite|default)\b",
        r"\balways\s+use\b",
        r"\bi\s+(?:want|need)\s+(?:it|things?|everything)\s+(?:in|with|to\s+be)\b",
        r"\bremember\s+(?:that\s+)?i\s+(?:prefer|like|want|use|need)\b",
        r"\bdon'?t\s+(?:ever\s+)?(?:use|suggest|recommend)\b",
        r"\bi\s+(?:hate|dislike|avoid)\b",
    ]

    # Decision pattern matching
    decision_patterns = [
        r"\blet'?s?\s+(?:go\s+with|use|switch\s+to|stick\s+with|move\s+to)\b",
        r"\bi\s+(?:decided?|chose|picked|went\s+with)\b",
        r"\bwe\s+(?:should|will|are\s+going\s+to)\s+(?:use|go\s+with|switch|adopt|implement)\b",
        r"\b(?:decision|approach|strategy):\s*\S",
        r"\binstead\s+of\s+\S+[,\s]+(?:use|let'?s|we'?ll)\b",
        r"\bfrom\s+now\s+on\b",
        r"\bremember\s+(?:that|this)\b",
    ]

    # Lesson pattern matching
    lesson_patterns = [
        r"\bi\s+learned\s+that\b",
        r"\bturns?\s+out\b",
        r"\bthe\s+trick\s+is\b",
        r"\bnote\s+to\s+self\b",
        r"\btil\b|\btoday\s+i\s+learned\b",
        r"\bthe\s+fix\s+was\b",
        r"\bthe\s+problem\s+was\b",
        r"\bdon'?t\s+forget\b",
        r"\bimportant:\s*\S",
        r"\bkey\s+(?:insight|takeaway|learning)\b",
        r"\bnever\s+(?:again|do|use)\b",
        r"\balways\s+(?:make\s+sure|remember|check)\b",
    ]

    prompt_lower = prompt.lower()
    is_preference = any(re.search(pat, prompt_lower) for pat in preference_patterns)
    is_decision = any(re.search(pat, prompt_lower) for pat in decision_patterns)
    is_lesson = any(re.search(pat, prompt_lower) for pat in lesson_patterns)

    if not is_preference and not is_decision and not is_lesson:
        # Pass through any accumulated coord/router output even if no capture
        passthrough = "\n".join(filter(None, [coord_output, router_output]))
        return {"output": passthrough, "error": None}

    # Preference > Decision > Lesson priority
    if is_preference:
        event_type = "user_preference"
        content_prefix = "Preference"
    elif is_decision:
        event_type = "decision"
        content_prefix = "Decision"
    else:
        event_type = "lesson_learned"
        content_prefix = "Lesson"

    # Lesson quality gate: min 50 chars, >= 7 words
    # Pattern match already signals intent — no secondary tech signal required
    if event_type == "lesson_learned":
        if len(prompt) < 50 or len(prompt.split()) < 7:
            return {"output": "", "error": None}

    try:
        from omega.bridge import auto_capture

        meta = {"source": "auto_capture_hook", "project": cwd}
        if classified_intent:
            meta["intent"] = classified_intent
        auto_capture(
            content=f"{content_prefix}: {prompt[:500]}",
            event_type=event_type,
            metadata=meta,
            session_id=session_id,
            project=cwd,
            entity_id=entity_id,
        )
    except Exception as e:
        _log_hook_error("auto_capture", e)
        return {"output": "\n".join(filter(None, [coord_output, router_output])), "error": None}

    # User-visible confirmation of what was captured
    capture_line = f"[CAPTURED] {content_prefix.lower()}: {prompt[:80].replace(chr(10), ' ').strip()}"
    combined = "\n".join(filter(None, [capture_line, coord_output, router_output]))
    return {"output": combined, "error": None}


# ---------------------------------------------------------------------------
# Pre-push guard (git divergence + branch claims)
# ---------------------------------------------------------------------------


def _get_current_branch(project: str) -> str | None:
    """Get the current git branch name."""
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        return result.stdout.strip() if result.returncode == 0 else None
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return None


def _parse_checkout_target(command: str) -> str | None:
    """Parse the target branch from git checkout/switch commands.

    Returns None for new branch creation (-b/-B/-c/-C/--orphan)
    and file restores (-- separator).
    """
    for segment in re.split(r"&&|\|\||;", command):
        segment = segment.strip()
        match = re.match(r"git\s+(checkout|switch)\s+(.*)", segment)
        if not match:
            continue
        args_str = match.group(2).strip()

        if re.search(r"(?:^|\s)-[bBcC]\b", args_str):
            return None
        if "--orphan" in args_str:
            return None
        if " -- " in args_str or args_str.startswith("-- "):
            return None

        for token in args_str.split():
            if token.startswith("-"):
                continue
            return token

    return None


def handle_pre_push_guard(payload: dict) -> dict:
    """Git push divergence guard + branch claim check.

    Enforces:
      1. git push: blocks if origin has advanced (divergence guard)
      2. git checkout/switch: blocks if target branch is claimed by another agent
      3. git commit: blocks if current branch is claimed by another agent

    Returns exit_code=2 when blocking. Fail-open on errors.
    """
    tool_name = payload.get("tool_name", "")
    if tool_name != "Bash":
        return {"output": "", "error": None}

    input_data = _parse_tool_input(payload)
    command = input_data.get("command", "")
    session_id = payload.get("session_id", "")
    project = payload.get("project", "")

    if re.search(r"\bgit\s+push\b", command):
        result = _handle_push_divergence(command, project, session_id)
        if result.get("exit_code"):
            return result
        _handle_auto_claim_branch(command, session_id, project)
        return {"output": "", "error": None}

    if re.search(r"\bgit\s+(?:checkout|switch|commit)\b", command):
        return _handle_branch_claims(command, session_id, project)

    return {"output": "", "error": None}


def _handle_push_divergence(command: str, project: str, session_id: str) -> dict:
    """Check for push divergence. Returns exit_code=2 if origin has advanced."""
    if not project:
        return {"output": "", "error": None}

    try:
        result = subprocess.run(
            ["git", "rev-parse", "--is-inside-work-tree"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        if result.returncode != 0:
            return {"output": "", "error": None}

        subprocess.run(
            ["git", "fetch", "origin", "--quiet"],
            capture_output=True,
            text=True,
            timeout=15,
            cwd=project,
        )

        branch = _get_current_branch(project) or "main"

        behind_result = subprocess.run(
            ["git", "log", f"HEAD..origin/{branch}", "--oneline", "--no-decorate", "-10"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        if behind_result.returncode != 0 or not behind_result.stdout.strip():
            # Not behind — log push event
            _log_push_event(project, branch, session_id)
            return {"output": "", "error": None}

        behind_lines = behind_result.stdout.strip().split("\n")
        count = len(behind_lines)

        # Log divergence to coordination
        try:
            from omega.coordination import get_manager

            mgr = get_manager()
            mgr.log_git_event(
                project=project,
                event_type="push_divergence_warning",
                branch=branch,
                message=f"{count} upstream commit(s) detected before push",
                session_id=session_id,
            )
        except Exception:
            pass

        lines = [f"\n[GIT-GUARD] BLOCKED: origin/{branch} has {count} commit(s) not in HEAD:"]
        for line in behind_lines[:5]:
            lines.append(f"  {line}")
        if count > 5:
            lines.append(f"  ... and {count - 5} more")
        lines.append("  Run 'git pull --rebase' before pushing to avoid conflicts.")

        return {"output": "\n".join(lines), "error": None, "exit_code": 2}

    except (subprocess.TimeoutExpired, FileNotFoundError):
        return {"output": "", "error": None}
    except Exception as e:
        _log_hook_error("pre_push_guard", e)
        return {"output": "", "error": None}


def _handle_branch_claims(command: str, session_id: str, project: str) -> dict:
    """Check branch claims for checkout/switch/commit commands."""
    if not session_id:
        return {"output": "", "error": None}

    try:
        if "git commit" in command:
            branch = _get_current_branch(project)
            if branch:
                return _block_if_branch_claimed(session_id, project, branch)
            return {"output": "", "error": None}

        target = _parse_checkout_target(command)
        if target:
            return _block_if_branch_claimed(session_id, project, target)

    except Exception as e:
        _log_hook_error("branch_guard", e)

    return {"output": "", "error": None}


def _block_if_branch_claimed(session_id: str, project: str, branch: str) -> dict:
    """Block if the branch is claimed by another agent."""
    try:
        from omega.coordination import get_manager

        mgr = get_manager()
        info = mgr.check_branch(project, branch)

        if not info.get("claimed"):
            return {"output": "", "error": None}

        if info.get("session_id") == session_id:
            return {"output": "", "error": None}  # Self-claim

        owner_name = _agent_nickname(info.get("session_id", "unknown"))
        owner_task = info.get("task") or "unknown task"
        msg = (
            f"\n[BRANCH-GUARD] BLOCKED: branch '{branch}' is claimed by {owner_name} ({owner_task}).\n"
            f"  Options:\n"
            f"    1. Wait for the other agent to finish\n"
            f"    2. Ask other agent to call omega_branch_release\n"
            f"    3. Use a different feature branch"
        )
        return {"output": msg, "error": None, "exit_code": 2}

    except ImportError:
        return {"output": "", "error": None}
    except Exception as e:
        _log_hook_error("branch_guard", e)
        return {"output": "", "error": None}


def _handle_auto_claim_branch(command: str, session_id: str, project: str):
    """Auto-claim the current branch before a push succeeds."""
    if not session_id or not project or not os.path.isdir(project):
        return
    try:
        branch = _get_current_branch(project)
        if not branch or branch == "HEAD":
            return
        from omega.coordination import get_manager

        mgr = get_manager()
        mgr.claim_branch(project=project, branch=branch, session_id=session_id, task="pushing to remote")
    except Exception:
        pass


def _log_push_event(project: str, branch: str, session_id: str):
    """Log a push event to coordination."""
    try:
        from omega.coordination import get_manager

        mgr = get_manager()

        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=project,
        )
        commit_hash = result.stdout.strip() if result.returncode == 0 else None

        mgr.log_git_event(
            project=project,
            event_type="push",
            commit_hash=commit_hash,
            branch=branch,
            session_id=session_id,
        )
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Handler dispatch table — core handlers are always available; additional
# handlers (coordination, etc.) are loaded when present or provided by plugins.
# ---------------------------------------------------------------------------

# Core memory handlers — always shipped with omega-memory
_CORE_HOOK_HANDLERS = {
    "session_start": handle_session_start,
    "session_stop": handle_session_stop,
    "surface_memories": handle_surface_memories,
    "auto_capture": handle_auto_capture,
}

# Additional handlers — loaded when available, or provided by plugins
_EXTENDED_HOOK_HANDLERS = {
    "coord_session_start": handle_coord_session_start,
    "coord_session_stop": handle_coord_session_stop,
    "coord_heartbeat": handle_coord_heartbeat,
    "auto_claim_file": handle_auto_claim_file,
    "pre_file_guard": handle_pre_file_guard,
    "pre_task_guard": handle_pre_task_guard,
    "pre_push_guard": handle_pre_push_guard,
}

# Build the dispatch table: core + extended (if available) + plugins
HOOK_HANDLERS = dict(_CORE_HOOK_HANDLERS)

# Coordination handlers: only register if the coordination module is available.
# In core-only installs (PyPI omega), these remain defined but unreachable —
# hooks-core.json won't wire them, and this guard prevents accidental dispatch.
try:
    import omega.coordination  # noqa: F401
    HOOK_HANDLERS.update(_EXTENDED_HOOK_HANDLERS)
except ImportError:
    pass

# Discover plugin-provided hook handlers (e.g. omega-pro adds its own hooks)
try:
    from omega.plugins import discover_plugins

    for _plugin in discover_plugins():
        if _plugin.HOOK_HANDLERS:
            HOOK_HANDLERS.update(_plugin.HOOK_HANDLERS)
except Exception:
    pass


def register_hook_handler(name: str, handler):
    """Register a hook handler at runtime (for plugins)."""
    HOOK_HANDLERS[name] = handler


# ---------------------------------------------------------------------------
# UDS Server
# ---------------------------------------------------------------------------


async def handle_connection(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    """Handle a single hook client connection."""
    t0 = time.monotonic()
    hook_name = "unknown"
    try:
        # Read until EOF — client calls shutdown(SHUT_WR) after sendall()
        chunks = []
        while True:
            chunk = await asyncio.wait_for(reader.read(65536), timeout=10.0)
            if not chunk:
                break
            chunks.append(chunk)
        data = b"".join(chunks)
        if not data:
            writer.close()
            return

        request = json.loads(data.decode("utf-8").strip())

        # Batch mode: {"hooks": ["a", "b", ...], ...}
        # Single mode: {"hook": "a", ...}
        hook_names = request.pop("hooks", None)
        if hook_names:
            hook_name = "+".join(hook_names)
            loop = asyncio.get_running_loop()
            results = []
            for name in hook_names:
                handler = HOOK_HANDLERS.get(name)
                if not handler:
                    results.append({"output": "", "error": f"Unknown hook: {name}"})
                else:
                    r = await loop.run_in_executor(None, handler, request)
                    results.append(r)
                    # Short-circuit on block — skip remaining hooks
                    if r.get("exit_code"):
                        break
            response = {"results": results}
        else:
            hook_name = request.pop("hook", "unknown")
            handler = HOOK_HANDLERS.get(hook_name)
            if not handler:
                response = {"output": "", "error": f"Unknown hook: {hook_name}"}
            else:
                # Run handler in executor to avoid blocking the event loop
                loop = asyncio.get_running_loop()
                response = await loop.run_in_executor(None, handler, request)

        writer.write(json.dumps(response).encode("utf-8"))
        await writer.drain()
    except asyncio.TimeoutError:
        try:
            writer.write(json.dumps({"output": "", "error": "timeout"}).encode("utf-8"))
            await writer.drain()
        except Exception:
            pass
    except Exception as e:
        _log_hook_error(f"connection/{hook_name}", e)
        try:
            writer.write(json.dumps({"output": "", "error": str(e)}).encode("utf-8"))
            await writer.drain()
        except Exception:
            pass
    finally:
        elapsed_ms = (time.monotonic() - t0) * 1000
        _log_timing(hook_name, elapsed_ms)
        try:
            writer.close()
            await writer.wait_closed()
        except Exception:
            pass


_hook_server: asyncio.Server | None = None


async def start_hook_server() -> asyncio.Server | None:
    """Start the UDS hook server. Returns the server instance."""
    global _hook_server

    # Ensure directory exists
    SOCK_PATH.parent.mkdir(parents=True, exist_ok=True)

    # Remove stale socket from previous run
    if SOCK_PATH.exists():
        SOCK_PATH.unlink()

    try:
        _hook_server = await asyncio.start_unix_server(handle_connection, path=str(SOCK_PATH))
        # Make socket accessible
        SOCK_PATH.chmod(0o600)
        logger.info("Hook server listening on %s", SOCK_PATH)
        return _hook_server
    except Exception as e:
        logger.error("Failed to start hook server: %s", e)
        return None


async def stop_hook_server(srv: asyncio.Server | None = None):
    """Stop the hook server and clean up socket.

    Only deletes the socket file if this process owns the server,
    to avoid breaking another MCP server's active socket.
    """
    global _hook_server
    server = srv or _hook_server
    if server:
        server.close()
        await server.wait_closed()
        _hook_server = None

        # Only unlink if we were the ones serving
        if SOCK_PATH.exists():
            try:
                SOCK_PATH.unlink()
            except Exception:
                pass