Hooks

Where it lives

• hooks/assistant_capture.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA Stop hook fallback — capture high-value assistant responses.
  
  Fires on every Stop event when the hook daemon is unavailable.
  Detects fix, decision, lesson, and recommendation patterns in
  ``last_assistant_message`` and stores them via bridge.auto_capture.
  
  This is the cold-path fallback. The fast path routes through the hook
  daemon to ``handle_assistant_capture`` in the hook_server package.
  """
  import json
  import re
  import sys
  
  # Patterns (mirrors assistant.py in hook_server)
  FIX_PATTERNS = [
      r"the (?:fix|issue|problem|bug) was\b",
      r"root cause (?:was|is)\b",
      r"the error (?:occurred|happens|was caused) because\b",
      r"fixed (?:by|this by)\b",
  ]
  
  DECISION_PATTERNS = [
      r"(?:decided|choosing) to\b",
      r"going with\b",
      r"switched to\b",
      r"using \S+ instead of\b",
      r"chose \S+ because\b",
  ]
  
  LESSON_PATTERNS = [
      r"(?:note|notice) that\b",
      r"important:\s",
      r"be careful\b",
      r"gotcha:\s",
      r"caveat:\s",
      r"key takeaway\b",
  ]
  
  MIN_MESSAGE_LENGTH = 200
  MIN_CONTENT_CHARS = 40
  MIN_CONTENT_WORDS = 8
  
  _FENCED_CODE_RE = re.compile(r"```[\s\S]*?```", re.DOTALL)
  _SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+|\n+")
  _INSIGHT_OPEN_RE = re.compile(r"[★✦]\s*Insight\s*─+", re.IGNORECASE)
  _INSIGHT_CLOSE_RE = re.compile(r"─{10,}")
  
  _captured_count = 0
  MAX_CAPTURES = 10
  
  
  def _clean(text):
      return _FENCED_CODE_RE.sub("", text).strip()
  
  
  def _find_match(text, patterns):
      for pat in patterns:
          compiled = re.compile(pat, re.IGNORECASE)
          for sentence in _SENTENCE_SPLIT_RE.split(text):
              sentence = sentence.strip()
              if compiled.search(sentence):
                  if len(sentence) >= MIN_CONTENT_CHARS and len(sentence.split()) >= MIN_CONTENT_WORDS:
                      return sentence
      return None
  
  
  def _extract_insight_blocks(text):
      """Extract ★ Insight delimited blocks from assistant text."""
      blocks = []
      search_start = 0
      while True:
          open_match = _INSIGHT_OPEN_RE.search(text, search_start)
          if not open_match:
              break
          body_start = open_match.end()
          close_match = _INSIGHT_CLOSE_RE.search(text, body_start)
          if not close_match:
              body = text[body_start:body_start + 2000].strip()
          else:
              body = text[body_start:close_match.start()].strip()
          if body and len(body) >= MIN_CONTENT_CHARS:
              blocks.append(body[:2000])
          search_start = close_match.end() if close_match else len(text)
      return blocks
  
  
  def main(data=None):
      global _captured_count
      if _captured_count >= MAX_CAPTURES:
          return
  
      if data is None:
          try:
              raw = sys.stdin.read()
              if not raw.strip():
                  return
              data = json.loads(raw)
          except (json.JSONDecodeError, Exception):
              return
  
      message = data.get("last_assistant_message", "")
      if not message or len(message) < MIN_MESSAGE_LENGTH:
          return
  
      session_id = data.get("session_id", "")
      cwd = data.get("cwd", data.get("project", ""))
  
      # Pre-pass: detect ★ Insight delimited blocks
      insight_blocks = _extract_insight_blocks(message)
      if insight_blocks and _captured_count < MAX_CAPTURES:
          for block in insight_blocks:
              if _captured_count >= MAX_CAPTURES:
                  break
              try:
                  from omega.bridge import auto_capture
  
                  auto_capture(
                      content=f"Insight: {block}",
                      event_type="advisor_insight",
                      metadata={"source": "assistant_capture_hook", "project": cwd, "capture_confidence": "high", "category": "system_insight"},
                      session_id=session_id,
                      project=cwd,
                  )
                  _captured_count += 1
                  preview = block[:80].replace("\n", " ").strip()
                  print(f"[LEARNED] insight: {preview}")
              except ImportError:
                  pass
              except Exception:
                  pass
          return
  
      cleaned = _clean(message)
      if not cleaned:
          return
  
      # Try pattern groups in priority order
      for label, event_type, patterns in [
          ("fix", "lesson_learned", FIX_PATTERNS),
          ("decision", "decision", DECISION_PATTERNS),
          ("lesson", "lesson_learned", LESSON_PATTERNS),
      ]:
          content = _find_match(cleaned, patterns)
          if content:
              try:
                  from omega.bridge import auto_capture
  
                  auto_capture(
                      content=f"Assistant {label}: {content[:500]}",
                      event_type=event_type,
                      metadata={"source": "assistant_capture_hook", "project": cwd},
                      session_id=session_id,
                      project=cwd,
                  )
                  _captured_count += 1
                  preview = content[:80].replace("\n", " ").strip()
                  print(f"[LEARNED] {label}: {preview}")
              except ImportError:
                  pass
              except Exception:
                  pass
              return
  
  
  if __name__ == "__main__":
      main()
  

• hooks/auto_capture.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA UserPromptSubmit hook — Auto-capture decisions and lessons from user prompts.
  
  Fires on every user prompt. Detects decision and lesson patterns and stores them
  as 'decision' or 'lesson_learned' event type in OMEGA memory. Uses conservative
  matching to avoid noise.
  """
  import json
  import re
  import sys
  
  
  # Decision indicators (case-insensitive patterns)
  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|prefer)\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 indicators (case-insensitive patterns)
  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",
  ]
  
  # Minimum prompt length to avoid matching on short commands
  MIN_PROMPT_LENGTH = 20
  
  # Maximum prompts to process per session (avoid runaway storage)
  _captured_count = 0
  MAX_CAPTURES_PER_SESSION = 20
  
  
  def _summarize_content(prompt: str, max_len: int = 60) -> str:
      """Extract a concise summary from the prompt for the echo line."""
      # Strip common prefixes like "Decision: " or "Lesson: "
      text = re.sub(r"^(Decision|Lesson):\s*", "", prompt, flags=re.IGNORECASE).strip()
      # Take first sentence or first max_len chars
      first_sentence = re.split(r"[.!?\n]", text)[0].strip()
      if len(first_sentence) <= max_len:
          return first_sentence
      return first_sentence[:max_len].rsplit(" ", 1)[0] + "..."
  
  
  def _echo_capture(result: str, event_type: str, prompt: str):
      """Print a 1-line capture confirmation visible to the user.
  
      Parses bridge.auto_capture() return value to distinguish:
      - New capture → [OMEGA] Captured: decision about X
      - Evolution   → [OMEGA] Memory evolved: added insight to existing memory
      - Dedup/Block → silent (no output)
      """
      if not result:
          return
  
      summary = _summarize_content(prompt)
  
      if "Memory Evolved" in result:
          # Extract evolution number from "Evolution #N"
          evo_match = re.search(r"Evolution #(\d+)", result)
          evo_num = evo_match.group(1) if evo_match else "?"
          print(f"[OMEGA] Memory evolved: {event_type} updated (evolution #{evo_num}) — {summary}")
      elif "Memory Captured" in result:
          print(f"[OMEGA] Captured: {event_type} — {summary}")
      # Dedup/Blocked → stay silent
  
  
  def _detect_decision(prompt: str) -> bool:
      """Check if prompt contains a decision pattern."""
      if len(prompt) < MIN_PROMPT_LENGTH:
          return False
      prompt_lower = prompt.lower()
      return any(re.search(pat, prompt_lower) for pat in DECISION_PATTERNS)
  
  
  def _detect_lesson(prompt: str) -> bool:
      """Check if prompt contains a lesson/insight pattern."""
      if len(prompt) < MIN_PROMPT_LENGTH:
          return False
      prompt_lower = prompt.lower()
      return any(re.search(pat, prompt_lower) for pat in LESSON_PATTERNS)
  
  
  def main():
      global _captured_count
      if _captured_count >= MAX_CAPTURES_PER_SESSION:
          return
  
      # Read hook input from stdin
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return
          data = json.loads(raw)
      except (json.JSONDecodeError, Exception):
          return
  
      prompt = data.get("prompt", "")
      session_id = data.get("session_id", "")
      cwd = data.get("cwd", "")
  
      if not prompt:
          return
  
      # Decision takes priority if both match
      if _detect_decision(prompt):
          try:
              from omega.bridge import auto_capture
              result = auto_capture(
                  content=f"Decision: {prompt[:500]}",
                  event_type="decision",
                  metadata={"source": "auto_capture_hook", "project": cwd},
                  session_id=session_id,
                  project=cwd,
              )
              _captured_count += 1
              _echo_capture(result, "decision", prompt)
          except ImportError:
              pass
          except Exception:
              pass
          return
  
      if _detect_lesson(prompt):
          # Lesson quality gate: min 60 chars, >= 8 words, substance validation
          if len(prompt) < 60 or len(prompt.split()) < 8:
              return
          _tech_signals = ["/", "`", "Error", "error", ".py", ".js", ".ts", "import ", "def ", "class "]
          if len(prompt) < 100 and not any(s in prompt for s in _tech_signals):
              return
  
          try:
              from omega.bridge import auto_capture
              result = auto_capture(
                  content=f"Lesson: {prompt[:500]}",
                  event_type="lesson_learned",
                  metadata={"source": "auto_capture_hook", "project": cwd},
                  session_id=session_id,
                  project=cwd,
              )
              _captured_count += 1
              _echo_capture(result, "lesson", prompt)
          except ImportError:
              pass
          except Exception:
              pass
  
  
  if __name__ == "__main__":
      main()
  

