Coverage for astrocyte/harness/hooks.py: 96%
355 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""``astrocyte hook <event> [--host codex]`` — agent lifecycle hooks for automatic memory.
3================= ======================================================
4``session-start`` Starts the agent daemon (waits for it: once per session)
5 and injects a short summary of this project's memory.
6``prompt`` Injects memories relevant to the prompt — gated on
7 similarity, deduplicated per session, hard deadline. A
8 trivial prompt never reaches the daemon at all.
9``stop`` Spools the turn that just finished (~0.35 s; synchronous,
10 because headless `claude -p` kills background hooks).
11``edit`` Codex: notes the files an ``apply_patch`` changed, for
12 the turn ``stop`` captures.
13``file`` Claude Code, opt-in: memories of earlier turns that
14 touched the file the agent just read or edited.
15================= ======================================================
17Claude Code and Codex fire these under the same event names with the same
18input and output contract. They differ in where the finished turn comes from:
19Claude Code's Stop points at a transcript we read incrementally; Codex's Stop
20carries the reply itself (``last_assistant_message``), so the prompt is kept
21from ``prompt`` until the turn ends.
23Contract with the agent: always exit 0, print nothing but valid hook JSON,
24never load a model in this process, never wait past the deadline. A broken or
25missing daemon means "no memory this turn", never a blocked prompt.
27Interactive sessions only, by default. Hooks are global, so they also fire for
28every headless ``claude -p`` / ``codex exec`` on the machine: scripts,
29benchmarks, and LLM provider calls (including Astrocyte's own ``claude_cli``
30provider). Capturing those pollutes memory, injecting into them perturbs
31automated runs, and the provider case recursed (observed on first real
32install). A session whose agent process runs non-interactively is left alone.
34``ASTROCYTE_HOOKS=off`` pauses everything; ``ASTROCYTE_HOOKS=all`` also
35serves headless sessions.
36"""
38from __future__ import annotations
40import json
41import logging
42import os
43import re
44import subprocess
45import sys
46from collections.abc import Callable
47from dataclasses import dataclass
48from datetime import datetime, timezone
49from pathlib import Path
51from . import agentd
52from .paths import config_path, state_dir
53from .project import project_bank, project_root
54from .transcript import Turn, antigravity_prompt, read_new_turns
56SESSION_START_WAIT = 8.0 # cold start incl. model load measured at ~1.5 s
57PROMPT_DEADLINE = 2.5 # the user is waiting; warm recall measured at ~10 ms
59_WORD = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.\-/]+")
60_SMALL_TALK = frozenset(
61 "the a an and or but if is are was were be to of in on at for with this that it its "
62 "please thanks thank you ok okay yes no sure great good nice cool lgtm continue go ahead "
63 "do it can could would should will i me my we our us hi hello hey".split()
64)
66logger = logging.getLogger("astrocyte.hooks")
69def worth_recalling(prompt: str) -> bool:
70 """Skip the daemon for prompts with nothing to recall against —
71 "thanks!", "continue", slash commands."""
72 text = prompt.strip()
73 if not text or text.startswith("/"):
74 return False
75 terms = [w for w in _WORD.findall(text.lower()) if w not in _SMALL_TALK and len(w) > 2]
76 return len(terms) >= 2
79# Codex options that consume the next argument, so ``codex -m o3 exec`` is
80# read as subcommand ``exec`` rather than ``o3``.
81_CODEX_VALUE_FLAGS = frozenset(
82 "-c --config -m --model -p --profile -s --sandbox -a --ask-for-approval -C --cd -i --image "
83 "--enable --disable --add-dir --local-provider".split()
84)
85# `app-server` is deliberately absent: it backs Codex's desktop app and IDE
86# extension, which are interactive.
87_CODEX_HEADLESS = frozenset({"exec", "e", "review", "mcp-server"})
90def _claude_headless(argv: list[str]) -> bool:
91 # The desktop app and the interactive CLI never pass -p/--print
92 # (verified against the desktop app's own invocation).
93 return any(a in ("-p", "--print") for a in argv[1:])
96def _codex_headless(argv: list[str]) -> bool:
97 args = iter(argv[1:])
98 for arg in args:
99 if arg in _CODEX_VALUE_FLAGS:
100 next(args, None)
101 elif not arg.startswith("-"):
102 return arg in _CODEX_HEADLESS
103 return False
106def _print_flag_headless(argv: list[str]) -> bool:
107 """agy and copilot: -p / --print / --prompt run one prompt and exit."""
108 return any(a in ("-p", "--print", "--prompt") for a in argv[1:])
111def _claude_emit(event: str, context: str) -> None:
112 if context:
113 print(json.dumps({"hookSpecificOutput": {"hookEventName": event, "additionalContext": context}}))
116def _copilot_emit(_event: str, context: str) -> None:
117 if context:
118 print(json.dumps({"additionalContext": context}))
121def _antigravity_emit(_event: str, context: str) -> None:
122 if context: # main() prints {} when nothing was emitted
123 print(json.dumps({"injectSteps": [{"ephemeralMessage": context}]}))
126@dataclass(frozen=True)
127class Dialect:
128 """How one harness speaks the hook contract."""
130 source: str # recorded on every captured memory
131 binary: str # process name of the agent, found above the hook
132 headless: Callable[[list[str]], bool] # agent argv → runs non-interactively?
133 # Where the finished turn comes from at Stop: Claude Code's transcript,
134 # or the payload itself (the prompt kept from UserPromptSubmit).
135 turn_from_payload: bool
136 emit: Callable[[str, str], None] = _claude_emit
137 # Antigravity has no prompt-submitted event: its PreInvocation hook runs
138 # before every model call and carries no prompt, so the prompt is read
139 # from the transcript and recall runs once per new user message.
140 prompt_from_transcript: bool = False
141 antigravity_transcript: bool = False
142 captures_turns: bool = True # False: this agent's hooks only recall
144 @property
145 def turn_source(self) -> str | None:
146 """"payload", "transcript", or None when turns aren't captured."""
147 if not self.captures_turns:
148 return None
149 return "payload" if self.turn_from_payload else "transcript"
152DIALECTS: dict[str, Dialect] = {
153 "claude": Dialect("claude-code", "claude", _claude_headless, turn_from_payload=False),
154 "codex": Dialect("codex", "codex", _codex_headless, turn_from_payload=True),
155}
157# Added after v0.16.0. Kept out of DIALECTS, whose released value is part of
158# the public API; look dialects up with dialect_for().
159ANTIGRAVITY_DIALECT = Dialect(
160 "antigravity", "agy", _print_flag_headless, turn_from_payload=False,
161 emit=_antigravity_emit, prompt_from_transcript=True, antigravity_transcript=True,
162)
163COPILOT_DIALECT = Dialect(
164 "copilot", "copilot", _print_flag_headless, turn_from_payload=False, emit=_copilot_emit, captures_turns=False,
165)
166_MORE_DIALECTS = {"antigravity": ANTIGRAVITY_DIALECT, "copilot": COPILOT_DIALECT}
169def dialect_for(host: str) -> Dialect | None:
170 """The hook dialect of ``host`` (``astrocyte hook --host``), if known."""
171 return DIALECTS.get(host) or _MORE_DIALECTS.get(host)
174def _agent_argv(argv: list[str], binary: str) -> list[str] | None:
175 """``argv`` if it is the ``binary`` agent, run directly or by Node."""
176 if argv and binary in Path(argv[0]).name.lower():
177 return argv
178 # Node CLIs run as `node /path/to/copilot …`; on Windows npm's shims run
179 # `node …/node_modules/@anthropic-ai/claude-code/cli.js`, so the agent's
180 # name is in the script's path rather than its file name.
181 if len(argv) > 1 and Path(argv[0]).name.lower().startswith("node"):
182 parts = re.split(r"[\\/]", argv[1].lower())
183 if any(binary in part for part in parts):
184 return argv[1:]
185 return None
188def _agent_ancestor_args(binary: str, max_depth: int = 5) -> list[str] | None:
189 """argv of the nearest ``binary`` process above this hook, if any.
191 Hooks may be run through a shell, so walk a few levels up. Environment
192 variables can't answer "is this headless?": a ``claude -p`` started from
193 an interactive session inherits ``CLAUDE_CODE_SESSION_ATTENDED=1`` and
194 ``CLAUDE_CODE_ENTRYPOINT`` unchanged (measured).
195 """
196 walk = _windows_ancestor_args if os.name == "nt" else _posix_ancestor_args
197 return walk(binary, max_depth)
200def _posix_ancestor_args(binary: str, max_depth: int) -> list[str] | None:
201 pid = os.getppid()
202 for _ in range(max_depth):
203 if pid <= 1:
204 return None
205 try:
206 out = subprocess.run(["ps", "-o", "ppid=", "-o", "args=", "-p", str(pid)],
207 capture_output=True, text=True, timeout=2, check=False).stdout.strip()
208 except (OSError, subprocess.SubprocessError):
209 return None
210 if not out:
211 return None
212 ppid, _, args = out.partition(" ")
213 if found := _agent_argv(args.split(), binary):
214 return found
215 try:
216 pid = int(ppid)
217 except ValueError:
218 return None
219 return None
222def _windows_ancestor_args(binary: str, max_depth: int) -> list[str] | None:
223 """Windows has no `ps`; psutil reads each parent's command line (the
224 headless flags live there, which process names alone don't show)."""
225 try:
226 import psutil
227 except ImportError: # a Windows install without the dependency: treat as interactive
228 return None
229 try:
230 proc = psutil.Process(os.getppid())
231 for _ in range(max_depth):
232 if found := _agent_argv(proc.cmdline(), binary):
233 return found
234 proc = proc.parent()
235 if proc is None:
236 return None
237 except (psutil.Error, OSError):
238 return None
239 return None
242def headless_session(dialect: str = "claude") -> bool:
243 """Is the agent session running non-interactively (``claude -p``,
244 ``codex exec``, ``agy -p``, ``copilot -p``)? The Antigravity app runs
245 no ``agy`` process, so its sessions are interactive."""
246 d = dialect_for(dialect)
247 if d is None:
248 return False
249 argv = _agent_ancestor_args(d.binary)
250 return bool(argv) and d.headless(argv)
253def _provider_scratch(cwd: str) -> bool:
254 """Provider subprocesses run in a private ``astrocyte-claude-cli-*`` temp
255 dir. Older provider code didn't disable hooks, so recognise it directly."""
256 return Path(cwd).name.startswith("astrocyte-claude-cli-")
259def _session_file(session_id: str, suffix: str = "") -> Path:
260 safe = re.sub(r"[^A-Za-z0-9._-]", "_", session_id) or "unknown"
261 return state_dir() / "sessions" / f"{safe}{suffix}.json"
264def _write_state(path: Path, data: dict) -> None:
265 path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
266 path.write_text(json.dumps(data), encoding="utf-8")
269def _session_start(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
270 if not agentd.ensure_running(cfg, wait=SESSION_START_WAIT):
271 return
272 reply = agentd.request(
273 "boot", {"bank": bank, "session_id": session, "source": payload.get("source")}, timeout=5
274 )
275 dialect.emit("SessionStart", (reply or {}).get("context", ""))
278def _latest_user_message(transcript: object) -> tuple[int | None, str]:
279 """The last user message in an Antigravity transcript, with its step index."""
280 if not isinstance(transcript, str) or not transcript:
281 return None, ""
282 step, text = None, ""
283 try:
284 with open(transcript, encoding="utf-8") as fh:
285 for raw in fh:
286 try:
287 line = json.loads(raw)
288 except ValueError:
289 continue
290 if isinstance(line, dict) and (prompt := antigravity_prompt(line)) is not None:
291 step, text = line.get("step_index"), prompt
292 except OSError:
293 return None, ""
294 return step, text
297def _prompt_from_transcript(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
298 """Antigravity: PreInvocation runs before every model call of a turn, so
299 act once per new user message; and since there is no session-start event,
300 the project summary rides on a conversation's first model call."""
301 step, prompt = _latest_user_message(payload.get("transcriptPath") or payload.get("transcript_path"))
302 if step is None:
303 return
304 marker = _session_file(session, ".asked")
305 try:
306 first_sight = False
307 if json.loads(marker.read_text(encoding="utf-8")).get("step") == step:
308 return # a later model call in the same turn
309 except (OSError, ValueError, AttributeError):
310 first_sight = True
311 _write_state(marker, {"step": step})
312 parts: list[str] = []
313 if first_sight and agentd.ensure_running(cfg, wait=SESSION_START_WAIT):
314 boot = agentd.request("boot", {"bank": bank, "session_id": session}, timeout=5)
315 parts.append((boot or {}).get("context", ""))
316 if worth_recalling(prompt):
317 reply = agentd.request("recall", {"bank": bank, "session_id": session, "prompt": prompt},
318 timeout=PROMPT_DEADLINE)
319 if reply is None:
320 agentd.spawn(cfg)
321 else:
322 parts.append(reply.get("context", ""))
323 dialect.emit("UserPromptSubmit", "\n\n".join(p for p in parts if p))
326def _prompt(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
327 if dialect.prompt_from_transcript:
328 return _prompt_from_transcript(payload, cfg, bank, session, dialect)
329 prompt = payload.get("prompt") or ""
330 if dialect.turn_source == "payload" and prompt.strip() and agentd.supported():
331 # Kept for Stop, which reports only the reply. Every prompt, even
332 # one too slight to recall against: it is still half of a turn.
333 _write_state(_session_file(session, ".prompt"),
334 {"prompt": redact_secrets(prompt), "at": datetime.now(timezone.utc).isoformat()})
335 _session_file(session, ".files").unlink(missing_ok=True) # an unfinished turn's edits
336 if not worth_recalling(prompt):
337 return
338 reply = agentd.request("recall", {"bank": bank, "session_id": session, "prompt": prompt}, timeout=PROMPT_DEADLINE)
339 if reply is None:
340 agentd.spawn(cfg) # warm for the next prompt; never make this one wait
341 return
342 dialect.emit("UserPromptSubmit", reply.get("context", ""))
345def _noop() -> None:
346 pass
349# Each source returns the turns to capture and a commit that marks them taken.
350# The commit runs only after the spool write succeeded: a crash in between
351# captures the same turn again rather than losing it.
352TurnSource = tuple[list[Turn], Callable[[], None]]
355def _turns_from_transcript(payload: dict, session: str, *, antigravity: bool = False) -> TurnSource:
356 transcript = payload.get("transcript_path") or payload.get("transcriptPath")
357 if not transcript:
358 return [], _noop
359 marker = _session_file(session)
360 try:
361 offset = int(json.loads(marker.read_text(encoding="utf-8"))["offset"])
362 first_sight = False
363 except (OSError, ValueError, KeyError, TypeError):
364 offset, first_sight = 0, True
365 turns, resume = read_new_turns(transcript, offset, antigravity=antigravity)
366 if first_sight:
367 # A session we have never seen — already running when the hooks were
368 # installed (Claude Code hot-reloads settings), or resumed from before
369 # — would otherwise have its entire history captured in one burst
370 # (observed: 359 turns from one session in seconds). Memory starts now.
371 turns = turns[-1:]
372 if resume == offset:
373 return turns, _noop
374 return turns, lambda: _write_state(marker, {"offset": resume, "transcript": transcript})
377def _turn_from_payload(payload: dict, session: str) -> TurnSource:
378 pending = _session_file(session, ".prompt")
379 edited = _session_file(session, ".files")
381 def commit() -> None:
382 pending.unlink(missing_ok=True)
383 edited.unlink(missing_ok=True)
385 try:
386 kept = json.loads(pending.read_text(encoding="utf-8"))
387 except (OSError, ValueError):
388 return [], _noop # the prompt predates the hooks, or this turn was already taken
389 answer = payload.get("last_assistant_message")
390 if not isinstance(answer, str) or not answer.strip() or not isinstance(kept, dict):
391 return [], commit
392 try:
393 started = datetime.fromisoformat(kept.get("at") or "")
394 except (TypeError, ValueError):
395 started = None
396 try:
397 files = list(dict.fromkeys(line for line in edited.read_text(encoding="utf-8").splitlines() if line))
398 except OSError:
399 files = []
400 return [Turn(user=str(kept.get("prompt") or ""), assistant=[answer], started_at=started, files=files)], commit
403def redact_secrets(text: str) -> str:
404 """The retain barrier's credential scrub, imported only when a turn is
405 kept: the barrier module pulls in every type (~20 ms on each hook)."""
406 from astrocyte.policy.barriers import redact_secrets as scrub
408 return scrub(text)
411# Kept on each captured memory: enough to tie a turn to its code, small
412# enough for the metadata limit (4 KB).
413MAX_TOUCHED_FILES = 20
414MAX_TOUCHED_CHARS = 1_500
417def _project_root_of(payload: dict) -> Path | None:
418 workspaces = payload.get("workspacePaths") # Antigravity names the workspace, not a cwd
419 cwd = payload.get("cwd") or (workspaces[0] if isinstance(workspaces, list) and workspaces else None)
420 return project_root(cwd) if isinstance(cwd, str) and cwd else None
423def _shown_paths(files: list[str], root: Path | None) -> list[str]:
424 """Touched files, relative to the project where they're inside it (so a
425 teammate's clone at another path names them the same), capped."""
426 shown, used = [], 0
427 for raw in files[:MAX_TOUCHED_FILES]:
428 path = Path(raw)
429 if root is not None:
430 try:
431 path = path.resolve().relative_to(root.resolve())
432 except (ValueError, OSError):
433 path = Path(raw) # outside the project: as the agent gave it
434 text = path.as_posix()
435 if used + len(text) + 1 > MAX_TOUCHED_CHARS:
436 break
437 shown.append(text)
438 used += len(text) + 1
439 return shown
442def _stop(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
443 if not agentd.supported() or dialect.turn_source is None:
444 return # without a daemon to drain it, the spool would only grow
445 if dialect.turn_source == "payload":
446 turns, commit = _turn_from_payload(payload, session)
447 else:
448 turns, commit = _turns_from_transcript(payload, session, antigravity=dialect.antigravity_transcript)
449 if turns:
450 # Scrubbed before it touches disk: the retain barrier redacts again,
451 # but the spool is plaintext and outlives a crashed daemon.
452 root = _project_root_of(payload)
453 agentd.spool_capture(
454 bank, session, dialect.source,
455 [{"content": redact_secrets(t.render()), "started_at": t.started_at.isoformat() if t.started_at else None,
456 "files": _shown_paths(t.files, root)}
457 for t in turns],
458 )
459 commit()
460 if turns and agentd.request("capture", timeout=0.5) is None:
461 agentd.spawn(cfg)
464def _file(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
465 """After a file tool (opt-in): memories of earlier turns that touched this
466 file, added next to the tool's result. Claude Code only for now."""
467 if dialect.source != "claude-code" or not agentd.supported():
468 return
469 tool_input = payload.get("tool_input") or {}
470 raw = tool_input.get("file_path") or tool_input.get("notebook_path") if isinstance(tool_input, dict) else None
471 if not isinstance(raw, str) or not raw:
472 return
473 [path] = _shown_paths([raw], _project_root_of(payload)) or [None]
474 if not path:
475 return
476 reply = agentd.request("file", {"bank": bank, "session_id": session, "path": path}, timeout=PROMPT_DEADLINE)
477 if reply is None:
478 agentd.spawn(cfg) # warm for the next file; never make this one wait
479 return
480 dialect.emit("PostToolUse", reply.get("context", ""))
483# The file lines of an apply_patch envelope (Codex's only file-writing tool).
484_PATCH_FILE = re.compile(r"^\*\*\* (?:Add File|Update File|Delete File|Move to): (.+?)\s*$", re.MULTILINE)
487def patch_paths(patch: str) -> list[str]:
488 """The files an ``apply_patch`` envelope adds, updates, deletes or moves to."""
489 return list(dict.fromkeys(_PATCH_FILE.findall(patch)))
492def _edit(payload: dict, cfg: Path, bank: str, session: str, dialect: Dialect) -> None:
493 """After a Codex ``apply_patch`` succeeds: note the files it changed, for
494 Stop to keep on the turn. Codex reads files through shell commands, which
495 name no path reliably, so only edits are recorded."""
496 if dialect.turn_source != "payload" or payload.get("tool_name") != "apply_patch" or not agentd.supported():
497 return
498 tool_input = payload.get("tool_input")
499 patch = tool_input.get("command") if isinstance(tool_input, dict) else None
500 if not isinstance(patch, str):
501 return
502 cwd = payload.get("cwd")
503 base = Path(cwd) if isinstance(cwd, str) and cwd else None
504 paths = [str(base / p) if base is not None and not Path(p).is_absolute() else p for p in patch_paths(patch)]
505 if not paths:
506 return
507 marker = _session_file(session, ".files") # one path per line, despite the name
508 marker.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
509 # Appended, one path per line: patches applied in parallel each add theirs.
510 with marker.open("a", encoding="utf-8") as f:
511 f.write("".join(f"{p}\n" for p in paths))
514_HANDLERS = {"session-start": _session_start, "prompt": _prompt, "stop": _stop, "file": _file, "edit": _edit}
517PONG = {"astrocyte": "pong"}
520def _is_ping(stdin_text: str) -> bool:
521 try:
522 payload = json.loads(stdin_text or "{}")
523 except ValueError:
524 return False
525 return isinstance(payload, dict) and payload.get("ping") is True
528def run(event: str, stdin_text: str, host: str = "claude") -> int:
529 # `astrocyte doctor` runs each registered command through the shells an
530 # agent may use, with {"ping": true}: answering proves the command line
531 # reached this code, without touching memory. Before ASTROCYTE_HOOKS, so a
532 # paused install still checks out.
533 if _is_ping(stdin_text):
534 print(json.dumps(PONG))
535 return 0
536 mode = os.environ.get("ASTROCYTE_HOOKS", "").strip().lower()
537 if mode in ("0", "off", "false", "no"):
538 return 0
539 handler = _HANDLERS.get(event)
540 dialect = dialect_for(host)
541 if handler is None or dialect is None:
542 return 0
543 try:
544 payload = json.loads(stdin_text or "{}")
545 if not isinstance(payload, dict):
546 return 0
547 cfg = config_path()
548 if not cfg.is_file():
549 return 0 # not set up: stay out of the way
550 workspaces = payload.get("workspacePaths")
551 cwd = str(payload.get("cwd") or (workspaces[0] if isinstance(workspaces, list) and workspaces else "")
552 or os.getcwd())
553 if _provider_scratch(cwd) or (mode != "all" and headless_session(host)):
554 return 0
555 session = str(payload.get("session_id") or payload.get("sessionId") or payload.get("conversationId")
556 or "unknown")
557 bank = project_bank(cwd)
558 handler(payload, cfg, bank, session, dialect)
559 except Exception: # noqa: BLE001 — a hook must never break the agent
560 try:
561 d = state_dir()
562 d.mkdir(parents=True, exist_ok=True, mode=0o700)
563 # A handler of our own, not basicConfig: that is a no-op once
564 # anything has configured the root logger.
565 handler = logging.FileHandler(d / "hooks.log")
566 handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
567 logger.addHandler(handler)
568 try:
569 logger.exception("hook %s failed", event)
570 finally:
571 logger.removeHandler(handler)
572 handler.close()
573 except Exception: # noqa: BLE001
574 pass
575 return 0
578def main(event: str, host: str = "claude") -> int:
579 if host == "antigravity":
580 # Antigravity parses every hook's stdout as JSON: say "nothing" with {}
581 # whenever the hook had nothing to add (or stayed out of the way).
582 import contextlib
583 import io
585 out = io.StringIO()
586 with contextlib.redirect_stdout(out):
587 code = run(event, sys.stdin.read() if not sys.stdin.isatty() else "", host)
588 print(out.getvalue().strip() or "{}")
589 return code
590 return run(event, sys.stdin.read() if not sys.stdin.isatty() else "", host)