Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279"""Kernel-local Python callbacks and user-level Claude Code command hooks."""
from __future__ import annotations
import asyncioimport hashlibimport inspectimport jsonimport osimport reimport signalimport subprocessimport tempfilefrom collections.abc import Callablefrom concurrent.futures import ThreadPoolExecutorfrom pathlib import Path
import albedo_proc
EVENTS = ( "SessionStart", "SessionEnd", "UserPromptSubmit", "PreToolUse", "PostToolUse", "Stop",)Callback = Callable[[dict], object]_callbacks: list[tuple[str, str | None, Callback]] = []enabled: bool = Falsestarted: bool = False # a prompt has been seen: this kernel is a live session_session, _cwd = "", None
def register(event: str, callback: Callback, matcher: str | None = None): if event not in EVENTS or not callable(callback): raise ValueError("expected a supported event and callable callback") if matcher not in (None, "", "*"): re.compile(matcher) entry = (event, matcher, callback) _callbacks.append(entry)
def remove(): if entry in _callbacks: _callbacks.remove(entry)
return remove
registered, clear = _callbacks.copy, _callbacks.clear
def _matches(matcher, subject): return matcher in (None, "", "*") or re.search(matcher, subject) is not None
def _empty() -> dict: return dict(block=None, halt=None, updated_input=None, context=[], messages=[])
def _merge(answer, extra): for key in ("block", "halt", "updated_input"): if answer[key] is None: answer[key] = extra[key] for key in ("context", "messages"): answer[key].extend(extra[key])
def _clip(text): return text if len(text) <= 10000 else text[:9988] + "\n[truncated]"
def _normal(output): if not isinstance(output, dict): raise ValueError("hook output must be an object") answer = _empty() specific = output.get("hookSpecificOutput") or {} if output.get("continue") is False: answer["halt"] = str(output.get("stopReason") or "stopped by a hook") if output.get("decision") == "block": answer["block"] = str(output.get("reason") or "blocked by a hook") if specific.get("permissionDecision") == "deny": answer["block"] = str(specific.get("permissionDecisionReason") or "denied") if isinstance(specific.get("updatedInput"), dict): answer["updated_input"] = specific["updatedInput"] if specific.get("additionalContext"): answer["context"] = [_clip(str(specific["additionalContext"]))] if output.get("systemMessage"): answer["messages"] = [str(output["systemMessage"])] return answer
def _settled(value): if not inspect.isawaitable(value): return value
async def wait(): return await value
with ThreadPoolExecutor(max_workers=1) as pool: return pool.submit(asyncio.run, wait()).result()
def _capture(command, payload, timeout): """Run one hook command: the payload on stdin, output in temp files so a backgrounded child holding a pipe cannot hang us, the whole process group killed afterwards.""" with tempfile.TemporaryFile() as stdin, tempfile.TemporaryFile() as out: with tempfile.TemporaryFile() as err: stdin.write(json.dumps(payload).encode()) stdin.seek(0) proc = subprocess.Popen( command, shell=True, cwd=payload["cwd"], stdin=stdin, stdout=out, stderr=err, start_new_session=True, ) group = albedo_proc.Group(proc.pid, albedo_proc.leader_token(proc.pid)) try: proc.wait(timeout) except subprocess.TimeoutExpired: raise TimeoutError("command hook timed out") from None finally: albedo_proc.send_signal(group, signal.SIGKILL) proc.wait() out.seek(0) err.seek(0) text = lambda f: f.read(1 << 20).decode("utf-8", errors="replace") # noqa: E731 return proc.returncode, text(out), text(err)
def _command(definition, payload): try: if definition.get("type") != "command" or not isinstance( definition.get("command"), str ): raise ValueError("expected a command hook") timeout = definition.get("timeout", 600) if payload["hook_event_name"] == "SessionEnd": timeout = min(timeout, 3) # the kernel is shutting down if type(timeout) not in (int, float) or not 0 < timeout < float("inf"): raise ValueError("timeout must be a positive finite number") status, stdout, stderr = _capture(definition["command"], payload, timeout) answer = _empty() if status == 2: answer["block"] = stderr.strip() or "blocked by a hook" elif status != 0: raise ValueError(f"command hook exited {status}: {stderr.strip()}") elif stdout.strip(): try: output = json.loads(stdout) except ValueError: if payload["hook_event_name"] not in ( "UserPromptSubmit", "SessionStart", ): raise ValueError("invalid hook JSON") from None answer["context"] = [_clip(stdout.strip())] else: answer = _normal(output) return answer except Exception as error: return {**_empty(), "messages": [f"Hook failed open: {error}"]}
_warned: set[str] = set()
def _approved(path, answer) -> bool: """Whether the user approved this exact workspace hooks file: its sha256 sits in `hooks.sha256` beside it, which the user's `/hooks` page writes.""" digest = hashlib.sha256(path.read_bytes()).hexdigest() seal = path.with_name("hooks.sha256") if seal.exists() and seal.read_text().strip() == digest: return True if digest not in _warned: _warned.add(digest) answer["messages"].append( "This workspace's hooks are not approved, so they are skipped. " "Review and approve them with /hooks." ) return False
def fire(event: str, subject: str, fields: dict, session="", cwd=None) -> dict: payload = dict( fields, session_id=session or _session, cwd=cwd or _cwd or os.getcwd(), hook_event_name=event, ) payload.update(transcript_path="", permission_mode="default") answer = _empty() for kind, matcher, callback in list(_callbacks): if kind == event and _matches(matcher, subject): try: value = _settled(callback(dict(payload))) _merge(answer, _normal(value) if value is not None else _empty()) if answer["updated_input"] is not None: payload["tool_input"] = answer["updated_input"] except Exception as error: answer["messages"].append(f"Callback failed open: {error}") if answer["block"] is not None or answer["halt"] is not None: return answer home = Path(os.environ.get("ALBEDO_HOME") or Path.home() / ".albedo") workspace = home / "workspaces" / re.sub(r"[^A-Za-z0-9]", "-", payload["cwd"]) definitions = [] # The workspace's hooks come first, so they win conflicts. Its file is # model-writable, so (as in Codex) it runs only once its hash is approved; # the user's global hooks are their own and always run. for path, own in ((workspace / "hooks.json", False), (home / "hooks.json", True)): try: if not path.exists() or not (own or _approved(path, answer)): continue for group in json.loads(path.read_text()).get("hooks", {}).get(event, []): if _matches(group.get("matcher", ""), subject): definitions.extend(group["hooks"]) except Exception as error: answer["messages"].append(f"Hook config {path} failed open: {error}") with ThreadPoolExecutor(max_workers=8) as pool: for extra in pool.map(lambda d: _command(d, payload), definitions): _merge(answer, extra) return answer
def fire_json(event, subject, fields_json, session, cwd): global _session, _cwd, started _session, _cwd = session, cwd answer = fire(event, subject, json.loads(fields_json), session, cwd) if event == "UserPromptSubmit" and not started: # The first prompt of a kernel is when a session starts: SessionStart # context comes first, and it cannot block. started = True first = fire("SessionStart", "startup", {"source": "startup"}, session, cwd) first["block"] = None _merge(first, answer) answer = first if answer["updated_input"] is not None: answer["updated_input"] = json.dumps(answer["updated_input"]) return json.dumps(answer)
def end(): """SessionEnd, as the kernel shuts down.""" if started: fire("SessionEnd", "other", {"reason": "other"})
def _lines(event, answer): texts = answer["context"] + answer["messages"] texts += [v for v in (answer["block"], answer["halt"]) if v] return [f"[{event} hook] {text}" for text in texts]
def pre(tool, tool_input): if not enabled: return tool_input answer = fire("PreToolUse", tool, dict(tool_name=tool, tool_input=tool_input)) for line in _lines("PreToolUse", answer): print(line) if answer["halt"] or answer["block"]: raise PermissionError(f"blocked by a hook: {answer['halt'] or answer['block']}") updated = answer["updated_input"] return tool_input if updated is None else updated
def post_lines(tool, tool_input, response): fields = dict(tool_name=tool, tool_input=tool_input, tool_response=response) return _lines("PostToolUse", fire("PostToolUse", tool, fields))
def post(tool, tool_input, response): if enabled: for line in post_lines(tool, tool_input, response): print(line)