# Attractor Specification A DOT-based pipeline runner that uses directed graphs (defined in Graphviz DOT syntax) to orchestrate multi-stage AI workflows. Each node in the graph is an AI task (LLM call, human review, conditional branch, parallel fan-out, etc.) and edges define the flow between them. --- ## Table of Contents 1. [Overview and Goals](#1-overview-and-goals) 2. [DOT DSL Schema](#2-dot-dsl-schema) 3. [Pipeline Execution Engine](#3-pipeline-execution-engine) 4. [Node Handlers](#4-node-handlers) 5. [State and Context](#5-state-and-context) 6. [Human-in-the-Loop (Interviewer Pattern)](#6-human-in-the-loop-interviewer-pattern) 7. [Validation and Linting](#7-validation-and-linting) 8. [Model Stylesheet](#8-model-stylesheet) 9. [Transforms and Extensibility](#9-transforms-and-extensibility) 10. [Condition Expression Language](#10-condition-expression-language) 11. [Definition of Done](#11-definition-of-done) --- ## 1. Overview and Goals ### 1.1 Problem Statement AI-powered software workflows -- code generation, code review, testing, deployment planning -- often require multiple LLM calls chained together with conditional logic, human approvals, and parallel execution. Without a structured orchestration layer, developers either write fragile imperative scripts or build ad-hoc state machines that are difficult to visualize, version, or debug. Attractor solves this by letting pipeline authors define multi-stage AI workflows as directed graphs using Graphviz DOT syntax. The graph is the workflow: nodes are tasks, edges are transitions, and attributes configure behavior. The result is a declarative, visual, version-controllable pipeline definition that an execution engine can traverse deterministically. ### 1.2 Why DOT Syntax DOT is chosen as the pipeline definition format for several reasons: - **DOT is inherently a graph description language.** Workflow pipelines are directed graphs. Using DOT means the structure (nodes and edges) maps directly to the language's primary construct, rather than being encoded in a data format like YAML or JSON that has no native concept of graphs. - **Existing tooling.** DOT files can be rendered to SVG/PNG with standard Graphviz tooling, giving pipeline authors immediate visual feedback. Editors, linters, and parsers already exist. - **Declarative and human-readable.** A `.dot` file is a complete, self-contained workflow definition that can be version-controlled, diffed, and reviewed in pull requests. - **Constrained extensibility.** By restricting to a well-defined DOT subset (directed graphs only, typed attributes, no HTML labels), the DSL remains predictable while being extensible through custom attributes. For reference on DOT syntax, see the Graphviz DOT language specification: https://graphviz.org/doc/info/lang.html ### 1.3 Design Principles **Declarative pipelines.** The `.dot` file declares what the workflow looks like and what each stage should do. The execution engine decides how and when to run each stage. Pipeline authors do not write control flow; they declare graph structure. **Pluggable handlers.** Each node type (LLM call, human gate, parallel fan-out) is backed by a handler that implements a common interface. New node types are added by registering new handlers. The execution engine does not know about handler internals. **Checkpoint and resume.** After each node completes, the execution engine saves a serializable checkpoint. If the process crashes, execution resumes from the last checkpoint. **Human-in-the-loop.** The pipeline can pause at designated nodes, present choices to a human operator, and route based on the human's decision. This supports approval gates, code review, and manual override -- critical for AI workflows where automated judgment may not be sufficient. **Edge-based routing.** Transitions between nodes are controlled by conditions, labels, and weights on edges, with runtime condition evaluation. ### 1.4 Layering and LLM Backends Attractor defines the orchestration layer: graph definition, traversal, state management, and extensibility. It does NOT require any specific LLM integration. The codergen handler (Section 4.5) needs a way to call an LLM and get a response -- how you provide that is up to you. The codergen handler takes a backend that conforms to the `CodergenBackend` interface (Section 4.5). What that backend does internally is entirely up to the implementor -- use the companion [Coding Agent Loop](./coding-agent-loop-spec.md) and [Unified LLM Client](./unified-llm-spec.md) specs, spawn CLI agents (Claude Code, Codex, Gemini CLI) in subprocesses, run agents in tmux panes with a manager attaching to them, call an LLM API directly, or anything else. The pipeline definition (the DOT file) does not change regardless of backend choice. Attractor pipelines are driven by an event stream (Section 9.6). TUI, web, and IDE frontends consume events and submit human-in-the-loop answers. The pipeline engine is headless; the presentation layer is separate. --- ## 2. DOT DSL Schema ### 2.1 Supported Subset Attractor accepts a strict subset of the Graphviz DOT language. The restrictions exist for predictability: one graph per file, directed edges only, no HTML labels, and typed attributes with defaults. ### 2.2 BNF-Style Grammar ``` Graph ::= 'digraph' Identifier '{' Statement* '}' Statement ::= GraphAttrStmt | NodeDefaults | EdgeDefaults | SubgraphStmt | NodeStmt | EdgeStmt | GraphAttrDecl GraphAttrStmt ::= 'graph' AttrBlock ';'? NodeDefaults ::= 'node' AttrBlock ';'? EdgeDefaults ::= 'edge' AttrBlock ';'? GraphAttrDecl ::= Identifier '=' Value ';'? SubgraphStmt ::= 'subgraph' Identifier? '{' Statement* '}' NodeStmt ::= Identifier AttrBlock? ';'? EdgeStmt ::= Identifier ( '->' Identifier )+ AttrBlock? ';'? AttrBlock ::= '[' Attr ( ',' Attr )* ']' Attr ::= Key '=' Value Key ::= Identifier | QualifiedId QualifiedId ::= Identifier ( '.' Identifier )+ Value ::= String | Integer | Float | Boolean | Duration | BareValue Identifier ::= [A-Za-z_][A-Za-z0-9_]* String ::= '"' ( '\\"' | '\\n' | '\\t' | '\\\\' | [^"\\] )* '"' Integer ::= '-'? [0-9]+ Float ::= '-'? [0-9]* '.' [0-9]+ Boolean ::= 'true' | 'false' Duration ::= Integer ( 'ms' | 's' | 'm' | 'h' | 'd' ) BareValue ::= [A-Za-z_][A-Za-z0-9_.:-]* Direction ::= 'TB' | 'LR' | 'BT' | 'RL' ``` ### 2.3 Key Constraints - **One digraph per file.** Multiple graphs, undirected graphs, and `strict` modifiers are rejected. - **Bare identifiers for node IDs.** Node IDs must match `[A-Za-z_][A-Za-z0-9_]*`. Human-readable names go in the `label` attribute. - **Commas required between attributes.** Inside attribute blocks, commas separate key-value pairs for unambiguous parsing. - **Directed edges only.** `->` is the only edge operator. `--` (undirected) is rejected. - **Comments supported.** Both `// line` and `/* block */` comments are stripped before parsing. - **Semicolons optional.** Statement-terminating semicolons are accepted but not required. ### 2.4 Value Types | Type | Syntax | Examples | |----------|---------------------------------|--------------------------------------| | String | Double-quoted with escapes | `"Hello world"`, `"line1\nline2"` | | Integer | Optional sign, digits | `42`, `-1`, `0` | | Float | Decimal number | `0.5`, `-3.14` | | Boolean | Literal keywords | `true`, `false` | | Duration | Integer + unit suffix | `900s`, `15m`, `2h`, `250ms`, `1d` | ### 2.5 Graph-Level Attributes Graph attributes are declared in a `graph [ ... ]` block or as top-level `key = value` declarations. They configure the entire workflow. | Key | Type | Default | Description | |---------------------------|----------|-----------|-------------| | `goal` | String | `""` | Human-readable goal for the pipeline. Exposed as `$goal` in prompt templates and mirrored into the run context as `graph.goal`. | | `label` | String | `""` | Display name for the graph (used in visualization). | | `model_stylesheet` | String | `""` | CSS-like stylesheet for per-node LLM model/provider defaults. See Section 8. | | `default_max_retries` | Integer | `0` | Default additional retries for nodes that omit `max_retries`. Legacy alias: `default_max_retry`. | | `retry_target` | String | `""` | Node ID to jump to if exit is reached with unsatisfied goal gates. | | `fallback_retry_target` | String | `""` | Secondary jump target if `retry_target` is missing or invalid. | | `default_fidelity` | String | `""` | Default context fidelity mode when explicitly set. Empty string means "unset"; the runtime fallback is `compact` (see Section 5.4). | ### 2.6 Node Attributes | Key | Type | Default | Description | |---------------------|----------|-----------------|-------------| | `label` | String | node ID | Display name shown in UI, prompts, and telemetry. | | `shape` | String | `"box"` | Graphviz shape. Determines the default handler type (see mapping table below). | | `type` | String | `""` | Explicit handler type override. Takes precedence over shape-based resolution. | | `prompt` | String | `""` | Primary instruction for the stage. Supports `$goal` variable expansion. Falls back to `label` if empty for LLM stages. | | `max_retries` | Integer | inherited | Number of additional attempts beyond the initial execution. If omitted, inherits graph `default_max_retries`. `max_retries=3` means up to 4 total executions. | | `goal_gate` | Boolean | `false` | If `true`, this node must reach `SUCCESS` or `PARTIAL_SUCCESS` before the pipeline can exit. | | `retry_target` | String | `""` | Node ID to jump to if this node fails and retries are exhausted. | | `fallback_retry_target` | String | `""` | Secondary retry target. | | `fidelity` | String | inherited | Context fidelity mode for this node's LLM session. See Section 5.4. | | `thread_id` | String | derived | Explicit thread identifier for LLM session reuse under `full` fidelity. | | `class` | String | `""` | Comma-separated class names for model stylesheet targeting. | | `timeout` | Duration | unset | Maximum execution time for this node. | | `llm_model` | String | inherited | LLM model identifier. Overridable by stylesheet. | | `llm_provider` | String | auto-detected | LLM provider key. Auto-detected from model if unset. | | `reasoning_effort` | String | `"high"` | LLM reasoning effort: `low`, `medium`, `high`. | | `auto_status` | Boolean | `false` | If `true` and the handler writes no status, the engine auto-generates a SUCCESS outcome. | | `allow_partial` | Boolean | `false` | Accept PARTIAL_SUCCESS when retries are exhausted instead of failing. | The external DOT attribute name is `type`. Implementations may use an internal field name such as `node_type` to avoid reserved-word conflicts, but the externally visible behavior must remain identical. ### 2.7 Edge Attributes | Key | Type | Default | Description | |--------------|----------|---------|-------------| | `label` | String | `""` | Human-facing caption and routing key. Used for preferred-label matching in edge selection. | | `condition` | String | `""` | Boolean guard expression evaluated against the current context and outcome. See Section 10. | | `weight` | Integer | `0` | Numeric priority for edge selection. Higher weight wins among equally eligible edges. | | `fidelity` | String | unset | Override fidelity mode for the target node. Highest precedence in fidelity resolution. | | `thread_id` | String | unset | Override thread ID for session reuse at the target node. | | `loop_restart` | Boolean | `false` | When `true`, terminates the current run and re-launches with a fresh log directory. | ### 2.8 Shape-to-Handler-Type Mapping The `shape` attribute on a node determines which handler executes it, unless overridden by an explicit `type` attribute. This table defines the canonical mapping: | Shape | Handler Type | Description | |-------------------|-----------------------|-------------| | `Mdiamond` | `start` | Pipeline entry point. No-op handler. Every graph must have exactly one. | | `Msquare` | `exit` | Pipeline exit point. No-op handler. Goal gate enforcement is handled by the execution engine (Section 3.4). Every graph must have exactly one. | | `box` | `codergen` | LLM task (code generation, analysis, planning). The default for all nodes without an explicit shape. | | `hexagon` | `wait.human` | Human-in-the-loop gate. Blocks until a human selects an option. | | `diamond` | `conditional` | Conditional routing point. Routes based on edge conditions against current context. | | `component` | `parallel` | Parallel fan-out. Executes multiple branches concurrently. | | `tripleoctagon` | `parallel.fan_in` | Parallel fan-in. Waits for all branches and consolidates results. | | `parallelogram` | `tool` | External tool execution (shell command, API call). | | `house` | `stack.manager_loop` | Supervisor loop. Orchestrates observe/steer/wait cycles over a child pipeline. | ### 2.9 Chained Edges Chained edge declarations are syntactic sugar. The statement: ``` A -> B -> C [label="next"] ``` expands to two edges: ``` A -> B [label="next"] B -> C [label="next"] ``` Edge attributes in a chained declaration apply to all edges in the chain. ### 2.10 Subgraphs Subgraphs serve two purposes: **scoping defaults** and **deriving classes** for the model stylesheet. **Scoping defaults:** Attributes declared in a subgraph's `node [ ... ]` block apply to nodes within that subgraph unless the node explicitly overrides them. ``` subgraph cluster_loop { label = "Loop A" node [thread_id="loop-a", timeout="900s"] Plan [label="Plan next step"] Implement [label="Implement", timeout="1800s"] } ``` Here `Plan` inherits `thread_id="loop-a"` and `timeout="900s"`, while `Implement` inherits `thread_id` but overrides `timeout`. **Class derivation:** Subgraph labels can produce CSS-like classes for model stylesheet matching. Nodes inside a subgraph receive the derived class. The class name is derived by lowercasing the label, replacing spaces with hyphens, and stripping non-alphanumeric characters (except hyphens). For example, `label="Loop A"` yields class `loop-a`. ### 2.11 Node and Edge Default Blocks Default blocks set baseline attributes for all subsequent nodes or edges within their scope: ``` node [shape=box, timeout="900s"] edge [weight=0] ``` Explicit attributes on individual nodes or edges override these defaults. ### 2.12 Class Attribute The `class` attribute assigns one or more CSS-like class names to a node for model stylesheet targeting: ``` review_code [shape=box, class="code,critical", prompt="Review the code"] ``` Classes are comma-separated. They can be referenced in the model stylesheet with dot-prefix selectors (`.code`, `.critical`). ### 2.13 Minimal Examples **Simple linear workflow:** ``` digraph Simple { graph [goal="Run tests and report"] rankdir=LR start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] run_tests [label="Run Tests", prompt="Run the test suite and report results"] report [label="Report", prompt="Summarize the test results"] start -> run_tests -> report -> exit } ``` **Branching workflow with conditions:** ``` digraph Branch { graph [goal="Implement and validate a feature"] rankdir=LR node [shape=box, timeout="900s"] start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] plan [label="Plan", prompt="Plan the implementation"] implement [label="Implement", prompt="Implement the plan"] validate [label="Validate", prompt="Run tests"] gate [shape=diamond, label="Tests passing?"] start -> plan -> implement -> validate -> gate gate -> exit [label="Yes", condition="outcome=success"] gate -> implement [label="No", condition="outcome!=success"] } ``` **Human gate:** ``` digraph Review { rankdir=LR start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] review_gate [ shape=hexagon, label="Review Changes", type="wait.human" ] start -> review_gate review_gate -> ship_it [label="[A] Approve"] review_gate -> fixes [label="[F] Fix"] ship_it -> exit fixes -> review_gate } ``` --- ## 3. Pipeline Execution Engine ### 3.1 Run Lifecycle The execution lifecycle proceeds through six phases: ``` PARSE -> TRANSFORM -> VALIDATE -> INITIALIZE -> EXECUTE -> FINALIZE ``` 1. **Parse:** Read the `.dot` source and produce an in-memory Graph model (nodes, edges, attributes). 2. **Transform:** Apply parse-time transforms (stylesheet, variable expansion, and custom AST transforms). 3. **Validate:** Run lint rules (Section 7). Reject invalid graphs. Warn on suspicious patterns. 4. **Initialize:** Create the run directory, initial context, and checkpoint. Mirror graph attributes into the context. 5. **Execute:** Traverse the graph from the start node, executing handlers and selecting edges. 6. **Finalize:** Write the final checkpoint, emit completion events, and clean up resources (close sessions, release files). ### 3.2 Core Execution Loop The following pseudocode defines the execution engine's traversal algorithm. This is the heart of the system. ``` FUNCTION run(graph, config): context = new Context() mirror_graph_attributes(graph, context) checkpoint = new Checkpoint() completed_nodes = [] node_outcomes = {} current_node = find_start_node(graph) -- Resolves by: (1) shape=Mdiamond, (2) id="start" or "Start" -- Raises error if not found WHILE true: node = graph.nodes[current_node.id] -- Step 1: Check for terminal node IF is_terminal(node): gate_ok, failed_gate = check_goal_gates(graph, node_outcomes) IF NOT gate_ok AND failed_gate exists: retry_target = get_retry_target(failed_gate, graph) IF retry_target exists: current_node = graph.nodes[retry_target] CONTINUE ELSE: RETURN Outcome( status=FAIL, failure_reason="Goal gate unsatisfied and no retry target" ) RETURN Outcome(status=SUCCESS, notes="Pipeline completed") -- Step 2: Execute node handler with retry policy retry_policy = build_retry_policy(node, graph) outcome = execute_with_retry(node, context, graph, retry_policy) -- Step 3: Record completion completed_nodes.append(node.id) node_outcomes[node.id] = outcome -- Step 4: Apply context updates from outcome FOR EACH (key, value) IN outcome.context_updates: context.set(key, value) context.set("outcome", outcome.status) IF outcome.preferred_label is not empty: context.set("preferred_label", outcome.preferred_label) -- Step 5: Save checkpoint checkpoint = create_checkpoint(context, current_node.id, completed_nodes) save_checkpoint(checkpoint, logs_root) -- Step 6: Select next edge next_edge = select_edge(node, outcome, context, graph) IF next_edge is NONE: IF outcome.status == FAIL: RETURN outcome RETURN Outcome(status=SUCCESS, notes="Pipeline completed") -- Step 7: Handle loop_restart IF next_edge has loop_restart=true: restart_run(graph, config, start_at=next_edge.target) RETURN -- Step 8: Advance to next node current_node = graph.nodes[next_edge.to_node] RETURN Outcome(status=SUCCESS, notes="Pipeline completed") ``` ### 3.3 Edge Selection Algorithm After a node completes, the engine selects the next edge from the node's outgoing edges. The selection is deterministic and follows a five-step priority order: **Step 1: Condition-matching edges.** Evaluate each edge's `condition` expression (see Section 10) against the current context and outcome. Edges whose condition evaluates to `true` are eligible. Edges with no condition are not considered in this step; they proceed to later steps. **Step 2: Preferred label match.** If no condition-matching edges were found and the node's outcome includes a `preferred_label`, find the first unconditional edge whose `label` matches after normalization. Label normalization: lowercase, trim whitespace, strip accelerator prefixes (patterns like `[Y] `, `Y) `, `Y - `). **Step 3: Suggested next IDs.** If no label match and the outcome includes `suggested_next_ids`, find the first unconditional edge whose target node ID appears in the list. **Step 4: Highest weight.** Among remaining eligible unconditional edges, choose the one with the highest `weight` attribute (default 0). **Step 5: Lexical tiebreak.** If weights are equal, choose the edge whose target node ID comes first lexicographically. ``` FUNCTION select_edge(node, outcome, context, graph): edges = graph.outgoing_edges(node.id) IF edges is empty: RETURN NONE -- Step 1: Condition matching condition_matched = [] FOR EACH edge IN edges: IF edge.condition is not empty: IF evaluate_condition(edge.condition, outcome, context) == true: condition_matched.append(edge) IF condition_matched is not empty: RETURN best_by_weight_then_lexical(condition_matched) -- Step 2: Preferred label IF outcome.preferred_label is not empty: FOR EACH edge IN edges: IF edge.condition is empty AND normalize_label(edge.label) == normalize_label(outcome.preferred_label): RETURN edge -- Step 3: Suggested next IDs IF outcome.suggested_next_ids is not empty: FOR EACH suggested_id IN outcome.suggested_next_ids: FOR EACH edge IN edges: IF edge.condition is empty AND edge.to_node == suggested_id: RETURN edge -- Step 4 & 5: Weight with lexical tiebreak (unconditional edges only) unconditional = [e FOR e IN edges WHERE e.condition is empty] IF unconditional is not empty: RETURN best_by_weight_then_lexical(unconditional) RETURN NONE FUNCTION best_by_weight_then_lexical(edges): SORT edges BY (weight DESCENDING, to_node ASCENDING) RETURN edges[0] ``` ### 3.4 Goal Gate Enforcement Nodes with `goal_gate=true` represent critical stages that must succeed before the pipeline can exit. When the traversal reaches a terminal node (shape=Msquare): 1. Check all visited nodes that have `goal_gate=true`. 2. If any goal gate node has a non-success outcome (not SUCCESS or PARTIAL_SUCCESS), the pipeline cannot exit. 3. Instead, jump to the `retry_target` of the unsatisfied goal gate node. If that is not set, try `fallback_retry_target`. If that is also not set, try the graph-level `retry_target` and `fallback_retry_target`. 4. If no retry target exists at any level, the pipeline ends with a FAIL outcome. ``` FUNCTION check_goal_gates(graph, node_outcomes): FOR EACH (node_id, outcome) IN node_outcomes: node = graph.nodes[node_id] IF node.goal_gate == true: IF outcome.status NOT IN {SUCCESS, PARTIAL_SUCCESS}: RETURN (false, node) RETURN (true, NONE) ``` ### 3.5 Retry Logic Each node has a retry policy determined by: 1. Node attribute `max_retries` (if explicitly set) -- number of additional attempts beyond the initial execution 2. Graph attribute `default_max_retries` (fallback; legacy alias `default_max_retry` is accepted) If neither is set, the built-in default is 0 (no retries). The `max_retries` attribute specifies additional attempts. So `max_retries=3` means a total of 4 executions (1 initial + 3 retries). Internally this maps to `max_attempts = max_retries + 1`. ``` FUNCTION execute_with_retry(node, context, graph, retry_policy): FOR attempt FROM 1 TO retry_policy.max_attempts: TRY: outcome = handler.execute(node, context, graph, logs_root) CATCH exception: IF retry_policy.should_retry(exception) AND attempt < retry_policy.max_attempts: delay = retry_policy.backoff.delay_for_attempt(attempt) sleep(delay) CONTINUE ELSE: RETURN Outcome(status=FAIL, failure_reason=str(exception)) IF outcome.status IN {SUCCESS, PARTIAL_SUCCESS}: reset_retry_counter(node.id) RETURN outcome IF outcome.status == RETRY: IF attempt < retry_policy.max_attempts: increment_retry_counter(node.id) delay = retry_policy.backoff.delay_for_attempt(attempt) sleep(delay) CONTINUE ELSE: IF node.allow_partial == true: RETURN Outcome(status=PARTIAL_SUCCESS, notes="retries exhausted, partial accepted") RETURN Outcome(status=FAIL, failure_reason="max retries exceeded") IF outcome.status == FAIL: RETURN outcome RETURN Outcome(status=FAIL, failure_reason="max retries exceeded") ``` ### 3.6 Retry Policy ``` RetryPolicy: max_attempts : Integer -- minimum 1 (1 means no retries) backoff : BackoffConfig -- delay calculation between retries should_retry : Function(Error) -> Boolean -- predicate for retryable errors BackoffConfig: initial_delay_ms : Integer -- first retry delay in milliseconds (default: 200) backoff_factor : Float -- multiplier for subsequent delays (default: 2.0) max_delay_ms : Integer -- cap on delay in milliseconds (default: 60000) jitter : Boolean -- add random jitter to prevent thundering herd (default: true) ``` **Delay calculation:** ``` FUNCTION delay_for_attempt(attempt, config): -- attempt is 1-indexed (first retry is attempt=1) delay = config.initial_delay_ms * (config.backoff_factor ^ (attempt - 1)) delay = MIN(delay, config.max_delay_ms) IF config.jitter: delay = delay * random_uniform(0.5, 1.5) RETURN delay ``` **Preset policies:** | Name | Max Attempts | Initial Delay | Factor | Description | |--------------|-------------|---------------|--------|-------------| | `none` | 1 | -- | -- | No retries. Fail immediately on error. | | `standard` | 5 | 200ms | 2.0 | General-purpose. Delays: 200, 400, 800, 1600, 3200ms. | | `aggressive` | 5 | 500ms | 2.0 | For unreliable operations. Delays: 500, 1000, 2000, 4000, 8000ms. | | `linear` | 3 | 500ms | 1.0 | Fixed delay between attempts. Delays: 500, 500, 500ms. | | `patient` | 3 | 2000ms | 3.0 | Long-running operations. Delays: 2000, 6000, 18000ms. | **Default should_retry predicate:** Returns `true` for network errors, rate limit errors (HTTP 429), server errors (HTTP 5xx), and provider-reported transient failures. Returns `false` for authentication errors (HTTP 401, 403), bad request errors (HTTP 400), validation errors, and configuration errors. ### 3.7 Failure Routing When a stage returns FAIL (or retries are exhausted), the engine attempts failure routing in this order: 1. **Fail edge:** An outgoing edge with `condition="outcome=fail"`. If found, follow it. 2. **Retry target:** Node attribute `retry_target`. Jump to that node. 3. **Fallback retry target:** Node attribute `fallback_retry_target`. Jump to that node. 4. **Pipeline termination:** No failure route found. The pipeline fails with the stage's failure reason. ### 3.8 Concurrency Model The graph traversal is single-threaded. Only one node executes at a time in the top-level graph. This simplifies reasoning about context state and avoids race conditions. Parallelism exists within specific node handlers (`parallel`, `parallel.fan_in`) that manage concurrent execution internally. Each parallel branch receives an isolated clone of the context. Branch results are collected but individual branch context changes are not merged back into the parent -- only the handler's outcome and its `context_updates` are applied. --- ## 4. Node Handlers ### 4.1 Handler Interface Every node handler implements a common interface. The execution engine dispatches to the appropriate handler based on the node's `type` attribute (or shape-based resolution if `type` is empty). ``` INTERFACE Handler: FUNCTION execute(node, context, graph, logs_root) -> Outcome -- Parameters: -- node : The parsed Node with all its attributes -- context : The shared key-value Context for the pipeline run (read/write) -- graph : The full parsed Graph (for reading outgoing edges, etc.) -- logs_root : Filesystem path for this run's log/artifact directory -- Returns: -- Outcome : The result of execution (see Section 5.2) ``` ### 4.2 Handler Registry The handler registry maps type strings to handler instances. Resolution follows this order: 1. **Explicit `type` attribute** on the node (e.g., `type="wait.human"`) 2. **Shape-based resolution** using the shape-to-handler-type mapping table (Section 2.8) 3. **Default handler** (the codergen/LLM handler) ``` HandlerRegistry: handlers : Map -- type string -> handler instance default_handler : Handler -- fallback handler (typically codergen) FUNCTION register(type_string, handler): handlers[type_string] = handler -- Registering for an already-registered type replaces the previous handler FUNCTION resolve(node) -> Handler: -- 1. Explicit type attribute IF node.type is not empty AND node.type IN handlers: RETURN handlers[node.type] -- 2. Shape-based resolution handler_type = SHAPE_TO_TYPE[node.shape] IF handler_type IN handlers: RETURN handlers[handler_type] -- 3. Default RETURN default_handler ``` ### 4.3 Start Handler A no-op handler for the pipeline entry point. Returns SUCCESS immediately without performing any work. ``` StartHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: RETURN Outcome(status=SUCCESS) ``` Every graph must have exactly one start node (shape=Mdiamond or id matching `start`/`Start`). The lint rules enforce this. ### 4.4 Exit Handler A no-op handler for the pipeline exit point. Returns SUCCESS immediately. Goal gate enforcement is handled by the execution engine (Section 3.4), not by this handler. ``` ExitHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: RETURN Outcome(status=SUCCESS) ``` Every graph must have exactly one exit node (shape=Msquare or id matching `exit`/`end`). ### 4.5 Codergen Handler (LLM Task) The codergen handler is the default for all nodes that invoke an LLM. It reads the node's prompt, expands template variables, calls the LLM backend (see Section 1.4 for backend options), writes the prompt and response to the logs directory, and returns the outcome. ``` CodergenHandler: backend : CodergenBackend | None -- The LLM execution backend. Any implementation of the -- CodergenBackend interface (Section 4.5). None = simulation mode. FUNCTION execute(node, context, graph, logs_root) -> Outcome: -- 1. Build prompt prompt = node.prompt IF prompt is empty: prompt = node.label prompt = expand_variables(prompt, graph, context) -- 2. Write prompt to logs stage_dir = logs_root + "/" + node.id + "/" create_directory(stage_dir) write_file(stage_dir + "prompt.md", prompt) -- 3. Call LLM backend IF backend is not NONE: TRY: result = backend.run(node, prompt, context) IF result is an Outcome: write_status(stage_dir, result) RETURN result response_text = string(result) CATCH exception: RETURN Outcome(status=FAIL, failure_reason=str(exception)) ELSE: response_text = "[Simulated] Response for stage: " + node.id -- 4. Write response to logs write_file(stage_dir + "response.md", response_text) -- 5. Write status and return outcome outcome = Outcome( status=SUCCESS, notes="Stage completed: " + node.id, context_updates={ "last_stage": node.id, "last_response": truncate(response_text, 200) } ) write_status(stage_dir, outcome) RETURN outcome ``` **Variable expansion:** The only built-in template variable is `$goal`, which resolves to the graph-level `goal` attribute. Variable expansion is simple string replacement, not a templating engine. **Status file:** The handler writes `status.json` in the stage directory with the Outcome fields serialized as JSON. This file serves as an audit trail and enables the status-file contract: external tools or agents can write `status.json` to communicate outcomes back to the engine. #### CodergenBackend Interface ``` INTERFACE CodergenBackend: FUNCTION run(node: Node, prompt: String, context: Context) -> String | Outcome ``` How you implement this interface is up to you. The pipeline engine only cares that it gets a String or Outcome back. ### 4.6 Wait For Human Handler Blocks pipeline execution until a human selects an option derived from the node's outgoing edges. This implements the human-in-the-loop pattern (see Section 6 for the full Interviewer protocol). ``` WaitForHumanHandler: interviewer : Interviewer -- the human interaction frontend FUNCTION execute(node, context, graph, logs_root) -> Outcome: -- 1. Derive choices from outgoing edges edges = graph.outgoing_edges(node.id) choices = [] FOR EACH edge IN edges: label = edge.label OR edge.to_node key = parse_accelerator_key(label) choices.append(Choice(key=key, label=label, to=edge.to_node)) IF choices is empty: RETURN Outcome(status=FAIL, failure_reason="No outgoing edges for human gate") -- 2. Build question from choices options = [Option(key=c.key, label=c.label) FOR c IN choices] question = Question( text=node.label OR "Select an option:", type=MULTIPLE_CHOICE, options=options, stage=node.id ) -- 3. Present to interviewer and wait for answer answer = interviewer.ask(question) -- 4. Handle timeout/skip IF answer is TIMEOUT: default_choice = node.attrs["human.default_choice"] IF default_choice exists: -- Use default ELSE: RETURN Outcome(status=RETRY, failure_reason="human gate timeout, no default") IF answer is SKIPPED: RETURN Outcome(status=FAIL, failure_reason="human skipped interaction") -- 5. Find matching choice selected = find_choice_matching(answer, choices) IF selected is NONE: selected = choices[0] -- fallback to first -- 6. Record in context and return RETURN Outcome( status=SUCCESS, suggested_next_ids=[selected.to], context_updates={ "human.gate.selected": selected.key, "human.gate.label": selected.label } ) ``` **Accelerator key parsing** extracts shortcut keys from edge labels using these patterns: | Pattern | Example | Extracted Key | |-------------------|-------------------|---------------| | `[K] Label` | `[Y] Yes, deploy` | `Y` | | `K) Label` | `Y) Yes, deploy` | `Y` | | `K - Label` | `Y - Yes, deploy` | `Y` | | First character | `Yes, deploy` | `Y` | ### 4.7 Conditional Handler For diamond-shaped nodes that act as conditional routing points. The handler itself is a no-op that returns SUCCESS; the actual routing is handled by the execution engine's edge selection algorithm (Section 3.3), which evaluates conditions on outgoing edges. ``` ConditionalHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: RETURN Outcome( status=SUCCESS, notes="Conditional node evaluated: " + node.id ) ``` This design keeps routing logic in the engine (where it can be deterministic and inspectable) rather than in the handler. ### 4.8 Parallel Handler Fans out execution to multiple branches concurrently. Each parallel branch receives an isolated clone of the parent context and runs independently. The handler waits for all branches to complete (or applies a configurable join policy) before returning. ``` ParallelHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: -- 1. Identify fan-out edges (all outgoing edges from this node) branches = graph.outgoing_edges(node.id) -- 2. Determine join policy from node attributes join_policy = node.attrs.get("join_policy", "wait_all") max_parallel = integer(node.attrs.get("max_parallel", "4")) -- 3. Execute branches concurrently with bounded parallelism results = [] FOR EACH branch IN branches (up to max_parallel at a time): branch_context = context.clone() branch_outcome = execute_subgraph(branch.to_node, branch_context, graph, logs_root) results.append(branch_outcome) -- 4. Store results in context for downstream fan-in context.set("parallel.results", serialize_results(results)) -- 5. Evaluate join policy success_count = count(r FOR r IN results WHERE r.status == SUCCESS) fail_count = count(r FOR r IN results WHERE r.status == FAIL) IF join_policy == "wait_all": IF fail_count == 0: RETURN Outcome(status=SUCCESS) ELSE: RETURN Outcome(status=PARTIAL_SUCCESS) IF join_policy == "first_success": IF success_count > 0: RETURN Outcome(status=SUCCESS) ELSE: RETURN Outcome(status=FAIL) RETURN Outcome(status=SUCCESS) ``` **Join policies:** | Policy | Behavior | |------------------|----------| | `wait_all` | All branches must complete. Join satisfied when all are done. | | `first_success` | Join satisfied as soon as one branch succeeds. Others may be cancelled. | ### 4.9 Fan-In Handler Consolidates results from a preceding parallel node and selects the best candidate. ``` FanInHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: -- 1. Read parallel results results = context.get("parallel.results") IF results is empty: RETURN Outcome(status=FAIL, failure_reason="No parallel results to evaluate") -- 2. Evaluate candidates IF node.prompt is not empty: -- LLM-based evaluation: call LLM to rank candidates best = llm_evaluate(node.prompt, results) ELSE: -- Heuristic: rank by outcome status, then by score best = heuristic_select(results) -- 3. Record winner in context context_updates = { "parallel.fan_in.best_id": best.id, "parallel.fan_in.best_outcome": best.outcome } RETURN Outcome( status=SUCCESS, context_updates=context_updates, notes="Selected best candidate: " + best.id ) FUNCTION heuristic_select(candidates): outcome_rank = {SUCCESS: 0, PARTIAL_SUCCESS: 1, RETRY: 2, FAIL: 3} SORT candidates BY (outcome_rank[c.outcome], -c.score, c.id) RETURN candidates[0] ``` Fan-in runs even when some candidates failed, as long as at least one candidate is available. Only when all candidates fail does fan-in return FAIL. ### 4.10 Tool Handler Executes an external tool (shell command, API call, or other non-LLM operation) configured via node attributes. ``` ToolHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: command = node.attrs.get("tool_command", "") IF command is empty: RETURN Outcome(status=FAIL, failure_reason="No tool_command specified") -- Execute the command TRY: result = run_shell_command(command, timeout=node.timeout) RETURN Outcome( status=SUCCESS, context_updates={"tool.output": result.stdout}, notes="Tool completed: " + command ) CATCH exception: RETURN Outcome(status=FAIL, failure_reason=str(exception)) ``` ### 4.11 Manager Loop Handler Orchestrates sprint-based iteration by supervising a child pipeline. The manager observes the child's telemetry, evaluates progress via a guard function, and optionally steers the child through intervention. ``` ManagerLoopHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: child_dotfile = graph.attrs.get("stack.child_dotfile") poll_interval = parse_duration(node.attrs.get("manager.poll_interval", "45s")) max_cycles = integer(node.attrs.get("manager.max_cycles", "1000")) stop_condition = node.attrs.get("manager.stop_condition", "") actions = split(node.attrs.get("manager.actions", "observe,wait"), ",") -- 1. Auto-start child if configured IF node.attrs.get("stack.child_autostart", "true") == "true": start_child_pipeline(child_dotfile) -- 2. Observation loop FOR cycle FROM 1 TO max_cycles: IF "observe" IN actions: ingest_child_telemetry(context) IF "steer" IN actions AND steer_cooldown_elapsed(): steer_child(context, node) -- Evaluate stop conditions child_status = context.get_string("context.stack.child.status") IF child_status IN {"completed", "failed"}: child_outcome = context.get_string("context.stack.child.outcome") IF child_outcome == "success": RETURN Outcome(status=SUCCESS, notes="Child completed") IF child_status == "failed": RETURN Outcome(status=FAIL, failure_reason="Child failed") IF stop_condition is not empty: IF evaluate_condition(stop_condition, ..., context): RETURN Outcome(status=SUCCESS, notes="Stop condition satisfied") IF "wait" IN actions: sleep(poll_interval) RETURN Outcome(status=FAIL, failure_reason="Max cycles exceeded") ``` The manager pattern implements a **supervisor architecture** where: - **Observe** ingests worker telemetry (active stage, outcomes, retry counts, artifacts) - **Guard** scores worker progress and routes to continue, intervene, or escalate - **Steer** writes intervention instructions to the child's active stage directory ### 4.12 Custom Handlers New handler types are added by implementing the Handler interface and registering with the registry: ``` -- Define a custom handler MyCustomHandler: FUNCTION execute(node, context, graph, logs_root) -> Outcome: -- Custom logic here RETURN Outcome(status=SUCCESS) -- Register it registry.register("my_custom_type", MyCustomHandler()) -- Reference in DOT file my_node [type="my_custom_type", shape=box, custom_attr="value"] ``` **Handler contract:** - Handlers MUST be stateless or protect shared mutable state with synchronization. - Handler panics/exceptions MUST be caught by the engine and converted to FAIL outcomes. - Handlers SHOULD NOT embed provider-specific logic; LLM orchestration is delegated to the integrated SDK. --- ## 5. State and Context ### 5.1 Context The context is a thread-safe key-value store shared across all stages during a pipeline run. It is the primary mechanism for passing data between nodes. ``` Context: values : Map -- key-value store lock : ReadWriteLock -- thread safety for parallel access logs : List -- append-only run log FUNCTION set(key, value): ACQUIRE write lock values[key] = value RELEASE write lock FUNCTION get(key, default=NONE) -> Any: ACQUIRE read lock result = values.get(key, default) RELEASE read lock RETURN result FUNCTION get_string(key, default="") -> String: value = get(key) IF value is NONE: RETURN default RETURN string(value) FUNCTION append_log(entry): ACQUIRE write lock logs.append(entry) RELEASE write lock FUNCTION snapshot() -> Map: -- Returns a serializable deep copy of all values ACQUIRE read lock result = deep_copy(values) RELEASE read lock RETURN result FUNCTION clone() -> Context: -- Deep copy for parallel branch isolation ACQUIRE read lock new_context = new Context() new_context.values = deep_copy(values) new_context.logs = copy(logs) RELEASE read lock RETURN new_context FUNCTION apply_updates(updates): -- Merge a dictionary of updates into the context ACQUIRE write lock FOR EACH (key, value) IN updates: values[key] = value RELEASE write lock ``` **Built-in context keys set by the engine:** | Key | Type | Set By | Description | |---------------------------------------|---------|----------|-------------| | `outcome` | String | Engine | Last handler outcome status (`success`, `fail`, etc.) | | `preferred_label` | String | Engine | Last handler's preferred edge label | | `graph.goal` | String | Engine | Mirrored from graph `goal` attribute | | `current_node` | String | Engine | ID of the currently executing node | | `last_stage` | String | Handler | ID of the last completed stage | | `last_response` | String | Handler | Truncated text of the last LLM response | | `internal.retry_count.` | Integer | Engine | Retry counter for a specific node | **Context key namespace conventions:** | Prefix | Purpose | |---------------|------------------------------------------------| | `context.*` | Semantic state shared between nodes | | `graph.*` | Graph attributes mirrored at initialization | | `internal.*` | Engine bookkeeping (retry counters, timing) | | `parallel.*` | Parallel handler state (results, counts) | | `stack.*` | Supervisor/worker state | | `human.gate.*`| Human interaction state | | `work.*` | Per-item context for parallel work items | ### 5.2 Outcome The outcome is the result of executing a node handler. It drives routing decisions and state updates. ``` Outcome: status : StageStatus -- SUCCESS, FAIL, PARTIAL_SUCCESS, RETRY, SKIPPED preferred_label : String -- which edge label to follow (optional) suggested_next_ids : List -- explicit next node IDs (optional) context_updates : Map -- key-value pairs to merge into context notes : String -- human-readable execution summary failure_reason : String -- reason for failure (when status is FAIL or RETRY) ``` **StageStatus values:** | Status | Meaning | |--------------------|---------| | `SUCCESS` | Stage completed its work. Proceed to next edge. Reset retry counter. | | `PARTIAL_SUCCESS` | Stage completed with caveats. Treated as success for routing but notes describe what was incomplete. | | `RETRY` | Stage requests re-execution. Engine increments retry counter and re-executes if within limits. | | `FAIL` | Stage failed permanently. Engine looks for a fail edge or terminates the pipeline. | | `SKIPPED` | Stage was skipped (e.g., condition not met). Proceed without recording an outcome. | ### 5.3 Checkpoint A serializable snapshot of execution state, saved after each node completes. Enables crash recovery and resume. ``` Checkpoint: timestamp : Timestamp -- when this checkpoint was created current_node : String -- ID of the last completed node completed_nodes : List -- IDs of all completed nodes in order node_retries : Map -- retry counters per node context_values : Map -- serialized snapshot of the context logs : List -- run log entries FUNCTION save(path): -- Serialize to JSON and write to filesystem data = { "timestamp": timestamp, "current_node": current_node, "completed_nodes": completed_nodes, "node_retries": node_retries, "context": serialize_to_json(context_values), "logs": logs } write_json_file(path, data) FUNCTION load(path) -> Checkpoint: -- Deserialize from JSON file data = read_json_file(path) RETURN new Checkpoint from data ``` **Resume behavior:** 1. Load the checkpoint from `{logs_root}/checkpoint.json`. 2. Restore context state from `context_values`. 3. Restore `completed_nodes` to skip already-finished work. 4. Restore retry counters from `node_retries`. 5. Determine the next node to execute (the one after `current_node` in the traversal). 6. If the previous node used `full` fidelity, degrade to `summary:high` for the first resumed node, because in-memory LLM sessions cannot be serialized. After this one degraded hop, subsequent nodes may use `full` fidelity again. ### 5.4 Context Fidelity Context fidelity controls how much prior conversation and state is carried into the next node's LLM session. This is a core mechanism for managing context window usage across multi-stage pipelines. ``` FidelityMode ::= 'full' | 'truncate' | 'compact' | 'summary:low' | 'summary:medium' | 'summary:high' ``` | Mode | Session | Context Carried | Approximate Token Budget | |------------------|---------|---------------------------------------------------------|--------------------------| | `full` | Reused (same thread) | Full conversation history preserved | Unbounded (uses compaction) | | `truncate` | Fresh | Minimal: only graph goal and run ID | Minimal | | `compact` | Fresh | Structured bullet-point summary: completed stages, outcomes, key context values | Moderate | | `summary:low` | Fresh | Brief textual summary with minimal event counts | ~600 tokens | | `summary:medium` | Fresh | Moderate detail: recent stage outcomes, active context values, notable events | ~1500 tokens | | `summary:high` | Fresh | Detailed: many recent events, tool call summaries, comprehensive context | ~3000 tokens | **Fidelity resolution precedence (highest to lowest):** 1. Edge `fidelity` attribute (on the incoming edge) 2. Target node `fidelity` attribute 3. Graph `default_fidelity` attribute 4. Default when unset: `compact` **Thread resolution (for `full` fidelity):** When fidelity resolves to `full`, the engine determines a thread key for session reuse: 1. Target node `thread_id` attribute 2. Edge `thread_id` attribute 3. Graph-level default thread 4. Derived class from enclosing subgraph 5. Fallback: previous node ID Nodes that share the same thread key reuse the same LLM session. Nodes with different thread keys start fresh sessions. ### 5.5 Artifact Store The artifact store provides named, typed storage for large stage outputs that do not belong in the context (which should contain only small scalar values for routing and checkpoint serialization). ``` ArtifactStore: artifacts : Map lock : ReadWriteLock base_dir : String or NONE -- filesystem directory for file-backed artifacts FUNCTION store(artifact_id, name, data) -> ArtifactInfo: size = byte_size(data) is_file_backed = (size > FILE_BACKING_THRESHOLD) AND (base_dir is not NONE) IF is_file_backed: write data to "{base_dir}/artifacts/{artifact_id}.json" stored_data = file_path ELSE: stored_data = data info = ArtifactInfo(id=artifact_id, name=name, size=size, is_file_backed=is_file_backed) artifacts[artifact_id] = (info, stored_data) RETURN info FUNCTION retrieve(artifact_id) -> Any: IF artifact_id NOT IN artifacts: RAISE "Artifact not found" (info, data) = artifacts[artifact_id] IF info.is_file_backed: RETURN read_json_file(data) RETURN data FUNCTION has(artifact_id) -> Boolean FUNCTION list() -> List FUNCTION remove(artifact_id) FUNCTION clear() ArtifactInfo: id : String name : String size_bytes : Integer stored_at : Timestamp is_file_backed : Boolean ``` The default file-backing threshold is 100KB. Artifacts below this threshold are stored in memory; above it, they are written to disk. ### 5.6 Run Directory Structure Each pipeline execution produces a directory tree for logging, checkpoints, and artifacts: ``` {logs_root}/ checkpoint.json -- Serialized checkpoint after each node manifest.json -- Pipeline metadata (name, goal, start time) {node_id}/ status.json -- Node execution outcome prompt.md -- Rendered prompt sent to LLM response.md -- LLM response text artifacts/ {artifact_id}.json -- File-backed artifacts ``` --- ## 6. Human-in-the-Loop (Interviewer Pattern) ### 6.1 Interviewer Interface All human interaction in Attractor goes through an Interviewer interface. This abstraction allows the pipeline to present questions to a human and receive answers through any frontend: CLI, web UI, Slack bot, or a programmatic queue for testing. ``` INTERFACE Interviewer: FUNCTION ask(question: Question) -> Answer FUNCTION ask_multiple(questions: List) -> List FUNCTION inform(message: String, stage: String) -> Void ``` ### 6.2 Question Model ``` Question: text : String -- the question to present to the human type : QuestionType -- determines the UI and valid answers options : List