ae44be31d5
Includes AirPlan design documents, AircOding-alpha1-plan, AirPlanV2, AirPlan-ParaV2, AirPlan-Para V1 reference docs, and all working code changes across packages. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
107 lines
3.7 KiB
Python
Executable File
107 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""PostToolUse hook — lightweight ticker.
|
|
|
|
Responsibilities (all must complete in well under 100 ms):
|
|
1. Read AirContext/state.json. If paused, in-progress, or in cooldown, return.
|
|
2. Estimate active-chain token usage from transcript_path.
|
|
3. If usage / model_context_window >= trigger.threshold, fork compactor.py
|
|
in the background and return immediately.
|
|
|
|
The compactor runs detached so it never blocks Claude's main loop.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
_HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(_HERE))
|
|
|
|
from core.config_loader import load_config # noqa: E402
|
|
from core.state import StateFile # noqa: E402
|
|
from core.token_estimator import estimate_active_chain_tokens # noqa: E402
|
|
|
|
|
|
def _project_root(payload: dict) -> Path:
|
|
cwd = payload.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
|
|
return Path(cwd)
|
|
|
|
|
|
def _spawn_compactor(project: Path, transcript: Path, session_id: str) -> None:
|
|
"""Detach compactor.py so the hook returns immediately."""
|
|
compactor = _HERE / "core" / "compactor.py"
|
|
args = [sys.executable, str(compactor),
|
|
"--project", str(project),
|
|
"--transcript", str(transcript),
|
|
"--session-id", session_id]
|
|
log_dir = project / "AirContext" / "snapshots"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / "compactor.log"
|
|
|
|
if os.name == "nt":
|
|
DETACHED = 0x00000008 # DETACHED_PROCESS
|
|
NEW_GROUP = 0x00000200
|
|
creationflags = DETACHED | NEW_GROUP
|
|
with log_file.open("ab") as lf:
|
|
subprocess.Popen(args, stdout=lf, stderr=lf, stdin=subprocess.DEVNULL,
|
|
creationflags=creationflags, close_fds=True)
|
|
else:
|
|
with log_file.open("ab") as lf:
|
|
subprocess.Popen(args, stdout=lf, stderr=lf, stdin=subprocess.DEVNULL,
|
|
start_new_session=True, close_fds=True)
|
|
|
|
|
|
def main() -> int:
|
|
raw = sys.stdin.read() or "{}"
|
|
try:
|
|
payload = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return 0
|
|
|
|
project = _project_root(payload)
|
|
air = project / "AirContext"
|
|
config_path = air / "config.yaml"
|
|
if not config_path.exists():
|
|
return 0 # uninitialised — nothing to do
|
|
|
|
state = StateFile(air / "state.json")
|
|
snap = state.read()
|
|
if snap.get("paused") or snap.get("compaction_in_progress") or snap.get("compaction_ready"):
|
|
return 0
|
|
|
|
try:
|
|
cfg = load_config(config_path)
|
|
except Exception:
|
|
return 0 # config broken — UserPromptSubmit will surface it
|
|
|
|
cooldown = int(cfg.get("trigger", {}).get("cooldown_seconds", 300))
|
|
if time.time() - snap.get("last_compaction_unix", 0) < cooldown:
|
|
return 0
|
|
|
|
transcript = payload.get("transcript_path")
|
|
if not transcript or not Path(transcript).exists():
|
|
return 0
|
|
session_id = payload.get("session_id") or snap.get("last_session_id")
|
|
if not session_id:
|
|
return 0
|
|
|
|
threshold = float(cfg.get("trigger", {}).get("threshold", 0.6))
|
|
window = int(cfg.get("trigger", {}).get("model_context_window", 200000))
|
|
method = cfg.get("trigger", {}).get("estimate_method", "char_div_3_5")
|
|
|
|
used = estimate_active_chain_tokens(Path(transcript), method=method)
|
|
if used / max(window, 1) < threshold:
|
|
return 0
|
|
|
|
# Mark in-progress immediately so successive PostToolUse calls don't double-fire.
|
|
state.update(compaction_in_progress=True, last_session_id=session_id)
|
|
_spawn_compactor(project, Path(transcript), session_id)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|