7Hook System

Shell commands that fire before and after node execution, at graph lifecycle boundaries, enabling validation, logging, and policy enforcement without modifying the execution engine.

1. Overview

The hook system lets researchers attach shell commands to graph execution events. Hooks can observe, modify inputs/outputs, or block node execution — all without changing the graph nodes themselves.

Design principles: - Fail-open — hook errors (timeout, crash, malformed JSON) default to continue. Never crashes graph execution. - Zero-cost when unused — a single has_hooks() boolean check guards all hook logic. No overhead when graph.hooks is empty. - Serializable — hooks are pure JSON data in GraphDefinition. They travel with saved graphs.

2. Hook Events

2.1 PreNodeExecute Actions

The hook's response determines what happens:

2.2 PostNodeExecute Actions

PostNodeExecute receives an error field in the payload: null on success, the error message string on failure. This ensures observability hooks can track failures.

3. Hook Configuration

Hooks are defined in the GraphDefinition JSON:

{
  "name": "My Agent",
  "nodes": [...],
  "edges": [...],
  "hooks": [
    {
      "event": "PostNodeExecute",
      "command": "python log_node.py",
      "match_node_type": "llmCall",
      "timeout_ms": 1000,
      "enabled": true
    },
    {
      "event": "PreNodeExecute",
      "command": "python validate_env.py",
      "match_node_type": "env_habitat__*"
    },
    {
      "event": "GraphComplete",
      "command": "python export_metrics.py",
      "match_node_type": "*"
    }
  ]
}

3.1 HookDef Fields

3.2 Backend Data Model

@dataclass
class HookDef:
    event: str
    command: str
    match_node_type: str = "*"
    match_node_id: Optional[str] = None  # exact node instance match
    timeout_ms: int = 1000
    enabled: bool = True

Added to GraphDefinition as hooks: List[HookDef] — serialized only when non-empty for backward compatibility. Node-level hooks are stored in NodeDef.config["hooks"] and auto-get match_node_id set during merge.

4. Matching Rules

4.1 Type-based matching

The match_node_type field supports three patterns:

4.2 Instance-based matching

When match_node_id is set (non-null), the hook matches only the specific node instance with that ID. This takes precedence over match_node_type — type matching is skipped entirely.

Node-level hooks (from NodeDef.config["hooks"]) automatically get match_node_id set to the node's ID during extraction, ensuring they fire only for their owning node.

4.3 Implementation

def _matches(self, hook, node_type, node_id=None):
    # Node-level hooks: exact instance match
    if hook.match_node_id is not None:
        return node_id is not None and hook.match_node_id == node_id
    # Type-based matching (graph-level and global hooks)
    pattern = hook.match_node_type
    if pattern == "*":
        return True
    if node_type is None:
        return False
    if pattern.endswith("__*"):
        return node_type.startswith(pattern[:-1])
    return pattern == node_type

5. Execution Protocol

Hooks communicate with the executor via stdin/stdout JSON.

5.1 stdin (executor → hook)

{
  "event": "PreNodeExecute",
  "node_type": "llmCall",
  "node_id": "node_abc123",
  "inputs": {"prompt": "Navigate to the kitchen..."}
}

For PostNodeExecute, includes outputs and error (null or string). For lifecycle events (GraphStart, GraphComplete, GraphError), includes relevant metadata.

5.2 stdout (hook → executor)

{
  "action": "continue",
  "modified_data": null
}

• Empty stdout or no JSON → treated as "continue" (fail-open)
• Unknown action value → treated as "continue"
• Non-zero exit code → treated as "continue"
• Timeout → subprocess killed, treated as "continue"

5.3 Execution Order

Multiple hooks for the same event run sequentially in list order: - First "block" result wins — short-circuits, remaining hooks don't run - "modify" results accumulate — each hook can re-modify the previous hook's output - Errors in one hook don't prevent subsequent hooks from running

6. Hook Sources and Merge Order

Hooks come from three sources, merged in order from least to most specific:

global hooks  (workspace/hooks.json)
  + graph-level hooks  (GraphDefinition.hooks)
  + node-level hooks   (NodeDef.config.hooks, match_node_id auto-set)
  = merged list → HookRunner

6.1 Global Hooks

Project-wide hooks defined in workspace/hooks.json:

[
  {"event": "GraphComplete", "command": "python log_metrics.py", "match_node_type": "*"}
]

Loaded by WorkspaceComponentRegistry._load_global_hooks() at scan time. Passed to GraphExecutor via the API → LoopRunner → executor chain.

