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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247"""Hooks that gate the kernel's own tools: `run` and `files`.
The daemon fires hooks around the model's outer tool calls. A cell that calls`run()` or `files.edit()` hides those calls inside one `python` call, so thekernel asks the daemon's `hooks.fire` route itself, before and after, withthe Claude Code / Codex tool names scripts already match on: `Bash`, `Read`,`Edit`, `Write`. Nothing happens unless the hooks extension installed ablocking host call here."""
from __future__ import annotations
import asyncioimport inspectimport reimport threadingfrom collections.abc import Awaitable, Callablefrom typing import cast
_sync: Callable[[str, dict[str, object]], object] | None = None
# Python callables the model registered through `hooks.register`: they run in# the kernel, before the daemon's command hooks, and die with the kernel.Callback = Callable[[dict[str, object]], object]_callbacks: list[tuple[str, str | None, Callback]] = []
EVENTS = ("PreToolUse", "PostToolUse", "UserPromptSubmit", "Stop")
def register( event: str, callback: Callback, matcher: str | None = None) -> Callable[[], None]: """Add a callback for `event` on tools matching the regex `matcher`; answers the function that removes it.""" if event not in EVENTS: raise ValueError(f"event is one of {', '.join(EVENTS)}, not {event!r}") if not callable(callback): raise TypeError("callback must be callable") if matcher is not None: re.compile(matcher) entry = (event, matcher, callback) _callbacks.append(entry)
def remove() -> None: if entry in _callbacks: _callbacks.remove(entry)
return remove
def registered() -> list[tuple[str, str | None, Callback]]: return list(_callbacks)
def clear() -> None: _callbacks.clear()
def _matches(matcher: str | None, tool: str) -> bool: return matcher in (None, "", "*") or re.search(matcher, tool) is not None
def _settled(value: object) -> object: """A callback's answer, awaited off the (blocked) event loop if need be.""" if not inspect.isawaitable(value): return value box: list[object] = []
def work() -> None: try: box.append(asyncio.run(_await(cast(Awaitable[object], value)))) except BaseException as error: box.append(error)
thread = threading.Thread(target=work) thread.start() thread.join() if isinstance(box[0], BaseException): raise box[0] return box[0]
async def _await(value: Awaitable[object]) -> object: return await value
def _normal(output: object) -> dict[str, object]: """A callback's answer in the same shape the daemon's hooks answer: the Claude Code / Codex JSON output contract, read into block, halt, updated_input and context.""" if not isinstance(output, dict): return {} answer: dict[str, object] = {"context": []} specific = output.get("hookSpecificOutput") specific = specific if isinstance(specific, dict) else {} 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"] = [str(specific["additionalContext"])] return answer
def _callbacks_for( event: str, tool: str, tool_input: dict[str, object], response: object) -> dict[str, object]: fields: dict[str, object] = {"tool_name": tool, "tool_input": tool_input} if response is not None: fields["tool_response"] = response return run_event(event, tool, fields)
def run_event(event: str, subject: str, fields: dict[str, object]) -> dict[str, object]: """Every callback registered for `event` whose matcher fits `subject` (the tool name; empty for events without one), merged into one answer: block, halt, updated_input, context. A tool event's callbacks see each other's input rewrites.""" merged: dict[str, object] = {"context": []} context = cast(list[str], merged["context"]) current = dict(fields) for kind, matcher, callback in list(_callbacks): if kind != event or not _matches(matcher, subject): continue payload: dict[str, object] = {**current, "hook_event_name": event} try: answer = _normal(_settled(callback(payload))) except Exception as error: # a broken callback fails open, loudly print( f"[{event} hook] callback {getattr(callback, '__name__', callback)} failed: {error!r}" ) continue context.extend(cast(list[str], answer.get("context") or [])) for key in ("block", "halt"): if answer.get(key) and not merged.get(key): merged[key] = answer[key] if isinstance(answer.get("updated_input"), dict): current["tool_input"] = answer["updated_input"] merged["updated_input"] = answer["updated_input"] return merged
class Blocked(PermissionError): """A hook refused the call; the message is the hook's reason."""
def install(sync: Callable[[str, dict[str, object]], object] | None) -> None: global _sync _sync = sync
def active() -> bool: return _sync is not None or bool(_callbacks)
def _fire( event: str, tool: str, tool_input: dict[str, object], response: object = None, echo: bool = True,) -> dict[str, object]: answer = _callbacks_for(event, tool, tool_input, response) if _sync is not None and not (answer.get("block") or answer.get("halt")): current = cast(dict[str, object], answer.get("updated_input") or tool_input) args: dict[str, object] = { "event": event, "tool_name": tool, "tool_input": current, } if response is not None: args["tool_response"] = response try: host = cast(dict[str, object], _sync("hooks.fire", args)) except Exception as error: # a broken hook channel never blocks the tool print(f"[{event} hook] unavailable: {error}") host = {} # A callback's rewrite is the host's input already; the host may rewrite again. answer["context"] = list(cast(list[str], answer["context"])) + list( cast(list[str], host.get("context") or []) ) for key in ("block", "halt", "updated_input"): if host.get(key): answer[key] = host[key] if echo: for text in cast(list[str], answer.get("context") or []): print(f"[{event} hook] {text}") return answer
def pre(tool: str, tool_input: dict[str, object]) -> dict[str, object]: """Fire PreToolUse. Raises Blocked on a deny or halt; otherwise answers the input to use, which a hook may have rewritten.""" if not active(): return tool_input answer = _fire("PreToolUse", tool, tool_input) reason = answer.get("halt") or answer.get("block") if reason: raise Blocked(f"blocked by a hook: {reason}") updated = answer.get("updated_input") return cast(dict[str, object], updated) if isinstance(updated, dict) else tool_input
def post(tool: str, tool_input: dict[str, object], response: object) -> None: """Fire PostToolUse. A block cannot undo the call; its reason is printed.""" if not active(): return answer = _fire("PostToolUse", tool, tool_input, response) reason = answer.get("block") if reason: print(f"[PostToolUse hook] {reason}")
async def post_async( tool: str, tool_input: dict[str, object], response: object) -> list[str]: """PostToolUse for work that finishes after its cell did (a `run()` job): fired off the event loop, answering the lines the hooks had to say.""" if not active(): return []
def work() -> dict[str, object]: return _fire("PostToolUse", tool, tool_input, response, echo=False)
answer = await asyncio.to_thread(work) lines = [ f"[PostToolUse hook] {t}" for t in cast(list[str], answer.get("context") or []) ] if answer.get("block"): lines.append(f"[PostToolUse hook] {answer['block']}") return lines
def run_event_json(event: str, subject: str, fields_json: str) -> str: """`run_event` for the daemon, which fires the loop's own events (the outer tool call, the prompt, the end of a turn) through a hidden cell.""" import json
answer = run_event(event, subject, json.loads(fields_json)) if "updated_input" in answer: answer["updated_input"] = json.dumps(answer["updated_input"]) return json.dumps(answer)