• hooks/auto_claim_file.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA Coordination PostToolUse hook — Auto-claim files on Edit/Write.
  
  Fires after Edit|Write. Automatically claims the edited file so other agents
  see it as taken, without requiring explicit omega_file_claim calls.
  """
  import json
  import os
  import traceback
  from datetime import datetime
  from pathlib import Path
  
  
  def _log_hook_error(hook_name, error):
      try:
          log_path = Path.home() / ".omega" / "hooks.log"
          log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
          timestamp = datetime.now().isoformat(timespec="seconds")
          tb = traceback.format_exc()
          data = f"[{timestamp}] {hook_name}: {error}\n{tb}\n"
          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)
      except Exception:
          pass
  
  
  def main():
      tool_name = os.environ.get("TOOL_NAME", "")
  
      session_id = os.environ.get("SESSION_ID", "")
      if not session_id:
          return
  
      try:
          input_data = json.loads(os.environ.get("TOOL_INPUT", "{}"))
      except (json.JSONDecodeError, TypeError):
          return
  
      file_path = input_data.get("file_path", input_data.get("notebook_path", ""))
      if not file_path:
          return
  
      # On Read: warn if the file is claimed by another agent (no claim, just visibility)
      if tool_name == "Read":
          try:
              from omega.coordination import get_manager
              mgr = get_manager()
              info = mgr.check_file(file_path)
              if info.get("claimed") and info.get("session_id") != session_id:
                  owner = info.get("session_id", "unknown")[:16]
                  owner_task = info.get("task") or "unknown task"
                  print(
                      f"[FILE-INFO] {os.path.basename(file_path)} is being edited by "
                      f"session {owner} ({owner_task}). Coordinate before modifying."
                  )
          except (ImportError, Exception):
              pass
          return
  
      # On Edit/Write/NotebookEdit: auto-claim
      if tool_name not in ("Edit", "Write", "NotebookEdit"):
          return
  
      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 = result["claimed_by"][:20]
              owner_task = result.get("task") or "unknown task"
              print(
                  f"[CONFLICT] {os.path.basename(file_path)} is claimed by session "
                  f"{owner} ({owner_task}). Coordinate before editing."
              )
          elif result.get("success"):
              # 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
      except ImportError:
          pass
      except Exception as e:
          error_str = str(e)
          if "already claimed" not in error_str.lower():
              _log_hook_error("auto_claim_file", e)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/coord_heartbeat.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA Coordination PostToolUse hook — Update session heartbeat."""
  import os
  import time
  import traceback
  from datetime import datetime
  from pathlib import Path
  
  
  def _log_hook_error(hook_name, error):
      try:
          log_path = Path.home() / ".omega" / "hooks.log"
          log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
          timestamp = datetime.now().isoformat(timespec="seconds")
          tb = traceback.format_exc()
          data = f"[{timestamp}] {hook_name}: {error}\n{tb}\n"
          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)
      except Exception:
          pass
  
  
  def main():
      session_id = os.environ.get("SESSION_ID", "")
      if not session_id:
          return
  
      try:
          from omega.coordination import get_manager
          mgr = get_manager()
          mgr.heartbeat(session_id)
      except ImportError:
          pass
      except Exception as e:
          _log_hook_error("coord_heartbeat", e)
  
  
  def _log_timing(hook_name, elapsed_ms):
      try:
          log_path = Path.home() / ".omega" / "hooks.log"
          log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
          timestamp = datetime.now().isoformat(timespec="seconds")
          data = f"[{timestamp}] {hook_name}: OK ({elapsed_ms:.0f}ms)\n"
          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)
      except Exception:
          pass
  
  
  if __name__ == "__main__":
      _t0 = time.monotonic()
      main()
      _log_timing("coord_heartbeat", (time.monotonic() - _t0) * 1000)
  