6.2 Graph-Level Hooks

Defined in GraphDefinition.hooks — travel with saved graph JSON files.

6.3 Node-Level Hooks

Stored in NodeDef.config["hooks"]. Configured per-node via the Properties panel. During merge, extract_node_hooks() auto-sets match_node_type to the node's type and match_node_id to the node's ID.

6.4 Merge Utilities

# hooks.py
def extract_node_hooks(nodes: List[NodeDef]) -> List[HookDef]:
    """Extract hooks from NodeDef.config, auto-set match_node_id."""

def merge_hooks(global_hooks, graph_hooks, node_hooks) -> List[HookDef]:
    """Concatenate: global → graph → node."""

def load_hooks_file(path: str) -> List[HookDef]:
    """Load from JSON file. Returns [] on missing/malformed."""

6.5 Precedence Semantics

• block: first block wins (global can enforce project-wide policy)
• modify: last modify wins (node-level has final say)

7. Integration Points

7.1 GraphExecutor.run()

async def run(self, graph, orchestrator, ..., global_hooks=None):
    # Merge 3 hook sources
    node_hooks = extract_node_hooks(graph.nodes)
    merged = merge_hooks(global_hooks or [], graph.hooks, node_hooks)
    self._hook_runner = HookRunner(merged)

    # GraphStart
    await self._hook_runner.run_hooks("GraphStart", payload={...})

    try:
        # ... dataflow loop ...
    except Exception as e:
        await self._hook_runner.run_hooks("GraphError", payload={"error": str(e)})
        raise
    finally:
        await self._hook_runner.run_hooks("GraphComplete", payload={...})

7.2 GraphExecutor._fire_node()

async def _fire_node(self, node, orchestrator):
    # PreNodeExecute — can block or modify inputs
    if self._hook_runner.has_hooks():
        pre_result = await self._hook_runner.run_hooks(
            "PreNodeExecute", node_type=node.type, node_id=node.id,
            payload={"node_id": node.id, "inputs": safe_serialize(node.pending_inputs)},
        )
        if pre_result.action == "block":
            return {}
        if pre_result.action == "modify" and pre_result.modified_data:
            node.pending_inputs.update(pre_result.modified_data)

    # Execute with error capture
    exec_error = None
    try:
        result = await instance.forward(node.pending_inputs, ctx)
    except Exception as e:
        exec_error = e
        result = {"error": str(e)}

    # PostNodeExecute — fires on BOTH success and error
    if self._hook_runner.has_hooks():
        post_result = await self._hook_runner.run_hooks(
            "PostNodeExecute", node_type=node.type, node_id=node.id,
            payload={"node_id": node.id, "outputs": safe_serialize(result),
                     "error": str(exec_error) if exec_error else None},
        )
        if post_result.action == "modify" and post_result.modified_data and not exec_error:
            result = post_result.modified_data

    if exec_error:
        raise exec_error
    return result

8. Properties Panel UI

When a node is selected on the canvas, the Properties panel shows a Hooks section at the bottom:

• Add button creates a new hook with defaults (PostNodeExecute, empty command, 1000ms timeout, enabled)
• Each hook row shows: event dropdown (PreNodeExecute/PostNodeExecute), command text input, timeout input (ms), enabled toggle, remove button
• Changes are stored in node.data.hooks which maps to NodeDef.config.hooks via fromFlowNodes
• The match_node_type and match_node_id fields are implicit — auto-set during server-side extraction

The UI is implemented as HooksSection + HookRow components in PropertiesPanel.tsx.

9. safe_serialize()

Hook payloads must be JSON-safe. The safe_serialize() utility converts complex Python objects to metadata-only representations — no full data (images, arrays) passes through the hook stdin pipe.

Prior art: _broadcast_step() in graph_executor.py handles similar conversions for WebSocket broadcast.

10. Performance Considerations

Shell hooks have subprocess startup overhead (~5-20ms per invocation). This is significant for large graphs:

Mitigations: 1. Use exact or prefix matching instead of wildcard ("llmCall" vs "*") 2. Reduce step_budget during development 3. Set enabled: false to disable hooks without removing config 4. Future V1.1: persistent-process hooks that avoid subprocess respawn

Zero-cost guarantee: When graph.hooks is empty (the common case), HookRunner.has_hooks() returns False and all hook code is skipped — no async overhead, no function calls, just a single boolean check.

11. Key Files

Status