Hooks

Where it lives

• hooks/evidence_gate.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Evidence gate Stop hook. Blocks session end when the EXECUTE phase has
  in-progress tasks with incomplete CDD evidence (fewer evidence_files than
  subtasks).
  
  Reads Stop JSON from stdin, returns a decision on stdout.
  Never crashes — all parsing wrapped in try/except.
  Short-circuits when stop_hook_active is True to avoid infinite block loops
  when Claude Code re-invokes Stop hooks after a block decision.
  No explicit process termination — main() returns and the process ends naturally.
  """
  import json
  import sys
  from pathlib import Path
  
  
  def main():
      try:
          payload = json.loads(sys.stdin.read())
      except Exception:
          # never crash the hook — bad stdin is a no-op (implicit allow)
          print(json.dumps({}))
          return
  
      try:
          # Short-circuit: if Claude Code is re-invoking Stop hooks after a block,
          # always allow to prevent infinite loops.
          if payload.get("stop_hook_active") is True:
              print(json.dumps({}))
              return
  
          pipeline_path = Path(".atlas-ai/state/pipeline.json")
          if not pipeline_path.is_file():
              print(json.dumps({}))
              return
  
          try:
              pipeline = json.loads(pipeline_path.read_text())
          except Exception:
              print(json.dumps({}))
              return
  
          if pipeline.get("current_phase") != "EXECUTE":
              print(json.dumps({}))
              return
  
          tasks_path = Path(".atlas-ai/taskmaster/tasks/tasks.json")
          if not tasks_path.is_file():
              print(json.dumps({}))
              return
  
          try:
              tasks_data = json.loads(tasks_path.read_text())
          except Exception:
              print(json.dumps({}))
              return
  
          tasks = tasks_data.get("master", {}).get("tasks", []) or []
          offenders = []
          for t in tasks:
              if not isinstance(t, dict):
                  continue
              if t.get("status") != "in-progress":
                  continue
              subtasks = t.get("subtasks") or []
              evidence_files = t.get("evidence_files") or []
              if len(subtasks) > 0 and len(evidence_files) < len(subtasks):
                  offenders.append(
                      f"task {t.get('id')}: {len(evidence_files)}/{len(subtasks)} evidence"
                  )
  
          if offenders:
              reason = (
                  f"Incomplete CDD evidence — {len(offenders)} tasks have subtasks "
                  f"without matching evidence: {'; '.join(offenders)}. "
                  f"Run execute-task loop to completion or mark blocked."
              )
              print(json.dumps({"decision": "block", "reason": reason}))
          else:
              print(json.dumps({}))
      except Exception:
          # any unexpected failure — allow, never crash the hook
          print(json.dumps({}))
          return
  
  
  if __name__ == "__main__":
      main()
  

• hooks/gate_enforcer.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Gate enforcer hook. Blocks mcp__...__advance_phase calls without gate evidence.
  Reads PreToolUse JSON from stdin, returns permissionDecision on stdout.
  Never crashes — all JSON parsing wrapped in try/except.
  No explicit process termination — main() returns and the process ends naturally.
  """
  import json
  import sys
  
  
  def main():
      try:
          payload = json.loads(sys.stdin.read())
      except Exception:
          # never crash the hook — bad stdin is a no-op (implicit allow)
          print(json.dumps({}))
          return
  
      tool_name = payload.get("tool_name", "")
      if not tool_name.endswith("__advance_phase"):
          print(json.dumps({}))
          return
  
      tool_input = payload.get("tool_input", {})
      target = tool_input.get("target")
      evidence = tool_input.get("evidence", {})
  
      # Mirror the same gate logic as pipeline.check_gate (simplified for hook)
      violations = []
      if target == "SETUP":
          vs = evidence.get("validate_setup", {})
          if not vs.get("ready") or vs.get("critical_failures", 1) > 0:
              violations.append("validate_setup must report ready=true and critical_failures==0; evidence must contain validate_setup block")
      elif target == "DISCOVER":
          if not (
              evidence.get("user_approved")
              or (
                  evidence.get("auto_classification") == "CLEAR"
                  and evidence.get("assumptions_documented")
              )
          ):
              violations.append(
                  "DISCOVER gate requires user_approved=true OR (auto_classification==CLEAR AND assumptions_documented=true); evidence missing"
              )
      elif target == "GENERATE":
          if evidence.get("validation_grade") not in ("EXCELLENT", "GOOD"):
              violations.append("validation_grade must be EXCELLENT or GOOD; evidence insufficient")
          if evidence.get("task_count", 0) == 0:
              violations.append("task_count must be > 0; evidence shows no tasks parsed")
          if evidence.get("subtask_coverage", 0) < 1.0:
              violations.append("subtask_coverage must be >= 1.0; evidence shows subtasks incomplete")
      elif target == "HANDOFF":
          if not evidence.get("user_mode_choice"):
              violations.append("user_mode_choice missing from evidence; user must select execution mode before HANDOFF")
          if not evidence.get("plan_file_exists"):
              violations.append("plan_file_exists missing from evidence; plan file must be written before HANDOFF")
      elif target == "EXECUTE":
          # EXECUTE is terminal — individual tasks track own status; no gate check needed
          pass
  
      if violations:
          reason = f"Gate for {target} not passed: {'; '.join(violations)}"
          print(json.dumps({
              "hookSpecificOutput": {
                  "hookEventName": "PreToolUse",
                  "permissionDecision": "deny",
                  "permissionDecisionReason": reason,
              }
          }))
      else:
          print(json.dumps({}))
  
  
  if __name__ == "__main__":
      main()
  

• hooks/mode_d_blocker.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Mode D blocker hook. Blocks mcp__atlas-cdd and mcp__atlas-loop tool calls.
  Mode D (full CDD + atlas-phoenix) is preview-alpha and requires waitlist opt-in.
  Reads PreToolUse JSON from stdin, returns permissionDecision on stdout.
  Never crashes — all JSON parsing wrapped in try/except.
  No explicit process termination — main() returns and the process ends naturally.
  """
  import json
  import sys
  
  
  def main():
      try:
          payload = json.loads(sys.stdin.read())
      except Exception:
          # never crash the hook — bad stdin is a no-op (implicit allow)
          print(json.dumps({}))
          return
  
      tool_name = payload.get("tool_name", "")
  
      # Check if tool is from blocked Mode D namespaces
      if tool_name.startswith("mcp__atlas-cdd__") or tool_name.startswith("mcp__atlas-loop__"):
          reason = (
              "Mode D (full CDD + atlas-phoenix integration) is preview-alpha and requires waitlist opt-in. "
              "Please join the waitlist at https://atlas-ai.au/waitlist/mode-d to access this feature."
          )
          print(json.dumps({
              "hookSpecificOutput": {
                  "hookEventName": "PreToolUse",
                  "permissionDecision": "deny",
                  "permissionDecisionReason": reason,
              }
          }))
      else:
          print(json.dumps({}))
  
  
  if __name__ == "__main__":
      main()
  

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.