• hooks/coord_session_start.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA Coordination SessionStart hook — Register agent session."""
  import os
  import time
  import traceback
  from datetime import datetime, timezone
  from pathlib import Path
  
  
  def _log_hook_error(hook_name, error):
      try:
          log_path = Path.home() / ".omega" / "hooks.log"
          log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
          timestamp = datetime.now().isoformat(timespec="seconds")
          tb = traceback.format_exc()
          data = f"[{timestamp}] {hook_name}: {error}\n{tb}\n"
          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)
      except Exception:
          pass
  
  
  def _kill_orphaned_mcp_servers():
      """Kill OMEGA MCP server processes whose parent has exited (PPID=1)."""
      import signal
      import subprocess
  
      try:
          # Find all omega MCP server processes
          result = subprocess.run(
              ["pgrep", "-f", "omega.server.mcp_server"],
              capture_output=True, text=True, timeout=5,
          )
          if result.returncode != 0 or not result.stdout.strip():
              return
  
          my_pid = os.getpid()
          pids = [int(p) for p in result.stdout.strip().split("\n") if p.strip()]
  
          killed = 0
          for pid in pids:
              if pid == my_pid:
                  continue
              # Check if this process is orphaned (PPID=1 on macOS means parent exited)
              ps_result = subprocess.run(
                  ["ps", "-o", "ppid=", "-p", str(pid)],
                  capture_output=True, text=True, timeout=5,
              )
              if ps_result.returncode != 0:
                  continue
              ppid = ps_result.stdout.strip()
              if ppid == "1":
                  try:
                      os.kill(pid, signal.SIGTERM)
                      killed += 1
                  except ProcessLookupError:
                      pass
                  except PermissionError:
                      pass
  
          if killed > 0:
              _log_hook_error("orphan_cleanup", f"Killed {killed} orphaned MCP server(s)")
  
      except subprocess.TimeoutExpired:
          pass
      except FileNotFoundError:
          pass  # pgrep not available
      except Exception as e:
          _log_hook_error("orphan_cleanup", e)
  
  
  def _clean_stale_socket():
      """Delete hook.sock if it exists but nothing is listening."""
      import socket as _socket
      sock_path = os.path.expanduser("~/.omega/hook.sock")
      if not os.path.exists(sock_path):
          return
      s = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
      s.settimeout(1.0)
      try:
          s.connect(sock_path)
          s.close()  # Socket is alive, leave it
      except (ConnectionRefusedError, OSError, _socket.timeout):
          try:
              os.unlink(sock_path)
          except OSError:
              pass
      finally:
          try:
              s.close()
          except Exception:
              pass
  
  
  def main():
      session_id = os.environ.get("SESSION_ID", "")
      project = os.environ.get("PROJECT_DIR", os.getcwd())
  
      if not session_id:
          return
  
      _clean_stale_socket()
      _kill_orphaned_mcp_servers()
  
      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.getppid(),
              project=project,
          )
          peers = result.get("peers_on_project", 0)
  
          # Rich peer roster (matches daemon quality)
          if peers > 0:
              print(f"[STANDUP] {peers} peer{'s' if peers != 1 else ''} active")
              _print_peer_roster(session_id, project, mgr)
  
          # Surface structured handoff from predecessor (takes priority)
          _surface_handoff(session_id, project, mgr)
  
          # Surface unread inbox messages (show content, not just count)
          if peers > 0:
              _surface_inbox(session_id, mgr)
  
          # Surface recent decisions from other sessions (catch-up mechanism)
          _surface_recent_peer_decisions(session_id, project)
  
          # Git sync check: detect upstream commits from uncoordinated agents
          # Run in background — git fetch is network I/O that shouldn't block session start
          import threading
          threading.Thread(
              target=_check_git_sync,
              args=(session_id, project, mgr),
              daemon=True,
          ).start()
  
          # Check for predecessor session snapshots
          _session_resume(session_id, project, mgr)
  
          # Surface running background processes (benchmarks, long scripts)
          _check_running_processes(project)
  
          # Surface pending tasks for this project, with staleness detection
          try:
              tasks = mgr.list_tasks(project=project, status="pending")
              if tasks:
                  # Cross-reference tasks against recent decisions to detect completed ones
                  stale_ids = set()
                  try:
                      from omega import bridge
                      for t in tasks[:5]:
                          title = t.get("title", "")
                          if not title:
                              continue
                          result_str = bridge.query(
                              title, limit=2, event_type="decision",
                              project=project,
                          )
                          if result_str and "No matching memories" not in result_str:
                              lower = result_str.lower()
                              if any(w in lower for w in [
                                  "committed", "shipped", "deployed", "done",
                                  "completed", "merged", "published", "live",
                                  "verified", "set up", "configured", "added",
                              ]):
                                  stale_ids.add(t["id"])
                  except Exce

• hooks/coord_session_stop.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """OMEGA Coordination Stop hook — Deregister session and release all claims."""
  import os
  import sys
  import time
  import traceback
  from datetime import datetime
  from pathlib import Path
  
  
  def _log_hook_error(hook_name, error):
      try:
          log_path = Path.home() / ".omega" / "hooks.log"
          log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
          timestamp = datetime.now().isoformat(timespec="seconds")
          tb = traceback.format_exc()
          data = f"[{timestamp}] {hook_name}: {error}\n{tb}\n"
          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)
      except Exception:
          pass
  
  
  def _broadcast_session_end(session_id, project, mgr):
      """Broadcast session summary to active peers before deregistering."""
      try:
          # Check if there are active peers to notify
          sessions = mgr.list_sessions(auto_clean=False)
          peers = [s for s in sessions if s.get("session_id") != session_id]
          if not peers:
              return
  
          # Build a compact summary of what this session did
          summary = _build_end_summary(session_id, project)
          if not summary:
              return
  
          # Broadcast to project (all active peers will see it in their inbox)
          mgr.send_message(
              from_session=session_id,
              subject=f"Session ended: {summary}",
              msg_type="complete",
              project=project,
              body=summary,
              ttl_minutes=120,  # 2 hours — enough for next session start
          )
      except Exception as e:
          _log_hook_error("broadcast_session_end", e)
  
  
  def _build_end_summary(session_id, project):
      """Build a compact summary of session activity for the broadcast."""
      try:
          from omega.bridge import query_structured
      except ImportError:
          return None
  
      decisions = query_structured(
          query_text="decisions made",
          limit=3,
          session_id=session_id,
          project=project,
          event_type="decision",
      )
  
      if not decisions:
          return None
  
      items = []
      for d in decisions[:3]:
          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
          first_line = content.split("\n")[0].strip()
          if first_line and len(first_line) > 10:
              items.append(first_line[:120])
  
      if not items:
          return None
  
      return "; ".join(items)[:400]
  
  
  def _auto_handoff(session_id, project, mgr):
      """Auto-generate a structured handoff from session state. Best-effort."""
      try:
          from omega.bridge import query_structured
  
          # Gather decisions from this session
          decisions_raw = query_structured(
              query_text="decisions made",
              limit=5,
              session_id=session_id,
              project=project,
              event_type="decision",
          )
          decisions = []
          for d in (decisions_raw or []):
              content = d.get("content", "")
              for prefix in ("Plan/decision captured: ", "Decision: "):
                  if content.startswith(prefix):
                      content = content[len(prefix):]
              stripped = content.lstrip()
              if stripped.startswith(("{", "[", '"filePath')):
                  continue
              first_line = content.split("\n")[0].strip()
              if first_line and len(first_line) > 10:
                  decisions.append(first_line[:200])
              if len(decisions) >= 5:
                  break
  
          # Gather completed OMEGA tasks
          completed = []
          try:
              tasks = mgr.list_tasks(project=project, status="completed")
              for t in (tasks or [])[:5]:
                  if t.get("session_id") == session_id:
                      completed.append(t["title"])
          except Exception:
              pass
  
          mgr.create_handoff(
              session_id=session_id,
              project=project,
              completed_tasks=completed or None,
              decisions_made=decisions or None,
          )
      except ImportError:
          pass
      except Exception as e:
          _log_hook_error("auto_handoff", e)
  
  
  def _nudge_handoff(session_id, project, mgr):
      """Nudge agent about incomplete work being orphaned."""
      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]
          if my_tasks:
              task_list = ", ".join(f"#{t['id']} {t['title']}" for t in my_tasks[:3])
              print(
                  f"[HANDOFF] Active work returned to queue: {task_list}\n"
                  "  Next time, use omega_handoff(action='create', ...) before ending "
                  "to give your successor structured context."
              )
      except Exception as e:
          _log_hook_error("nudge_handoff", e)
  
  
  def _extract_project_entity(file_path):
      """Extract project name from a file path."""
      import re
      match = re.search(r'/Projects/([^/]+)', file_path)
      return match.group(1) if match else None
  
  
  def _build_entity_links(claims, current_project):
      """Build cross-project entity links from file claims."""
      projects = set()
      for claim in claims:
          proj = _extract_project_entity(claim.get("file_path", ""))
          if proj and proj != current_project:
              projects.add(proj)
  
      return [
          {"from": current_project, "to": proj, "relationship": "depends_on"}
          for proj in sorted(projects)
      ]
  
  
  def _detect_drift(original_task, files_modified, commits):
      """Detect if session work drifted from declared task via keyword overlap."""
      if n

• hooks/fast_hook.pyGitHub ↗
• hooks/post_edit_test.pyGitHub ↗
• hooks/pre_alignment_gate.pyGitHub ↗
• hooks/pre_commit_guard.pyGitHub ↗
• hooks/pre_deploy_guard.pyGitHub ↗
• hooks/pre_edit_surface.pyGitHub ↗
• hooks/pre_file_guard.pyGitHub ↗
• hooks/pre_protocol_gate.pyGitHub ↗
• hooks/pre_push_guard.pyGitHub ↗
• hooks/pre_review.pyGitHub ↗
• hooks/pre_task_guard.pyGitHub ↗
• hooks/session_start.pyGitHub ↗
• hooks/session_stop.pyGitHub ↗
• hooks/surface_memories.pyGitHub ↗
• hooks/trace_capture.pyGitHub ↗
• hooks/track_file_read.pyGitHub ↗

All 22 scripts are listed above. The source is inlined for 6 of them, starting with whatever hooks.json actually runs. See all of them in the repo.

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.