Hooks

Where it lives

• hooks/compact_run.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  # /// script
  # requires-python = ">=3.12"
  # ///
  """
  compact_run.py — Lightweight output compressor for Claude Code (cross-platform)
  
  Runs a command, captures output, and applies compression:
    - Git ops:    success -> one-liner summary, failure -> full stderr
    - Log cmds:   dedup repeated lines + tail
    - Test cmds:  success -> summary line, failure -> failures only
  
  Install: Part of workflow-orchestrator plugin (hooks/compact_run.py)
  Called by token_rewrite_hook.py, never directly by Claude.
  """
  
  import io
  import os
  import re
  import subprocess
  import sys
  
  # Force UTF-8 output on Windows (fixes encoding errors)
  if sys.platform == "win32":
      sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
      sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
  
  # --- Config ---
  MAX_LINES = 150  # Truncation safety net
  LOG_TAIL = 50  # Max log lines to show
  LOG_DEDUP = True  # Deduplicate log lines
  CMD_TIMEOUT = int(
      os.environ.get("COMPACT_RUN_TIMEOUT", "120")
  )  # Max seconds (env-configurable)
  
  
  def truncated_output(content: str) -> str:
      """Apply truncation safety net."""
      lines = content.splitlines()
      total = len(lines)
      if total > MAX_LINES:
          truncated = lines[-MAX_LINES:]
          return (
              f"[truncated: {total} lines total, showing last {MAX_LINES}]\n"
              + "\n".join(truncated)
          )
      return content
  
  
  def emit_failure(stdout: str, stderr: str, exit_code: int) -> int:
      """On failure, show stderr + truncated stdout, then exit."""
      if stderr:
          print(stderr, file=sys.stderr)  # noqa: T201
      if stdout:
          print(truncated_output(stdout))  # noqa: T201
      return exit_code
  
  
  def handle_git(args: list[str], stdout: str, stderr: str, exit_code: int) -> int:
      """Handle git command compression."""
      if exit_code != 0:
          return emit_failure(stdout, stderr, exit_code)
  
      second = args[1] if len(args) > 1 else ""
      combined = stdout + stderr
  
      if second == "push":
          # Extract branch from "-> branch" pattern
          match = re.search(r"-> (\S+)", combined)
          if match:
              branch = match.group(1)
          else:
              result = subprocess.run(  # noqa: S603
                  ["git", "branch", "--show-current"],  # noqa: S607
                  capture_output=True,
                  text=True,
              )
              branch = result.stdout.strip() if result.returncode == 0 else "?"
          print(f"ok \u2192 {branch}")  # noqa: T201
  
      elif second == "pull":
          match = re.search(r"\s*(\d+ files? changed.*)", stdout)
          if match:
              print(f"ok \u2192 {match.group(1).strip()}")  # noqa: T201
          elif "Already up to date" in stdout:
              print("ok \u2192 already up to date")  # noqa: T201
          else:
              print("ok")  # noqa: T201
  
      elif second == "commit":
          match = re.search(r"\[.+ ([a-f0-9]{7,})\] (.+)", combined)
          if match:
              hash_val = match.group(1)
              msg = match.group(2)
              print(f'ok \u2192 {hash_val} "{msg}"')  # noqa: T201
          else:
              print("ok")  # noqa: T201
  
      elif second == "add":
          print("ok")  # noqa: T201
  
      elif second == "fetch":
          new_refs = len(re.findall(r"^\s*(From|\[new|->)", combined, re.MULTILINE))
          if new_refs > 0:
              print(f"ok \u2192 {new_refs} new refs")  # noqa: T201
          else:
              print("ok")  # noqa: T201
  
      elif second == "merge":
          if "Already up to date" in stdout:
              print("ok \u2192 already up to date")  # noqa: T201
          else:
              match = re.search(r"\s*(\d+ files? changed.*)", stdout)
              summary = match.group(1).strip() if match else "merged"
              print(f"ok \u2192 {summary}")  # noqa: T201
  
      elif second == "rebase":
          match = re.search(r"Successfully rebased.*", combined)
          summary = match.group(0) if match else "rebased"
          print(f"ok \u2192 {summary}")  # noqa: T201
  
      elif second == "stash":
          if re.search(r"saved working directory", stdout, re.IGNORECASE):
              print("ok \u2192 stashed")  # noqa: T201
          elif re.search(r"dropped", stdout, re.IGNORECASE):
              print("ok \u2192 dropped")  # noqa: T201
          elif re.search(r"no local changes", stdout, re.IGNORECASE):
              print("ok \u2192 nothing to stash")  # noqa: T201
          else:
              # stash list, stash show, etc. — pass through
              print(truncated_output(stdout))  # noqa: T201
  
      else:
          print(truncated_output(stdout))  # noqa: T201
  
      return exit_code
  
  
  def handle_container_logs(stdout: str, stderr: str, exit_code: int) -> int:
      """Handle docker/podman/kubectl logs compression."""
      if exit_code != 0:
          return emit_failure(stdout, stderr, exit_code)
  
      if LOG_DEDUP:
          # Deduplicate consecutive identical lines
          lines = stdout.splitlines()
          deduped: list[str] = []
          prev = None
          count = 0
          for line in lines:
              if line == prev:
                  count += 1
              else:
                  if count > 1:
                      deduped.append(f"  [repeated {count} times]")
                  if prev is not None or line != "":
                      deduped.append(line)
                  prev = line
                  count = 1
          if count > 1:
              deduped.append(f"  [repeated {count} times]")
  
          total = len(deduped)
          if total > LOG_TAIL:
              print(f"[showing last {LOG_TAIL} of {total} lines, duplicates collapsed]")  # noqa: T201
              print("\n".join(deduped[-LOG_TAIL:]))  # noqa: T201
          else:
              print("\n".join(deduped))  # noqa: T201
      else:
          lines = stdout.splitlines()
          print("\n".join(lines[-LOG_TAIL:]))  # noqa: T201
  
      return exit_code
  
  
  def handle_pytest(stdout: str, stderr: str, exit_code: int) -> int:
      """Handle pytest output compressi

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.