API Reference

OpenSymbolicAI Core Runtime.

CheckpointStatus

Bases: str, Enum

Status of an execution checkpoint.

Source code in src/opensymbolicai/checkpoint.py

class CheckpointStatus(str, Enum):
    """Status of an execution checkpoint."""

    PENDING = "pending"  # Not yet started
    RUNNING = "running"  # Currently executing
    PAUSED = "paused"  # Manually paused
    AWAITING_APPROVAL = "awaiting_approval"  # Waiting for mutation approval
    COMPLETED = "completed"  # Successfully finished
    FAILED = "failed"  # Execution failed

CheckpointStore

Bases: Protocol

Protocol for checkpoint storage backends.

Implement this protocol to store checkpoints in your preferred backend (PostgreSQL, Redis, MongoDB, filesystem, etc.).

Example

class RedisCheckpointStore: def init(self, redis_client): self.redis = redis_client

def save(self, checkpoint: ExecutionCheckpoint) -> None:
    self.redis.set(
        f"checkpoint:{checkpoint.checkpoint_id}",
        checkpoint.model_dump_json()
    )

def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
    data = self.redis.get(f"checkpoint:{checkpoint_id}")
    if data:
        return ExecutionCheckpoint.model_validate_json(data)
    return None

def delete(self, checkpoint_id: str) -> None:
    self.redis.delete(f"checkpoint:{checkpoint_id}")

def list_by_status(self, status: CheckpointStatus) -> list[str]:
    # Implementation depends on your indexing strategy
    ...

Source code in src/opensymbolicai/checkpoint.py

@runtime_checkable
class CheckpointStore(Protocol):
    """Protocol for checkpoint storage backends.

    Implement this protocol to store checkpoints in your preferred backend
    (PostgreSQL, Redis, MongoDB, filesystem, etc.).

    Example:
        class RedisCheckpointStore:
            def __init__(self, redis_client):
                self.redis = redis_client

            def save(self, checkpoint: ExecutionCheckpoint) -> None:
                self.redis.set(
                    f"checkpoint:{checkpoint.checkpoint_id}",
                    checkpoint.model_dump_json()
                )

            def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
                data = self.redis.get(f"checkpoint:{checkpoint_id}")
                if data:
                    return ExecutionCheckpoint.model_validate_json(data)
                return None

            def delete(self, checkpoint_id: str) -> None:
                self.redis.delete(f"checkpoint:{checkpoint_id}")

            def list_by_status(self, status: CheckpointStatus) -> list[str]:
                # Implementation depends on your indexing strategy
                ...
    """

    def save(self, checkpoint: ExecutionCheckpoint) -> None:
        """Save a checkpoint to storage."""
        ...

    def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
        """Load a checkpoint from storage. Returns None if not found."""
        ...

    def delete(self, checkpoint_id: str) -> None:
        """Delete a checkpoint from storage."""
        ...

    def list_by_status(self, status: CheckpointStatus) -> list[str]:
        """List checkpoint IDs with the given status."""
        ...

delete(checkpoint_id)

Delete a checkpoint from storage.

Source code in src/opensymbolicai/checkpoint.py

def delete(self, checkpoint_id: str) -> None:
    """Delete a checkpoint from storage."""
    ...

list_by_status(status)

List checkpoint IDs with the given status.

Source code in src/opensymbolicai/checkpoint.py

def list_by_status(self, status: CheckpointStatus) -> list[str]:
    """List checkpoint IDs with the given status."""
    ...

load(checkpoint_id)

Load a checkpoint from storage. Returns None if not found.

Source code in src/opensymbolicai/checkpoint.py

def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
    """Load a checkpoint from storage. Returns None if not found."""
    ...

save(checkpoint)

Save a checkpoint to storage.

Source code in src/opensymbolicai/checkpoint.py

def save(self, checkpoint: ExecutionCheckpoint) -> None:
    """Save a checkpoint to storage."""
    ...

DecompositionInfo

Bases: BaseModel

Metadata describing a decomposition method, passed to PromptProvider selectors.

Source code in src/opensymbolicai/models.py

class DecompositionInfo(BaseModel):
    """Metadata describing a decomposition method, passed to PromptProvider selectors."""

    name: str = Field(..., description="Method name.")
    intent: str = Field(default="", description="High-level intent of the decomposition.")
    expanded_intent: str = Field(default="", description="Step-by-step breakdown of the approach.")
    parameters: list[ParameterInfo] = Field(
        default_factory=list, description="Positional parameters (excluding self)."
    )
    return_type: str = Field(default="Any", description="Return type annotation as a string.")
    source: str = Field(default="", description="Source code body of the decomposition.")

DesignExecute

Bases: PlanExecute

Agent that generates and executes Python plans with control flow.

DesignExecute extends PlanExecute to allow for, while, if/elif/else, try/except, and raise in LLM-generated plans while maintaining safety (loop limits, blocked dangerous ops) and full traceability (every primitive call is recorded).

Plans may include: - Assignment statements: result = add(1, 2) - For loops: for item in items: - While loops: while condition: - Conditionals: if/elif/else - Try/except blocks: try: ... except ValueError: ... - Raise statements: raise ValueError("message") - break/continue (configurable)

Plans may NOT include: - Function/class definitions - Import statements - With blocks - exec/eval/open or other dangerous builtins - Private/dunder attribute access

Tracing works by wrapping all primitives with instrumentation wrappers that record each call (name, args, result, timing) into a flat trace.

Example::

class DataProcessor(DesignExecute):
    @primitive(read_only=True)
    def process(self, item: str) -> str:
        return item.upper()

    @primitive(read_only=True)
    def summarize(self, items: list[str]) -> str:
        return ", ".join(items)

Source code in src/opensymbolicai/blueprints/design_execute.py

class DesignExecute(PlanExecute):
    """Agent that generates and executes Python plans with control flow.

    DesignExecute extends PlanExecute to allow for, while, if/elif/else,
    try/except, and raise in LLM-generated plans while maintaining safety
    (loop limits, blocked dangerous ops) and full traceability (every
    primitive call is recorded).

    Plans may include:
    - Assignment statements: ``result = add(1, 2)``
    - For loops: ``for item in items:``
    - While loops: ``while condition:``
    - Conditionals: ``if/elif/else``
    - Try/except blocks: ``try: ... except ValueError: ...``
    - Raise statements: ``raise ValueError("message")``
    - break/continue (configurable)

    Plans may NOT include:
    - Function/class definitions
    - Import statements
    - With blocks
    - exec/eval/open or other dangerous builtins
    - Private/dunder attribute access

    Tracing works by wrapping all primitives with instrumentation wrappers
    that record each call (name, args, result, timing) into a flat trace.

    Example::

        class DataProcessor(DesignExecute):
            @primitive(read_only=True)
            def process(self, item: str) -> str:
                return item.upper()

            @primitive(read_only=True)
            def summarize(self, items: list[str]) -> str:
                return ", ".join(items)
    """

    # Exception types available in plans for raise/try-except
    ALLOWED_EXCEPTIONS: dict[str, type] = {
        "ValueError": ValueError,
        "RuntimeError": RuntimeError,
        "TypeError": TypeError,
        "KeyError": KeyError,
    }

    def __init__(
        self,
        llm: LLM | LLMConfig,
        name: str = "",
        description: str = "",
        config: DesignExecuteConfig | None = None,
    ) -> None:
        super().__init__(
            llm=llm,
            name=name,
            description=description,
            config=config or DesignExecuteConfig(),
        )
        # Make exception types available for validation and execution
        self.allowed_builtins.update(self.ALLOWED_EXCEPTIONS)

    @property
    def design_config(self) -> DesignExecuteConfig:
        """Get the DesignExecute-specific config."""
        assert isinstance(self.config, DesignExecuteConfig)
        return self.config

    @property
    def blueprint_type(self) -> str:
        """The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'."""
        return "DesignExecute"

    # -------------------------------------------------------------------------
    # Prompt Building (Override)
    # -------------------------------------------------------------------------

    def build_plan_prompt(self, task: str, feedback: str | None = None) -> str:
        """Build prompt that allows control flow.

        Fully overrides the parent prompt to replace the rules section
        with DesignExecute-specific rules permitting loops and conditionals.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        # Build primitive documentation
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        # Build decomposition examples
        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build conversation history section if in multi-turn mode
        history_section = ""
        if self.config.multi_turn and self._history:
            history_section = f"""
## Conversation History

Previous turns in this conversation. You can reference variables from previous turns.

{self._format_history_for_prompt()}

"""

        # Build feedback section if retrying after a failed plan
        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        max_iters = self.design_config.max_loop_iterations

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

Here are examples of how to compose primitives:

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{history_section}{feedback_section}## Task

Generate Python code to accomplish this task: {task}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. You can use assignment statements, for loops, while loops, if/elif/else, try/except, and raise
2. You can ONLY call the primitive methods listed above
3. Do NOT use imports, function definitions, class definitions, or with statements
4. Do NOT use any dangerous operations (exec, eval, open, etc.)
5. While loops MUST have a clear termination condition (max {max_iters} iterations)
6. Assign the final result to a variable named `result`
7. Call primitives directly (e.g. `lookup_price(item=item)`), do NOT use `self.`
8. Use loops when you need to process collections or repeat operations
9. Use conditionals when the task requires branching logic
10. Use raise ValueError("message") to signal errors (e.g. missing required input)
11. Available exception types: ValueError, RuntimeError, TypeError, KeyError

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

    # -------------------------------------------------------------------------
    # Plan Validation (Override)
    # -------------------------------------------------------------------------

    def validate_plan(self, plan: str) -> None:
        """Validate plan allowing control flow but blocking dangerous ops.

        Compared to PlanExecute, this allows For, While, If, AugAssign,
        and Expr (bare function calls) at the top level.
        """
        try:
            tree = ast.parse(plan)
        except SyntaxError as e:
            raise ValueError(f"Invalid Python syntax: {e}") from e

        primitive_names = self._get_primitive_names()

        # These are still disallowed even in DesignExecute
        disallowed_statements: tuple[type, ...] = (
            ast.With,
            ast.FunctionDef,
            ast.AsyncFunctionDef,
            ast.ClassDef,
            ast.Import,
            ast.ImportFrom,
            ast.Global,
            ast.Nonlocal,
            ast.Assert,
            ast.Delete,
        )
        if hasattr(ast, "Match"):
            disallowed_statements = (*disallowed_statements, ast.Match)

        # Optionally disallow break/continue
        if not self.design_config.allow_break_continue:
            disallowed_statements = (*disallowed_statements, ast.Break, ast.Continue)

        for node in ast.walk(tree):
            if isinstance(node, disallowed_statements):
                node_type = type(node).__name__
                raise ValueError(f"{node_type} statements are not allowed in plans")

        # Validate that top-level statements are allowed types
        allowed_top_level = (
            ast.Assign,
            ast.AnnAssign,
            ast.AugAssign,
            ast.For,
            ast.While,
            ast.If,
            ast.Expr,
            ast.Try,
            ast.Raise,
        )
        for stmt in tree.body:
            if not isinstance(stmt, allowed_top_level):
                stmt_type = type(stmt).__name__
                raise ValueError(
                    f"Statement type '{stmt_type}' is not allowed at top level. "
                    f"Allowed: assignments, for, while, if, try/except, raise, expressions."
                )

        self._validate_ast_nodes(tree, primitive_names)

    # -------------------------------------------------------------------------
    # AST Transformation: Loop Guard Injection
    # -------------------------------------------------------------------------

    def _inject_loop_guards(self, tree: ast.Module) -> ast.Module:
        """Transform AST to inject loop iteration counters.

        For every ``for`` and ``while`` loop, injects a counter variable
        and a limit check that calls ``__loop_guard_raise__()`` if exceeded.
        """
        transformer = _LoopGuardTransformer()
        new_tree = transformer.visit(tree)
        ast.fix_missing_locations(new_tree)
        return new_tree  # type: ignore[no-any-return]

    # -------------------------------------------------------------------------
    # Primitive Instrumentation
    # -------------------------------------------------------------------------

    def _create_traced_primitive(
        self,
        name: str,
        method: Any,
        trace_list: list[ExecutionStep],
        step_counter: list[int],
        read_only_map: dict[str, bool],
        namespace: dict[str, Any],
        reserved_names: set[str],
    ) -> Any:
        """Create a wrapper around a primitive that records each call."""
        sig = inspect.signature(method)
        param_names = [p for p in sig.parameters if p != "self"]

        def traced_wrapper(*args: Any, **kwargs: Any) -> Any:
            step_number = step_counter[0]
            step_counter[0] += 1

            # Check total call limit
            if step_number > self.design_config.max_total_primitive_calls:
                raise RuntimeError(
                    f"Exceeded maximum total primitive calls "
                    f"({self.design_config.max_total_primitive_calls}). "
                    f"Plan may contain too many iterations."
                )

            # Build args dict for tracing
            traced_args: dict[str, ArgumentValue] = {}
            for i, arg_val in enumerate(args):
                pname = param_names[i] if i < len(param_names) else f"arg{i}"
                traced_args[pname] = ArgumentValue(
                    expression=repr(arg_val),
                    resolved_value=arg_val,
                )
            for kw_name, kw_val in kwargs.items():
                traced_args[kw_name] = ArgumentValue(
                    expression=repr(kw_val),
                    resolved_value=kw_val,
                )

            # Namespace snapshot before
            namespace_before = self._snapshot_namespace(namespace, reserved_names)

            # Check mutation hook
            is_mutation = not read_only_map.get(name, True)
            if is_mutation and self.config.on_mutation is not None:
                plain_args = {k: v.resolved_value for k, v in traced_args.items()}
                hook_context = MutationHookContext(
                    method_name=name,
                    args=plain_args,
                    result=None,
                    step=None,
                )
                rejection_reason = self.config.on_mutation(hook_context)
                if rejection_reason is not None:
                    step = ExecutionStep(
                        step_number=step_number,
                        statement=f"{name}(...)",
                        primitive_called=name,
                        args=traced_args,
                        namespace_before=namespace_before,
                        namespace_after=namespace_before,
                        time_seconds=0.0,
                        success=False,
                        error=f"{MUTATION_REJECTED_PREFIX}: {rejection_reason}",
                    )
                    trace_list.append(step)
                    raise RuntimeError(f"{MUTATION_REJECTED_PREFIX}: {rejection_reason}")

            # Execute the actual primitive
            start_time = time.perf_counter()
            try:
                result = method(*args, **kwargs)
                elapsed = time.perf_counter() - start_time

                namespace_after = self._snapshot_namespace(namespace, reserved_names)

                result_json = (
                    NULL_JSON
                    if self.config.skip_result_serialization
                    else json.dumps(result, default=str)
                )

                arg_strs = [
                    f"{k}={v.expression}" for k, v in traced_args.items()
                ]
                step = ExecutionStep(
                    step_number=step_number,
                    statement=f"{name}({', '.join(arg_strs)})",
                    primitive_called=name,
                    args=traced_args,
                    namespace_before=namespace_before,
                    namespace_after=namespace_after,
                    result_type=(
                        type(result).__name__ if result is not None else NONE_TYPE_NAME
                    ),
                    result_value=result,
                    result_json=result_json,
                    time_seconds=elapsed,
                    success=True,
                )
                trace_list.append(step)

                if self._tracer and self._tracer.config.capture_execution_steps:
                    self._tracer.emit(
                        EventType.EXECUTION_STEP,
                        self._tracer.filter.execution_step(step.model_dump()),
                    )

                return result

            except Exception as e:
                elapsed = time.perf_counter() - start_time
                namespace_after = self._snapshot_namespace(namespace, reserved_names)
                step = ExecutionStep(
                    step_number=step_number,
                    statement=f"{name}(...)",
                    primitive_called=name,
                    args=traced_args,
                    namespace_before=namespace_before,
                    namespace_after=namespace_after,
                    time_seconds=elapsed,
                    success=False,
                    error=str(e),
                )
                trace_list.append(step)
                raise

        return traced_wrapper

    # -------------------------------------------------------------------------
    # Execution (Override)
    # -------------------------------------------------------------------------

    def execute(self, plan: str) -> ExecutionResult:
        """Execute a plan with control flow support and full tracing.

        Unlike PlanExecute which executes statement-by-statement, this
        instruments all primitives with tracing wrappers and executes
        the entire plan as a single block. Loop guards are injected via
        AST transformation to prevent infinite loops.
        """
        self.validate_plan(plan)

        exec_span: str | None = None
        if self._tracer:
            exec_span = self._tracer.start_span(
                EventType.EXECUTION_START,
                {"plan": plan} if self._tracer.config.capture_plan_source else {},
            )

        tree = ast.parse(plan)

        # Inject loop guards for safety
        tree = self._inject_loop_guards(tree)

        # Build execution namespace — primitives are added below as traced
        # wrappers, NOT as raw methods, so the LLM cannot bypass tracing.
        namespace: dict[str, Any] = {}
        namespace.update(self.allowed_builtins)
        if self.config.multi_turn:
            namespace.update(self._persisted_namespace)

        # Setup tracing infrastructure
        trace_steps: list[ExecutionStep] = []
        step_counter = [1]  # mutable counter
        read_only_map = self._get_primitive_read_only_map()

        # Reserved names include internal loop guard variables
        reserved_names = (
            {"__loop_limit__", "__loop_guard_raise__"}
            | self._build_reserved_names()
        )

        # Instrument primitives with tracing wrappers
        for prim_name, method in self._get_primitive_methods():
            traced = self._create_traced_primitive(
                prim_name,
                method,
                trace_steps,
                step_counter,
                read_only_map,
                namespace,
                reserved_names,
            )
            namespace[prim_name] = traced

        # Add loop guard infrastructure to namespace
        def _loop_guard_raise() -> None:
            raise RuntimeError(_LOOP_LIMIT_EXCEEDED_MSG)

        namespace["__loop_guard_raise__"] = _loop_guard_raise
        namespace["__loop_limit__"] = self.design_config.max_loop_iterations

        total_start = time.perf_counter()

        try:
            exec(  # noqa: S102
                compile(tree, PLAN_COMPILE_SOURCE, "exec"),
                empty_builtins(),
                namespace,
            )
        except RuntimeError as e:
            err_msg = str(e)
            if MUTATION_REJECTED_PREFIX not in err_msg:
                if _LOOP_LIMIT_EXCEEDED_MSG in err_msg:
                    trace_steps.append(
                        ExecutionStep(
                            step_number=step_counter[0],
                            statement="<loop limit exceeded>",
                            success=False,
                            error=f"Loop exceeded maximum iterations "
                            f"({self.design_config.max_loop_iterations})",
                        )
                    )
                else:
                    trace_steps.append(
                        ExecutionStep(
                            step_number=step_counter[0],
                            statement="<runtime error>",
                            success=False,
                            error=err_msg,
                        )
                    )
        except Exception as e:
            trace_steps.append(
                ExecutionStep(
                    step_number=step_counter[0],
                    statement="<execution error>",
                    success=False,
                    error=str(e),
                )
            )

        total_elapsed = time.perf_counter() - total_start

        self._persist_user_variables(namespace, reserved_names)

        trace = ExecutionTrace(
            steps=trace_steps,
            total_time_seconds=total_elapsed,
        )

        # Determine final result
        result_value = None
        result_type = NONE_TYPE_NAME
        result_name = ""
        result_json = NULL_JSON

        # Priority 1: explicit 'result' variable
        if "result" in namespace and "result" not in reserved_names:
            result_value = namespace["result"]
            result_name = "result"
        else:
            # Priority 2: last user-assigned variable in namespace
            for key in reversed(list(namespace.keys())):
                if key not in reserved_names and not key.startswith("__loop_guard_"):
                    result_value = namespace[key]
                    result_name = key
                    break

        if result_value is not None:
            result_type = type(result_value).__name__
            if not self.config.skip_result_serialization:
                try:
                    result_json = json.dumps(result_value, default=str)
                except (TypeError, ValueError):
                    result_json = NULL_JSON

        result = ExecutionResult(
            value_type=result_type,
            value_name=result_name,
            value_json=result_json,
            trace=trace,
        )

        if self._tracer and exec_span:
            self._tracer.end_span(
                exec_span,
                EventType.EXECUTION_COMPLETE,
                ExecutionSummary(
                    value_type=result.value_type,
                    value_name=result.value_name,
                    step_count=trace.step_count,
                    all_succeeded=trace.all_succeeded,
                    total_time_seconds=total_elapsed,
                ).model_dump(),
            )

        return result

    # -------------------------------------------------------------------------
    # Checkpoint-based Execution (Not Supported)
    # -------------------------------------------------------------------------

    def execute_stepwise(
        self,
        task: str,
        plan_context: PlanContext | None = None,
        serializer: SerializerRegistry | None = None,
        checkpoint_id: str | None = None,
    ) -> Iterator[ExecutionCheckpoint]:
        """Not supported for DesignExecute.

        Stepwise checkpoint execution is incompatible with block-based
        control flow execution. Use ``execute()`` or ``run()`` instead.
        """
        raise NotImplementedError(
            "DesignExecute does not support stepwise execution with control flow. "
            "Use execute() or run() instead."
        )

blueprint_type property

The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'.

design_config property

Get the DesignExecute-specific config.

build_plan_prompt(task, feedback=None)

Build prompt that allows control flow.

Fully overrides the parent prompt to replace the rules section with DesignExecute-specific rules permitting loops and conditionals.

Source code in src/opensymbolicai/blueprints/design_execute.py

    def build_plan_prompt(self, task: str, feedback: str | None = None) -> str:
        """Build prompt that allows control flow.

        Fully overrides the parent prompt to replace the rules section
        with DesignExecute-specific rules permitting loops and conditionals.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        # Build primitive documentation
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        # Build decomposition examples
        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build conversation history section if in multi-turn mode
        history_section = ""
        if self.config.multi_turn and self._history:
            history_section = f"""
## Conversation History

Previous turns in this conversation. You can reference variables from previous turns.

{self._format_history_for_prompt()}

"""

        # Build feedback section if retrying after a failed plan
        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        max_iters = self.design_config.max_loop_iterations

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

Here are examples of how to compose primitives:

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{history_section}{feedback_section}## Task

Generate Python code to accomplish this task: {task}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. You can use assignment statements, for loops, while loops, if/elif/else, try/except, and raise
2. You can ONLY call the primitive methods listed above
3. Do NOT use imports, function definitions, class definitions, or with statements
4. Do NOT use any dangerous operations (exec, eval, open, etc.)
5. While loops MUST have a clear termination condition (max {max_iters} iterations)
6. Assign the final result to a variable named `result`
7. Call primitives directly (e.g. `lookup_price(item=item)`), do NOT use `self.`
8. Use loops when you need to process collections or repeat operations
9. Use conditionals when the task requires branching logic
10. Use raise ValueError("message") to signal errors (e.g. missing required input)
11. Available exception types: ValueError, RuntimeError, TypeError, KeyError

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

execute(plan)

Execute a plan with control flow support and full tracing.

Unlike PlanExecute which executes statement-by-statement, this instruments all primitives with tracing wrappers and executes the entire plan as a single block. Loop guards are injected via AST transformation to prevent infinite loops.

Source code in src/opensymbolicai/blueprints/design_execute.py

def execute(self, plan: str) -> ExecutionResult:
    """Execute a plan with control flow support and full tracing.

    Unlike PlanExecute which executes statement-by-statement, this
    instruments all primitives with tracing wrappers and executes
    the entire plan as a single block. Loop guards are injected via
    AST transformation to prevent infinite loops.
    """
    self.validate_plan(plan)

    exec_span: str | None = None
    if self._tracer:
        exec_span = self._tracer.start_span(
            EventType.EXECUTION_START,
            {"plan": plan} if self._tracer.config.capture_plan_source else {},
        )

    tree = ast.parse(plan)

    # Inject loop guards for safety
    tree = self._inject_loop_guards(tree)

    # Build execution namespace — primitives are added below as traced
    # wrappers, NOT as raw methods, so the LLM cannot bypass tracing.
    namespace: dict[str, Any] = {}
    namespace.update(self.allowed_builtins)
    if self.config.multi_turn:
        namespace.update(self._persisted_namespace)

    # Setup tracing infrastructure
    trace_steps: list[ExecutionStep] = []
    step_counter = [1]  # mutable counter
    read_only_map = self._get_primitive_read_only_map()

    # Reserved names include internal loop guard variables
    reserved_names = (
        {"__loop_limit__", "__loop_guard_raise__"}
        | self._build_reserved_names()
    )

    # Instrument primitives with tracing wrappers
    for prim_name, method in self._get_primitive_methods():
        traced = self._create_traced_primitive(
            prim_name,
            method,
            trace_steps,
            step_counter,
            read_only_map,
            namespace,
            reserved_names,
        )
        namespace[prim_name] = traced

    # Add loop guard infrastructure to namespace
    def _loop_guard_raise() -> None:
        raise RuntimeError(_LOOP_LIMIT_EXCEEDED_MSG)

    namespace["__loop_guard_raise__"] = _loop_guard_raise
    namespace["__loop_limit__"] = self.design_config.max_loop_iterations

    total_start = time.perf_counter()

    try:
        exec(  # noqa: S102
            compile(tree, PLAN_COMPILE_SOURCE, "exec"),
            empty_builtins(),
            namespace,
        )
    except RuntimeError as e:
        err_msg = str(e)
        if MUTATION_REJECTED_PREFIX not in err_msg:
            if _LOOP_LIMIT_EXCEEDED_MSG in err_msg:
                trace_steps.append(
                    ExecutionStep(
                        step_number=step_counter[0],
                        statement="<loop limit exceeded>",
                        success=False,
                        error=f"Loop exceeded maximum iterations "
                        f"({self.design_config.max_loop_iterations})",
                    )
                )
            else:
                trace_steps.append(
                    ExecutionStep(
                        step_number=step_counter[0],
                        statement="<runtime error>",
                        success=False,
                        error=err_msg,
                    )
                )
    except Exception as e:
        trace_steps.append(
            ExecutionStep(
                step_number=step_counter[0],
                statement="<execution error>",
                success=False,
                error=str(e),
            )
        )

    total_elapsed = time.perf_counter() - total_start

    self._persist_user_variables(namespace, reserved_names)

    trace = ExecutionTrace(
        steps=trace_steps,
        total_time_seconds=total_elapsed,
    )

    # Determine final result
    result_value = None
    result_type = NONE_TYPE_NAME
    result_name = ""
    result_json = NULL_JSON

    # Priority 1: explicit 'result' variable
    if "result" in namespace and "result" not in reserved_names:
        result_value = namespace["result"]
        result_name = "result"
    else:
        # Priority 2: last user-assigned variable in namespace
        for key in reversed(list(namespace.keys())):
            if key not in reserved_names and not key.startswith("__loop_guard_"):
                result_value = namespace[key]
                result_name = key
                break

    if result_value is not None:
        result_type = type(result_value).__name__
        if not self.config.skip_result_serialization:
            try:
                result_json = json.dumps(result_value, default=str)
            except (TypeError, ValueError):
                result_json = NULL_JSON

    result = ExecutionResult(
        value_type=result_type,
        value_name=result_name,
        value_json=result_json,
        trace=trace,
    )

    if self._tracer and exec_span:
        self._tracer.end_span(
            exec_span,
            EventType.EXECUTION_COMPLETE,
            ExecutionSummary(
                value_type=result.value_type,
                value_name=result.value_name,
                step_count=trace.step_count,
                all_succeeded=trace.all_succeeded,
                total_time_seconds=total_elapsed,
            ).model_dump(),
        )

    return result

execute_stepwise(task, plan_context=None, serializer=None, checkpoint_id=None)

Not supported for DesignExecute.

Stepwise checkpoint execution is incompatible with block-based control flow execution. Use execute() or run() instead.

Source code in src/opensymbolicai/blueprints/design_execute.py

def execute_stepwise(
    self,
    task: str,
    plan_context: PlanContext | None = None,
    serializer: SerializerRegistry | None = None,
    checkpoint_id: str | None = None,
) -> Iterator[ExecutionCheckpoint]:
    """Not supported for DesignExecute.

    Stepwise checkpoint execution is incompatible with block-based
    control flow execution. Use ``execute()`` or ``run()`` instead.
    """
    raise NotImplementedError(
        "DesignExecute does not support stepwise execution with control flow. "
        "Use execute() or run() instead."
    )

validate_plan(plan)

Validate plan allowing control flow but blocking dangerous ops.

Compared to PlanExecute, this allows For, While, If, AugAssign, and Expr (bare function calls) at the top level.

Source code in src/opensymbolicai/blueprints/design_execute.py

def validate_plan(self, plan: str) -> None:
    """Validate plan allowing control flow but blocking dangerous ops.

    Compared to PlanExecute, this allows For, While, If, AugAssign,
    and Expr (bare function calls) at the top level.
    """
    try:
        tree = ast.parse(plan)
    except SyntaxError as e:
        raise ValueError(f"Invalid Python syntax: {e}") from e

    primitive_names = self._get_primitive_names()

    # These are still disallowed even in DesignExecute
    disallowed_statements: tuple[type, ...] = (
        ast.With,
        ast.FunctionDef,
        ast.AsyncFunctionDef,
        ast.ClassDef,
        ast.Import,
        ast.ImportFrom,
        ast.Global,
        ast.Nonlocal,
        ast.Assert,
        ast.Delete,
    )
    if hasattr(ast, "Match"):
        disallowed_statements = (*disallowed_statements, ast.Match)

    # Optionally disallow break/continue
    if not self.design_config.allow_break_continue:
        disallowed_statements = (*disallowed_statements, ast.Break, ast.Continue)

    for node in ast.walk(tree):
        if isinstance(node, disallowed_statements):
            node_type = type(node).__name__
            raise ValueError(f"{node_type} statements are not allowed in plans")

    # Validate that top-level statements are allowed types
    allowed_top_level = (
        ast.Assign,
        ast.AnnAssign,
        ast.AugAssign,
        ast.For,
        ast.While,
        ast.If,
        ast.Expr,
        ast.Try,
        ast.Raise,
    )
    for stmt in tree.body:
        if not isinstance(stmt, allowed_top_level):
            stmt_type = type(stmt).__name__
            raise ValueError(
                f"Statement type '{stmt_type}' is not allowed at top level. "
                f"Allowed: assignments, for, while, if, try/except, raise, expressions."
            )

    self._validate_ast_nodes(tree, primitive_names)

DesignExecuteConfig

Bases: PlanExecuteConfig

Configuration for DesignExecute agents with control flow support.

Source code in src/opensymbolicai/models.py

class DesignExecuteConfig(PlanExecuteConfig):
    """Configuration for DesignExecute agents with control flow support."""

    max_loop_iterations: int = Field(
        default=100,
        description="Maximum iterations per loop (for/while). Prevents infinite loops.",
    )
    max_total_primitive_calls: int = Field(
        default=1000,
        description="Maximum total primitive calls across the entire plan execution.",
    )
    allow_break_continue: bool = Field(
        default=True,
        description="Whether to allow break/continue in loops.",
    )

EventType

Bases: str, Enum

Types of trace events emitted during agent execution.

Source code in src/opensymbolicai/observability/events.py

class EventType(str, Enum):
    """Types of trace events emitted during agent execution."""

    # Lifecycle
    RUN_START = "run.start"
    RUN_COMPLETE = "run.complete"
    RUN_ERROR = "run.error"

    # Planning
    PLAN_START = "plan.start"
    PLAN_LLM_REQUEST = "plan.llm_request"
    PLAN_LLM_RESPONSE = "plan.llm_response"
    PLAN_VALIDATION_ERROR = "plan.validation_error"
    PLAN_COMPLETE = "plan.complete"

    # Execution
    EXECUTION_START = "execution.start"
    EXECUTION_STEP = "execution.step"
    EXECUTION_COMPLETE = "execution.complete"
    MUTATION_BLOCKED = "execution.mutation_blocked"

    # Goal seeking
    GOAL_SEEK_START = "goal.seek_start"
    GOAL_ITERATION_START = "goal.iteration_start"
    GOAL_EVALUATION = "goal.evaluation"
    GOAL_EVALUATOR_LLM_REQUEST = "goal.evaluator_llm_request"
    GOAL_EVALUATOR_LLM_RESPONSE = "goal.evaluator_llm_response"
    GOAL_EVALUATOR_STEP = "goal.evaluator_step"
    GOAL_ITERATION_COMPLETE = "goal.iteration_complete"
    GOAL_SEEK_COMPLETE = "goal.seek_complete"

ExecutionCheckpoint

Bases: BaseModel

Serializable execution state for pause/resume across distributed workers.

Source code in src/opensymbolicai/checkpoint.py

class ExecutionCheckpoint(BaseModel):
    """Serializable execution state for pause/resume across distributed workers."""

    # Identity
    checkpoint_id: str = Field(
        default_factory=lambda: str(uuid.uuid4()),
        description="Unique checkpoint identifier",
    )
    task: str = Field(..., description="The original task description")
    plan: str = Field(..., description="The Python plan being executed")

    # Plan generation context (LLM history)
    plan_context: PlanContext | None = Field(
        default=None,
        description="Full context of how the plan was generated, including LLM interactions",
    )

    # Execution progress
    current_step: int = Field(default=0, description="Next step to execute (0-indexed)")
    total_steps: int = Field(..., description="Total number of steps in the plan")
    status: CheckpointStatus = Field(
        default=CheckpointStatus.PENDING, description="Current execution status"
    )

    # State
    namespace_snapshot: dict[str, SerializedValue] = Field(
        default_factory=dict,
        description="Serialized namespace variables",
    )
    completed_steps: list[ExecutionStep] = Field(
        default_factory=list, description="Steps that have been executed"
    )

    # Mutation approval
    pending_mutation: PendingMutation | None = Field(
        default=None,
        description="Mutation awaiting approval (if status is awaiting_approval)",
    )

    # Error tracking
    error: str | None = Field(
        default=None, description="Error message if status is failed"
    )

    # Metadata
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        description="When the checkpoint was created",
    )
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        description="When the checkpoint was last updated",
    )
    worker_id: str | None = Field(
        default=None, description="ID of the worker that created/last updated this"
    )

    # Result (when completed)
    result_value: SerializedValue | None = Field(
        default=None, description="The final result value (when status is completed)"
    )
    result_variable: str = Field(
        default="", description="Name of the variable containing the result"
    )

    def touch(self, worker_id: str | None = None) -> None:
        """Update the updated_at timestamp and optionally worker_id."""
        self.updated_at = datetime.now(UTC)
        if worker_id is not None:
            self.worker_id = worker_id

    @property
    def is_resumable(self) -> bool:
        """Check if this checkpoint can be resumed."""
        return self.status in (
            CheckpointStatus.PENDING,
            CheckpointStatus.RUNNING,
            CheckpointStatus.PAUSED,
            CheckpointStatus.AWAITING_APPROVAL,
        )

    @property
    def is_terminal(self) -> bool:
        """Check if this checkpoint is in a terminal state."""
        return self.status in (CheckpointStatus.COMPLETED, CheckpointStatus.FAILED)

    @property
    def progress_fraction(self) -> float:
        """Get execution progress as a fraction (0.0 to 1.0)."""
        if self.total_steps == 0:
            return 0.0
        return self.current_step / self.total_steps

    model_config = {"arbitrary_types_allowed": True}

is_resumable property

Check if this checkpoint can be resumed.

is_terminal property

Check if this checkpoint is in a terminal state.

progress_fraction property

Get execution progress as a fraction (0.0 to 1.0).

touch(worker_id=None)

Update the updated_at timestamp and optionally worker_id.

Source code in src/opensymbolicai/checkpoint.py

def touch(self, worker_id: str | None = None) -> None:
    """Update the updated_at timestamp and optionally worker_id."""
    self.updated_at = datetime.now(UTC)
    if worker_id is not None:
        self.worker_id = worker_id

ExecutionError

Bases: Exception

Base exception for all primitive execution errors.

This is the root of the exception hierarchy. All custom exceptions should inherit from this class. When raised in a primitive, execution will stop and the error details will be captured in the execution trace.

Attributes:

Name	Type	Description
message		Human-readable error message.
code		Optional error code for programmatic handling.
details		Optional dictionary with additional context.
halt_execution		Whether this exception should stop plan execution.

Source code in src/opensymbolicai/exceptions.py

class ExecutionError(Exception):
    """Base exception for all primitive execution errors.

    This is the root of the exception hierarchy. All custom exceptions
    should inherit from this class. When raised in a primitive, execution
    will stop and the error details will be captured in the execution trace.

    Attributes:
        message: Human-readable error message.
        code: Optional error code for programmatic handling.
        details: Optional dictionary with additional context.
        halt_execution: Whether this exception should stop plan execution.
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = None,
        details: dict[str, Any] | None = None,
        halt_execution: bool = True,
    ) -> None:
        """Initialize the exception.

        Args:
            message: Human-readable error message.
            code: Optional error code for programmatic handling (e.g., "INVALID_INPUT").
            details: Optional dictionary with additional context about the error.
            halt_execution: Whether this exception should stop plan execution.
                           Defaults to True.
        """
        super().__init__(message)
        self.message = message
        self.code = code
        self.details = details or {}
        self.halt_execution = halt_execution

    def __str__(self) -> str:
        """Return string representation of the exception."""
        parts = [self.message]
        if self.code:
            parts.insert(0, f"[{self.code}]")
        return " ".join(parts)

    def __repr__(self) -> str:
        """Return detailed representation of the exception."""
        return (
            f"{self.__class__.__name__}("
            f"message={self.message!r}, "
            f"code={self.code!r}, "
            f"details={self.details!r}, "
            f"halt_execution={self.halt_execution!r})"
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert exception to a dictionary for serialization.

        Returns:
            Dictionary containing all exception attributes.
        """
        return {
            "type": self.__class__.__name__,
            "message": self.message,
            "code": self.code,
            "details": self.details,
            "halt_execution": self.halt_execution,
        }

__init__(message, *, code=None, details=None, halt_execution=True)

Initialize the exception.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Optional error code for programmatic handling (e.g., "INVALID_INPUT").	None
details	dict[str, Any] | None	Optional dictionary with additional context about the error.	None
halt_execution	bool	Whether this exception should stop plan execution. Defaults to True.	True

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    details: dict[str, Any] | None = None,
    halt_execution: bool = True,
) -> None:
    """Initialize the exception.

    Args:
        message: Human-readable error message.
        code: Optional error code for programmatic handling (e.g., "INVALID_INPUT").
        details: Optional dictionary with additional context about the error.
        halt_execution: Whether this exception should stop plan execution.
                       Defaults to True.
    """
    super().__init__(message)
    self.message = message
    self.code = code
    self.details = details or {}
    self.halt_execution = halt_execution

__repr__()

Return detailed representation of the exception.

Source code in src/opensymbolicai/exceptions.py

def __repr__(self) -> str:
    """Return detailed representation of the exception."""
    return (
        f"{self.__class__.__name__}("
        f"message={self.message!r}, "
        f"code={self.code!r}, "
        f"details={self.details!r}, "
        f"halt_execution={self.halt_execution!r})"
    )

__str__()

Return string representation of the exception.

Source code in src/opensymbolicai/exceptions.py

def __str__(self) -> str:
    """Return string representation of the exception."""
    parts = [self.message]
    if self.code:
        parts.insert(0, f"[{self.code}]")
    return " ".join(parts)

to_dict()

Convert exception to a dictionary for serialization.

Returns:

Type	Description
dict[str, Any]	Dictionary containing all exception attributes.

Source code in src/opensymbolicai/exceptions.py

def to_dict(self) -> dict[str, Any]:
    """Convert exception to a dictionary for serialization.

    Returns:
        Dictionary containing all exception attributes.
    """
    return {
        "type": self.__class__.__name__,
        "message": self.message,
        "code": self.code,
        "details": self.details,
        "halt_execution": self.halt_execution,
    }

ExecutionMetrics

Bases: BaseModel

Metrics from a complete orchestration run.

Source code in src/opensymbolicai/models.py

class ExecutionMetrics(BaseModel):
    """Metrics from a complete orchestration run."""

    plan_tokens: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens used for planning"
    )
    plan_time_seconds: float = Field(default=0.0, description="Time for planning")
    execute_time_seconds: float = Field(default=0.0, description="Time for execution")
    steps_executed: int = Field(default=0, description="Number of steps executed")
    provider: str = Field(default="", description="LLM provider used")
    model: str = Field(default="", description="Model used")

    @property
    def total_time_seconds(self) -> float:
        """Total time for planning and execution."""
        return self.plan_time_seconds + self.execute_time_seconds

total_time_seconds property

Total time for planning and execution.

ExecutionResult

Bases: BaseModel

Result from executing a plan.

Source code in src/opensymbolicai/models.py

class ExecutionResult(BaseModel):
    """Result from executing a plan."""

    value_type: str = Field(..., description="Python type name of the result")
    value_name: str = Field(default="", description="Variable name of the result")
    value_json: str = Field(default=NULL_JSON, description="JSON-serialized value")
    trace: ExecutionTrace = Field(
        default_factory=ExecutionTrace, description="Step-by-step execution trace"
    )

    def get_value(self) -> Any:
        """Deserialize and return the actual value."""
        return json.loads(self.value_json)

get_value()

Deserialize and return the actual value.

Source code in src/opensymbolicai/models.py

def get_value(self) -> Any:
    """Deserialize and return the actual value."""
    return json.loads(self.value_json)

ExecutionStep

Bases: BaseModel

A single step in plan execution.

Source code in src/opensymbolicai/models.py

class ExecutionStep(BaseModel):
    """A single step in plan execution."""

    step_number: int = Field(..., description="1-based step number")
    statement: str = Field(..., description="The Python statement executed")
    variable_name: str = Field(default="", description="Variable being assigned")
    primitive_called: str | None = Field(
        default=None, description="Name of primitive method called"
    )
    args: dict[str, ArgumentValue] = Field(
        default_factory=dict,
        description="Arguments passed to the primitive with both expression and resolved value",
    )
    namespace_before: dict[str, Any] = Field(
        default_factory=dict, description="Namespace snapshot before execution"
    )
    namespace_after: dict[str, Any] = Field(
        default_factory=dict, description="Namespace snapshot after execution"
    )
    result_type: str = Field(default="", description="Type of the result")
    result_value: Any = Field(default=None, description="The computed value")
    result_json: str = Field(default=NULL_JSON, description="JSON-serialized result")
    time_seconds: float = Field(default=0.0, description="Time taken for this step")
    success: bool = Field(default=True, description="Whether the step succeeded")
    error: str | None = Field(default=None, description="Error message if failed")

    model_config = {"arbitrary_types_allowed": True}

ExecutionTrace

Bases: BaseModel

Complete trace of plan execution.

Source code in src/opensymbolicai/models.py

class ExecutionTrace(BaseModel):
    """Complete trace of plan execution."""

    steps: list[ExecutionStep] = Field(
        default_factory=list, description="All executed steps"
    )
    total_time_seconds: float = Field(default=0.0, description="Total execution time")

    @property
    def step_count(self) -> int:
        """Number of steps executed."""
        return len(self.steps)

    @property
    def successful_steps(self) -> list[ExecutionStep]:
        """Get all successful steps."""
        return [s for s in self.steps if s.success]

    @property
    def failed_steps(self) -> list[ExecutionStep]:
        """Get all failed steps."""
        return [s for s in self.steps if not s.success]

    @property
    def all_succeeded(self) -> bool:
        """Check if all steps succeeded."""
        return all(s.success for s in self.steps)

    @property
    def last_step(self) -> ExecutionStep | None:
        """Get the last executed step."""
        return self.steps[-1] if self.steps else None

    @property
    def primitives_called(self) -> list[str]:
        """Get list of all primitives called."""
        return [s.primitive_called for s in self.steps if s.primitive_called]

all_succeeded property

Check if all steps succeeded.

failed_steps property

Get all failed steps.

last_step property

Get the last executed step.

primitives_called property

Get list of all primitives called.

step_count property

Number of steps executed.

successful_steps property

Get all successful steps.

FileCheckpointStore

File-based checkpoint store for simple persistence.

Source code in src/opensymbolicai/checkpoint.py

class FileCheckpointStore:
    """File-based checkpoint store for simple persistence."""

    def __init__(self, directory: str) -> None:
        """Initialize with a directory path for storing checkpoints.

        Args:
            directory: Path to directory where checkpoint JSON files are stored.
        """
        import os

        self.directory = directory
        os.makedirs(directory, exist_ok=True)

    def _path(self, checkpoint_id: str) -> str:
        import os

        # Sanitize checkpoint_id to prevent path traversal
        safe_id = "".join(c for c in checkpoint_id if c.isalnum() or c in "-_")
        return os.path.join(self.directory, f"{safe_id}.json")

    def save(self, checkpoint: ExecutionCheckpoint) -> None:
        """Save a checkpoint to a JSON file."""
        checkpoint.touch()
        with open(self._path(checkpoint.checkpoint_id), "w") as f:
            f.write(checkpoint.model_dump_json(indent=2))

    def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
        """Load a checkpoint from a JSON file."""
        import os

        path = self._path(checkpoint_id)
        if not os.path.exists(path):
            return None
        with open(path) as f:
            return ExecutionCheckpoint.model_validate_json(f.read())

    def delete(self, checkpoint_id: str) -> None:
        """Delete a checkpoint file."""
        import os

        path = self._path(checkpoint_id)
        if os.path.exists(path):
            os.remove(path)

    def list_by_status(self, status: CheckpointStatus) -> list[str]:
        """List checkpoint IDs with the given status."""
        import os

        result = []
        for filename in os.listdir(self.directory):
            if filename.endswith(".json"):
                checkpoint_id = filename[:-5]
                checkpoint = self.load(checkpoint_id)
                if checkpoint and checkpoint.status == status:
                    result.append(checkpoint_id)
        return result

    def list_all(self) -> list[str]:
        """List all checkpoint IDs."""
        import os

        return [f[:-5] for f in os.listdir(self.directory) if f.endswith(".json")]

__init__(directory)

Initialize with a directory path for storing checkpoints.

Parameters:

Name	Type	Description	Default
directory	str	Path to directory where checkpoint JSON files are stored.	required

Source code in src/opensymbolicai/checkpoint.py

def __init__(self, directory: str) -> None:
    """Initialize with a directory path for storing checkpoints.

    Args:
        directory: Path to directory where checkpoint JSON files are stored.
    """
    import os

    self.directory = directory
    os.makedirs(directory, exist_ok=True)

delete(checkpoint_id)

Delete a checkpoint file.

Source code in src/opensymbolicai/checkpoint.py

def delete(self, checkpoint_id: str) -> None:
    """Delete a checkpoint file."""
    import os

    path = self._path(checkpoint_id)
    if os.path.exists(path):
        os.remove(path)

list_all()

List all checkpoint IDs.

Source code in src/opensymbolicai/checkpoint.py

def list_all(self) -> list[str]:
    """List all checkpoint IDs."""
    import os

    return [f[:-5] for f in os.listdir(self.directory) if f.endswith(".json")]

list_by_status(status)

List checkpoint IDs with the given status.

Source code in src/opensymbolicai/checkpoint.py

def list_by_status(self, status: CheckpointStatus) -> list[str]:
    """List checkpoint IDs with the given status."""
    import os

    result = []
    for filename in os.listdir(self.directory):
        if filename.endswith(".json"):
            checkpoint_id = filename[:-5]
            checkpoint = self.load(checkpoint_id)
            if checkpoint and checkpoint.status == status:
                result.append(checkpoint_id)
    return result

load(checkpoint_id)

Load a checkpoint from a JSON file.

Source code in src/opensymbolicai/checkpoint.py

def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
    """Load a checkpoint from a JSON file."""
    import os

    path = self._path(checkpoint_id)
    if not os.path.exists(path):
        return None
    with open(path) as f:
        return ExecutionCheckpoint.model_validate_json(f.read())

save(checkpoint)

Save a checkpoint to a JSON file.

Source code in src/opensymbolicai/checkpoint.py

def save(self, checkpoint: ExecutionCheckpoint) -> None:
    """Save a checkpoint to a JSON file."""
    checkpoint.touch()
    with open(self._path(checkpoint.checkpoint_id), "w") as f:
        f.write(checkpoint.model_dump_json(indent=2))

FileTransport

Appends events as newline-delimited JSON to a file.

Each event is written as a single JSON line (JSONL format).

Parameters:

Name	Type	Description	Default
path	str	File path to write to. Parent directories are created if needed.	required

Source code in src/opensymbolicai/observability/transports/file.py

class FileTransport:
    """Appends events as newline-delimited JSON to a file.

    Each event is written as a single JSON line (JSONL format).

    Args:
        path: File path to write to. Parent directories are created if needed.
    """

    def __init__(self, path: str) -> None:
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._file: IO[str] | None = None

    def _ensure_open(self) -> IO[str]:
        if self._file is None or self._file.closed:
            self._file = open(self._path, "a")  # noqa: SIM115
        return self._file

    def send(self, events: list[TraceEvent]) -> None:
        """Append events as JSONL to the file."""
        f = self._ensure_open()
        for event in events:
            line = event.model_dump_json()
            f.write(line + "\n")
        f.flush()

    def close(self) -> None:
        """Close the file handle."""
        if self._file is not None and not self._file.closed:
            self._file.close()

close()

Close the file handle.

Source code in src/opensymbolicai/observability/transports/file.py

def close(self) -> None:
    """Close the file handle."""
    if self._file is not None and not self._file.closed:
        self._file.close()

send(events)

Append events as JSONL to the file.

Source code in src/opensymbolicai/observability/transports/file.py

def send(self, events: list[TraceEvent]) -> None:
    """Append events as JSONL to the file."""
    f = self._ensure_open()
    for event in events:
        line = event.model_dump_json()
        f.write(line + "\n")
    f.flush()

GoalContext

Bases: BaseModel

Accumulated structured knowledge across iterations.

This is the introspection boundary. Subclass to add domain-specific insight fields that update_context() populates from raw execution results. The planner and evaluator only see these fields — never raw results.

Source code in src/opensymbolicai/models.py

class GoalContext(BaseModel):
    """Accumulated structured knowledge across iterations.

    This is the introspection boundary. Subclass to add domain-specific
    insight fields that update_context() populates from raw execution results.
    The planner and evaluator only see these fields — never raw results.
    """

    goal: str = Field(..., description="The original goal")

    iterations: list[Iteration] = Field(
        default_factory=list, description="All completed iterations"
    )

    @property
    def iteration_count(self) -> int:
        """Number of completed iterations."""
        return len(self.iterations)

    @property
    def last_evaluation(self) -> GoalEvaluation | None:
        """Get the evaluation from the last iteration."""
        if self.iterations:
            return self.iterations[-1].evaluation
        return None

iteration_count property

Number of completed iterations.

last_evaluation property

Get the evaluation from the last iteration.

GoalEvaluation

Bases: BaseModel

Result of evaluating progress toward a goal.

Subclass to add domain-specific fields (findings, confidence, etc.)

Source code in src/opensymbolicai/models.py

class GoalEvaluation(BaseModel):
    """Result of evaluating progress toward a goal.

    Subclass to add domain-specific fields (findings, confidence, etc.)
    """

    goal_achieved: bool
    """Whether the goal has been achieved."""

goal_achieved instance-attribute

Whether the goal has been achieved.

GoalSeeking

Bases: DesignExecute

Agent that iteratively pursues a goal through plan-execute-evaluate cycles.

GoalSeeking extends DesignExecute with an iterative loop that: 1. Plans the next step toward a goal 2. Executes the plan 3. Introspects results into structured context (the introspection boundary) 4. Evaluates progress toward the goal 5. Repeats until the goal is achieved or termination conditions are met

Evaluation uses a two-tier approach: - Tier 1 (static): An @evaluator-decorated method on the agent - Tier 2 (dynamic): LLM-generated evaluator code from the goal

Subclasses should: 1. Define @primitive methods for available operations 2. Optionally define an @evaluator method for static evaluation 3. Override update_context() to extract insights from execution results 4. Override create_context() to return a custom GoalContext subclass

Source code in src/opensymbolicai/blueprints/goal_seeking.py

class GoalSeeking(DesignExecute):
    """Agent that iteratively pursues a goal through plan-execute-evaluate cycles.

    GoalSeeking extends DesignExecute with an iterative loop that:
    1. Plans the next step toward a goal
    2. Executes the plan
    3. Introspects results into structured context (the introspection boundary)
    4. Evaluates progress toward the goal
    5. Repeats until the goal is achieved or termination conditions are met

    Evaluation uses a two-tier approach:
    - Tier 1 (static): An @evaluator-decorated method on the agent
    - Tier 2 (dynamic): LLM-generated evaluator code from the goal

    Subclasses should:
    1. Define @primitive methods for available operations
    2. Optionally define an @evaluator method for static evaluation
    3. Override update_context() to extract insights from execution results
    4. Override create_context() to return a custom GoalContext subclass
    """

    def __init__(
        self,
        llm: LLM | LLMConfig,
        name: str = "",
        description: str = "",
        config: GoalSeekingConfig | None = None,
    ) -> None:
        """Initialize the GoalSeeking agent.

        Args:
            llm: LLM instance or config for plan generation.
            name: Agent name for prompts.
            description: Agent description for prompts.
            config: GoalSeeking-specific configuration.
        """
        cfg = config or GoalSeekingConfig()
        super().__init__(llm=llm, name=name, description=description, config=cfg)
        self.goal_config = cfg

    @property
    def blueprint_type(self) -> str:
        """The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'."""
        return "GoalSeeking"

    # -------------------------------------------------------------------------
    # Evaluator Introspection
    # -------------------------------------------------------------------------

    def _get_evaluator_method(
        self,
    ) -> Callable[[str, GoalContext], GoalEvaluation] | None:
        """Find the @evaluator-decorated method on this agent.

        Returns:
            The bound evaluator method, or None if no @evaluator is defined.
        """
        for name in dir(self):
            if name.startswith("__"):
                continue
            method = getattr(self, name, None)
            if (
                callable(method)
                and hasattr(method, "__method_type__")
                and method.__method_type__ == MethodType.EVALUATOR
            ):
                return method  # type: ignore[no-any-return]
        return None

    # -------------------------------------------------------------------------
    # Prompt Building
    # -------------------------------------------------------------------------

    def build_goal_prompt(
        self, goal: str, context: GoalContext, feedback: str | None = None
    ) -> str:
        """Build the prompt for planning the next iteration.

        Includes the original goal, available primitives, decomposition examples,
        and structured insights from context (NOT raw execution results).

        Args:
            goal: The goal being pursued.
            context: Accumulated context with structured insights.
            feedback: Error feedback from a failed plan attempt (for retry).

        Returns:
            Complete prompt for plan generation.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build context summary from structured insights (not raw results)
        context_section = ""
        if context.iteration_count > 0:
            context_section = f"""
## Previous Iterations

{context.iteration_count} iteration(s) completed so far.
"""
            # Include any custom fields from context subclasses
            custom_fields = _get_custom_context_fields(context)
            if custom_fields:
                context_section += "\n## Accumulated Knowledge\n\n"
                for field_name, field_value in custom_fields.items():
                    context_section += f"- {field_name}: {field_value!r}\n"

        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans to achieve a goal.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Goal

{goal}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{context_section}{feedback_section}## Task

Generate Python code for the NEXT step toward achieving the goal: {goal}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. You can use: assignments, for/while loops, if/elif/else, try/except
2. You can ONLY call the primitive methods listed above
3. Do NOT use imports, function definitions, class definitions, or with statements
4. Do NOT use any dangerous operations (exec, eval, open, etc.)
5. Call primitives directly (e.g. `search(query="...")`), do NOT use `self.`
6. Use if/else to handle missing data or check results before proceeding
7. Use try/except around search() or read_page() to handle errors gracefully
8. The last assigned variable will be the final result

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

    def build_evaluator_prompt(self, goal: str) -> str:
        """Build the prompt for evaluator code generation.

        Args:
            goal: The goal to generate evaluator code for.

        Returns:
            Prompt for the LLM to generate evaluator code.
        """
        primitives = self._get_prompt_primitives()
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        return f"""You are generating Python evaluation code for a goal-seeking agent.

{PROMPT_DEFINITIONS_BEGIN}

## Goal

{goal}

## Available Primitives

```python
{chr(10).join(primitive_docs)}
```

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}

## Task

Generate Python code that evaluates whether the goal has been achieved.

The code has access to these variables:
- `goal` (str): The goal being pursued
- `context` (GoalContext): Accumulated context with structured insights
- `self`: The agent instance (can call primitives)
- `GoalEvaluation`: The evaluation result class

The code MUST assign `result = GoalEvaluation(goal_achieved=...)`.

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. Output ONLY Python assignment statements
2. Check context fields (insights), NOT raw execution results
3. The final statement must be: `result = GoalEvaluation(goal_achieved=...)`
4. Do NOT use imports, loops, conditionals, or function definitions

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""

    # -------------------------------------------------------------------------
    # Planning
    # -------------------------------------------------------------------------

    def plan_iteration(
        self, goal: str, context: GoalContext, feedback: str | None = None
    ) -> PlanResult:
        """Generate a plan for the next iteration.

        Uses build_goal_prompt() directly (not self.plan()) to avoid
        double-wrapping the prompt through build_plan_prompt().

        Args:
            goal: The goal being pursued.
            context: Accumulated context from previous iterations.
            feedback: Error feedback from a failed plan attempt (for retry).

        Returns:
            PlanResult with the generated plan.
        """
        plan_span: str | None = None
        if self._tracer:
            plan_span = self._tracer.start_span(
                EventType.PLAN_START, {"task": goal, "feedback": feedback}
            )

        prompt = self.build_goal_prompt(goal, context, feedback=feedback)

        llm_span: str | None = None
        if self._tracer:
            llm_payload: dict[str, Any] = {}
            if self._tracer.config.capture_llm_prompts:
                llm_payload["prompt"] = prompt
            llm_span = self._tracer.start_span(
                EventType.PLAN_LLM_REQUEST, llm_payload, defer=True
            )

        start_time = time.perf_counter()
        response = self._llm.generate(prompt)
        elapsed = time.perf_counter() - start_time

        raw_response = response.text
        extracted_code = self._extract_code_block(raw_response)
        plan_text = self.on_code_extracted(raw_response, extracted_code)

        llm_interaction = LLMInteraction(
            prompt=prompt,
            response=raw_response,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            time_seconds=elapsed,
            provider=response.provider,
            model=response.model,
        )
        plan_generation = PlanGeneration(
            llm_interaction=llm_interaction,
            extracted_code=extracted_code,
        )

        if self._tracer and llm_span:
            response_payload = self._tracer.filter.llm_interaction(
                llm_interaction.model_dump()
            ) if self._tracer.config.capture_llm_responses else {}
            self._tracer.end_span(
                llm_span, EventType.PLAN_LLM_RESPONSE, response_payload
            )

        plan_result = PlanResult(
            plan=plan_text,
            usage=TokenUsage(
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
            ),
            time_seconds=elapsed,
            provider=response.provider,
            model=response.model,
            plan_generation=plan_generation,
        )

        if self._tracer and plan_span:
            self._tracer.end_span(
                plan_span,
                EventType.PLAN_COMPLETE,
                self._tracer.filter.plan_result(plan_result.model_dump()),
            )

        return plan_result

    # -------------------------------------------------------------------------
    # Evaluation
    # -------------------------------------------------------------------------

    def plan_evaluator(self, goal: str) -> str:
        """Generate evaluator code from the goal using the LLM.

        Called once before the loop starts. The generated code must assign
        `result = GoalEvaluation(goal_achieved=...)`.

        Args:
            goal: The goal to generate evaluator code for.

        Returns:
            Python code string for evaluation.
        """
        prompt = self.build_evaluator_prompt(goal)

        llm_span: str | None = None
        if self._tracer:
            llm_payload: dict[str, Any] = {}
            if self._tracer.config.capture_llm_prompts:
                llm_payload["prompt"] = prompt
            llm_span = self._tracer.start_span(
                EventType.GOAL_EVALUATOR_LLM_REQUEST, llm_payload, defer=True
            )

        start_time = time.perf_counter()
        response = self._llm.generate(prompt)
        elapsed = time.perf_counter() - start_time

        code = self._extract_code_block(response.text)

        if self._tracer and llm_span:
            response_payload: dict[str, Any] = {}
            if self._tracer.config.capture_llm_responses:
                llm_interaction = LLMInteraction(
                    prompt=prompt,
                    response=response.text,
                    input_tokens=response.usage.input_tokens,
                    output_tokens=response.usage.output_tokens,
                    time_seconds=elapsed,
                    provider=response.provider,
                    model=response.model,
                )
                response_payload = self._tracer.filter.llm_interaction(
                    llm_interaction.model_dump()
                )
            self._tracer.end_span(
                llm_span, EventType.GOAL_EVALUATOR_LLM_RESPONSE, response_payload
            )

        return code

    def run_evaluator(
        self,
        evaluator_code: str,
        goal: str,
        context: GoalContext,
    ) -> EvaluatorResult:
        """Execute generated evaluator code in a sandboxed namespace.

        Args:
            evaluator_code: The Python evaluator code to execute.
            goal: The goal being pursued.
            context: Accumulated context.

        Returns:
            EvaluatorResult with the evaluation and any traced primitive steps.
        """
        namespace: dict[str, Any] = {
            "self": self,
            "goal": goal,
            "context": context,
            "GoalEvaluation": GoalEvaluation,
        }
        trace_steps: list[ExecutionStep] = []
        # Add primitives to namespace — traced if observability is active
        if self._tracer and self._tracer.config.capture_execution_steps:
            step_counter = [1]
            read_only_map = self._get_primitive_read_only_map()
            reserved_names = (
                {"self", "goal", "context", "GoalEvaluation", "result"}
                | self._build_reserved_names()
            )
            for name, method in self._get_primitive_methods():
                traced = self._create_traced_evaluator_primitive(
                    name, method, trace_steps, step_counter,
                    read_only_map, namespace, reserved_names,
                )
                namespace[name] = traced
        else:
            for name, method in self._get_primitive_methods():
                namespace[name] = method
        namespace.update(self.allowed_builtins)

        exec(  # noqa: S102
            compile(evaluator_code, EVALUATOR_COMPILE_SOURCE, "exec"),
            empty_builtins(),
            namespace,
        )

        result = namespace.get("result")
        evaluation = (
            result if isinstance(result, GoalEvaluation)
            else GoalEvaluation(goal_achieved=False)
        )
        return EvaluatorResult(evaluation=evaluation, trace_steps=trace_steps)

    def _create_traced_evaluator_primitive(
        self,
        name: str,
        method: Any,
        trace_list: list[ExecutionStep],
        step_counter: list[int],
        read_only_map: dict[str, bool],
        namespace: dict[str, Any],
        reserved_names: set[str],
    ) -> Any:
        """Create a wrapper around a primitive called from evaluator code.

        Similar to DesignExecute._create_traced_primitive but emits
        GOAL_EVALUATOR_STEP events instead of EXECUTION_STEP.
        """
        assert self._tracer is not None
        tracer = self._tracer

        sig = inspect.signature(method)
        param_names = [p for p in sig.parameters if p != "self"]

        def traced_wrapper(*args: Any, **kwargs: Any) -> Any:
            step_number = step_counter[0]
            step_counter[0] += 1

            traced_args: dict[str, ArgumentValue] = {}
            for i, arg_val in enumerate(args):
                pname = param_names[i] if i < len(param_names) else f"arg{i}"
                traced_args[pname] = ArgumentValue(
                    expression=repr(arg_val),
                    resolved_value=arg_val,
                )
            for kw_name, kw_val in kwargs.items():
                traced_args[kw_name] = ArgumentValue(
                    expression=repr(kw_val),
                    resolved_value=kw_val,
                )

            namespace_before = self._snapshot_namespace(namespace, reserved_names)

            # Check mutation hook
            is_mutation = not read_only_map.get(name, True)
            if is_mutation and self.goal_config.on_mutation is not None:
                plain_args = {k: v.resolved_value for k, v in traced_args.items()}
                hook_context = MutationHookContext(
                    method_name=name,
                    args=plain_args,
                    result=None,
                    step=None,
                )
                rejection_reason = self.goal_config.on_mutation(hook_context)
                if rejection_reason is not None:
                    step = ExecutionStep(
                        step_number=step_number,
                        statement=f"{name}(...)",
                        primitive_called=name,
                        args=traced_args,
                        namespace_before=namespace_before,
                        namespace_after=namespace_before,
                        time_seconds=0.0,
                        success=False,
                        error=f"{MUTATION_REJECTED_PREFIX}: {rejection_reason}",
                    )
                    trace_list.append(step)
                    tracer.emit(
                        EventType.GOAL_EVALUATOR_STEP,
                        tracer.filter.execution_step(step.model_dump()),
                    )
                    raise RuntimeError(f"{MUTATION_REJECTED_PREFIX}: {rejection_reason}")

            start_time = time.perf_counter()
            try:
                result = method(*args, **kwargs)
                elapsed = time.perf_counter() - start_time

                namespace_after = self._snapshot_namespace(namespace, reserved_names)

                arg_strs = [
                    f"{k}={v.expression}" for k, v in traced_args.items()
                ]
                step = ExecutionStep(
                    step_number=step_number,
                    statement=f"{name}({', '.join(arg_strs)})",
                    primitive_called=name,
                    args=traced_args,
                    namespace_before=namespace_before,
                    namespace_after=namespace_after,
                    result_type=(
                        type(result).__name__ if result is not None else "NoneType"
                    ),
                    result_value=result,
                    time_seconds=elapsed,
                    success=True,
                )
                trace_list.append(step)

                tracer.emit(
                    EventType.GOAL_EVALUATOR_STEP,
                    tracer.filter.execution_step(step.model_dump()),
                )

                return result

            except Exception as e:
                elapsed = time.perf_counter() - start_time
                namespace_after = self._snapshot_namespace(namespace, reserved_names)
                step = ExecutionStep(
                    step_number=step_number,
                    statement=f"{name}(...)",
                    primitive_called=name,
                    args=traced_args,
                    namespace_before=namespace_before,
                    namespace_after=namespace_after,
                    time_seconds=elapsed,
                    success=False,
                    error=str(e),
                )
                trace_list.append(step)

                tracer.emit(
                    EventType.GOAL_EVALUATOR_STEP,
                    tracer.filter.execution_step(step.model_dump()),
                )

                raise

        return traced_wrapper

    # -------------------------------------------------------------------------
    # Termination Logic
    # -------------------------------------------------------------------------

    def should_continue(
        self,
        context: GoalContext,
        evaluation: GoalEvaluation,
    ) -> tuple[bool, GoalStatus]:
        """Determine if the loop should continue.

        Args:
            context: The current goal context.
            evaluation: The latest evaluation result.

        Returns:
            Tuple of (should_continue, status_if_stopping).
        """
        if evaluation.goal_achieved:
            return False, GoalStatus.ACHIEVED

        if context.iteration_count >= self.goal_config.max_iterations:
            return False, GoalStatus.MAX_ITERATIONS

        return True, GoalStatus.PURSUING

    # -------------------------------------------------------------------------
    # Hooks
    # -------------------------------------------------------------------------

    def on_iteration_start(self, iteration_number: int, context: GoalContext) -> None:
        """Hook called at the start of each iteration.

        Override for logging, metrics, or custom setup.
        """

    def on_iteration_complete(
        self, iteration: Iteration, context: GoalContext
    ) -> None:
        """Hook called after each iteration completes.

        Override for logging, metrics, or triggering side effects.
        """

    def on_goal_achieved(self, result: GoalSeekingResult) -> None:
        """Hook called when goal is achieved.

        Override for notifications, cleanup, or celebration.
        """

    def update_context(
        self, context: GoalContext, execution_result: ExecutionResult
    ) -> None:
        """THE INTROSPECTION BOUNDARY. Called after each execution.

        This is where raw ExecutionResult is introspected into structured
        insights on the context. The planner and evaluator only see what
        this method writes into context — never the raw result.

        Override to extract domain-specific insights from the execution result.
        """

    def create_context(self, goal: str) -> GoalContext:
        """Factory hook to create the initial context.

        Override to return a custom context subclass.
        """
        return GoalContext(goal=goal)

    # -------------------------------------------------------------------------
    # Answer Extraction
    # -------------------------------------------------------------------------

    def _extract_final_answer(self, context: GoalContext) -> Any:
        """Extract the final answer from context.

        Default: returns last execution result value.
        Override for custom answer extraction logic.
        """
        if context.iterations:
            last_step = context.iterations[-1].execution_result.trace.last_step
            if last_step is not None:
                return last_step.result_value
        return None

    # -------------------------------------------------------------------------
    # Main Orchestration
    # -------------------------------------------------------------------------

    def seek(self, goal: str) -> GoalSeekingResult:
        """Pursue a goal through iterative plan-execute-evaluate cycles.

        Args:
            goal: The goal to achieve (natural language).

        Returns:
            GoalSeekingResult with final answer and iteration history.
        """
        seek_span: str | None = None
        if self._tracer:
            self._tracer.new_trace()
            seek_span = self._tracer.start_span(
                EventType.GOAL_SEEK_START,
                GoalSeekStartPayload(
                    goal=goal,
                    max_iterations=self.goal_config.max_iterations,
                ).model_dump(),
            )

        context = self.create_context(goal)

        # Resolve evaluator: static (@evaluator) or dynamic (LLM-generated)
        static_evaluator = self._get_evaluator_method()
        evaluator_code: str | None = None
        if not static_evaluator:
            evaluator_code = self.plan_evaluator(goal)

        max_plan_attempts = 1 + self.goal_config.max_plan_retries

        while True:
            iteration_number = context.iteration_count + 1

            iter_span: str | None = None
            if self._tracer:
                iter_span = self._tracer.start_span(
                    EventType.GOAL_ITERATION_START,
                    {"iteration": iteration_number},
                    parent_span_id=seek_span,
                )

            # Hook: iteration start
            self.on_iteration_start(iteration_number, context)

            # 1. Plan + Execute with retry on validation errors
            plan_result: PlanResult | None = None
            exec_result: ExecutionResult | None = None
            plan_attempts: list[PlanAttempt] = []
            feedback: str | None = None

            for attempt in range(max_plan_attempts):
                plan_result = self.plan_iteration(goal, context, feedback=feedback)

                plan_attempt = PlanAttempt(
                    attempt_number=attempt + 1,
                    plan_generation=self._plan_generation_from_result(plan_result),
                    feedback=feedback,
                )

                try:
                    exec_result = self.execute(plan_result.plan)
                    plan_attempt.success = True
                    plan_attempts.append(plan_attempt)
                    break
                except ValueError as e:
                    # Validation error — retry with feedback if attempts remain
                    error_msg = str(e)
                    plan_attempt.validation_error = error_msg
                    plan_attempt.success = False
                    plan_attempts.append(plan_attempt)

                    if self._tracer:
                        self._tracer.emit(
                            EventType.PLAN_VALIDATION_ERROR,
                            {"error": error_msg, "attempt": attempt + 1},
                        )

                    if attempt < max_plan_attempts - 1:
                        feedback = error_msg
                        continue

                    # All retries exhausted — create a synthetic failed result
                    exec_result = ExecutionResult(
                        value_type="error",
                        trace=ExecutionTrace(
                            steps=[
                                ExecutionStep(
                                    step_number=1,
                                    statement="<plan validation failed>",
                                    success=False,
                                    error=f"Plan validation failed after "
                                    f"{max_plan_attempts} attempts: {error_msg}",
                                )
                            ]
                        ),
                    )

            assert plan_result is not None
            assert exec_result is not None

            # 2. Introspect: derive structured insights from raw result
            self.update_context(context, exec_result)

            # 3. Evaluate progress (checks context insights, not raw result)
            evaluator_result: EvaluatorResult | None = None
            if static_evaluator:
                evaluation = static_evaluator(goal, context)
            else:
                assert evaluator_code is not None
                evaluator_result = self.run_evaluator(evaluator_code, goal, context)
                evaluation = evaluator_result.evaluation

            if self._tracer:
                payload = evaluation.model_dump()
                if evaluator_result and evaluator_result.trace_steps:
                    payload["evaluator_step_count"] = len(evaluator_result.trace_steps)
                self._tracer.emit(
                    EventType.GOAL_EVALUATION,
                    payload,
                    parent_span_id=iter_span,
                )

            # 4. Record iteration (raw result preserved for traceability)
            iteration = Iteration(
                iteration_number=iteration_number,
                plan_result=plan_result,
                execution_result=exec_result,
                evaluation=evaluation,
                plan_attempts=plan_attempts,
            )
            context.iterations.append(iteration)

            # Hook: iteration complete
            self.on_iteration_complete(iteration, context)

            if self._tracer and iter_span:
                self._tracer.end_span(
                    iter_span,
                    EventType.GOAL_ITERATION_COMPLETE,
                    GoalIterationSummary(
                        iteration=iteration_number,
                        goal_achieved=evaluation.goal_achieved,
                    ).model_dump(),
                )

            # 5. Check termination
            should_cont, status = self.should_continue(context, evaluation)

            if not should_cont:
                result = GoalSeekingResult(
                    goal=goal,
                    status=status,
                    final_answer=self._extract_final_answer(context),
                    iterations=context.iterations,
                )

                if self._tracer and seek_span:
                    self._tracer.end_span(
                        seek_span,
                        EventType.GOAL_SEEK_COMPLETE,
                        GoalSeekSummary(
                            status=status.value,
                            iteration_count=result.iteration_count,
                        ).model_dump(),
                    )
                    self._tracer.flush()

                if status == GoalStatus.ACHIEVED:
                    self.on_goal_achieved(result)

                _emit_seek_telemetry(self, result)
                return result

blueprint_type property

The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'.

__init__(llm, name='', description='', config=None)

Initialize the GoalSeeking agent.

Parameters:

Name	Type	Description	Default
llm	LLM | LLMConfig	LLM instance or config for plan generation.	required
name	str	Agent name for prompts.	''
description	str	Agent description for prompts.	''
config	GoalSeekingConfig | None	GoalSeeking-specific configuration.	None

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def __init__(
    self,
    llm: LLM | LLMConfig,
    name: str = "",
    description: str = "",
    config: GoalSeekingConfig | None = None,
) -> None:
    """Initialize the GoalSeeking agent.

    Args:
        llm: LLM instance or config for plan generation.
        name: Agent name for prompts.
        description: Agent description for prompts.
        config: GoalSeeking-specific configuration.
    """
    cfg = config or GoalSeekingConfig()
    super().__init__(llm=llm, name=name, description=description, config=cfg)
    self.goal_config = cfg

build_evaluator_prompt(goal)

Build the prompt for evaluator code generation.

Parameters:

Name	Type	Description	Default
goal	str	The goal to generate evaluator code for.	required

Returns:

Type	Description
str	Prompt for the LLM to generate evaluator code.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

    def build_evaluator_prompt(self, goal: str) -> str:
        """Build the prompt for evaluator code generation.

        Args:
            goal: The goal to generate evaluator code for.

        Returns:
            Prompt for the LLM to generate evaluator code.
        """
        primitives = self._get_prompt_primitives()
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        return f"""You are generating Python evaluation code for a goal-seeking agent.

{PROMPT_DEFINITIONS_BEGIN}

## Goal

{goal}

## Available Primitives

```python
{chr(10).join(primitive_docs)}
```

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}

## Task

Generate Python code that evaluates whether the goal has been achieved.

The code has access to these variables:
- `goal` (str): The goal being pursued
- `context` (GoalContext): Accumulated context with structured insights
- `self`: The agent instance (can call primitives)
- `GoalEvaluation`: The evaluation result class

The code MUST assign `result = GoalEvaluation(goal_achieved=...)`.

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. Output ONLY Python assignment statements
2. Check context fields (insights), NOT raw execution results
3. The final statement must be: `result = GoalEvaluation(goal_achieved=...)`
4. Do NOT use imports, loops, conditionals, or function definitions

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""

build_goal_prompt(goal, context, feedback=None)

Build the prompt for planning the next iteration.

Includes the original goal, available primitives, decomposition examples, and structured insights from context (NOT raw execution results).

Parameters:

Name	Type	Description	Default
goal	str	The goal being pursued.	required
context	GoalContext	Accumulated context with structured insights.	required
feedback	str | None	Error feedback from a failed plan attempt (for retry).	None

Returns:

Type	Description
str	Complete prompt for plan generation.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

    def build_goal_prompt(
        self, goal: str, context: GoalContext, feedback: str | None = None
    ) -> str:
        """Build the prompt for planning the next iteration.

        Includes the original goal, available primitives, decomposition examples,
        and structured insights from context (NOT raw execution results).

        Args:
            goal: The goal being pursued.
            context: Accumulated context with structured insights.
            feedback: Error feedback from a failed plan attempt (for retry).

        Returns:
            Complete prompt for plan generation.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build context summary from structured insights (not raw results)
        context_section = ""
        if context.iteration_count > 0:
            context_section = f"""
## Previous Iterations

{context.iteration_count} iteration(s) completed so far.
"""
            # Include any custom fields from context subclasses
            custom_fields = _get_custom_context_fields(context)
            if custom_fields:
                context_section += "\n## Accumulated Knowledge\n\n"
                for field_name, field_value in custom_fields.items():
                    context_section += f"- {field_name}: {field_value!r}\n"

        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans to achieve a goal.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Goal

{goal}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{context_section}{feedback_section}## Task

Generate Python code for the NEXT step toward achieving the goal: {goal}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. You can use: assignments, for/while loops, if/elif/else, try/except
2. You can ONLY call the primitive methods listed above
3. Do NOT use imports, function definitions, class definitions, or with statements
4. Do NOT use any dangerous operations (exec, eval, open, etc.)
5. Call primitives directly (e.g. `search(query="...")`), do NOT use `self.`
6. Use if/else to handle missing data or check results before proceeding
7. Use try/except around search() or read_page() to handle errors gracefully
8. The last assigned variable will be the final result

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

create_context(goal)

Factory hook to create the initial context.

Override to return a custom context subclass.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def create_context(self, goal: str) -> GoalContext:
    """Factory hook to create the initial context.

    Override to return a custom context subclass.
    """
    return GoalContext(goal=goal)

on_goal_achieved(result)

Hook called when goal is achieved.

Override for notifications, cleanup, or celebration.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def on_goal_achieved(self, result: GoalSeekingResult) -> None:
    """Hook called when goal is achieved.

    Override for notifications, cleanup, or celebration.
    """

on_iteration_complete(iteration, context)

Hook called after each iteration completes.

Override for logging, metrics, or triggering side effects.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def on_iteration_complete(
    self, iteration: Iteration, context: GoalContext
) -> None:
    """Hook called after each iteration completes.

    Override for logging, metrics, or triggering side effects.
    """

on_iteration_start(iteration_number, context)

Hook called at the start of each iteration.

Override for logging, metrics, or custom setup.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def on_iteration_start(self, iteration_number: int, context: GoalContext) -> None:
    """Hook called at the start of each iteration.

    Override for logging, metrics, or custom setup.
    """

plan_evaluator(goal)

Generate evaluator code from the goal using the LLM.

Called once before the loop starts. The generated code must assign result = GoalEvaluation(goal_achieved=...).

Parameters:

Name	Type	Description	Default
goal	str	The goal to generate evaluator code for.	required

Returns:

Type	Description
str	Python code string for evaluation.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def plan_evaluator(self, goal: str) -> str:
    """Generate evaluator code from the goal using the LLM.

    Called once before the loop starts. The generated code must assign
    `result = GoalEvaluation(goal_achieved=...)`.

    Args:
        goal: The goal to generate evaluator code for.

    Returns:
        Python code string for evaluation.
    """
    prompt = self.build_evaluator_prompt(goal)

    llm_span: str | None = None
    if self._tracer:
        llm_payload: dict[str, Any] = {}
        if self._tracer.config.capture_llm_prompts:
            llm_payload["prompt"] = prompt
        llm_span = self._tracer.start_span(
            EventType.GOAL_EVALUATOR_LLM_REQUEST, llm_payload, defer=True
        )

    start_time = time.perf_counter()
    response = self._llm.generate(prompt)
    elapsed = time.perf_counter() - start_time

    code = self._extract_code_block(response.text)

    if self._tracer and llm_span:
        response_payload: dict[str, Any] = {}
        if self._tracer.config.capture_llm_responses:
            llm_interaction = LLMInteraction(
                prompt=prompt,
                response=response.text,
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
                time_seconds=elapsed,
                provider=response.provider,
                model=response.model,
            )
            response_payload = self._tracer.filter.llm_interaction(
                llm_interaction.model_dump()
            )
        self._tracer.end_span(
            llm_span, EventType.GOAL_EVALUATOR_LLM_RESPONSE, response_payload
        )

    return code

plan_iteration(goal, context, feedback=None)

Generate a plan for the next iteration.

Uses build_goal_prompt() directly (not self.plan()) to avoid double-wrapping the prompt through build_plan_prompt().

Parameters:

Name	Type	Description	Default
goal	str	The goal being pursued.	required
context	GoalContext	Accumulated context from previous iterations.	required
feedback	str | None	Error feedback from a failed plan attempt (for retry).	None

Returns:

Type	Description
PlanResult	PlanResult with the generated plan.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def plan_iteration(
    self, goal: str, context: GoalContext, feedback: str | None = None
) -> PlanResult:
    """Generate a plan for the next iteration.

    Uses build_goal_prompt() directly (not self.plan()) to avoid
    double-wrapping the prompt through build_plan_prompt().

    Args:
        goal: The goal being pursued.
        context: Accumulated context from previous iterations.
        feedback: Error feedback from a failed plan attempt (for retry).

    Returns:
        PlanResult with the generated plan.
    """
    plan_span: str | None = None
    if self._tracer:
        plan_span = self._tracer.start_span(
            EventType.PLAN_START, {"task": goal, "feedback": feedback}
        )

    prompt = self.build_goal_prompt(goal, context, feedback=feedback)

    llm_span: str | None = None
    if self._tracer:
        llm_payload: dict[str, Any] = {}
        if self._tracer.config.capture_llm_prompts:
            llm_payload["prompt"] = prompt
        llm_span = self._tracer.start_span(
            EventType.PLAN_LLM_REQUEST, llm_payload, defer=True
        )

    start_time = time.perf_counter()
    response = self._llm.generate(prompt)
    elapsed = time.perf_counter() - start_time

    raw_response = response.text
    extracted_code = self._extract_code_block(raw_response)
    plan_text = self.on_code_extracted(raw_response, extracted_code)

    llm_interaction = LLMInteraction(
        prompt=prompt,
        response=raw_response,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        time_seconds=elapsed,
        provider=response.provider,
        model=response.model,
    )
    plan_generation = PlanGeneration(
        llm_interaction=llm_interaction,
        extracted_code=extracted_code,
    )

    if self._tracer and llm_span:
        response_payload = self._tracer.filter.llm_interaction(
            llm_interaction.model_dump()
        ) if self._tracer.config.capture_llm_responses else {}
        self._tracer.end_span(
            llm_span, EventType.PLAN_LLM_RESPONSE, response_payload
        )

    plan_result = PlanResult(
        plan=plan_text,
        usage=TokenUsage(
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
        ),
        time_seconds=elapsed,
        provider=response.provider,
        model=response.model,
        plan_generation=plan_generation,
    )

    if self._tracer and plan_span:
        self._tracer.end_span(
            plan_span,
            EventType.PLAN_COMPLETE,
            self._tracer.filter.plan_result(plan_result.model_dump()),
        )

    return plan_result

run_evaluator(evaluator_code, goal, context)

Execute generated evaluator code in a sandboxed namespace.

Parameters:

Name	Type	Description	Default
evaluator_code	str	The Python evaluator code to execute.	required
goal	str	The goal being pursued.	required
context	GoalContext	Accumulated context.	required

Returns:

Type	Description
EvaluatorResult	EvaluatorResult with the evaluation and any traced primitive steps.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def run_evaluator(
    self,
    evaluator_code: str,
    goal: str,
    context: GoalContext,
) -> EvaluatorResult:
    """Execute generated evaluator code in a sandboxed namespace.

    Args:
        evaluator_code: The Python evaluator code to execute.
        goal: The goal being pursued.
        context: Accumulated context.

    Returns:
        EvaluatorResult with the evaluation and any traced primitive steps.
    """
    namespace: dict[str, Any] = {
        "self": self,
        "goal": goal,
        "context": context,
        "GoalEvaluation": GoalEvaluation,
    }
    trace_steps: list[ExecutionStep] = []
    # Add primitives to namespace — traced if observability is active
    if self._tracer and self._tracer.config.capture_execution_steps:
        step_counter = [1]
        read_only_map = self._get_primitive_read_only_map()
        reserved_names = (
            {"self", "goal", "context", "GoalEvaluation", "result"}
            | self._build_reserved_names()
        )
        for name, method in self._get_primitive_methods():
            traced = self._create_traced_evaluator_primitive(
                name, method, trace_steps, step_counter,
                read_only_map, namespace, reserved_names,
            )
            namespace[name] = traced
    else:
        for name, method in self._get_primitive_methods():
            namespace[name] = method
    namespace.update(self.allowed_builtins)

    exec(  # noqa: S102
        compile(evaluator_code, EVALUATOR_COMPILE_SOURCE, "exec"),
        empty_builtins(),
        namespace,
    )

    result = namespace.get("result")
    evaluation = (
        result if isinstance(result, GoalEvaluation)
        else GoalEvaluation(goal_achieved=False)
    )
    return EvaluatorResult(evaluation=evaluation, trace_steps=trace_steps)

seek(goal)

Pursue a goal through iterative plan-execute-evaluate cycles.

Parameters:

Name	Type	Description	Default
goal	str	The goal to achieve (natural language).	required

Returns:

Type	Description
GoalSeekingResult	GoalSeekingResult with final answer and iteration history.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def seek(self, goal: str) -> GoalSeekingResult:
    """Pursue a goal through iterative plan-execute-evaluate cycles.

    Args:
        goal: The goal to achieve (natural language).

    Returns:
        GoalSeekingResult with final answer and iteration history.
    """
    seek_span: str | None = None
    if self._tracer:
        self._tracer.new_trace()
        seek_span = self._tracer.start_span(
            EventType.GOAL_SEEK_START,
            GoalSeekStartPayload(
                goal=goal,
                max_iterations=self.goal_config.max_iterations,
            ).model_dump(),
        )

    context = self.create_context(goal)

    # Resolve evaluator: static (@evaluator) or dynamic (LLM-generated)
    static_evaluator = self._get_evaluator_method()
    evaluator_code: str | None = None
    if not static_evaluator:
        evaluator_code = self.plan_evaluator(goal)

    max_plan_attempts = 1 + self.goal_config.max_plan_retries

    while True:
        iteration_number = context.iteration_count + 1

        iter_span: str | None = None
        if self._tracer:
            iter_span = self._tracer.start_span(
                EventType.GOAL_ITERATION_START,
                {"iteration": iteration_number},
                parent_span_id=seek_span,
            )

        # Hook: iteration start
        self.on_iteration_start(iteration_number, context)

        # 1. Plan + Execute with retry on validation errors
        plan_result: PlanResult | None = None
        exec_result: ExecutionResult | None = None
        plan_attempts: list[PlanAttempt] = []
        feedback: str | None = None

        for attempt in range(max_plan_attempts):
            plan_result = self.plan_iteration(goal, context, feedback=feedback)

            plan_attempt = PlanAttempt(
                attempt_number=attempt + 1,
                plan_generation=self._plan_generation_from_result(plan_result),
                feedback=feedback,
            )

            try:
                exec_result = self.execute(plan_result.plan)
                plan_attempt.success = True
                plan_attempts.append(plan_attempt)
                break
            except ValueError as e:
                # Validation error — retry with feedback if attempts remain
                error_msg = str(e)
                plan_attempt.validation_error = error_msg
                plan_attempt.success = False
                plan_attempts.append(plan_attempt)

                if self._tracer:
                    self._tracer.emit(
                        EventType.PLAN_VALIDATION_ERROR,
                        {"error": error_msg, "attempt": attempt + 1},
                    )

                if attempt < max_plan_attempts - 1:
                    feedback = error_msg
                    continue

                # All retries exhausted — create a synthetic failed result
                exec_result = ExecutionResult(
                    value_type="error",
                    trace=ExecutionTrace(
                        steps=[
                            ExecutionStep(
                                step_number=1,
                                statement="<plan validation failed>",
                                success=False,
                                error=f"Plan validation failed after "
                                f"{max_plan_attempts} attempts: {error_msg}",
                            )
                        ]
                    ),
                )

        assert plan_result is not None
        assert exec_result is not None

        # 2. Introspect: derive structured insights from raw result
        self.update_context(context, exec_result)

        # 3. Evaluate progress (checks context insights, not raw result)
        evaluator_result: EvaluatorResult | None = None
        if static_evaluator:
            evaluation = static_evaluator(goal, context)
        else:
            assert evaluator_code is not None
            evaluator_result = self.run_evaluator(evaluator_code, goal, context)
            evaluation = evaluator_result.evaluation

        if self._tracer:
            payload = evaluation.model_dump()
            if evaluator_result and evaluator_result.trace_steps:
                payload["evaluator_step_count"] = len(evaluator_result.trace_steps)
            self._tracer.emit(
                EventType.GOAL_EVALUATION,
                payload,
                parent_span_id=iter_span,
            )

        # 4. Record iteration (raw result preserved for traceability)
        iteration = Iteration(
            iteration_number=iteration_number,
            plan_result=plan_result,
            execution_result=exec_result,
            evaluation=evaluation,
            plan_attempts=plan_attempts,
        )
        context.iterations.append(iteration)

        # Hook: iteration complete
        self.on_iteration_complete(iteration, context)

        if self._tracer and iter_span:
            self._tracer.end_span(
                iter_span,
                EventType.GOAL_ITERATION_COMPLETE,
                GoalIterationSummary(
                    iteration=iteration_number,
                    goal_achieved=evaluation.goal_achieved,
                ).model_dump(),
            )

        # 5. Check termination
        should_cont, status = self.should_continue(context, evaluation)

        if not should_cont:
            result = GoalSeekingResult(
                goal=goal,
                status=status,
                final_answer=self._extract_final_answer(context),
                iterations=context.iterations,
            )

            if self._tracer and seek_span:
                self._tracer.end_span(
                    seek_span,
                    EventType.GOAL_SEEK_COMPLETE,
                    GoalSeekSummary(
                        status=status.value,
                        iteration_count=result.iteration_count,
                    ).model_dump(),
                )
                self._tracer.flush()

            if status == GoalStatus.ACHIEVED:
                self.on_goal_achieved(result)

            _emit_seek_telemetry(self, result)
            return result

should_continue(context, evaluation)

Determine if the loop should continue.

Parameters:

Name	Type	Description	Default
context	GoalContext	The current goal context.	required
evaluation	GoalEvaluation	The latest evaluation result.	required

Returns:

Type	Description
tuple[bool, GoalStatus]	Tuple of (should_continue, status_if_stopping).

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def should_continue(
    self,
    context: GoalContext,
    evaluation: GoalEvaluation,
) -> tuple[bool, GoalStatus]:
    """Determine if the loop should continue.

    Args:
        context: The current goal context.
        evaluation: The latest evaluation result.

    Returns:
        Tuple of (should_continue, status_if_stopping).
    """
    if evaluation.goal_achieved:
        return False, GoalStatus.ACHIEVED

    if context.iteration_count >= self.goal_config.max_iterations:
        return False, GoalStatus.MAX_ITERATIONS

    return True, GoalStatus.PURSUING

update_context(context, execution_result)

THE INTROSPECTION BOUNDARY. Called after each execution.

This is where raw ExecutionResult is introspected into structured insights on the context. The planner and evaluator only see what this method writes into context — never the raw result.

Override to extract domain-specific insights from the execution result.

Source code in src/opensymbolicai/blueprints/goal_seeking.py

def update_context(
    self, context: GoalContext, execution_result: ExecutionResult
) -> None:
    """THE INTROSPECTION BOUNDARY. Called after each execution.

    This is where raw ExecutionResult is introspected into structured
    insights on the context. The planner and evaluator only see what
    this method writes into context — never the raw result.

    Override to extract domain-specific insights from the execution result.
    """

GoalSeekingConfig

Bases: DesignExecuteConfig

Configuration for GoalSeeking agents.

Source code in src/opensymbolicai/models.py

class GoalSeekingConfig(DesignExecuteConfig):
    """Configuration for GoalSeeking agents."""

    max_iterations: int = Field(
        default=10, description="Maximum iterations before stopping"
    )
    max_plan_retries: int = Field(
        default=2,
        description="Maximum number of times to retry plan generation if validation fails. "
        "When a plan fails validation, the error is fed back to the LLM for regeneration.",
    )

GoalSeekingResult

Bases: BaseModel

Result from a complete goal-seeking run.

Subclass to add domain-specific result fields.

Source code in src/opensymbolicai/models.py

class GoalSeekingResult(BaseModel):
    """Result from a complete goal-seeking run.

    Subclass to add domain-specific result fields.
    """

    goal: str = Field(..., description="The original goal")

    status: GoalStatus = Field(..., description="Final status")

    final_answer: Any = Field(default=None, description="The final result/answer")

    iterations: list[Iteration] = Field(
        default_factory=list, description="All iterations performed"
    )

    model_config = {"arbitrary_types_allowed": True}

    @property
    def iteration_count(self) -> int:
        """Number of iterations performed."""
        return len(self.iterations)

    @property
    def succeeded(self) -> bool:
        """Whether the goal was achieved."""
        return self.status == GoalStatus.ACHIEVED

iteration_count property

Number of iterations performed.

succeeded property

Whether the goal was achieved.

GoalStatus

Bases: str, Enum

Status of goal pursuit.

Source code in src/opensymbolicai/models.py

class GoalStatus(str, Enum):
    """Status of goal pursuit."""

    PURSUING = "pursuing"
    ACHIEVED = "achieved"
    FAILED = "failed"
    MAX_ITERATIONS = "max_iterations"

HttpTransport

Batched HTTP transport using stdlib urllib.

Events are queued in memory and flushed by a background thread either when the batch size is reached or on close().

On HTTP failure the batch is dropped with a warning to stderr. No retries — keep it simple.

Parameters:

Name	Type	Description	Default
url	str	Collector endpoint (e.g. http://localhost:8100/events).	required
batch_size	int	Flush after this many events are queued.	50
flush_interval_seconds	float	Periodic flush interval.	5.0
headers	dict[str, str] | None	Extra HTTP headers sent with every request (e.g. {"X-API-Key": "os-…-am"}).	None

Source code in src/opensymbolicai/observability/transports/http.py

class HttpTransport:
    """Batched HTTP transport using stdlib urllib.

    Events are queued in memory and flushed by a background thread
    either when the batch size is reached or on ``close()``.

    On HTTP failure the batch is dropped with a warning to stderr.
    No retries — keep it simple.

    Args:
        url: Collector endpoint (e.g. ``http://localhost:8100/events``).
        batch_size: Flush after this many events are queued.
        flush_interval_seconds: Periodic flush interval.
        headers: Extra HTTP headers sent with every request
            (e.g. ``{"X-API-Key": "os-…-am"}``).
    """

    def __init__(
        self,
        url: str,
        batch_size: int = 50,
        flush_interval_seconds: float = 5.0,
        headers: dict[str, str] | None = None,
    ) -> None:
        self._url = url
        self._batch_size = batch_size
        self._flush_interval = flush_interval_seconds
        self._extra_headers = headers or {}
        self._queue: Queue[TraceEvent] = Queue()
        self._closed = False
        self._lock = threading.Lock()

        # Background flusher thread
        self._thread = threading.Thread(
            target=self._flush_loop, daemon=True, name="observability-http"
        )
        self._thread.start()

    def send(self, events: list[TraceEvent]) -> None:
        """Queue events for batched sending."""
        if self._closed:
            return
        for event in events:
            self._queue.put(event)

    def flush(self) -> None:
        """Flush any buffered events without closing the transport."""
        self._flush_batch()

    def close(self) -> None:
        """Flush remaining events and stop the background thread."""
        self._closed = True
        self._flush_batch()
        self._thread.join(timeout=5.0)

    def _flush_loop(self) -> None:
        """Background loop that flushes batches periodically."""
        while not self._closed:
            try:
                event = self._queue.get(timeout=self._flush_interval)
                # Re-queue so _flush_batch drains it along with everything else.
                self._queue.put(event)
            except Empty:
                pass
            self._flush_batch()

    def _flush_batch(self) -> None:
        """Drain the queue and send all buffered events."""
        batch: list[TraceEvent] = []
        while True:
            try:
                batch.append(self._queue.get_nowait())
            except Empty:
                break

        if not batch:
            return

        self._send_http(batch)

    def _send_http(self, events: list[TraceEvent]) -> None:
        """POST a batch of events to the collector."""
        body = json.dumps(
            {"events": [e.model_dump(mode="json") for e in events]}
        ).encode("utf-8")

        req = urllib.request.Request(
            self._url,
            data=body,
            headers={"Content-Type": "application/json", **self._extra_headers},
            method="POST",
        )

        try:
            with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
                resp.read()
        except (urllib.error.URLError, OSError, TimeoutError) as e:
            print(
                f"[opensymbolicai.observability] Failed to send {len(events)} events: {e}",
                file=sys.stderr,
            )

close()

Flush remaining events and stop the background thread.

Source code in src/opensymbolicai/observability/transports/http.py

def close(self) -> None:
    """Flush remaining events and stop the background thread."""
    self._closed = True
    self._flush_batch()
    self._thread.join(timeout=5.0)

flush()

Flush any buffered events without closing the transport.

Source code in src/opensymbolicai/observability/transports/http.py

def flush(self) -> None:
    """Flush any buffered events without closing the transport."""
    self._flush_batch()

send(events)

Queue events for batched sending.

Source code in src/opensymbolicai/observability/transports/http.py

def send(self, events: list[TraceEvent]) -> None:
    """Queue events for batched sending."""
    if self._closed:
        return
    for event in events:
        self._queue.put(event)

InMemoryCheckpointStore

In-memory checkpoint store for testing and development.

Source code in src/opensymbolicai/checkpoint.py

class InMemoryCheckpointStore:
    """In-memory checkpoint store for testing and development."""

    def __init__(self) -> None:
        self._checkpoints: dict[str, ExecutionCheckpoint] = {}

    def save(self, checkpoint: ExecutionCheckpoint) -> None:
        """Save a checkpoint to memory."""
        checkpoint.touch()
        # Store a copy to prevent mutation issues
        self._checkpoints[checkpoint.checkpoint_id] = checkpoint.model_copy(deep=True)

    def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
        """Load a checkpoint from memory."""
        cp = self._checkpoints.get(checkpoint_id)
        if cp:
            return cp.model_copy(deep=True)
        return None

    def delete(self, checkpoint_id: str) -> None:
        """Delete a checkpoint from memory."""
        self._checkpoints.pop(checkpoint_id, None)

    def list_by_status(self, status: CheckpointStatus) -> list[str]:
        """List checkpoint IDs with the given status."""
        return [
            cp.checkpoint_id for cp in self._checkpoints.values() if cp.status == status
        ]

    def list_all(self) -> list[str]:
        """List all checkpoint IDs."""
        return list(self._checkpoints.keys())

    def clear(self) -> None:
        """Clear all checkpoints."""
        self._checkpoints.clear()

clear()

Clear all checkpoints.

Source code in src/opensymbolicai/checkpoint.py

def clear(self) -> None:
    """Clear all checkpoints."""
    self._checkpoints.clear()

delete(checkpoint_id)

Delete a checkpoint from memory.

Source code in src/opensymbolicai/checkpoint.py

def delete(self, checkpoint_id: str) -> None:
    """Delete a checkpoint from memory."""
    self._checkpoints.pop(checkpoint_id, None)

list_all()

List all checkpoint IDs.

Source code in src/opensymbolicai/checkpoint.py

def list_all(self) -> list[str]:
    """List all checkpoint IDs."""
    return list(self._checkpoints.keys())

list_by_status(status)

List checkpoint IDs with the given status.

Source code in src/opensymbolicai/checkpoint.py

def list_by_status(self, status: CheckpointStatus) -> list[str]:
    """List checkpoint IDs with the given status."""
    return [
        cp.checkpoint_id for cp in self._checkpoints.values() if cp.status == status
    ]

load(checkpoint_id)

Load a checkpoint from memory.

Source code in src/opensymbolicai/checkpoint.py

def load(self, checkpoint_id: str) -> ExecutionCheckpoint | None:
    """Load a checkpoint from memory."""
    cp = self._checkpoints.get(checkpoint_id)
    if cp:
        return cp.model_copy(deep=True)
    return None

save(checkpoint)

Save a checkpoint to memory.

Source code in src/opensymbolicai/checkpoint.py

def save(self, checkpoint: ExecutionCheckpoint) -> None:
    """Save a checkpoint to memory."""
    checkpoint.touch()
    # Store a copy to prevent mutation issues
    self._checkpoints[checkpoint.checkpoint_id] = checkpoint.model_copy(deep=True)

InMemoryTransport

Stores trace events in memory.

Useful for testing and for inspecting events in the same process.

Source code in src/opensymbolicai/observability/transports/memory.py

class InMemoryTransport:
    """Stores trace events in memory.

    Useful for testing and for inspecting events in the same process.
    """

    def __init__(self) -> None:
        self.events: list[TraceEvent] = []

    def send(self, events: list[TraceEvent]) -> None:
        """Append events to the in-memory list."""
        self.events.extend(events)

    def close(self) -> None:
        """No-op for in-memory transport."""

close()

No-op for in-memory transport.

Source code in src/opensymbolicai/observability/transports/memory.py

def close(self) -> None:
    """No-op for in-memory transport."""

send(events)

Append events to the in-memory list.

Source code in src/opensymbolicai/observability/transports/memory.py

def send(self, events: list[TraceEvent]) -> None:
    """Append events to the in-memory list."""
    self.events.extend(events)

Iteration

Bases: BaseModel

A single iteration of the goal-seeking loop.

Source code in src/opensymbolicai/models.py

class Iteration(BaseModel):
    """A single iteration of the goal-seeking loop."""

    iteration_number: int = Field(..., description="1-based iteration number")

    plan_result: PlanResult = Field(
        ..., description="The plan generated for this iteration"
    )

    execution_result: ExecutionResult = Field(
        ..., description="Result of executing the plan"
    )

    evaluation: GoalEvaluation = Field(
        ..., description="Evaluation of progress toward goal"
    )

    plan_attempts: list[PlanAttempt] = Field(
        default_factory=list,
        description="Plan generation attempts including retries (populated when max_plan_retries > 0)",
    )

MethodType

Bases: Enum

Classification of agent methods.

Source code in src/opensymbolicai/core.py

class MethodType(Enum):
    """Classification of agent methods."""

    PRIMITIVE = "primitive"
    DECOMPOSITION = "decomposition"
    EVALUATOR = "evaluator"

MutationHookContext

Bases: BaseModel

Context passed to mutation hooks when a non-read-only primitive is called.

Source code in src/opensymbolicai/models.py

class MutationHookContext(BaseModel):
    """Context passed to mutation hooks when a non-read-only primitive is called."""

    method_name: str = Field(..., description="Name of the primitive method called")
    args: dict[str, Any] = Field(
        default_factory=dict, description="Arguments passed to the method"
    )
    result: Any = Field(default=None, description="Result returned by the method")
    step: ExecutionStep | None = Field(
        default=None, description="The execution step containing full details"
    )

    model_config = {"arbitrary_types_allowed": True}

ObservabilityConfig

Bases: BaseModel

Controls what gets captured and where it goes.

When enabled is False (the default), no tracing overhead is incurred.

Source code in src/opensymbolicai/observability/config.py

class ObservabilityConfig(BaseModel):
    """Controls what gets captured and where it goes.

    When ``enabled`` is False (the default), no tracing overhead is incurred.
    """

    enabled: bool = Field(default=False, description="Master switch for observability")

    # Granular capture filters
    capture_llm_prompts: bool = Field(
        default=True, description="Capture LLM prompts (can be large)"
    )
    capture_llm_responses: bool = Field(
        default=True, description="Capture LLM responses"
    )
    capture_namespace_snapshots: bool = Field(
        default=False, description="Capture namespace before/after each step (large)"
    )
    capture_execution_steps: bool = Field(
        default=True, description="Capture individual execution steps"
    )
    capture_plan_source: bool = Field(
        default=True, description="Capture generated plan source code"
    )

    # Metadata attached to all events
    session_id: str = Field(
        default_factory=lambda: uuid.uuid4().hex,
        description="Groups multiple traces into a logical session. "
        "Auto-generated if not provided. Share a session_id across "
        "multiple agent.run()/seek() calls to correlate their traces.",
    )
    tags: dict[str, str] = Field(
        default_factory=dict, description="Custom metadata attached to all events"
    )

    # Transport config
    collector_url: str | None = Field(
        default=None, description="URL for HttpTransport (e.g. http://localhost:8000/api/events)"
    )
    collector_headers: dict[str, str] = Field(
        default_factory=dict,
        description="Extra HTTP headers for HttpTransport (e.g. X-API-Key)",
    )
    output_path: str | None = Field(
        default=None, description="File path for FileTransport (JSONL)"
    )

    # Custom transport takes precedence over collector_url / output_path
    transport: TraceTransport | None = Field(
        default=None,
        description="Custom TraceTransport instance. Takes precedence over collector_url/output_path.",
        exclude=True,
    )

    model_config = {"arbitrary_types_allowed": True}

OperationError

Bases: ExecutionError

Exception raised when an operation fails during execution.

Use this for runtime errors that occur during the actual operation, not input validation.

Example

raise OperationError( "File write failed", code="WRITE_FAILED", details={"path": "/tmp/output.txt", "reason": "disk full"} )

Source code in src/opensymbolicai/exceptions.py

class OperationError(ExecutionError):
    """Exception raised when an operation fails during execution.

    Use this for runtime errors that occur during the actual operation,
    not input validation.

    Example:
        raise OperationError(
            "File write failed",
            code="WRITE_FAILED",
            details={"path": "/tmp/output.txt", "reason": "disk full"}
        )
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = "OPERATION_ERROR",
        details: dict[str, Any] | None = None,
        operation: str | None = None,
    ) -> None:
        """Initialize operation error.

        Args:
            message: Human-readable error message.
            code: Error code, defaults to "OPERATION_ERROR".
            details: Additional context about the error.
            operation: Optional name of the operation that failed.
        """
        if operation:
            details = details or {}
            details["operation"] = operation
        super().__init__(message, code=code, details=details)
        self.operation = operation

__init__(message, *, code='OPERATION_ERROR', details=None, operation=None)

Initialize operation error.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Error code, defaults to "OPERATION_ERROR".	'OPERATION_ERROR'
details	dict[str, Any] | None	Additional context about the error.	None
operation	str | None	Optional name of the operation that failed.	None

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = "OPERATION_ERROR",
    details: dict[str, Any] | None = None,
    operation: str | None = None,
) -> None:
    """Initialize operation error.

    Args:
        message: Human-readable error message.
        code: Error code, defaults to "OPERATION_ERROR".
        details: Additional context about the error.
        operation: Optional name of the operation that failed.
    """
    if operation:
        details = details or {}
        details["operation"] = operation
    super().__init__(message, code=code, details=details)
    self.operation = operation

OrchestrationResult

Bases: BaseModel

Result from a complete plan-and-execute run.

Source code in src/opensymbolicai/models.py

class OrchestrationResult(BaseModel):
    """Result from a complete plan-and-execute run."""

    success: bool = Field(..., description="Whether the orchestration succeeded")
    result: Any = Field(default=None, description="The computed result")
    error: str | None = Field(default=None, description="Error message if failed")
    metrics: ExecutionMetrics | None = Field(
        default=None, description="Execution metrics"
    )
    plan: str | None = Field(default=None, description="The generated plan")
    trace: ExecutionTrace | None = Field(
        default=None, description="Step-by-step execution trace"
    )
    plan_attempts: list[PlanAttempt] = Field(
        default_factory=list,
        description="All plan generation attempts including retries",
    )
    task: str = Field(default="", description="The original task description")

ParameterInfo

Bases: BaseModel

Metadata for a single parameter of a primitive or decomposition method.

Source code in src/opensymbolicai/models.py

class ParameterInfo(BaseModel):
    """Metadata for a single parameter of a primitive or decomposition method."""

    name: str = Field(..., description="Parameter name.")
    type: str = Field(..., description="Type annotation as a string (e.g. 'float', 'list[str]').")
    default: str | None = Field(
        default=None,
        description="String representation of the default value, or None if required.",
    )

PendingMutation

Bases: BaseModel

Information about a mutation awaiting approval.

Source code in src/opensymbolicai/checkpoint.py

class PendingMutation(BaseModel):
    """Information about a mutation awaiting approval."""

    method_name: str = Field(..., description="Name of the mutating primitive")
    args: dict[str, Any] = Field(
        default_factory=dict, description="Arguments to the mutation"
    )
    statement: str = Field(..., description="The full statement to execute")
    step_number: int = Field(..., description="Which step this mutation is")
    variable_name: str = Field(
        default="", description="Variable being assigned the result"
    )

PlanAnalysis

Bases: BaseModel

Analysis of a plan's structure.

Source code in src/opensymbolicai/models.py

class PlanAnalysis(BaseModel):
    """Analysis of a plan's structure."""

    calls: list[PrimitiveCall] = Field(
        default_factory=list, description="All primitive calls in the plan"
    )

    @property
    def has_mutations(self) -> bool:
        """Check if any calls are not read-only."""
        return any(not call.read_only for call in self.calls)

    @property
    def method_names(self) -> list[str]:
        """Get list of all method names called."""
        return [call.method_name for call in self.calls]

has_mutations property

Check if any calls are not read-only.

method_names property

Get list of all method names called.

PlanContext

Bases: BaseModel

Context about how the plan was generated.

Source code in src/opensymbolicai/checkpoint.py

class PlanContext(BaseModel):
    """Context about how the plan was generated."""

    plan_attempts: list[PlanAttempt] = Field(
        default_factory=list,
        description="All plan generation attempts including retries",
    )
    final_plan: str = Field(default="", description="The final validated plan")
    generation_time_seconds: float = Field(
        default=0.0, description="Total time spent generating the plan"
    )

    @property
    def attempt_count(self) -> int:
        """Number of plan generation attempts."""
        return len(self.plan_attempts)

    @property
    def had_retries(self) -> bool:
        """Whether the plan required retries."""
        return len(self.plan_attempts) > 1

    @property
    def all_llm_interactions(self) -> list[LLMInteraction]:
        """Get all LLM interactions from all attempts."""
        return [
            attempt.plan_generation.llm_interaction for attempt in self.plan_attempts
        ]

all_llm_interactions property

Get all LLM interactions from all attempts.

attempt_count property

Number of plan generation attempts.

had_retries property

Whether the plan required retries.

PlanExecute

Bases: Planner

Agent that generates and executes Python plans using LLMs.

PlanExecute orchestrators use an LLM to generate Python code that calls primitive methods, then execute that code step-by-step with full tracing.

Subclasses should: 1. Define primitive methods using the @primitive decorator 2. Optionally define decomposition examples using @decomposition

Example

class Calculator(PlanExecute): @primitive(read_only=True) def add(self, a: float, b: float) -> float: return a + b

@decomposition(intent="What is 2 + 3?")
def _example_add(self) -> float:
    result = self.add(2, 3)
    return result

Source code in src/opensymbolicai/blueprints/plan_execute.py

class PlanExecute(Planner):
    """Agent that generates and executes Python plans using LLMs.

    PlanExecute orchestrators use an LLM to generate Python code that calls
    primitive methods, then execute that code step-by-step with full tracing.

    Subclasses should:
    1. Define primitive methods using the @primitive decorator
    2. Optionally define decomposition examples using @decomposition

    Example:
        class Calculator(PlanExecute):
            @primitive(read_only=True)
            def add(self, a: float, b: float) -> float:
                return a + b

            @decomposition(intent="What is 2 + 3?")
            def _example_add(self) -> float:
                result = self.add(2, 3)
                return result
    """

    DANGEROUS_BUILTINS: set[str] = {
        "exec",
        "eval",
        "compile",
        "open",
        "__import__",
        "globals",
        "locals",
        "vars",
        "dir",
        "getattr",
        "setattr",
        "delattr",
        "hasattr",
        "type",
        "issubclass",
        "callable",
        "breakpoint",
        "input",
        "memoryview",
        "object",
    }

    def __init__(
        self,
        llm: LLM | LLMConfig,
        name: str = "",
        description: str = "",
        config: PlanExecuteConfig | None = None,
    ) -> None:
        """Initialize the PlanExecute agent.

        Args:
            llm: LLM instance or config for plan generation.
            name: Agent name for prompts.
            description: Agent description for prompts.
            config: Extended configuration options.
        """
        if isinstance(llm, LLMConfig):
            self._llm = create_llm(llm)
        else:
            self._llm = llm

        self.name = name or self.__class__.__name__
        self.description = description or self.__class__.__doc__ or ""
        self.config = config or PlanExecuteConfig()
        self.allowed_builtins = (
            self.config.allowed_builtins
            if self.config.allowed_builtins is not None
            else DEFAULT_ALLOWED_BUILTINS.copy()
        )

        # Multi-turn state
        self._history: list[ConversationTurn] = []
        self._persisted_namespace: dict[str, Any] = {}

        # Observability
        self._tracer: Tracer | None = None
        if self.config.observability and self.config.observability.enabled:
            self._tracer = Tracer(self.config.observability, type(self).__name__)

    @property
    def history(self) -> list[ConversationTurn]:
        """Get the conversation history (multi-turn mode only)."""
        return self._history

    @property
    def persisted_variables(self) -> dict[str, Any]:
        """Get the persisted variables from previous turns (multi-turn mode only)."""
        return self._persisted_namespace.copy()

    @property
    def blueprint_type(self) -> str:
        """The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'."""
        return "PlanExecute"

    # -------------------------------------------------------------------------
    # Introspection: Extract primitives and decompositions
    # -------------------------------------------------------------------------

    def _get_primitive_methods(self) -> list[tuple[str, Callable[..., Any]]]:
        """Get all methods decorated with @primitive."""
        primitives = []
        for name in dir(self):
            if name.startswith("_"):
                continue
            method = getattr(self, name, None)
            if (
                callable(method)
                and hasattr(method, "__method_type__")
                and method.__method_type__ == MethodType.PRIMITIVE
            ):
                primitives.append((name, method))
        return primitives

    def _get_decomposition_methods(
        self,
    ) -> list[tuple[str, Callable[..., Any], str, str]]:
        """Get all methods decorated with @decomposition.

        Returns:
            List of (name, method, intent, expanded_intent) tuples.
        """
        decompositions = []
        for name in dir(self):
            method = getattr(self, name, None)
            if (
                callable(method)
                and hasattr(method, "__method_type__")
                and method.__method_type__ == MethodType.DECOMPOSITION
            ):
                intent = getattr(method, "__decomposition_intent__", "")
                expanded = getattr(method, "__decomposition_expanded_intent__", "")
                decompositions.append((name, method, intent, expanded))
        return decompositions

    # -------------------------------------------------------------------------
    # Prompt-filtered introspection (respects PromptProvider)
    # -------------------------------------------------------------------------

    @staticmethod
    def _annotation_to_str(annotation: Any) -> str:
        """Convert a type annotation to a human-readable string.

        Handles generics like ``list[str]`` correctly (via ``str()``),
        falling back to ``__name__`` only for simple non-generic types.
        """
        if annotation is inspect.Parameter.empty or annotation is inspect.Signature.empty:
            return "Any"
        # Generic aliases (list[str], dict[str, int], X | None) have __args__
        if hasattr(annotation, "__args__"):
            return str(annotation)
        if hasattr(annotation, "__name__"):
            return annotation.__name__
        return str(annotation)

    def _extract_signature_metadata(
        self, method: Callable[..., Any]
    ) -> tuple[list[ParameterInfo], str]:
        """Extract parameter info and return type from a method signature."""
        sig = inspect.signature(method)
        params = []
        for pname, param in sig.parameters.items():
            if pname == "self":
                continue
            type_str = self._annotation_to_str(param.annotation)
            default = repr(param.default) if param.default != inspect.Parameter.empty else None
            params.append(ParameterInfo(name=pname, type=type_str, default=default))
        return_type = self._annotation_to_str(sig.return_annotation)
        return params, return_type

    def _build_primitive_info(self, name: str, method: Callable[..., Any]) -> PrimitiveInfo:
        """Build a PrimitiveInfo from a primitive method."""
        params, return_type = self._extract_signature_metadata(method)
        return PrimitiveInfo(
            name=name,
            docstring=inspect.getdoc(method) or "",
            read_only=getattr(method, "__primitive_read_only__", False),
            deterministic=getattr(method, "__primitive_deterministic__", True),
            parameters=params,
            return_type=return_type,
        )

    def _build_decomposition_info(
        self, name: str, method: Callable[..., Any], intent: str, expanded_intent: str
    ) -> DecompositionInfo:
        """Build a DecompositionInfo from a decomposition method."""
        params, return_type = self._extract_signature_metadata(method)
        return DecompositionInfo(
            name=name,
            intent=intent,
            expanded_intent=expanded_intent,
            parameters=params,
            return_type=return_type,
            source=self._get_decomposition_source(method),
        )

    def _get_prompt_primitives(self) -> list[tuple[str, Callable[..., Any]]]:
        """Get primitives filtered by the configured PromptProvider."""
        all_primitives = self._get_primitive_methods()
        provider = self.config.prompt_provider
        if provider is None:
            return all_primitives
        infos = [self._build_primitive_info(n, m) for n, m in all_primitives]
        selected = set(provider.select_primitives(infos))
        return [(n, m) for n, m in all_primitives if n in selected]

    def _get_prompt_decompositions(
        self,
    ) -> list[tuple[str, Callable[..., Any], str, str]]:
        """Get decompositions filtered by the configured PromptProvider."""
        all_decomps = self._get_decomposition_methods()
        provider = self.config.prompt_provider
        if provider is None:
            return all_decomps
        infos = [self._build_decomposition_info(n, m, i, e) for n, m, i, e in all_decomps]
        selected = set(provider.select_decompositions(infos))
        return [(n, m, i, e) for n, m, i, e in all_decomps if n in selected]

    def _get_primitive_names(self) -> set[str]:
        """Get the names of all primitive methods."""
        return {name for name, _ in self._get_primitive_methods()}

    def _get_primitive_read_only_map(self) -> dict[str, bool]:
        """Get a mapping of primitive names to their read_only status."""
        return {
            name: getattr(method, "__primitive_read_only__", False)
            for name, method in self._get_primitive_methods()
        }

    def _get_primitive_determinism_map(self) -> dict[str, bool]:
        """Get a mapping of primitive names to their deterministic status."""
        return {
            name: getattr(method, "__primitive_deterministic__", True)
            for name, method in self._get_primitive_methods()
        }

    def compute_signature_hash(self) -> str:
        """Compute a hash of all primitive signatures and decomposition examples.

        A changed hash means the agent's interface has changed and any
        downstream artifacts (e.g. fine-tuned adapters) need regeneration.

        Returns:
            A 16-character hex digest.
        """
        primitives = self._get_primitive_methods()
        decompositions = self._get_decomposition_methods()

        parts: list[str] = []
        for name, method in sorted(primitives, key=lambda x: x[0]):
            sig = inspect.signature(method)
            doc = inspect.getdoc(method) or ""
            read_only = getattr(method, "__primitive_read_only__", False)
            deterministic = getattr(method, "__primitive_deterministic__", True)
            parts.append(
                f"PRIM:{name}:{sig}:{doc}:{read_only}:{deterministic}"
            )

        for name, method, intent, expanded in sorted(
            decompositions, key=lambda x: x[0]
        ):
            source = self._get_decomposition_source(method)
            parts.append(f"DECOMP:{name}:{intent}:{expanded}:{source}")

        combined = "\n".join(parts)
        return hashlib.sha256(combined.encode()).hexdigest()[:16]

    def _format_primitive_signature(self, name: str, method: Callable[..., Any]) -> str:
        """Format a primitive method's signature and docstring for the prompt."""
        sig = inspect.signature(method)
        params = []
        for param_name, param in sig.parameters.items():
            if param_name == "self":
                continue
            annotation = (
                param.annotation.__name__
                if hasattr(param.annotation, "__name__")
                else str(param.annotation)
            )
            if param.annotation == inspect.Parameter.empty:
                annotation = "Any"
            if param.default != inspect.Parameter.empty:
                params.append(f"{param_name}: {annotation} = {param.default!r}")
            else:
                params.append(f"{param_name}: {annotation}")

        return_annotation = sig.return_annotation
        if return_annotation == inspect.Signature.empty:
            return_type = "Any"
        elif hasattr(return_annotation, "__name__"):
            return_type = return_annotation.__name__
        else:
            return_type = str(return_annotation)

        signature_str = f"{name}({', '.join(params)}) -> {return_type}"
        docstring = inspect.getdoc(method) or ""
        return f'{signature_str}\n    """{docstring}"""'

    def _get_decomposition_source(self, method: Callable[..., Any]) -> str:
        """Extract the source code body of a decomposition method."""
        try:
            source = inspect.getsource(method)
            # Dedent to remove class-level indentation before parsing
            source = textwrap.dedent(source)
            # Parse and extract just the function body (skip decorator and def line)
            tree = ast.parse(source)
            if tree.body and isinstance(tree.body[0], ast.FunctionDef):
                func_def = tree.body[0]
                # Get the body statements as source
                body_lines = []
                for stmt in func_def.body:
                    # Skip docstrings
                    if isinstance(stmt, ast.Expr) and isinstance(
                        stmt.value, ast.Constant
                    ):
                        continue
                    body_lines.append(ast.unparse(stmt))
                # Strip self. prefixes — plans use bare function calls, not method calls
                return "\n".join(body_lines).replace("self.", "")
        except (OSError, TypeError):
            pass
        return ""

    # -------------------------------------------------------------------------
    # Prompt Building
    # -------------------------------------------------------------------------

    def _format_type_definitions(
        self, primitives: list[tuple[str, Callable[..., Any]]]
    ) -> str:
        """Build a ## Type Definitions prompt section for Pydantic BaseModel types.

        Scans parameter and return type annotations of all primitives,
        recursively unwraps generic wrappers (list[X], X | None, Union, etc.),
        collects any BaseModel subclasses found, and returns a formatted section
        listing each model's fields and types.  Returns "" when no models are
        referenced.
        """
        from typing import get_args, get_type_hints

        from pydantic import BaseModel

        found: set[type[BaseModel]] = set()

        def _extract_models(annotation: Any) -> None:
            if (
                isinstance(annotation, type)
                and issubclass(annotation, BaseModel)
                and annotation is not BaseModel
            ):
                found.add(annotation)
                return
            for arg in get_args(annotation):
                _extract_models(arg)

        for _name, method in primitives:
            try:
                hints = get_type_hints(method)
            except Exception:
                # Fallback: use inspect.signature annotations directly
                sig = inspect.signature(method)
                hints = {}
                for pname, param in sig.parameters.items():
                    if pname != "self" and param.annotation != inspect.Parameter.empty:
                        hints[pname] = param.annotation
                if sig.return_annotation != inspect.Signature.empty:
                    hints["return"] = sig.return_annotation

            for hint in hints.values():
                _extract_models(hint)

        if not found:
            return ""

        models = sorted(found, key=lambda cls: cls.__name__)
        lines: list[str] = []
        for model in models:
            fields: list[str] = []
            for field_name, field_info in model.model_fields.items():
                ann = field_info.annotation
                type_str = (
                    ann.__name__
                    if ann is not None and hasattr(ann, "__name__")
                    else str(ann)
                )
                fields.append(f"{field_name}: {type_str}")
            lines.append(f"{model.__name__}({', '.join(fields)})")

        return "\n## Type Definitions\n\n" + "\n".join(lines) + "\n"

    def _format_history_for_prompt(self) -> str:
        """Format conversation history for inclusion in the prompt."""
        if not self._history:
            return ""

        sections = []
        for i, turn in enumerate(self._history, 1):
            section = f"### Turn {i}\n"
            section += f"Task: {turn.task}\n"
            section += f"Plan:\n```python\n{turn.plan}\n```\n"
            if turn.success:
                section += f"Result: {turn.result!r}"
            else:
                section += f"Error: {turn.error}"
            sections.append(section)

        return "\n\n".join(sections)

    def build_plan_prompt(self, task: str, feedback: str | None = None) -> str:
        """Build the prompt for the LLM to generate a plan.

        Override this method in subclasses to customize the prompt sent to the LLM.

        Args:
            task: The task description to plan for.
            feedback: Optional error feedback from a previous failed plan attempt.

        Returns:
            The complete prompt string to send to the LLM.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        # Build primitive documentation
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        # Build decomposition examples
        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build conversation history section if in multi-turn mode
        history_section = ""
        if self.config.multi_turn and self._history:
            history_section = f"""
## Conversation History

Previous turns in this conversation. You can reference variables from previous turns.

{self._format_history_for_prompt()}

"""

        # Build feedback section if retrying after a failed plan
        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

Here are examples of how to compose primitives:

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{history_section}{feedback_section}## Task

Generate Python code to accomplish this task: {task}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. Output ONLY Python assignment statements
2. Each statement must assign a result to a variable
3. You can ONLY call the primitive methods listed above
4. Do NOT use imports, loops, conditionals, or function definitions
5. Do NOT use any dangerous operations (exec, eval, open, etc.)
6. Call primitives directly (e.g. `add(a=1, b=2)`), do NOT use `self.`
7. The last assigned variable will be the final result

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

    def _extract_code_block(self, response_text: str) -> str:
        """Extract code from markdown code blocks in an LLM response.

        Args:
            response_text: The raw text response from the LLM.

        Returns:
            The extracted code string.
        """
        text = response_text.strip()
        code = text
        if "```python" in text:
            # Extract code between ```python and ```
            start = text.find("```python") + len("```python")
            end = text.find("```", start)
            if end > start:
                code = text[start:end]
        elif "```" in text:
            # Generic code block
            start = text.find("```") + 3
            end = text.find("```", start)
            if end > start:
                code = text[start:end]

        # Normalize indentation - some models (especially smaller local ones)
        # may return code with unexpected leading indentation
        code = textwrap.dedent(code).strip()

        # Strip trailing markdown code block delimiters that weren't properly matched
        # This handles cases where the LLM adds a closing ``` without proper opening
        if code.endswith("```"):
            code = code[:-3].rstrip()

        return code

    def on_code_extracted(self, raw_response: str, extracted_code: str) -> str:
        """Hook called after code is extracted from the LLM response.

        Override this method in subclasses to observe and/or modify the
        extracted code before it is used for planning.

        Args:
            raw_response: The original raw text response from the LLM.
            extracted_code: The code extracted by _extract_code_block.

        Returns:
            The final code to use (can be modified from extracted_code).
        """
        return extracted_code

    # -------------------------------------------------------------------------
    # Plan Generation
    # -------------------------------------------------------------------------

    def plan(self, task: str, feedback: str | None = None) -> PlanResult:
        """Generate a plan (Python statements) for a task.

        Args:
            task: The task description to plan for.
            feedback: Optional error feedback from a previous failed plan attempt.

        Returns:
            PlanResult containing the generated plan and metrics.
        """
        plan_span: str | None = None
        if self._tracer:
            plan_span = self._tracer.start_span(
                EventType.PLAN_START, {"task": task, "feedback": feedback}
            )

        prompt = self.build_plan_prompt(task, feedback=feedback)

        llm_span: str | None = None
        if self._tracer:
            llm_payload: dict[str, Any] = {}
            if self._tracer.config.capture_llm_prompts:
                llm_payload["prompt"] = prompt
            llm_span = self._tracer.start_span(
                EventType.PLAN_LLM_REQUEST, llm_payload, defer=True
            )

        start_time = time.perf_counter()
        response = self._llm.generate(prompt)
        elapsed = time.perf_counter() - start_time

        # Extract code from response (handle markdown code blocks)
        raw_response = response.text
        extracted_code = self._extract_code_block(raw_response)
        plan_text = self.on_code_extracted(raw_response, extracted_code)

        # Build full LLM interaction details
        llm_interaction = LLMInteraction(
            prompt=prompt,
            response=raw_response,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            time_seconds=elapsed,
            provider=response.provider,
            model=response.model,
        )
        plan_generation = PlanGeneration(
            llm_interaction=llm_interaction,
            extracted_code=extracted_code,
        )

        if self._tracer and llm_span:
            response_payload = self._tracer.filter.llm_interaction(
                llm_interaction.model_dump()
            ) if self._tracer.config.capture_llm_responses else {}
            self._tracer.end_span(
                llm_span, EventType.PLAN_LLM_RESPONSE, response_payload
            )

        plan_result = PlanResult(
            plan=plan_text,
            usage=TokenUsage(
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
            ),
            time_seconds=elapsed,
            provider=response.provider,
            model=response.model,
            plan_generation=plan_generation,
        )

        if self._tracer and plan_span:
            self._tracer.end_span(
                plan_span,
                EventType.PLAN_COMPLETE,
                self._tracer.filter.plan_result(plan_result.model_dump()),
            )

        return plan_result

    # -------------------------------------------------------------------------
    # Plan Analysis and Validation
    # -------------------------------------------------------------------------

    def analyze_plan(self, plan: str) -> PlanAnalysis:
        """Analyze a plan to extract primitive calls.

        Args:
            plan: The Python statements to analyze.

        Returns:
            PlanAnalysis containing all primitive calls found.

        Raises:
            ValueError: If the plan has invalid syntax.
        """
        try:
            tree = ast.parse(plan)
        except SyntaxError as e:
            raise ValueError(f"Invalid Python syntax: {e}") from e

        primitive_names = self._get_primitive_names()
        read_only_map = self._get_primitive_read_only_map()
        calls: list[PrimitiveCall] = []

        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue

            method_name: str | None = None

            # Direct call: method_name(...)
            if isinstance(node.func, ast.Name) and node.func.id in primitive_names:
                method_name = node.func.id

            if method_name is not None:
                args: dict[str, Any] = {}
                for kw in node.keywords:
                    if kw.arg is not None:
                        try:
                            args[kw.arg] = ast.literal_eval(kw.value)
                        except (ValueError, TypeError):
                            args[kw.arg] = ast.unparse(kw.value)

                for i, arg in enumerate(node.args):
                    try:
                        args[f"arg{i}"] = ast.literal_eval(arg)
                    except (ValueError, TypeError):
                        args[f"arg{i}"] = ast.unparse(arg)

                calls.append(
                    PrimitiveCall(
                        method_name=method_name,
                        read_only=read_only_map.get(method_name, False),
                        args=args,
                    )
                )

        return PlanAnalysis(calls=calls)

    def validate_plan(self, plan: str) -> None:
        """Validate that a plan only uses allowed operations.

        Args:
            plan: The Python statements to validate.

        Raises:
            ValueError: If the plan contains disallowed operations.
        """
        try:
            tree = ast.parse(plan)
        except SyntaxError as e:
            raise ValueError(f"Invalid Python syntax: {e}") from e

        primitive_names = self._get_primitive_names()

        disallowed_statements: tuple[type, ...] = (
            ast.If,
            ast.For,
            ast.While,
            ast.Try,
            ast.With,
            ast.FunctionDef,
            ast.AsyncFunctionDef,
            ast.ClassDef,
            ast.Import,
            ast.ImportFrom,
            ast.Global,
            ast.Nonlocal,
            ast.Raise,
            ast.Assert,
            ast.Delete,
        )
        if hasattr(ast, "Match"):
            disallowed_statements = (*disallowed_statements, ast.Match)

        for node in ast.walk(tree):
            if isinstance(node, disallowed_statements):
                node_type = type(node).__name__
                raise ValueError(f"{node_type} statements are not allowed in plans")

        # Every top-level statement must be an assignment
        for stmt in tree.body:
            if not isinstance(stmt, (ast.Assign, ast.AnnAssign)):
                stmt_type = type(stmt).__name__
                raise ValueError(
                    f"Every statement must be an assignment. Found: {stmt_type}"
                )

        self._validate_ast_nodes(tree, primitive_names)

    def _validate_ast_nodes(self, tree: ast.Module, primitive_names: set[str]) -> None:
        """Validate AST nodes for dangerous ops, self. prefix, and private attrs.

        Shared between PlanExecute and DesignExecute validation.
        """
        for node in ast.walk(tree):
            if isinstance(node, ast.Attribute) and node.attr.startswith("_"):
                raise ValueError(
                    f"Accessing private/dunder attributes not allowed: {node.attr}"
                )

            if isinstance(node, ast.Call):
                if isinstance(node.func, ast.Name):
                    func_name = node.func.id
                    # Check allowed_builtins first so user config can override DANGEROUS_BUILTINS
                    if func_name in self.DANGEROUS_BUILTINS and func_name not in self.allowed_builtins:
                        raise ValueError(f"Calling '{func_name}' is not allowed")
                    allowed_names = primitive_names | set(self.allowed_builtins.keys())
                    if func_name not in allowed_names:
                        raise ValueError(
                            f"Function '{func_name}' is not allowed. "
                            f"Only primitive methods and allowed builtins can be called."
                        )

                if (
                    isinstance(node.func, ast.Attribute)
                    and isinstance(node.func.value, ast.Name)
                    and node.func.value.id == "self"
                ):
                    raise ValueError(
                        "Do not use 'self.' prefix — call primitives directly "
                        f"(e.g. `{node.func.attr}(...)` instead of "
                        f"`self.{node.func.attr}(...)`)."
                    )

    # -------------------------------------------------------------------------
    # Plan Generation Helpers
    # -------------------------------------------------------------------------

    @staticmethod
    def _plan_generation_from_result(plan_result: PlanResult) -> PlanGeneration:
        """Get a PlanGeneration from a PlanResult, creating a fallback if needed."""
        if plan_result.plan_generation is not None:
            return plan_result.plan_generation
        return PlanGeneration(
            llm_interaction=LLMInteraction(
                prompt="",
                response="",
                input_tokens=plan_result.usage.input_tokens,
                output_tokens=plan_result.usage.output_tokens,
                time_seconds=plan_result.time_seconds,
                provider=plan_result.provider,
                model=plan_result.model,
            ),
            extracted_code=plan_result.plan,
        )

    # -------------------------------------------------------------------------
    # Mutation Detection (for checkpoint-based execution)
    # -------------------------------------------------------------------------

    @staticmethod
    def _detect_mutation(
        stmt: ast.stmt, read_only_map: dict[str, bool]
    ) -> MutationDetection:
        """Detect whether an AST statement is a non-read-only primitive call."""
        result = MutationDetection()

        if isinstance(stmt, ast.Assign):
            if stmt.targets and isinstance(stmt.targets[0], ast.Name):
                result.variable_name = stmt.targets[0].id
            if isinstance(stmt.value, ast.Call):
                call = stmt.value
                if isinstance(call.func, ast.Name):
                    result.method_name = call.func.id
                elif isinstance(call.func, ast.Attribute):
                    result.method_name = call.func.attr

                if result.method_name and not read_only_map.get(result.method_name, True):
                    result.is_mutation = True
                    for idx, arg in enumerate(call.args):
                        try:
                            result.args[f"arg{idx}"] = ast.literal_eval(arg)
                        except (ValueError, TypeError):
                            result.args[f"arg{idx}"] = ast.unparse(arg)
                    for kw in call.keywords:
                        if kw.arg:
                            try:
                                result.args[kw.arg] = ast.literal_eval(kw.value)
                            except (ValueError, TypeError):
                                result.args[kw.arg] = ast.unparse(kw.value)

        return result

    # -------------------------------------------------------------------------
    # Namespace Helpers
    # -------------------------------------------------------------------------

    def _build_reserved_names(self) -> set[str]:
        """Names that should be excluded from namespace snapshots."""
        return (
            set(self.allowed_builtins.keys())
            | {name for name, _ in self._get_primitive_methods()}
        )

    def _build_namespace(self) -> dict[str, Any]:
        """Build the execution namespace with primitives, builtins, and persisted vars.

        Note: adds *raw* primitive methods.  ``DesignExecute`` intentionally
        does NOT call this because it wraps primitives in traced wrappers.
        """
        namespace: dict[str, Any] = {}
        for name, method in self._get_primitive_methods():
            namespace[name] = method
        namespace.update(self.allowed_builtins)
        if self.config.multi_turn:
            namespace.update(self._persisted_namespace)
        return namespace

    def _persist_user_variables(
        self, namespace: dict[str, Any], reserved_names: set[str]
    ) -> None:
        """Persist user-defined variables for multi-turn mode."""
        if self.config.multi_turn:
            for key, value in namespace.items():
                if key not in reserved_names:
                    self._persisted_namespace[key] = value

    # -------------------------------------------------------------------------
    # Step-by-Step Execution
    # -------------------------------------------------------------------------

    def _snapshot_namespace(
        self, namespace: dict[str, Any], reserved_names: set[str]
    ) -> dict[str, Any]:
        """Create a JSON-serializable snapshot of user-defined variables in the namespace."""
        snapshot: dict[str, Any] = {}
        for key, value in namespace.items():
            if key in reserved_names:
                continue
            try:
                # Try to JSON serialize to ensure it's capturable
                json.dumps(value, default=str)
                snapshot[key] = value
            except (TypeError, ValueError):
                snapshot[key] = f"<{type(value).__name__}>"
        return snapshot

    def _resolve_arg_value(
        self, node: ast.expr, namespace: dict[str, Any]
    ) -> ArgumentValue:
        """Resolve an argument AST node to an ArgumentValue with expression and resolved value."""
        expression = ast.unparse(node)
        variable_reference: str | None = None
        resolved_value: Any = None

        # Check if it's a simple variable reference
        if isinstance(node, ast.Name):
            variable_reference = node.id
            resolved_value = namespace.get(node.id)
        else:
            # Try to evaluate the expression in the namespace
            try:
                resolved_value = ast.literal_eval(node)
            except (ValueError, TypeError):
                # Try to eval in the namespace for more complex expressions
                try:
                    resolved_value = eval(  # noqa: S307
                        expression, empty_builtins(), namespace
                    )
                except Exception:
                    resolved_value = None

        return ArgumentValue(
            expression=expression,
            resolved_value=resolved_value,
            variable_reference=variable_reference,
        )

    def _execute_statement(
        self,
        stmt: ast.stmt,
        namespace: dict[str, Any],
        step_number: int,
        read_only_map: dict[str, bool] | None = None,
        reserved_names: set[str] | None = None,
    ) -> ExecutionStep:
        """Execute a single statement and return the step result."""
        statement_str = ast.unparse(stmt)
        start_time = time.perf_counter()

        if reserved_names is None:
            reserved_names = set()

        # Capture namespace before execution
        namespace_before = self._snapshot_namespace(namespace, reserved_names)

        # Determine variable name and extract call info
        variable_name = ""
        primitive_called = None
        args: dict[str, ArgumentValue] = {}

        if isinstance(stmt, ast.Assign):
            if stmt.targets and isinstance(stmt.targets[0], ast.Name):
                variable_name = stmt.targets[0].id
            # Check if value is a Call
            if isinstance(stmt.value, ast.Call):
                call = stmt.value
                if isinstance(call.func, ast.Name):
                    primitive_called = call.func.id
                elif isinstance(call.func, ast.Attribute):
                    primitive_called = call.func.attr

                # Extract positional args
                for i, arg in enumerate(call.args):
                    args[f"arg{i}"] = self._resolve_arg_value(arg, namespace)

                # Extract keyword args
                for kw in call.keywords:
                    if kw.arg is not None:
                        args[kw.arg] = self._resolve_arg_value(kw.value, namespace)

        elif isinstance(stmt, ast.AnnAssign):
            if isinstance(stmt.target, ast.Name):
                variable_name = stmt.target.id
            # Check if value is a Call (annotated assignments can also have calls)
            if stmt.value is not None and isinstance(stmt.value, ast.Call):
                call = stmt.value
                if isinstance(call.func, ast.Name):
                    primitive_called = call.func.id
                elif isinstance(call.func, ast.Attribute):
                    primitive_called = call.func.attr

                # Extract positional args
                for i, arg in enumerate(call.args):
                    args[f"arg{i}"] = self._resolve_arg_value(arg, namespace)

                # Extract keyword args
                for kw in call.keywords:
                    if kw.arg is not None:
                        args[kw.arg] = self._resolve_arg_value(kw.value, namespace)

        # Check mutation hook BEFORE execution for non-read-only primitives
        is_mutation = (
            primitive_called is not None
            and read_only_map is not None
            and not read_only_map.get(primitive_called, True)
        )

        if is_mutation and self.config.on_mutation is not None:
            assert primitive_called is not None  # Guaranteed by is_mutation check
            # Convert ArgumentValue dict to plain dict for hook context
            plain_args = {k: v.resolved_value for k, v in args.items()}
            hook_context = MutationHookContext(
                method_name=primitive_called,
                args=plain_args,
                result=None,  # Not yet executed
                step=None,  # Not yet created
            )
            rejection_reason = self.config.on_mutation(hook_context)

            if rejection_reason is not None:
                # Hook rejected the mutation - fail without executing
                elapsed = time.perf_counter() - start_time
                return ExecutionStep(
                    step_number=step_number,
                    statement=statement_str,
                    variable_name=variable_name,
                    primitive_called=primitive_called,
                    args=args,
                    namespace_before=namespace_before,
                    namespace_after=namespace_before,  # No change since rejected
                    result_type="",
                    result_value=None,
                    result_json=NULL_JSON,
                    time_seconds=elapsed,
                    success=False,
                    error=f"{MUTATION_REJECTED_PREFIX}: {rejection_reason}",
                )

        try:
            # Execute the single statement
            exec(  # noqa: S102
                compile(ast.Module(body=[stmt], type_ignores=[]), PLAN_COMPILE_SOURCE, "exec"),
                empty_builtins(),
                namespace,
            )
            elapsed = time.perf_counter() - start_time

            # Capture namespace after execution
            namespace_after = self._snapshot_namespace(namespace, reserved_names)

            # Get the result value
            result_value = namespace.get(variable_name) if variable_name else None

            result_json = (
                NULL_JSON
                if self.config.skip_result_serialization
                else json.dumps(result_value, default=str)
            )

            step = ExecutionStep(
                step_number=step_number,
                statement=statement_str,
                variable_name=variable_name,
                primitive_called=primitive_called,
                args=args,
                namespace_before=namespace_before,
                namespace_after=namespace_after,
                result_type=type(result_value).__name__
                if result_value is not None
                else NONE_TYPE_NAME,
                result_value=result_value,
                result_json=result_json,
                time_seconds=elapsed,
                success=True,
            )

            return step

        except Exception as e:
            elapsed = time.perf_counter() - start_time
            # Capture namespace after failed execution (may have partial changes)
            namespace_after = self._snapshot_namespace(namespace, reserved_names)
            return ExecutionStep(
                step_number=step_number,
                statement=statement_str,
                variable_name=variable_name,
                primitive_called=primitive_called,
                args=args,
                namespace_before=namespace_before,
                namespace_after=namespace_after,
                result_type="",
                result_value=None,
                result_json=NULL_JSON,
                time_seconds=elapsed,
                success=False,
                error=str(e),
            )

    def execute(self, plan: str) -> ExecutionResult:
        """Execute a plan step-by-step with full tracing.

        Args:
            plan: The Python statements to execute.

        Returns:
            ExecutionResult containing the final value and execution trace.

        Raises:
            ValueError: If the plan contains disallowed operations.
        """
        self.validate_plan(plan)

        exec_span: str | None = None
        if self._tracer:
            exec_span = self._tracer.start_span(
                EventType.EXECUTION_START,
                {"plan": plan} if self._tracer.config.capture_plan_source else {},
            )

        tree = ast.parse(plan)
        steps: list[ExecutionStep] = []
        total_start = time.perf_counter()

        namespace = self._build_namespace()
        read_only_map = self._get_primitive_read_only_map()
        reserved_names = self._build_reserved_names()

        # Execute statement by statement
        for i, stmt in enumerate(tree.body, 1):
            step = self._execute_statement(
                stmt, namespace, i, read_only_map, reserved_names
            )
            steps.append(step)

            if self._tracer and self._tracer.config.capture_execution_steps:
                self._tracer.emit(
                    EventType.EXECUTION_STEP,
                    self._tracer.filter.execution_step(step.model_dump()),
                )

            # Stop on first error
            if not step.success:
                break

        self._persist_user_variables(namespace, reserved_names)

        total_elapsed = time.perf_counter() - total_start

        trace = ExecutionTrace(
            steps=steps,
            total_time_seconds=total_elapsed,
        )

        # Determine final result
        if steps and steps[-1].success:
            last = steps[-1]
            result = ExecutionResult(
                value_type=last.result_type,
                value_name=last.variable_name,
                value_json=last.result_json,
                trace=trace,
            )
        else:
            result = ExecutionResult(
                value_type=NONE_TYPE_NAME,
                value_name="",
                value_json=NULL_JSON,
                trace=trace,
            )

        if self._tracer and exec_span:
            self._tracer.end_span(
                exec_span,
                EventType.EXECUTION_COMPLETE,
                ExecutionSummary(
                    value_type=result.value_type,
                    value_name=result.value_name,
                    step_count=trace.step_count,
                    all_succeeded=trace.all_succeeded,
                    total_time_seconds=total_elapsed,
                ).model_dump(),
            )

        return result

    # -------------------------------------------------------------------------
    # Checkpoint-based Execution (Distributed/Interruptible)
    # -------------------------------------------------------------------------

    def execute_stepwise(
        self,
        task: str,
        plan_context: PlanContext | None = None,
        serializer: SerializerRegistry | None = None,
        checkpoint_id: str | None = None,
    ) -> Iterator[ExecutionCheckpoint]:
        """Execute a plan step-by-step, yielding checkpoints for persistence.

        This method enables distributed execution by yielding a checkpoint after
        each step. The checkpoint can be persisted to a database and resumed on
        any worker.

        When `require_mutation_approval` is True (default), execution pauses
        before mutations and yields a checkpoint with status `awaiting_approval`.
        Call `resume_from_checkpoint()` with `approve_mutation=True` to continue.

        Args:
            task: The task description to accomplish.
            plan_context: Optional pre-computed plan context (from a previous
                planning phase). If None, will generate a new plan.
            serializer: Custom serializer registry. Defaults to the global registry.
            checkpoint_id: Optional ID for the checkpoint. Auto-generated if None.

        Yields:
            ExecutionCheckpoint after each step, allowing for persistence and
            distributed resume.

        Example:
            # Simple usage - collect all checkpoints
            for checkpoint in agent.execute_stepwise("do something"):
                store.save(checkpoint)
                if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
                    # Handle approval externally
                    break

            # Resume after approval
            checkpoint = store.load(checkpoint_id)
            for checkpoint in agent.resume_from_checkpoint(checkpoint, approve_mutation=True):
                store.save(checkpoint)
        """
        serializer = serializer or default_serializer_registry

        # Generate plan if not provided
        if plan_context is None:
            plan_result = self.plan(task)
            plan_context = PlanContext(
                plan_attempts=[
                    PlanAttempt(
                        attempt_number=1,
                        plan_generation=self._plan_generation_from_result(plan_result),
                        success=True,
                    )
                ],
                final_plan=plan_result.plan,
                generation_time_seconds=plan_result.time_seconds,
            )

        plan = plan_context.final_plan

        # Validate plan
        try:
            self.validate_plan(plan)
        except ValueError as e:
            checkpoint = ExecutionCheckpoint(
                task=task,
                plan=plan,
                plan_context=plan_context,
                current_step=0,
                total_steps=0,
                status=CheckpointStatus.FAILED,
                error=str(e),
                worker_id=self.config.worker_id,
            )
            if checkpoint_id is not None:
                checkpoint.checkpoint_id = checkpoint_id
            yield checkpoint
            return

        tree = ast.parse(plan)
        total_steps = len(tree.body)

        namespace = self._build_namespace()
        read_only_map = self._get_primitive_read_only_map()
        reserved_names = self._build_reserved_names()

        steps: list[ExecutionStep] = []

        # Create initial checkpoint
        checkpoint = ExecutionCheckpoint(
            task=task,
            plan=plan,
            plan_context=plan_context,
            current_step=0,
            total_steps=total_steps,
            status=CheckpointStatus.RUNNING,
            namespace_snapshot=serializer.serialize_namespace(
                namespace, reserved_names
            ),
            completed_steps=[],
            worker_id=self.config.worker_id,
        )
        if checkpoint_id is not None:
            checkpoint.checkpoint_id = checkpoint_id

        # Execute statement by statement
        for i, stmt in enumerate(tree.body):
            step_number = i + 1
            statement_str = ast.unparse(stmt)

            mutation = self._detect_mutation(stmt, read_only_map)

            # If mutation requires approval, pause and yield
            if mutation.is_mutation and self.config.require_mutation_approval:
                checkpoint.current_step = i
                checkpoint.status = CheckpointStatus.AWAITING_APPROVAL
                checkpoint.pending_mutation = PendingMutation(
                    method_name=mutation.method_name or "",
                    args=mutation.args,
                    statement=statement_str,
                    step_number=step_number,
                    variable_name=mutation.variable_name,
                )
                checkpoint.namespace_snapshot = serializer.serialize_namespace(
                    namespace, reserved_names
                )
                checkpoint.completed_steps = steps.copy()
                checkpoint.touch(self.config.worker_id)
                yield checkpoint
                return  # Caller must resume with approve_mutation=True

            # Execute the step
            step = self._execute_statement(
                stmt, namespace, step_number, read_only_map, reserved_names
            )
            steps.append(step)

            # Update checkpoint
            checkpoint.current_step = step_number
            checkpoint.completed_steps = steps.copy()
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )

            if not step.success:
                checkpoint.status = CheckpointStatus.FAILED
                checkpoint.error = step.error
                checkpoint.touch(self.config.worker_id)
                yield checkpoint
                return

            # Yield progress checkpoint
            checkpoint.status = CheckpointStatus.RUNNING
            checkpoint.touch(self.config.worker_id)
            yield checkpoint

        # Execution completed successfully
        checkpoint.status = CheckpointStatus.COMPLETED
        if steps:
            last_step = steps[-1]
            checkpoint.result_variable = last_step.variable_name
            result_value = namespace.get(last_step.variable_name)
            checkpoint.result_value = serializer.serialize(result_value)

        self._persist_user_variables(namespace, reserved_names)

        checkpoint.touch(self.config.worker_id)
        yield checkpoint

    def resume_from_checkpoint(
        self,
        checkpoint: ExecutionCheckpoint,
        approve_mutation: bool = False,
        serializer: SerializerRegistry | None = None,
    ) -> Iterator[ExecutionCheckpoint]:
        """Resume execution from a persisted checkpoint.

        This method reconstructs execution state from a checkpoint and continues
        execution from where it left off.

        Args:
            checkpoint: The checkpoint to resume from.
            approve_mutation: If True and checkpoint is awaiting approval,
                execute the pending mutation and continue.
            serializer: Custom serializer registry. Must match the one used
                when creating the checkpoint.

        Yields:
            ExecutionCheckpoint after each step.

        Raises:
            ValueError: If checkpoint is in a terminal state or approval is
                required but not provided.

        Example:
            # Load and resume
            checkpoint = store.load(checkpoint_id)
            if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
                # Get user approval somehow
                if user_approved:
                    for cp in agent.resume_from_checkpoint(checkpoint, approve_mutation=True):
                        store.save(cp)
        """
        serializer = serializer or default_serializer_registry

        if checkpoint.is_terminal:
            raise ValueError(
                f"Cannot resume from terminal checkpoint (status: {checkpoint.status})"
            )

        if (
            checkpoint.status == CheckpointStatus.AWAITING_APPROVAL
            and not approve_mutation
        ):
            raise ValueError(
                "Checkpoint is awaiting mutation approval. "
                "Set approve_mutation=True to continue."
            )

        # Parse the plan
        tree = ast.parse(checkpoint.plan)
        total_steps = len(tree.body)

        namespace = self._build_namespace()

        # Restore serialized variables (skip undeserializable values)
        import contextlib

        for var_name, serialized_val in checkpoint.namespace_snapshot.items():
            with contextlib.suppress(ValueError):
                namespace[var_name] = serializer.deserialize(serialized_val)

        read_only_map = self._get_primitive_read_only_map()
        reserved_names = self._build_reserved_names()

        # Copy completed steps
        steps = list(checkpoint.completed_steps)

        # Determine starting point
        start_step = checkpoint.current_step

        # If resuming from awaiting_approval, execute the pending mutation first
        if (
            checkpoint.status == CheckpointStatus.AWAITING_APPROVAL
            and approve_mutation
            and checkpoint.pending_mutation
        ):
            stmt = tree.body[start_step]
            step_number = start_step + 1

            step = self._execute_statement(
                stmt, namespace, step_number, read_only_map, reserved_names
            )
            steps.append(step)

            checkpoint.completed_steps = steps.copy()
            checkpoint.pending_mutation = None

            if not step.success:
                checkpoint.status = CheckpointStatus.FAILED
                checkpoint.error = step.error
                checkpoint.current_step = step_number
                checkpoint.namespace_snapshot = serializer.serialize_namespace(
                    namespace, reserved_names
                )
                checkpoint.touch(self.config.worker_id)
                yield checkpoint
                return

            checkpoint.current_step = step_number
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )
            checkpoint.status = CheckpointStatus.RUNNING
            checkpoint.touch(self.config.worker_id)
            yield checkpoint

            start_step += 1

        # Continue with remaining steps
        for i in range(start_step, total_steps):
            stmt = tree.body[i]
            step_number = i + 1
            statement_str = ast.unparse(stmt)

            mutation = self._detect_mutation(stmt, read_only_map)

            # Pause for mutation approval if required
            if mutation.is_mutation and self.config.require_mutation_approval:
                checkpoint.current_step = i
                checkpoint.status = CheckpointStatus.AWAITING_APPROVAL
                checkpoint.pending_mutation = PendingMutation(
                    method_name=mutation.method_name or "",
                    args=mutation.args,
                    statement=statement_str,
                    step_number=step_number,
                    variable_name=mutation.variable_name,
                )
                checkpoint.namespace_snapshot = serializer.serialize_namespace(
                    namespace, reserved_names
                )
                checkpoint.completed_steps = steps.copy()
                checkpoint.touch(self.config.worker_id)
                yield checkpoint
                return

            # Execute step
            step = self._execute_statement(
                stmt, namespace, step_number, read_only_map, reserved_names
            )
            steps.append(step)

            checkpoint.current_step = step_number
            checkpoint.completed_steps = steps.copy()
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )

            if not step.success:
                checkpoint.status = CheckpointStatus.FAILED
                checkpoint.error = step.error
                checkpoint.touch(self.config.worker_id)
                yield checkpoint
                return

            checkpoint.status = CheckpointStatus.RUNNING
            checkpoint.touch(self.config.worker_id)
            yield checkpoint

        # Completed
        checkpoint.status = CheckpointStatus.COMPLETED
        if steps:
            last_step = steps[-1]
            checkpoint.result_variable = last_step.variable_name
            result_value = namespace.get(last_step.variable_name)
            checkpoint.result_value = serializer.serialize(result_value)

        self._persist_user_variables(namespace, reserved_names)

        checkpoint.touch(self.config.worker_id)
        yield checkpoint

    def run_with_checkpoints(
        self,
        task: str,
        store: CheckpointStore,
        serializer: SerializerRegistry | None = None,
        auto_approve: bool = False,
    ) -> ExecutionCheckpoint:
        """Run execution with automatic checkpoint persistence.

        Convenience method that handles checkpoint saving automatically.

        Args:
            task: The task to execute.
            store: Checkpoint store for persistence.
            serializer: Custom serializer registry.
            auto_approve: If True, automatically approve all mutations.

        Returns:
            The final checkpoint (completed or failed).
        """
        checkpoint: ExecutionCheckpoint | None = None

        for checkpoint in self.execute_stepwise(task, serializer=serializer):
            store.save(checkpoint)

            if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
                if auto_approve:
                    # Resume with approval - consume all checkpoints from resume
                    current_cp = checkpoint
                    while current_cp.status == CheckpointStatus.AWAITING_APPROVAL:
                        for next_cp in self.resume_from_checkpoint(
                            current_cp, approve_mutation=True, serializer=serializer
                        ):
                            store.save(next_cp)
                            current_cp = next_cp
                    checkpoint = current_cp
                else:
                    # Return checkpoint for external approval
                    return checkpoint

        if checkpoint is None:
            raise RuntimeError("No checkpoint was generated")

        return checkpoint

    # -------------------------------------------------------------------------
    # Main Orchestration
    # -------------------------------------------------------------------------

    def _emit_run_end(
        self, run_span: str, result: OrchestrationResult
    ) -> None:
        """Emit the RUN_COMPLETE or RUN_ERROR event for a run span."""
        if not self._tracer:
            return
        if result.success:
            self._tracer.end_span(
                run_span,
                EventType.RUN_COMPLETE,
                RunCompleteSummary(
                    result_type=type(result.result).__name__
                    if result.result is not None
                    else NONE_TYPE_NAME,
                    steps_executed=result.metrics.steps_executed
                    if result.metrics
                    else 0,
                ).model_dump(),
            )
        else:
            self._tracer.end_span(
                run_span,
                EventType.RUN_ERROR,
                RunErrorSummary(
                    error=result.error or "Unknown error",
                ).model_dump(),
            )

    def run(self, task: str) -> OrchestrationResult:
        """Run the complete plan-and-execute cycle.

        Args:
            task: The task description to accomplish.

        Returns:
            OrchestrationResult containing the outcome and metrics.
        """
        run_span: str | None = None
        if self._tracer:
            self._tracer.new_trace()
            run_span = self._tracer.start_span(
                EventType.RUN_START,
                {"task": task, "config": self.config.model_dump(exclude={"observability"})},
            )

        result: OrchestrationResult | None = None
        try:
            result = self._run_inner(task, run_span)
            return result
        finally:
            if self._tracer:
                self._tracer.flush()
            _emit_telemetry(self, success=result is not None and result.success)

    def _run_inner(
        self, task: str, run_span: str | None
    ) -> OrchestrationResult:
        """Core run logic, separated so ``run()`` can wrap with try/finally."""
        plan_result = None
        feedback: str | None = None
        max_attempts = 1 + self.config.max_plan_retries
        plan_attempts: list[PlanAttempt] = []

        for attempt in range(max_attempts):
            try:
                # Generate plan (with feedback if retrying)
                plan_result = self.plan(task, feedback=feedback)

                # Record the plan attempt
                plan_attempt = PlanAttempt(
                    attempt_number=attempt + 1,
                    plan_generation=self._plan_generation_from_result(plan_result),
                    feedback=feedback,
                    validation_error=None,
                    success=True,
                )

                # Validate plan before execution (to catch validation errors for retry)
                try:
                    self.validate_plan(plan_result.plan)
                except ValueError as validation_error:
                    plan_attempt.validation_error = str(validation_error)
                    plan_attempt.success = False
                    plan_attempts.append(plan_attempt)
                    if self._tracer:
                        self._tracer.emit(
                            EventType.PLAN_VALIDATION_ERROR,
                            {"error": str(validation_error), "attempt": attempt + 1},
                        )
                    raise

                plan_attempts.append(plan_attempt)

                # Execute plan
                exec_start = time.perf_counter()
                exec_result = self.execute(plan_result.plan)
                exec_time = time.perf_counter() - exec_start

                metrics = ExecutionMetrics(
                    plan_tokens=plan_result.usage,
                    plan_time_seconds=plan_result.time_seconds,
                    execute_time_seconds=exec_time,
                    steps_executed=exec_result.trace.step_count,
                    provider=plan_result.provider,
                    model=plan_result.model,
                )

                # Check if execution succeeded
                if exec_result.trace.all_succeeded:
                    result_value = exec_result.get_value()

                    # Track history in multi-turn mode
                    if self.config.multi_turn:
                        self._history.append(
                            ConversationTurn(
                                task=task,
                                plan=plan_result.plan,
                                result=result_value,
                                success=True,
                            )
                        )

                    orch_result = OrchestrationResult(
                        success=True,
                        result=result_value,
                        metrics=metrics,
                        plan=plan_result.plan,
                        trace=exec_result.trace,
                        plan_attempts=plan_attempts,
                        task=task,
                    )
                    if run_span:
                        self._emit_run_end(run_span, orch_result)
                    return orch_result
                else:
                    # Get error from failed step
                    failed = (
                        exec_result.trace.failed_steps[0]
                        if exec_result.trace.failed_steps
                        else None
                    )
                    error_msg = failed.error if failed else "Unknown execution error"

                    # Track history in multi-turn mode (even for failures)
                    if self.config.multi_turn:
                        self._history.append(
                            ConversationTurn(
                                task=task,
                                plan=plan_result.plan,
                                success=False,
                                error=error_msg,
                            )
                        )

                    orch_result = OrchestrationResult(
                        success=False,
                        error=error_msg,
                        metrics=metrics,
                        plan=plan_result.plan,
                        trace=exec_result.trace,
                        plan_attempts=plan_attempts,
                        task=task,
                    )
                    if run_span:
                        self._emit_run_end(run_span, orch_result)
                    return orch_result

            except ValueError as e:
                # Validation error - retry if attempts remaining
                error_msg = str(e)
                if attempt < max_attempts - 1:
                    feedback = error_msg
                    continue

                # No more retries - return failure
                if self.config.multi_turn and plan_result is not None:
                    self._history.append(
                        ConversationTurn(
                            task=task,
                            plan=plan_result.plan,
                            success=False,
                            error=error_msg,
                        )
                    )

                orch_result = OrchestrationResult(
                    success=False,
                    error=error_msg,
                    plan=plan_result.plan if plan_result else None,
                    plan_attempts=plan_attempts,
                    task=task,
                )
                if run_span:
                    self._emit_run_end(run_span, orch_result)
                return orch_result

            except Exception as e:
                error_msg = str(e)

                # Track history in multi-turn mode (even for validation/other failures)
                if self.config.multi_turn and plan_result is not None:
                    self._history.append(
                        ConversationTurn(
                            task=task,
                            plan=plan_result.plan,
                            success=False,
                            error=error_msg,
                        )
                    )

                orch_result = OrchestrationResult(
                    success=False,
                    error=error_msg,
                    plan=plan_result.plan if plan_result else None,
                    plan_attempts=plan_attempts,
                    task=task,
                )
                if run_span:
                    self._emit_run_end(run_span, orch_result)
                return orch_result

        # Should not reach here, but satisfy type checker
        return OrchestrationResult(
            success=False,
            error="Unexpected: exhausted all plan retry attempts",
            plan=plan_result.plan if plan_result else None,
            plan_attempts=plan_attempts,
            task=task,
        )

blueprint_type property

The blueprint type: 'PlanExecute', 'DesignExecute', or 'GoalSeeking'.

history property

Get the conversation history (multi-turn mode only).

persisted_variables property

Get the persisted variables from previous turns (multi-turn mode only).

__init__(llm, name='', description='', config=None)

Initialize the PlanExecute agent.

Parameters:

Name	Type	Description	Default
llm	LLM | LLMConfig	LLM instance or config for plan generation.	required
name	str	Agent name for prompts.	''
description	str	Agent description for prompts.	''
config	PlanExecuteConfig | None	Extended configuration options.	None

Source code in src/opensymbolicai/blueprints/plan_execute.py

def __init__(
    self,
    llm: LLM | LLMConfig,
    name: str = "",
    description: str = "",
    config: PlanExecuteConfig | None = None,
) -> None:
    """Initialize the PlanExecute agent.

    Args:
        llm: LLM instance or config for plan generation.
        name: Agent name for prompts.
        description: Agent description for prompts.
        config: Extended configuration options.
    """
    if isinstance(llm, LLMConfig):
        self._llm = create_llm(llm)
    else:
        self._llm = llm

    self.name = name or self.__class__.__name__
    self.description = description or self.__class__.__doc__ or ""
    self.config = config or PlanExecuteConfig()
    self.allowed_builtins = (
        self.config.allowed_builtins
        if self.config.allowed_builtins is not None
        else DEFAULT_ALLOWED_BUILTINS.copy()
    )

    # Multi-turn state
    self._history: list[ConversationTurn] = []
    self._persisted_namespace: dict[str, Any] = {}

    # Observability
    self._tracer: Tracer | None = None
    if self.config.observability and self.config.observability.enabled:
        self._tracer = Tracer(self.config.observability, type(self).__name__)

analyze_plan(plan)

Analyze a plan to extract primitive calls.

Parameters:

Name	Type	Description	Default
plan	str	The Python statements to analyze.	required

Returns:

Type	Description
PlanAnalysis	PlanAnalysis containing all primitive calls found.

Raises:

Type	Description
ValueError	If the plan has invalid syntax.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def analyze_plan(self, plan: str) -> PlanAnalysis:
    """Analyze a plan to extract primitive calls.

    Args:
        plan: The Python statements to analyze.

    Returns:
        PlanAnalysis containing all primitive calls found.

    Raises:
        ValueError: If the plan has invalid syntax.
    """
    try:
        tree = ast.parse(plan)
    except SyntaxError as e:
        raise ValueError(f"Invalid Python syntax: {e}") from e

    primitive_names = self._get_primitive_names()
    read_only_map = self._get_primitive_read_only_map()
    calls: list[PrimitiveCall] = []

    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue

        method_name: str | None = None

        # Direct call: method_name(...)
        if isinstance(node.func, ast.Name) and node.func.id in primitive_names:
            method_name = node.func.id

        if method_name is not None:
            args: dict[str, Any] = {}
            for kw in node.keywords:
                if kw.arg is not None:
                    try:
                        args[kw.arg] = ast.literal_eval(kw.value)
                    except (ValueError, TypeError):
                        args[kw.arg] = ast.unparse(kw.value)

            for i, arg in enumerate(node.args):
                try:
                    args[f"arg{i}"] = ast.literal_eval(arg)
                except (ValueError, TypeError):
                    args[f"arg{i}"] = ast.unparse(arg)

            calls.append(
                PrimitiveCall(
                    method_name=method_name,
                    read_only=read_only_map.get(method_name, False),
                    args=args,
                )
            )

    return PlanAnalysis(calls=calls)

build_plan_prompt(task, feedback=None)

Build the prompt for the LLM to generate a plan.

Override this method in subclasses to customize the prompt sent to the LLM.

Parameters:

Name	Type	Description	Default
task	str	The task description to plan for.	required
feedback	str | None	Optional error feedback from a previous failed plan attempt.	None

Returns:

Type	Description
str	The complete prompt string to send to the LLM.

Source code in src/opensymbolicai/blueprints/plan_execute.py

    def build_plan_prompt(self, task: str, feedback: str | None = None) -> str:
        """Build the prompt for the LLM to generate a plan.

        Override this method in subclasses to customize the prompt sent to the LLM.

        Args:
            task: The task description to plan for.
            feedback: Optional error feedback from a previous failed plan attempt.

        Returns:
            The complete prompt string to send to the LLM.
        """
        primitives = self._get_prompt_primitives()
        decompositions = self._get_prompt_decompositions()

        # Build primitive documentation
        primitive_docs = [
            self._format_primitive_signature(name, method)
            for name, method in primitives
        ]

        # Build decomposition examples
        examples = []
        for _name, method, intent, expanded in decompositions:
            source = self._get_decomposition_source(method)
            if source:
                example = f"Intent: {intent}"
                if expanded:
                    example += f"\nApproach: {expanded}"
                example += f"\nPython:\n{source}"
                examples.append(example)

        # Build type definitions section for Pydantic models
        type_defs_section = self._format_type_definitions(primitives)

        # Build conversation history section if in multi-turn mode
        history_section = ""
        if self.config.multi_turn and self._history:
            history_section = f"""
## Conversation History

Previous turns in this conversation. You can reference variables from previous turns.

{self._format_history_for_prompt()}

"""

        # Build feedback section if retrying after a failed plan
        feedback_section = ""
        if feedback:
            feedback_section = f"""
## Previous Attempt Failed

Your previous plan was invalid. Please fix the following error and regenerate:

{feedback}

"""

        prompt = f"""You are {self.name}, an AI agent that generates Python code plans.

{self.description}

{PROMPT_DEFINITIONS_BEGIN}

## Available Primitive Methods

You can ONLY call these methods:

```python
{chr(10).join(primitive_docs)}
```
{type_defs_section}
## Example Decompositions

Here are examples of how to compose primitives:

{chr(10).join(f"### Example {i + 1}{chr(10)}{ex}" for i, ex in enumerate(examples)) if examples else "No examples available."}

{PROMPT_DEFINITIONS_END}

{PROMPT_CONTEXT_BEGIN}
{history_section}{feedback_section}## Task

Generate Python code to accomplish this task: {task}

{PROMPT_CONTEXT_END}

{PROMPT_INSTRUCTIONS_BEGIN}

## Rules

1. Output ONLY Python assignment statements
2. Each statement must assign a result to a variable
3. You can ONLY call the primitive methods listed above
4. Do NOT use imports, loops, conditionals, or function definitions
5. Do NOT use any dangerous operations (exec, eval, open, etc.)
6. Call primitives directly (e.g. `add(a=1, b=2)`), do NOT use `self.`
7. The last assigned variable will be the final result

## Output

```python

{PROMPT_INSTRUCTIONS_END}
"""
        return prompt

compute_signature_hash()

Compute a hash of all primitive signatures and decomposition examples.

A changed hash means the agent's interface has changed and any downstream artifacts (e.g. fine-tuned adapters) need regeneration.

Returns:

Type	Description
str	A 16-character hex digest.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def compute_signature_hash(self) -> str:
    """Compute a hash of all primitive signatures and decomposition examples.

    A changed hash means the agent's interface has changed and any
    downstream artifacts (e.g. fine-tuned adapters) need regeneration.

    Returns:
        A 16-character hex digest.
    """
    primitives = self._get_primitive_methods()
    decompositions = self._get_decomposition_methods()

    parts: list[str] = []
    for name, method in sorted(primitives, key=lambda x: x[0]):
        sig = inspect.signature(method)
        doc = inspect.getdoc(method) or ""
        read_only = getattr(method, "__primitive_read_only__", False)
        deterministic = getattr(method, "__primitive_deterministic__", True)
        parts.append(
            f"PRIM:{name}:{sig}:{doc}:{read_only}:{deterministic}"
        )

    for name, method, intent, expanded in sorted(
        decompositions, key=lambda x: x[0]
    ):
        source = self._get_decomposition_source(method)
        parts.append(f"DECOMP:{name}:{intent}:{expanded}:{source}")

    combined = "\n".join(parts)
    return hashlib.sha256(combined.encode()).hexdigest()[:16]

execute(plan)

Execute a plan step-by-step with full tracing.

Parameters:

Name	Type	Description	Default
plan	str	The Python statements to execute.	required

Returns:

Type	Description
ExecutionResult	ExecutionResult containing the final value and execution trace.

Raises:

Type	Description
ValueError	If the plan contains disallowed operations.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def execute(self, plan: str) -> ExecutionResult:
    """Execute a plan step-by-step with full tracing.

    Args:
        plan: The Python statements to execute.

    Returns:
        ExecutionResult containing the final value and execution trace.

    Raises:
        ValueError: If the plan contains disallowed operations.
    """
    self.validate_plan(plan)

    exec_span: str | None = None
    if self._tracer:
        exec_span = self._tracer.start_span(
            EventType.EXECUTION_START,
            {"plan": plan} if self._tracer.config.capture_plan_source else {},
        )

    tree = ast.parse(plan)
    steps: list[ExecutionStep] = []
    total_start = time.perf_counter()

    namespace = self._build_namespace()
    read_only_map = self._get_primitive_read_only_map()
    reserved_names = self._build_reserved_names()

    # Execute statement by statement
    for i, stmt in enumerate(tree.body, 1):
        step = self._execute_statement(
            stmt, namespace, i, read_only_map, reserved_names
        )
        steps.append(step)

        if self._tracer and self._tracer.config.capture_execution_steps:
            self._tracer.emit(
                EventType.EXECUTION_STEP,
                self._tracer.filter.execution_step(step.model_dump()),
            )

        # Stop on first error
        if not step.success:
            break

    self._persist_user_variables(namespace, reserved_names)

    total_elapsed = time.perf_counter() - total_start

    trace = ExecutionTrace(
        steps=steps,
        total_time_seconds=total_elapsed,
    )

    # Determine final result
    if steps and steps[-1].success:
        last = steps[-1]
        result = ExecutionResult(
            value_type=last.result_type,
            value_name=last.variable_name,
            value_json=last.result_json,
            trace=trace,
        )
    else:
        result = ExecutionResult(
            value_type=NONE_TYPE_NAME,
            value_name="",
            value_json=NULL_JSON,
            trace=trace,
        )

    if self._tracer and exec_span:
        self._tracer.end_span(
            exec_span,
            EventType.EXECUTION_COMPLETE,
            ExecutionSummary(
                value_type=result.value_type,
                value_name=result.value_name,
                step_count=trace.step_count,
                all_succeeded=trace.all_succeeded,
                total_time_seconds=total_elapsed,
            ).model_dump(),
        )

    return result

execute_stepwise(task, plan_context=None, serializer=None, checkpoint_id=None)

Execute a plan step-by-step, yielding checkpoints for persistence.

This method enables distributed execution by yielding a checkpoint after each step. The checkpoint can be persisted to a database and resumed on any worker.

When require_mutation_approval is True (default), execution pauses before mutations and yields a checkpoint with status awaiting_approval. Call resume_from_checkpoint() with approve_mutation=True to continue.

Parameters:

Name	Type	Description	Default
task	str	The task description to accomplish.	required
plan_context	PlanContext | None	Optional pre-computed plan context (from a previous planning phase). If None, will generate a new plan.	None
serializer	SerializerRegistry | None	Custom serializer registry. Defaults to the global registry.	None
checkpoint_id	str | None	Optional ID for the checkpoint. Auto-generated if None.	None

Yields:

Type	Description
ExecutionCheckpoint	ExecutionCheckpoint after each step, allowing for persistence and
ExecutionCheckpoint	distributed resume.

Example

Simple usage - collect all checkpoints

for checkpoint in agent.execute_stepwise("do something"): store.save(checkpoint) if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL: # Handle approval externally break

Resume after approval

checkpoint = store.load(checkpoint_id) for checkpoint in agent.resume_from_checkpoint(checkpoint, approve_mutation=True): store.save(checkpoint)

Source code in src/opensymbolicai/blueprints/plan_execute.py

def execute_stepwise(
    self,
    task: str,
    plan_context: PlanContext | None = None,
    serializer: SerializerRegistry | None = None,
    checkpoint_id: str | None = None,
) -> Iterator[ExecutionCheckpoint]:
    """Execute a plan step-by-step, yielding checkpoints for persistence.

    This method enables distributed execution by yielding a checkpoint after
    each step. The checkpoint can be persisted to a database and resumed on
    any worker.

    When `require_mutation_approval` is True (default), execution pauses
    before mutations and yields a checkpoint with status `awaiting_approval`.
    Call `resume_from_checkpoint()` with `approve_mutation=True` to continue.

    Args:
        task: The task description to accomplish.
        plan_context: Optional pre-computed plan context (from a previous
            planning phase). If None, will generate a new plan.
        serializer: Custom serializer registry. Defaults to the global registry.
        checkpoint_id: Optional ID for the checkpoint. Auto-generated if None.

    Yields:
        ExecutionCheckpoint after each step, allowing for persistence and
        distributed resume.

    Example:
        # Simple usage - collect all checkpoints
        for checkpoint in agent.execute_stepwise("do something"):
            store.save(checkpoint)
            if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
                # Handle approval externally
                break

        # Resume after approval
        checkpoint = store.load(checkpoint_id)
        for checkpoint in agent.resume_from_checkpoint(checkpoint, approve_mutation=True):
            store.save(checkpoint)
    """
    serializer = serializer or default_serializer_registry

    # Generate plan if not provided
    if plan_context is None:
        plan_result = self.plan(task)
        plan_context = PlanContext(
            plan_attempts=[
                PlanAttempt(
                    attempt_number=1,
                    plan_generation=self._plan_generation_from_result(plan_result),
                    success=True,
                )
            ],
            final_plan=plan_result.plan,
            generation_time_seconds=plan_result.time_seconds,
        )

    plan = plan_context.final_plan

    # Validate plan
    try:
        self.validate_plan(plan)
    except ValueError as e:
        checkpoint = ExecutionCheckpoint(
            task=task,
            plan=plan,
            plan_context=plan_context,
            current_step=0,
            total_steps=0,
            status=CheckpointStatus.FAILED,
            error=str(e),
            worker_id=self.config.worker_id,
        )
        if checkpoint_id is not None:
            checkpoint.checkpoint_id = checkpoint_id
        yield checkpoint
        return

    tree = ast.parse(plan)
    total_steps = len(tree.body)

    namespace = self._build_namespace()
    read_only_map = self._get_primitive_read_only_map()
    reserved_names = self._build_reserved_names()

    steps: list[ExecutionStep] = []

    # Create initial checkpoint
    checkpoint = ExecutionCheckpoint(
        task=task,
        plan=plan,
        plan_context=plan_context,
        current_step=0,
        total_steps=total_steps,
        status=CheckpointStatus.RUNNING,
        namespace_snapshot=serializer.serialize_namespace(
            namespace, reserved_names
        ),
        completed_steps=[],
        worker_id=self.config.worker_id,
    )
    if checkpoint_id is not None:
        checkpoint.checkpoint_id = checkpoint_id

    # Execute statement by statement
    for i, stmt in enumerate(tree.body):
        step_number = i + 1
        statement_str = ast.unparse(stmt)

        mutation = self._detect_mutation(stmt, read_only_map)

        # If mutation requires approval, pause and yield
        if mutation.is_mutation and self.config.require_mutation_approval:
            checkpoint.current_step = i
            checkpoint.status = CheckpointStatus.AWAITING_APPROVAL
            checkpoint.pending_mutation = PendingMutation(
                method_name=mutation.method_name or "",
                args=mutation.args,
                statement=statement_str,
                step_number=step_number,
                variable_name=mutation.variable_name,
            )
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )
            checkpoint.completed_steps = steps.copy()
            checkpoint.touch(self.config.worker_id)
            yield checkpoint
            return  # Caller must resume with approve_mutation=True

        # Execute the step
        step = self._execute_statement(
            stmt, namespace, step_number, read_only_map, reserved_names
        )
        steps.append(step)

        # Update checkpoint
        checkpoint.current_step = step_number
        checkpoint.completed_steps = steps.copy()
        checkpoint.namespace_snapshot = serializer.serialize_namespace(
            namespace, reserved_names
        )

        if not step.success:
            checkpoint.status = CheckpointStatus.FAILED
            checkpoint.error = step.error
            checkpoint.touch(self.config.worker_id)
            yield checkpoint
            return

        # Yield progress checkpoint
        checkpoint.status = CheckpointStatus.RUNNING
        checkpoint.touch(self.config.worker_id)
        yield checkpoint

    # Execution completed successfully
    checkpoint.status = CheckpointStatus.COMPLETED
    if steps:
        last_step = steps[-1]
        checkpoint.result_variable = last_step.variable_name
        result_value = namespace.get(last_step.variable_name)
        checkpoint.result_value = serializer.serialize(result_value)

    self._persist_user_variables(namespace, reserved_names)

    checkpoint.touch(self.config.worker_id)
    yield checkpoint

on_code_extracted(raw_response, extracted_code)

Hook called after code is extracted from the LLM response.

Override this method in subclasses to observe and/or modify the extracted code before it is used for planning.

Parameters:

Name	Type	Description	Default
raw_response	str	The original raw text response from the LLM.	required
extracted_code	str	The code extracted by _extract_code_block.	required

Returns:

Type	Description
str	The final code to use (can be modified from extracted_code).

Source code in src/opensymbolicai/blueprints/plan_execute.py

def on_code_extracted(self, raw_response: str, extracted_code: str) -> str:
    """Hook called after code is extracted from the LLM response.

    Override this method in subclasses to observe and/or modify the
    extracted code before it is used for planning.

    Args:
        raw_response: The original raw text response from the LLM.
        extracted_code: The code extracted by _extract_code_block.

    Returns:
        The final code to use (can be modified from extracted_code).
    """
    return extracted_code

plan(task, feedback=None)

Generate a plan (Python statements) for a task.

Parameters:

Name	Type	Description	Default
task	str	The task description to plan for.	required
feedback	str | None	Optional error feedback from a previous failed plan attempt.	None

Returns:

Type	Description
PlanResult	PlanResult containing the generated plan and metrics.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def plan(self, task: str, feedback: str | None = None) -> PlanResult:
    """Generate a plan (Python statements) for a task.

    Args:
        task: The task description to plan for.
        feedback: Optional error feedback from a previous failed plan attempt.

    Returns:
        PlanResult containing the generated plan and metrics.
    """
    plan_span: str | None = None
    if self._tracer:
        plan_span = self._tracer.start_span(
            EventType.PLAN_START, {"task": task, "feedback": feedback}
        )

    prompt = self.build_plan_prompt(task, feedback=feedback)

    llm_span: str | None = None
    if self._tracer:
        llm_payload: dict[str, Any] = {}
        if self._tracer.config.capture_llm_prompts:
            llm_payload["prompt"] = prompt
        llm_span = self._tracer.start_span(
            EventType.PLAN_LLM_REQUEST, llm_payload, defer=True
        )

    start_time = time.perf_counter()
    response = self._llm.generate(prompt)
    elapsed = time.perf_counter() - start_time

    # Extract code from response (handle markdown code blocks)
    raw_response = response.text
    extracted_code = self._extract_code_block(raw_response)
    plan_text = self.on_code_extracted(raw_response, extracted_code)

    # Build full LLM interaction details
    llm_interaction = LLMInteraction(
        prompt=prompt,
        response=raw_response,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        time_seconds=elapsed,
        provider=response.provider,
        model=response.model,
    )
    plan_generation = PlanGeneration(
        llm_interaction=llm_interaction,
        extracted_code=extracted_code,
    )

    if self._tracer and llm_span:
        response_payload = self._tracer.filter.llm_interaction(
            llm_interaction.model_dump()
        ) if self._tracer.config.capture_llm_responses else {}
        self._tracer.end_span(
            llm_span, EventType.PLAN_LLM_RESPONSE, response_payload
        )

    plan_result = PlanResult(
        plan=plan_text,
        usage=TokenUsage(
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
        ),
        time_seconds=elapsed,
        provider=response.provider,
        model=response.model,
        plan_generation=plan_generation,
    )

    if self._tracer and plan_span:
        self._tracer.end_span(
            plan_span,
            EventType.PLAN_COMPLETE,
            self._tracer.filter.plan_result(plan_result.model_dump()),
        )

    return plan_result

resume_from_checkpoint(checkpoint, approve_mutation=False, serializer=None)

Resume execution from a persisted checkpoint.

This method reconstructs execution state from a checkpoint and continues execution from where it left off.

Parameters:

Name	Type	Description	Default
checkpoint	ExecutionCheckpoint	The checkpoint to resume from.	required
approve_mutation	bool	If True and checkpoint is awaiting approval, execute the pending mutation and continue.	False
serializer	SerializerRegistry | None	Custom serializer registry. Must match the one used when creating the checkpoint.	None

Yields:

Type	Description
ExecutionCheckpoint	ExecutionCheckpoint after each step.

Raises:

Type	Description
ValueError	If checkpoint is in a terminal state or approval is required but not provided.

Example

Load and resume

checkpoint = store.load(checkpoint_id) if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL: # Get user approval somehow if user_approved: for cp in agent.resume_from_checkpoint(checkpoint, approve_mutation=True): store.save(cp)

Source code in src/opensymbolicai/blueprints/plan_execute.py

def resume_from_checkpoint(
    self,
    checkpoint: ExecutionCheckpoint,
    approve_mutation: bool = False,
    serializer: SerializerRegistry | None = None,
) -> Iterator[ExecutionCheckpoint]:
    """Resume execution from a persisted checkpoint.

    This method reconstructs execution state from a checkpoint and continues
    execution from where it left off.

    Args:
        checkpoint: The checkpoint to resume from.
        approve_mutation: If True and checkpoint is awaiting approval,
            execute the pending mutation and continue.
        serializer: Custom serializer registry. Must match the one used
            when creating the checkpoint.

    Yields:
        ExecutionCheckpoint after each step.

    Raises:
        ValueError: If checkpoint is in a terminal state or approval is
            required but not provided.

    Example:
        # Load and resume
        checkpoint = store.load(checkpoint_id)
        if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
            # Get user approval somehow
            if user_approved:
                for cp in agent.resume_from_checkpoint(checkpoint, approve_mutation=True):
                    store.save(cp)
    """
    serializer = serializer or default_serializer_registry

    if checkpoint.is_terminal:
        raise ValueError(
            f"Cannot resume from terminal checkpoint (status: {checkpoint.status})"
        )

    if (
        checkpoint.status == CheckpointStatus.AWAITING_APPROVAL
        and not approve_mutation
    ):
        raise ValueError(
            "Checkpoint is awaiting mutation approval. "
            "Set approve_mutation=True to continue."
        )

    # Parse the plan
    tree = ast.parse(checkpoint.plan)
    total_steps = len(tree.body)

    namespace = self._build_namespace()

    # Restore serialized variables (skip undeserializable values)
    import contextlib

    for var_name, serialized_val in checkpoint.namespace_snapshot.items():
        with contextlib.suppress(ValueError):
            namespace[var_name] = serializer.deserialize(serialized_val)

    read_only_map = self._get_primitive_read_only_map()
    reserved_names = self._build_reserved_names()

    # Copy completed steps
    steps = list(checkpoint.completed_steps)

    # Determine starting point
    start_step = checkpoint.current_step

    # If resuming from awaiting_approval, execute the pending mutation first
    if (
        checkpoint.status == CheckpointStatus.AWAITING_APPROVAL
        and approve_mutation
        and checkpoint.pending_mutation
    ):
        stmt = tree.body[start_step]
        step_number = start_step + 1

        step = self._execute_statement(
            stmt, namespace, step_number, read_only_map, reserved_names
        )
        steps.append(step)

        checkpoint.completed_steps = steps.copy()
        checkpoint.pending_mutation = None

        if not step.success:
            checkpoint.status = CheckpointStatus.FAILED
            checkpoint.error = step.error
            checkpoint.current_step = step_number
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )
            checkpoint.touch(self.config.worker_id)
            yield checkpoint
            return

        checkpoint.current_step = step_number
        checkpoint.namespace_snapshot = serializer.serialize_namespace(
            namespace, reserved_names
        )
        checkpoint.status = CheckpointStatus.RUNNING
        checkpoint.touch(self.config.worker_id)
        yield checkpoint

        start_step += 1

    # Continue with remaining steps
    for i in range(start_step, total_steps):
        stmt = tree.body[i]
        step_number = i + 1
        statement_str = ast.unparse(stmt)

        mutation = self._detect_mutation(stmt, read_only_map)

        # Pause for mutation approval if required
        if mutation.is_mutation and self.config.require_mutation_approval:
            checkpoint.current_step = i
            checkpoint.status = CheckpointStatus.AWAITING_APPROVAL
            checkpoint.pending_mutation = PendingMutation(
                method_name=mutation.method_name or "",
                args=mutation.args,
                statement=statement_str,
                step_number=step_number,
                variable_name=mutation.variable_name,
            )
            checkpoint.namespace_snapshot = serializer.serialize_namespace(
                namespace, reserved_names
            )
            checkpoint.completed_steps = steps.copy()
            checkpoint.touch(self.config.worker_id)
            yield checkpoint
            return

        # Execute step
        step = self._execute_statement(
            stmt, namespace, step_number, read_only_map, reserved_names
        )
        steps.append(step)

        checkpoint.current_step = step_number
        checkpoint.completed_steps = steps.copy()
        checkpoint.namespace_snapshot = serializer.serialize_namespace(
            namespace, reserved_names
        )

        if not step.success:
            checkpoint.status = CheckpointStatus.FAILED
            checkpoint.error = step.error
            checkpoint.touch(self.config.worker_id)
            yield checkpoint
            return

        checkpoint.status = CheckpointStatus.RUNNING
        checkpoint.touch(self.config.worker_id)
        yield checkpoint

    # Completed
    checkpoint.status = CheckpointStatus.COMPLETED
    if steps:
        last_step = steps[-1]
        checkpoint.result_variable = last_step.variable_name
        result_value = namespace.get(last_step.variable_name)
        checkpoint.result_value = serializer.serialize(result_value)

    self._persist_user_variables(namespace, reserved_names)

    checkpoint.touch(self.config.worker_id)
    yield checkpoint

run(task)

Run the complete plan-and-execute cycle.

Parameters:

Name	Type	Description	Default
task	str	The task description to accomplish.	required

Returns:

Type	Description
OrchestrationResult	OrchestrationResult containing the outcome and metrics.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def run(self, task: str) -> OrchestrationResult:
    """Run the complete plan-and-execute cycle.

    Args:
        task: The task description to accomplish.

    Returns:
        OrchestrationResult containing the outcome and metrics.
    """
    run_span: str | None = None
    if self._tracer:
        self._tracer.new_trace()
        run_span = self._tracer.start_span(
            EventType.RUN_START,
            {"task": task, "config": self.config.model_dump(exclude={"observability"})},
        )

    result: OrchestrationResult | None = None
    try:
        result = self._run_inner(task, run_span)
        return result
    finally:
        if self._tracer:
            self._tracer.flush()
        _emit_telemetry(self, success=result is not None and result.success)

run_with_checkpoints(task, store, serializer=None, auto_approve=False)

Run execution with automatic checkpoint persistence.

Convenience method that handles checkpoint saving automatically.

Parameters:

Name	Type	Description	Default
task	str	The task to execute.	required
store	CheckpointStore	Checkpoint store for persistence.	required
serializer	SerializerRegistry | None	Custom serializer registry.	None
auto_approve	bool	If True, automatically approve all mutations.	False

Returns:

Type	Description
ExecutionCheckpoint	The final checkpoint (completed or failed).

Source code in src/opensymbolicai/blueprints/plan_execute.py

def run_with_checkpoints(
    self,
    task: str,
    store: CheckpointStore,
    serializer: SerializerRegistry | None = None,
    auto_approve: bool = False,
) -> ExecutionCheckpoint:
    """Run execution with automatic checkpoint persistence.

    Convenience method that handles checkpoint saving automatically.

    Args:
        task: The task to execute.
        store: Checkpoint store for persistence.
        serializer: Custom serializer registry.
        auto_approve: If True, automatically approve all mutations.

    Returns:
        The final checkpoint (completed or failed).
    """
    checkpoint: ExecutionCheckpoint | None = None

    for checkpoint in self.execute_stepwise(task, serializer=serializer):
        store.save(checkpoint)

        if checkpoint.status == CheckpointStatus.AWAITING_APPROVAL:
            if auto_approve:
                # Resume with approval - consume all checkpoints from resume
                current_cp = checkpoint
                while current_cp.status == CheckpointStatus.AWAITING_APPROVAL:
                    for next_cp in self.resume_from_checkpoint(
                        current_cp, approve_mutation=True, serializer=serializer
                    ):
                        store.save(next_cp)
                        current_cp = next_cp
                checkpoint = current_cp
            else:
                # Return checkpoint for external approval
                return checkpoint

    if checkpoint is None:
        raise RuntimeError("No checkpoint was generated")

    return checkpoint

validate_plan(plan)

Validate that a plan only uses allowed operations.

Parameters:

Name	Type	Description	Default
plan	str	The Python statements to validate.	required

Raises:

Type	Description
ValueError	If the plan contains disallowed operations.

Source code in src/opensymbolicai/blueprints/plan_execute.py

def validate_plan(self, plan: str) -> None:
    """Validate that a plan only uses allowed operations.

    Args:
        plan: The Python statements to validate.

    Raises:
        ValueError: If the plan contains disallowed operations.
    """
    try:
        tree = ast.parse(plan)
    except SyntaxError as e:
        raise ValueError(f"Invalid Python syntax: {e}") from e

    primitive_names = self._get_primitive_names()

    disallowed_statements: tuple[type, ...] = (
        ast.If,
        ast.For,
        ast.While,
        ast.Try,
        ast.With,
        ast.FunctionDef,
        ast.AsyncFunctionDef,
        ast.ClassDef,
        ast.Import,
        ast.ImportFrom,
        ast.Global,
        ast.Nonlocal,
        ast.Raise,
        ast.Assert,
        ast.Delete,
    )
    if hasattr(ast, "Match"):
        disallowed_statements = (*disallowed_statements, ast.Match)

    for node in ast.walk(tree):
        if isinstance(node, disallowed_statements):
            node_type = type(node).__name__
            raise ValueError(f"{node_type} statements are not allowed in plans")

    # Every top-level statement must be an assignment
    for stmt in tree.body:
        if not isinstance(stmt, (ast.Assign, ast.AnnAssign)):
            stmt_type = type(stmt).__name__
            raise ValueError(
                f"Every statement must be an assignment. Found: {stmt_type}"
            )

    self._validate_ast_nodes(tree, primitive_names)

PlanExecuteConfig

Bases: BaseModel

Extended configuration for PlanExecute agents.

Source code in src/opensymbolicai/models.py

class PlanExecuteConfig(BaseModel):
    """Extended configuration for PlanExecute agents."""

    allowed_builtins: dict[str, Any] | None = Field(
        default=None,
        description="Dict of builtins allowed in execution. None uses defaults.",
    )
    skip_result_serialization: bool = Field(
        default=False,
        description="Skip JSON serialization of results. Useful for large or non-serializable outputs.",
    )
    multi_turn: bool = Field(
        default=False,
        description="Enable multi-turn mode. Persists variables and tracks conversation history.",
    )
    on_mutation: MutationHook | None = Field(
        default=None,
        description="Hook called when a primitive with read_only=False is executed. "
        "Return None to allow, or a string rejection reason to block the mutation.",
    )
    max_plan_retries: int = Field(
        default=0,
        description="Maximum number of times to retry plan generation if validation fails. "
        "The LLM will receive the error message as feedback for regeneration.",
    )

    # Checkpoint/distributed execution settings
    require_mutation_approval: bool = Field(
        default=True,
        description="If True, execution pauses before mutations and yields a checkpoint "
        "for external approval. Use with execute_stepwise() for distributed execution.",
    )
    worker_id: str | None = Field(
        default=None,
        description="Identifier for this worker instance. Used in checkpoints to track "
        "which worker created/updated the checkpoint.",
    )
    observability: ObservabilityConfig | None = Field(
        default=None,
        description="Observability configuration. When set and enabled, trace events "
        "are emitted for planning, execution, and LLM interactions.",
    )
    prompt_provider: PromptProvider | None = Field(
        default=None,
        description="Controls which primitives and decompositions are included in the "
        "prompt. Subclass PromptProvider and override select_primitives / "
        "select_decompositions to filter what the LLM sees.",
    )

    model_config = {"arbitrary_types_allowed": True}

PlanResult

Bases: BaseModel

Result from the planning phase.

Source code in src/opensymbolicai/models.py

class PlanResult(BaseModel):
    """Result from the planning phase."""

    plan: str = Field(..., description="Generated Python statements")
    usage: TokenUsage = Field(default_factory=TokenUsage, description="Token usage")
    time_seconds: float = Field(default=0.0, description="Time taken to generate plan")
    provider: str = Field(default="", description="LLM provider used")
    model: str = Field(default="", description="Model used")
    plan_generation: PlanGeneration | None = Field(
        default=None, description="Full LLM interaction details for plan generation"
    )

Planner

Bases: ABC

Abstract base class for agents that plan and execute tasks.

The Planner defines the interface for agents that: 1. Generate plans from natural language tasks 2. Execute those plans step-by-step 3. Return structured results with execution traces

Subclasses must implement: - plan(): Generate a plan for a given task - execute(): Execute a plan and return results - run(): Complete plan-and-execute cycle

Source code in src/opensymbolicai/blueprints/planner.py

class Planner(ABC):
    """Abstract base class for agents that plan and execute tasks.

    The Planner defines the interface for agents that:
    1. Generate plans from natural language tasks
    2. Execute those plans step-by-step
    3. Return structured results with execution traces

    Subclasses must implement:
    - plan(): Generate a plan for a given task
    - execute(): Execute a plan and return results
    - run(): Complete plan-and-execute cycle
    """

    @abstractmethod
    def plan(self, task: str) -> PlanResult:
        """Generate a plan (Python statements) for a task.

        Args:
            task: The task description to plan for.

        Returns:
            PlanResult containing the generated plan and metrics.
        """
        ...

    @abstractmethod
    def execute(self, plan: str) -> ExecutionResult:
        """Execute a plan step-by-step.

        Args:
            plan: The Python statements to execute.

        Returns:
            ExecutionResult containing the final value and execution trace.
        """
        ...

    @abstractmethod
    def run(self, task: str) -> OrchestrationResult:
        """Run the complete plan-and-execute cycle.

        Args:
            task: The task description to accomplish.

        Returns:
            OrchestrationResult containing the outcome and metrics.
        """
        ...

execute(plan) abstractmethod

Execute a plan step-by-step.

Parameters:

Name	Type	Description	Default
plan	str	The Python statements to execute.	required

Returns:

Type	Description
ExecutionResult	ExecutionResult containing the final value and execution trace.

Source code in src/opensymbolicai/blueprints/planner.py

@abstractmethod
def execute(self, plan: str) -> ExecutionResult:
    """Execute a plan step-by-step.

    Args:
        plan: The Python statements to execute.

    Returns:
        ExecutionResult containing the final value and execution trace.
    """
    ...

plan(task) abstractmethod

Generate a plan (Python statements) for a task.

Parameters:

Name	Type	Description	Default
task	str	The task description to plan for.	required

Returns:

Type	Description
PlanResult	PlanResult containing the generated plan and metrics.

Source code in src/opensymbolicai/blueprints/planner.py

@abstractmethod
def plan(self, task: str) -> PlanResult:
    """Generate a plan (Python statements) for a task.

    Args:
        task: The task description to plan for.

    Returns:
        PlanResult containing the generated plan and metrics.
    """
    ...

run(task) abstractmethod

Run the complete plan-and-execute cycle.

Parameters:

Name	Type	Description	Default
task	str	The task description to accomplish.	required

Returns:

Type	Description
OrchestrationResult	OrchestrationResult containing the outcome and metrics.

Source code in src/opensymbolicai/blueprints/planner.py

@abstractmethod
def run(self, task: str) -> OrchestrationResult:
    """Run the complete plan-and-execute cycle.

    Args:
        task: The task description to accomplish.

    Returns:
        OrchestrationResult containing the outcome and metrics.
    """
    ...

PreconditionError

Bases: ExecutionError

Exception raised when a precondition for an operation is not met.

Use this when the operation cannot proceed due to missing prerequisites or invalid state (e.g., division by zero, empty collection).

Example

if divisor == 0: raise PreconditionError( "Cannot divide by zero", code="DIVISION_BY_ZERO", details={"dividend": dividend, "divisor": divisor} )

Source code in src/opensymbolicai/exceptions.py

class PreconditionError(ExecutionError):
    """Exception raised when a precondition for an operation is not met.

    Use this when the operation cannot proceed due to missing prerequisites
    or invalid state (e.g., division by zero, empty collection).

    Example:
        if divisor == 0:
            raise PreconditionError(
                "Cannot divide by zero",
                code="DIVISION_BY_ZERO",
                details={"dividend": dividend, "divisor": divisor}
            )
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = "PRECONDITION_FAILED",
        details: dict[str, Any] | None = None,
    ) -> None:
        """Initialize precondition error.

        Args:
            message: Human-readable error message.
            code: Error code, defaults to "PRECONDITION_FAILED".
            details: Additional context about the error.
        """
        super().__init__(message, code=code, details=details)

__init__(message, *, code='PRECONDITION_FAILED', details=None)

Initialize precondition error.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Error code, defaults to "PRECONDITION_FAILED".	'PRECONDITION_FAILED'
details	dict[str, Any] | None	Additional context about the error.	None

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = "PRECONDITION_FAILED",
    details: dict[str, Any] | None = None,
) -> None:
    """Initialize precondition error.

    Args:
        message: Human-readable error message.
        code: Error code, defaults to "PRECONDITION_FAILED".
        details: Additional context about the error.
    """
    super().__init__(message, code=code, details=details)

PrimitiveCall

Bases: BaseModel

Information about a primitive method call in a plan.

Source code in src/opensymbolicai/models.py

class PrimitiveCall(BaseModel):
    """Information about a primitive method call in a plan."""

    method_name: str = Field(..., description="Name of the primitive method")
    read_only: bool = Field(
        default=False, description="Whether the method is read-only"
    )
    args: dict[str, Any] = Field(default_factory=dict, description="Arguments passed")

PrimitiveInfo

Bases: BaseModel

Metadata describing a primitive method, passed to PromptProvider selectors.

Source code in src/opensymbolicai/models.py

class PrimitiveInfo(BaseModel):
    """Metadata describing a primitive method, passed to PromptProvider selectors."""

    name: str = Field(..., description="Method name.")
    docstring: str = Field(default="", description="The primitive's docstring.")
    read_only: bool = Field(default=False, description="Whether the primitive is read-only.")
    deterministic: bool = Field(default=True, description="Whether the primitive is deterministic.")
    parameters: list[ParameterInfo] = Field(
        default_factory=list, description="Positional parameters (excluding self)."
    )
    return_type: str = Field(default="Any", description="Return type annotation as a string.")

PromptProvider

Bases: BaseModel

Controls which primitives and decompositions are included in the prompt.

Subclass and override :meth:select_primitives and/or :meth:select_decompositions to filter what the LLM sees. The inheriting class only needs to choose method names — prompt construction is handled by the framework.

Each selector receives rich metadata so you can filter by any attribute (name, docstring, read_only, parameter types, etc.).

Example::

class ReadOnlyOnly(PromptProvider):
    def select_primitives(self, available: list[PrimitiveInfo]) -> list[str]:
        return [p.name for p in available if p.read_only]

Source code in src/opensymbolicai/models.py

class PromptProvider(BaseModel):
    """Controls which primitives and decompositions are included in the prompt.

    Subclass and override :meth:`select_primitives` and/or
    :meth:`select_decompositions` to filter what the LLM sees.  The
    inheriting class only needs to choose method names — prompt
    construction is handled by the framework.

    Each selector receives rich metadata so you can filter by any
    attribute (name, docstring, read_only, parameter types, etc.).

    Example::

        class ReadOnlyOnly(PromptProvider):
            def select_primitives(self, available: list[PrimitiveInfo]) -> list[str]:
                return [p.name for p in available if p.read_only]
    """

    model_config = {"frozen": False}

    def select_primitives(self, available: list[PrimitiveInfo]) -> list[str]:
        """Return the primitive names to include in the prompt.

        Args:
            available: Metadata for all primitive methods on the agent.

        Returns:
            The subset of names to expose to the LLM. Default: all.
        """
        return [p.name for p in available]

    def select_decompositions(self, available: list[DecompositionInfo]) -> list[str]:
        """Return the decomposition names to include in the prompt.

        Args:
            available: Metadata for all decomposition methods on the agent.

        Returns:
            The subset of names to expose to the LLM. Default: all.
        """
        return [d.name for d in available]

select_decompositions(available)

Return the decomposition names to include in the prompt.

Parameters:

Name	Type	Description	Default
available	list[DecompositionInfo]	Metadata for all decomposition methods on the agent.	required

Returns:

Type	Description
list[str]	The subset of names to expose to the LLM. Default: all.

Source code in src/opensymbolicai/models.py

def select_decompositions(self, available: list[DecompositionInfo]) -> list[str]:
    """Return the decomposition names to include in the prompt.

    Args:
        available: Metadata for all decomposition methods on the agent.

    Returns:
        The subset of names to expose to the LLM. Default: all.
    """
    return [d.name for d in available]

select_primitives(available)

Return the primitive names to include in the prompt.

Parameters:

Name	Type	Description	Default
available	list[PrimitiveInfo]	Metadata for all primitive methods on the agent.	required

Returns:

Type	Description
list[str]	The subset of names to expose to the LLM. Default: all.

Source code in src/opensymbolicai/models.py

def select_primitives(self, available: list[PrimitiveInfo]) -> list[str]:
    """Return the primitive names to include in the prompt.

    Args:
        available: Metadata for all primitive methods on the agent.

    Returns:
        The subset of names to expose to the LLM. Default: all.
    """
    return [p.name for p in available]

PromptSections

Bases: BaseModel

The three demarcated sections of an OpenSymbolicAI prompt.

Source code in src/opensymbolicai/prompt_utils.py

class PromptSections(BaseModel):
    """The three demarcated sections of an OpenSymbolicAI prompt."""

    preamble: str
    """Text before the DEFINITIONS marker (agent identity/description)."""

    definitions: str
    """Static per-agent content: primitive signatures, type defs, examples."""

    context: str
    """Dynamic per-call content: task, history, feedback."""

    instructions: str
    """Static per-blueprint content: rules and output format."""

context instance-attribute

Dynamic per-call content: task, history, feedback.

definitions instance-attribute

Static per-agent content: primitive signatures, type defs, examples.

instructions instance-attribute

Static per-blueprint content: rules and output format.

preamble instance-attribute

Text before the DEFINITIONS marker (agent identity/description).

ResourceError

Bases: ExecutionError

Exception raised when a required resource is unavailable.

Use this for missing files, network failures, or unavailable external services.

Example

raise ResourceError( "Database connection failed", code="DB_UNAVAILABLE", details={"host": "localhost", "port": 5432} )

Source code in src/opensymbolicai/exceptions.py

class ResourceError(ExecutionError):
    """Exception raised when a required resource is unavailable.

    Use this for missing files, network failures, or unavailable
    external services.

    Example:
        raise ResourceError(
            "Database connection failed",
            code="DB_UNAVAILABLE",
            details={"host": "localhost", "port": 5432}
        )
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = "RESOURCE_ERROR",
        details: dict[str, Any] | None = None,
        resource: str | None = None,
    ) -> None:
        """Initialize resource error.

        Args:
            message: Human-readable error message.
            code: Error code, defaults to "RESOURCE_ERROR".
            details: Additional context about the error.
            resource: Optional identifier for the unavailable resource.
        """
        if resource:
            details = details or {}
            details["resource"] = resource
        super().__init__(message, code=code, details=details)
        self.resource = resource

__init__(message, *, code='RESOURCE_ERROR', details=None, resource=None)

Initialize resource error.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Error code, defaults to "RESOURCE_ERROR".	'RESOURCE_ERROR'
details	dict[str, Any] | None	Additional context about the error.	None
resource	str | None	Optional identifier for the unavailable resource.	None

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = "RESOURCE_ERROR",
    details: dict[str, Any] | None = None,
    resource: str | None = None,
) -> None:
    """Initialize resource error.

    Args:
        message: Human-readable error message.
        code: Error code, defaults to "RESOURCE_ERROR".
        details: Additional context about the error.
        resource: Optional identifier for the unavailable resource.
    """
    if resource:
        details = details or {}
        details["resource"] = resource
    super().__init__(message, code=code, details=details)
    self.resource = resource

RetryableError

Bases: ExecutionError

Exception raised for errors that may succeed on retry.

Use this for transient failures like network timeouts or rate limits. By default, this does NOT halt execution to allow potential recovery.

Example

raise RetryableError( "API rate limit exceeded", code="RATE_LIMIT", details={"retry_after_seconds": 60} )

Source code in src/opensymbolicai/exceptions.py

class RetryableError(ExecutionError):
    """Exception raised for errors that may succeed on retry.

    Use this for transient failures like network timeouts or rate limits.
    By default, this does NOT halt execution to allow potential recovery.

    Example:
        raise RetryableError(
            "API rate limit exceeded",
            code="RATE_LIMIT",
            details={"retry_after_seconds": 60}
        )
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = "RETRYABLE_ERROR",
        details: dict[str, Any] | None = None,
        retry_after: float | None = None,
    ) -> None:
        """Initialize retryable error.

        Args:
            message: Human-readable error message.
            code: Error code, defaults to "RETRYABLE_ERROR".
            details: Additional context about the error.
            retry_after: Optional seconds to wait before retrying.
        """
        if retry_after is not None:
            details = details or {}
            details["retry_after_seconds"] = retry_after
        super().__init__(message, code=code, details=details, halt_execution=False)
        self.retry_after = retry_after

__init__(message, *, code='RETRYABLE_ERROR', details=None, retry_after=None)

Initialize retryable error.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Error code, defaults to "RETRYABLE_ERROR".	'RETRYABLE_ERROR'
details	dict[str, Any] | None	Additional context about the error.	None
retry_after	float | None	Optional seconds to wait before retrying.	None

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = "RETRYABLE_ERROR",
    details: dict[str, Any] | None = None,
    retry_after: float | None = None,
) -> None:
    """Initialize retryable error.

    Args:
        message: Human-readable error message.
        code: Error code, defaults to "RETRYABLE_ERROR".
        details: Additional context about the error.
        retry_after: Optional seconds to wait before retrying.
    """
    if retry_after is not None:
        details = details or {}
        details["retry_after_seconds"] = retry_after
    super().__init__(message, code=code, details=details, halt_execution=False)
    self.retry_after = retry_after

SerializedValue

Bases: BaseModel

A serialized value with type information for deserialization.

Source code in src/opensymbolicai/checkpoint.py

class SerializedValue(BaseModel):
    """A serialized value with type information for deserialization."""

    type_name: str = Field(..., description="Fully qualified type name")
    data: Any = Field(..., description="Serialized data (JSON-compatible)")
    is_primitive: bool = Field(
        default=True,
        description="Whether this is a JSON primitive (no custom deserializer needed)",
    )

SerializerRegistry

Registry for custom type serializers.

Allows registering serializers/deserializers for types that aren't natively JSON-serializable.

Example

registry = SerializerRegistry()

Register a custom class

registry.register( MyClass, serializer=lambda obj: {"field": obj.field}, deserializer=lambda data: MyClass(field=data["field"]) )

Or use decorators

@registry.register_serializer(MyClass) def serialize_myclass(obj: MyClass) -> dict: return {"field": obj.field}

@registry.register_deserializer("mymodule.MyClass") def deserialize_myclass(data: dict) -> MyClass: return MyClass(field=data["field"])

Source code in src/opensymbolicai/checkpoint.py

class SerializerRegistry:
    """Registry for custom type serializers.

    Allows registering serializers/deserializers for types that aren't
    natively JSON-serializable.

    Example:
        registry = SerializerRegistry()

        # Register a custom class
        registry.register(
            MyClass,
            serializer=lambda obj: {"field": obj.field},
            deserializer=lambda data: MyClass(field=data["field"])
        )

        # Or use decorators
        @registry.register_serializer(MyClass)
        def serialize_myclass(obj: MyClass) -> dict:
            return {"field": obj.field}

        @registry.register_deserializer("mymodule.MyClass")
        def deserialize_myclass(data: dict) -> MyClass:
            return MyClass(field=data["field"])
    """

    def __init__(self) -> None:
        self._serializers: dict[type, Serializer] = {}
        self._deserializers: dict[str, Deserializer] = {}
        self._register_defaults()

    def _register_defaults(self) -> None:
        """Register serializers for common types."""
        import base64

        # datetime
        self._serializers[datetime] = lambda dt: dt.isoformat()
        self._deserializers["datetime.datetime"] = lambda s: datetime.fromisoformat(s)

        # bytes
        self._serializers[bytes] = lambda b: base64.b64encode(b).decode("ascii")
        self._deserializers["builtins.bytes"] = lambda s: base64.b64decode(
            s.encode("ascii")
        )

        # set and frozenset
        self._serializers[set] = lambda s: list(s)
        self._deserializers["builtins.set"] = lambda lst: set(lst)
        self._serializers[frozenset] = lambda s: list(s)
        self._deserializers["builtins.frozenset"] = lambda lst: frozenset(lst)

        # tuple (preserve as tuple, not list)
        self._serializers[tuple] = lambda t: list(t)
        self._deserializers["builtins.tuple"] = lambda lst: tuple(lst)

    def register(
        self, type_: type, serializer: Serializer, deserializer: Deserializer
    ) -> None:
        """Register a serializer/deserializer pair for a type.

        Args:
            type_: The Python type to register.
            serializer: Function that converts instance to JSON-compatible data.
            deserializer: Function that converts JSON data back to instance.
        """
        type_name = f"{type_.__module__}.{type_.__qualname__}"
        self._serializers[type_] = serializer
        self._deserializers[type_name] = deserializer

    def register_serializer(self, type_: type) -> Callable[[Serializer], Serializer]:
        """Decorator to register a serializer for a type.

        Example:
            @registry.register_serializer(MyClass)
            def serialize(obj: MyClass) -> dict:
                return {"field": obj.field}
        """

        def decorator(fn: Serializer) -> Serializer:
            self._serializers[type_] = fn
            return fn

        return decorator

    def register_deserializer(
        self, type_name: str
    ) -> Callable[[Deserializer], Deserializer]:
        """Decorator to register a deserializer for a type name.

        Example:
            @registry.register_deserializer("mymodule.MyClass")
            def deserialize(data: dict) -> MyClass:
                return MyClass(field=data["field"])
        """

        def decorator(fn: Deserializer) -> Deserializer:
            self._deserializers[type_name] = fn
            return fn

        return decorator

    def _get_type_name(self, value_type: type) -> str:
        """Get the fully qualified type name."""
        return f"{value_type.__module__}.{value_type.__qualname__}"

    def serialize(self, value: Any) -> SerializedValue:
        """Serialize a value to a SerializedValue.

        Args:
            value: The value to serialize.

        Returns:
            SerializedValue with type info and serialized data.
        """
        # Handle None
        if value is None:
            return SerializedValue(type_name="NoneType", data=None, is_primitive=True)

        value_type = type(value)
        type_name = self._get_type_name(value_type)

        # Check for JSON primitives
        if isinstance(value, bool):  # Must check bool before int (bool is subclass)
            return SerializedValue(
                type_name="builtins.bool", data=value, is_primitive=True
            )
        if isinstance(value, int):
            return SerializedValue(
                type_name="builtins.int", data=value, is_primitive=True
            )
        if isinstance(value, float):
            return SerializedValue(
                type_name="builtins.float", data=value, is_primitive=True
            )
        if isinstance(value, str):
            return SerializedValue(
                type_name="builtins.str", data=value, is_primitive=True
            )

        # Check for registered serializer (before generic handling)
        if value_type in self._serializers:
            serializer = self._serializers[value_type]
            return SerializedValue(
                type_name=type_name, data=serializer(value), is_primitive=False
            )

        # Check for lists (recursively serialize)
        if isinstance(value, list):
            serialized_items = [self.serialize(item).model_dump() for item in value]
            return SerializedValue(
                type_name="builtins.list",
                data=serialized_items,
                is_primitive=False,
            )

        # Check for dicts (recursively serialize)
        if isinstance(value, dict):
            serialized_dict = {
                k: self.serialize(v).model_dump() for k, v in value.items()
            }
            return SerializedValue(
                type_name="builtins.dict",
                data=serialized_dict,
                is_primitive=False,
            )

        # Check if it's a Pydantic model
        if hasattr(value, "model_dump"):
            return SerializedValue(
                type_name=type_name,
                data={"__pydantic__": True, "data": value.model_dump()},
                is_primitive=False,
            )

        # Fallback: try to convert to string representation
        return SerializedValue(
            type_name=type_name,
            data={"__repr__": repr(value), "__unserializable__": True},
            is_primitive=False,
        )

    def deserialize(self, serialized: SerializedValue) -> Any:
        """Deserialize a SerializedValue back to its original value.

        Args:
            serialized: The SerializedValue to deserialize.

        Returns:
            The deserialized value.

        Raises:
            ValueError: If no deserializer is registered for the type.
        """
        # Handle None
        if serialized.type_name == "NoneType":
            return None

        # Handle primitives
        if serialized.is_primitive:
            return serialized.data

        # Handle lists
        if serialized.type_name == "builtins.list":
            return [
                self.deserialize(SerializedValue(**item)) for item in serialized.data
            ]

        # Handle dicts
        if serialized.type_name == "builtins.dict":
            return {
                k: self.deserialize(SerializedValue(**v))
                for k, v in serialized.data.items()
            }

        # Check for registered deserializer
        if serialized.type_name in self._deserializers:
            deserializer = self._deserializers[serialized.type_name]
            return deserializer(serialized.data)

        # Check for Pydantic model marker
        if isinstance(serialized.data, dict) and serialized.data.get("__pydantic__"):
            raise ValueError(
                f"Cannot deserialize Pydantic model '{serialized.type_name}': "
                f"register a deserializer that can reconstruct the model from: "
                f"{serialized.data['data']}"
            )

        # Check for __repr__ fallback (non-deserializable)
        if isinstance(serialized.data, dict) and serialized.data.get(
            "__unserializable__"
        ):
            raise ValueError(
                f"Cannot deserialize type '{serialized.type_name}': "
                f"no deserializer registered. Original repr: {serialized.data.get('__repr__', 'unknown')}"
            )

        raise ValueError(
            f"No deserializer registered for type '{serialized.type_name}'"
        )

    def can_deserialize(self, type_name: str) -> bool:
        """Check if a type can be deserialized."""
        if type_name in ("NoneType", "builtins.list", "builtins.dict"):
            return True
        # Primitives
        if type_name in (
            "builtins.bool",
            "builtins.int",
            "builtins.float",
            "builtins.str",
        ):
            return True
        return type_name in self._deserializers

    def serialize_namespace(
        self, namespace: dict[str, Any], exclude: set[str] | None = None
    ) -> dict[str, SerializedValue]:
        """Serialize an entire namespace dict.

        Args:
            namespace: The namespace to serialize.
            exclude: Keys to exclude from serialization.

        Returns:
            Dict mapping variable names to SerializedValues.
        """
        exclude = exclude or set()
        result: dict[str, SerializedValue] = {}
        for key, value in namespace.items():
            if key in exclude:
                continue
            result[key] = self.serialize(value)
        return result

    def deserialize_namespace(
        self, serialized: dict[str, SerializedValue]
    ) -> dict[str, Any]:
        """Deserialize a namespace back to regular values.

        Args:
            serialized: Dict of serialized values.

        Returns:
            Dict mapping variable names to deserialized values.
        """
        result: dict[str, Any] = {}
        for key, value in serialized.items():
            result[key] = self.deserialize(value)
        return result

can_deserialize(type_name)

Check if a type can be deserialized.

Source code in src/opensymbolicai/checkpoint.py

def can_deserialize(self, type_name: str) -> bool:
    """Check if a type can be deserialized."""
    if type_name in ("NoneType", "builtins.list", "builtins.dict"):
        return True
    # Primitives
    if type_name in (
        "builtins.bool",
        "builtins.int",
        "builtins.float",
        "builtins.str",
    ):
        return True
    return type_name in self._deserializers

deserialize(serialized)

Deserialize a SerializedValue back to its original value.

Parameters:

Name	Type	Description	Default
serialized	SerializedValue	The SerializedValue to deserialize.	required

Returns:

Type	Description
Any	The deserialized value.

Raises:

Type	Description
ValueError	If no deserializer is registered for the type.

Source code in src/opensymbolicai/checkpoint.py

def deserialize(self, serialized: SerializedValue) -> Any:
    """Deserialize a SerializedValue back to its original value.

    Args:
        serialized: The SerializedValue to deserialize.

    Returns:
        The deserialized value.

    Raises:
        ValueError: If no deserializer is registered for the type.
    """
    # Handle None
    if serialized.type_name == "NoneType":
        return None

    # Handle primitives
    if serialized.is_primitive:
        return serialized.data

    # Handle lists
    if serialized.type_name == "builtins.list":
        return [
            self.deserialize(SerializedValue(**item)) for item in serialized.data
        ]

    # Handle dicts
    if serialized.type_name == "builtins.dict":
        return {
            k: self.deserialize(SerializedValue(**v))
            for k, v in serialized.data.items()
        }

    # Check for registered deserializer
    if serialized.type_name in self._deserializers:
        deserializer = self._deserializers[serialized.type_name]
        return deserializer(serialized.data)

    # Check for Pydantic model marker
    if isinstance(serialized.data, dict) and serialized.data.get("__pydantic__"):
        raise ValueError(
            f"Cannot deserialize Pydantic model '{serialized.type_name}': "
            f"register a deserializer that can reconstruct the model from: "
            f"{serialized.data['data']}"
        )

    # Check for __repr__ fallback (non-deserializable)
    if isinstance(serialized.data, dict) and serialized.data.get(
        "__unserializable__"
    ):
        raise ValueError(
            f"Cannot deserialize type '{serialized.type_name}': "
            f"no deserializer registered. Original repr: {serialized.data.get('__repr__', 'unknown')}"
        )

    raise ValueError(
        f"No deserializer registered for type '{serialized.type_name}'"
    )

deserialize_namespace(serialized)

Deserialize a namespace back to regular values.

Parameters:

Name	Type	Description	Default
serialized	dict[str, SerializedValue]	Dict of serialized values.	required

Returns:

Type	Description
dict[str, Any]	Dict mapping variable names to deserialized values.

Source code in src/opensymbolicai/checkpoint.py

def deserialize_namespace(
    self, serialized: dict[str, SerializedValue]
) -> dict[str, Any]:
    """Deserialize a namespace back to regular values.

    Args:
        serialized: Dict of serialized values.

    Returns:
        Dict mapping variable names to deserialized values.
    """
    result: dict[str, Any] = {}
    for key, value in serialized.items():
        result[key] = self.deserialize(value)
    return result

register(type_, serializer, deserializer)

Register a serializer/deserializer pair for a type.

Parameters:

Name	Type	Description	Default
type_	type	The Python type to register.	required
serializer	Serializer	Function that converts instance to JSON-compatible data.	required
deserializer	Deserializer	Function that converts JSON data back to instance.	required

Source code in src/opensymbolicai/checkpoint.py

def register(
    self, type_: type, serializer: Serializer, deserializer: Deserializer
) -> None:
    """Register a serializer/deserializer pair for a type.

    Args:
        type_: The Python type to register.
        serializer: Function that converts instance to JSON-compatible data.
        deserializer: Function that converts JSON data back to instance.
    """
    type_name = f"{type_.__module__}.{type_.__qualname__}"
    self._serializers[type_] = serializer
    self._deserializers[type_name] = deserializer

register_deserializer(type_name)

Decorator to register a deserializer for a type name.

Example

@registry.register_deserializer("mymodule.MyClass") def deserialize(data: dict) -> MyClass: return MyClass(field=data["field"])

Source code in src/opensymbolicai/checkpoint.py

def register_deserializer(
    self, type_name: str
) -> Callable[[Deserializer], Deserializer]:
    """Decorator to register a deserializer for a type name.

    Example:
        @registry.register_deserializer("mymodule.MyClass")
        def deserialize(data: dict) -> MyClass:
            return MyClass(field=data["field"])
    """

    def decorator(fn: Deserializer) -> Deserializer:
        self._deserializers[type_name] = fn
        return fn

    return decorator

register_serializer(type_)

Decorator to register a serializer for a type.

Example

@registry.register_serializer(MyClass) def serialize(obj: MyClass) -> dict: return {"field": obj.field}

Source code in src/opensymbolicai/checkpoint.py

def register_serializer(self, type_: type) -> Callable[[Serializer], Serializer]:
    """Decorator to register a serializer for a type.

    Example:
        @registry.register_serializer(MyClass)
        def serialize(obj: MyClass) -> dict:
            return {"field": obj.field}
    """

    def decorator(fn: Serializer) -> Serializer:
        self._serializers[type_] = fn
        return fn

    return decorator

serialize(value)

Serialize a value to a SerializedValue.

Parameters:

Name	Type	Description	Default
value	Any	The value to serialize.	required

Returns:

Type	Description
SerializedValue	SerializedValue with type info and serialized data.

Source code in src/opensymbolicai/checkpoint.py

def serialize(self, value: Any) -> SerializedValue:
    """Serialize a value to a SerializedValue.

    Args:
        value: The value to serialize.

    Returns:
        SerializedValue with type info and serialized data.
    """
    # Handle None
    if value is None:
        return SerializedValue(type_name="NoneType", data=None, is_primitive=True)

    value_type = type(value)
    type_name = self._get_type_name(value_type)

    # Check for JSON primitives
    if isinstance(value, bool):  # Must check bool before int (bool is subclass)
        return SerializedValue(
            type_name="builtins.bool", data=value, is_primitive=True
        )
    if isinstance(value, int):
        return SerializedValue(
            type_name="builtins.int", data=value, is_primitive=True
        )
    if isinstance(value, float):
        return SerializedValue(
            type_name="builtins.float", data=value, is_primitive=True
        )
    if isinstance(value, str):
        return SerializedValue(
            type_name="builtins.str", data=value, is_primitive=True
        )

    # Check for registered serializer (before generic handling)
    if value_type in self._serializers:
        serializer = self._serializers[value_type]
        return SerializedValue(
            type_name=type_name, data=serializer(value), is_primitive=False
        )

    # Check for lists (recursively serialize)
    if isinstance(value, list):
        serialized_items = [self.serialize(item).model_dump() for item in value]
        return SerializedValue(
            type_name="builtins.list",
            data=serialized_items,
            is_primitive=False,
        )

    # Check for dicts (recursively serialize)
    if isinstance(value, dict):
        serialized_dict = {
            k: self.serialize(v).model_dump() for k, v in value.items()
        }
        return SerializedValue(
            type_name="builtins.dict",
            data=serialized_dict,
            is_primitive=False,
        )

    # Check if it's a Pydantic model
    if hasattr(value, "model_dump"):
        return SerializedValue(
            type_name=type_name,
            data={"__pydantic__": True, "data": value.model_dump()},
            is_primitive=False,
        )

    # Fallback: try to convert to string representation
    return SerializedValue(
        type_name=type_name,
        data={"__repr__": repr(value), "__unserializable__": True},
        is_primitive=False,
    )

serialize_namespace(namespace, exclude=None)

Serialize an entire namespace dict.

Parameters:

Name	Type	Description	Default
namespace	dict[str, Any]	The namespace to serialize.	required
exclude	set[str] | None	Keys to exclude from serialization.	None

Returns:

Type	Description
dict[str, SerializedValue]	Dict mapping variable names to SerializedValues.

Source code in src/opensymbolicai/checkpoint.py

def serialize_namespace(
    self, namespace: dict[str, Any], exclude: set[str] | None = None
) -> dict[str, SerializedValue]:
    """Serialize an entire namespace dict.

    Args:
        namespace: The namespace to serialize.
        exclude: Keys to exclude from serialization.

    Returns:
        Dict mapping variable names to SerializedValues.
    """
    exclude = exclude or set()
    result: dict[str, SerializedValue] = {}
    for key, value in namespace.items():
        if key in exclude:
            continue
        result[key] = self.serialize(value)
    return result

TokenUsage

Bases: BaseModel

Token usage statistics from LLM generation.

Source code in src/opensymbolicai/models.py

class TokenUsage(BaseModel):
    """Token usage statistics from LLM generation."""

    input_tokens: int = Field(default=0, description="Number of input/prompt tokens")
    output_tokens: int = Field(
        default=0, description="Number of output/completion tokens"
    )

    @property
    def total_tokens(self) -> int:
        """Total tokens used."""
        return self.input_tokens + self.output_tokens

total_tokens property

Total tokens used.

TraceEvent

Bases: BaseModel

A single trace event emitted during agent execution.

Events are grouped by trace_id (one per run/seek call) and form a tree via span_id / parent_span_id.

Source code in src/opensymbolicai/observability/events.py

class TraceEvent(BaseModel):
    """A single trace event emitted during agent execution.

    Events are grouped by trace_id (one per run/seek call) and form
    a tree via span_id / parent_span_id.
    """

    event_id: str = Field(..., description="Unique event identifier (UUID)")
    trace_id: str = Field(
        ..., description="Groups all events from one run()/seek() call"
    )
    session_id: str = Field(
        ..., description="Groups multiple traces into a logical session"
    )
    span_id: str = Field(..., description="Identifies this logical unit")
    parent_span_id: str | None = Field(
        default=None, description="Parent span for nesting"
    )
    event_type: EventType = Field(..., description="Type of event")
    timestamp: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        description="When the event occurred",
    )
    agent_class: str = Field(..., description="Agent class name")
    payload: dict[str, Any] = Field(
        default_factory=dict, description="Event-specific data"
    )
    tags: dict[str, str] = Field(
        default_factory=dict, description="User-supplied metadata"
    )

TraceTransport

Bases: Protocol

Protocol for sending trace events to a backend.

Implementations must provide send() and close().

Source code in src/opensymbolicai/observability/transports/protocol.py

@runtime_checkable
class TraceTransport(Protocol):
    """Protocol for sending trace events to a backend.

    Implementations must provide ``send()`` and ``close()``.
    """

    def send(self, events: list[TraceEvent]) -> None:
        """Send a batch of trace events.

        Args:
            events: List of events to send.
        """
        ...

    def close(self) -> None:
        """Flush any buffered events and release resources."""
        ...

close()

Flush any buffered events and release resources.

Source code in src/opensymbolicai/observability/transports/protocol.py

def close(self) -> None:
    """Flush any buffered events and release resources."""
    ...

send(events)

Send a batch of trace events.

Parameters:

Name	Type	Description	Default
events	list[TraceEvent]	List of events to send.	required

Source code in src/opensymbolicai/observability/transports/protocol.py

def send(self, events: list[TraceEvent]) -> None:
    """Send a batch of trace events.

    Args:
        events: List of events to send.
    """
    ...

ValidationError

Bases: ExecutionError

Exception raised when input validation fails.

Use this for invalid arguments, out-of-range values, or constraint violations in primitive inputs.

Example

if number <= 0: raise ValidationError( "Logarithm requires positive input", code="INVALID_INPUT", details={"received": number, "expected": "positive number"} )

Source code in src/opensymbolicai/exceptions.py

class ValidationError(ExecutionError):
    """Exception raised when input validation fails.

    Use this for invalid arguments, out-of-range values, or
    constraint violations in primitive inputs.

    Example:
        if number <= 0:
            raise ValidationError(
                "Logarithm requires positive input",
                code="INVALID_INPUT",
                details={"received": number, "expected": "positive number"}
            )
    """

    def __init__(
        self,
        message: str,
        *,
        code: str | None = "VALIDATION_ERROR",
        details: dict[str, Any] | None = None,
        field: str | None = None,
    ) -> None:
        """Initialize validation error.

        Args:
            message: Human-readable error message.
            code: Error code, defaults to "VALIDATION_ERROR".
            details: Additional context about the error.
            field: Optional name of the field that failed validation.
        """
        if field and details is None:
            details = {}
        if field:
            details = details or {}
            details["field"] = field
        super().__init__(message, code=code, details=details)
        self.field = field

__init__(message, *, code='VALIDATION_ERROR', details=None, field=None)

Initialize validation error.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Error code, defaults to "VALIDATION_ERROR".	'VALIDATION_ERROR'
details	dict[str, Any] | None	Additional context about the error.	None
field	str | None	Optional name of the field that failed validation.	None

Source code in src/opensymbolicai/exceptions.py

def __init__(
    self,
    message: str,
    *,
    code: str | None = "VALIDATION_ERROR",
    details: dict[str, Any] | None = None,
    field: str | None = None,
) -> None:
    """Initialize validation error.

    Args:
        message: Human-readable error message.
        code: Error code, defaults to "VALIDATION_ERROR".
        details: Additional context about the error.
        field: Optional name of the field that failed validation.
    """
    if field and details is None:
        details = {}
    if field:
        details = details or {}
        details["field"] = field
    super().__init__(message, code=code, details=details)
    self.field = field

decomposition(intent, expanded_intent='')

Mark a method as a decomposition example.

Decompositions demonstrate how to break down high-level intents into sequences of primitive operations. They serve as examples for the LLM to learn the patterns of composition.

Parameters:

Name	Type	Description	Default
intent	str	A high-level description of what this decomposition achieves. This is the natural language query that this example answers.	required
expanded_intent	str	An optional step-by-step breakdown of the approach. Provides additional context about the reasoning or methodology.	''

Returns:

Type	Description
Callable[[F], F]	A decorator that marks the function as a decomposition example.

Example

@decomposition( intent="What is sine of 90 degrees?", expanded_intent="First convert 90 degrees to radians, then calculate sine", ) def _example_sine_90(self) -> float: angle_rad: float = self.convert_degrees_to_radians(angle_in_degrees=90) sin_90: float = self.sine(angle_in_radians=angle_rad) return sin_90

Source code in src/opensymbolicai/core.py

def decomposition[F: Callable[..., Any]](intent: str, expanded_intent: str = "") -> Callable[[F], F]:
    """Mark a method as a decomposition example.

    Decompositions demonstrate how to break down high-level intents into
    sequences of primitive operations. They serve as examples for the LLM
    to learn the patterns of composition.

    Args:
        intent: A high-level description of what this decomposition achieves.
            This is the natural language query that this example answers.
        expanded_intent: An optional step-by-step breakdown of the approach.
            Provides additional context about the reasoning or methodology.

    Returns:
        A decorator that marks the function as a decomposition example.

    Example:
        @decomposition(
            intent="What is sine of 90 degrees?",
            expanded_intent="First convert 90 degrees to radians, then calculate sine",
        )
        def _example_sine_90(self) -> float:
            angle_rad: float = self.convert_degrees_to_radians(angle_in_degrees=90)
            sin_90: float = self.sine(angle_in_radians=angle_rad)
            return sin_90
    """

    def decorator(func: F) -> F:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)

        wrapper.__method_type__ = MethodType.DECOMPOSITION  # type: ignore[attr-defined]
        wrapper.__decomposition_intent__ = intent  # type: ignore[attr-defined]
        wrapper.__decomposition_expanded_intent__ = expanded_intent  # type: ignore[attr-defined]
        return cast(F, wrapper)

    return decorator

evaluator(func)

Mark a method as the goal evaluator.

The evaluator determines whether the goal has been achieved after each iteration in a GoalSeeking agent. Exactly one method per agent should be decorated with @evaluator.

The decorated method receives (goal, context) and must return a GoalEvaluation. The evaluator checks structured insights on the context — never raw ExecutionResult. By the time the evaluator runs, update_context() has already introspected the raw result into context fields.

Returns:

Type	Description
F	The decorated function marked as an evaluator.

Example

@evaluator def check_goal(self, goal: str, context: GoalContext) -> GoalEvaluation: has_enough = len(context.findings) >= 5 return GoalEvaluation(goal_achieved=has_enough)

Source code in src/opensymbolicai/core.py

def evaluator[F: Callable[..., Any]](func: F) -> F:
    """Mark a method as the goal evaluator.

    The evaluator determines whether the goal has been achieved after each
    iteration in a GoalSeeking agent. Exactly one method per agent should
    be decorated with @evaluator.

    The decorated method receives (goal, context) and must return a
    GoalEvaluation. The evaluator checks structured insights on the
    context — never raw ExecutionResult. By the time the evaluator runs,
    update_context() has already introspected the raw result into
    context fields.

    Returns:
        The decorated function marked as an evaluator.

    Example:
        @evaluator
        def check_goal(self, goal: str, context: GoalContext) -> GoalEvaluation:
            has_enough = len(context.findings) >= 5
            return GoalEvaluation(goal_achieved=has_enough)
    """

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        return func(*args, **kwargs)

    wrapper.__method_type__ = MethodType.EVALUATOR  # type: ignore[attr-defined]
    return cast(F, wrapper)

extract_context(full_prompt)

Extract only the CONTEXT section from a full prompt.

This is the dynamic per-call content, independent of the static DEFINITIONS and INSTRUCTIONS sections.

Parameters:

Name	Type	Description	Default
full_prompt	str	A prompt containing CONTEXT markers.	required

Returns:

Type	Description
str	The text between ###<<CONTEXT:BEGIN>>### and
str	###<<CONTEXT:END>>###.

Raises:

Type	Description
ValueError	If the prompt is missing CONTEXT markers.

Source code in src/opensymbolicai/prompt_utils.py

def extract_context(full_prompt: str) -> str:
    """Extract only the CONTEXT section from a full prompt.

    This is the dynamic per-call content, independent of the static
    DEFINITIONS and INSTRUCTIONS sections.

    Args:
        full_prompt: A prompt containing CONTEXT markers.

    Returns:
        The text between ``###<<CONTEXT:BEGIN>>###`` and
        ``###<<CONTEXT:END>>###``.

    Raises:
        ValueError: If the prompt is missing CONTEXT markers.
    """
    begin = full_prompt.find(PROMPT_CONTEXT_BEGIN)
    end = full_prompt.find(PROMPT_CONTEXT_END)
    if begin == -1 or end == -1:
        raise ValueError("Prompt missing CONTEXT markers")
    return full_prompt[begin + len(PROMPT_CONTEXT_BEGIN) : end].strip()

primitive(read_only=False, deterministic=True)

Mark a method as a primitive operation.

Primitives are atomic operations that the agent can directly execute. They serve as the building blocks for more complex behaviors.

Parameters:

Name	Type	Description	Default
read_only	bool	If True, indicates this primitive does not modify state.	False
deterministic	bool	If True (default), the primitive always returns the same output for the same inputs (pure function). Set to False for primitives that call LLMs, external APIs, or have side effects. Used by downstream tooling to decide whether to call the real implementation or mock-replay from captured traces during semantic validation.	True

Returns:

Type	Description
Callable[[F], F]	A decorator that marks the function as a primitive.

Example

@primitive(read_only=True) def add_numbers(self, a: float, b: float) -> float: return a + b

@primitive(deterministic=False) def resolve_name(self, name: str) -> str: return self._llm.generate(f"Resolve: {name}").text

Source code in src/opensymbolicai/core.py

def primitive[F: Callable[..., Any]](
    read_only: bool = False,
    deterministic: bool = True,
) -> Callable[[F], F]:
    """Mark a method as a primitive operation.

    Primitives are atomic operations that the agent can directly execute.
    They serve as the building blocks for more complex behaviors.

    Args:
        read_only: If True, indicates this primitive does not modify state.
        deterministic: If True (default), the primitive always returns the same
            output for the same inputs (pure function). Set to False for
            primitives that call LLMs, external APIs, or have side effects.
            Used by downstream tooling to decide whether to call
            the real implementation or mock-replay from captured traces during
            semantic validation.

    Returns:
        A decorator that marks the function as a primitive.

    Example:
        @primitive(read_only=True)
        def add_numbers(self, a: float, b: float) -> float:
            return a + b

        @primitive(deterministic=False)
        def resolve_name(self, name: str) -> str:
            return self._llm.generate(f"Resolve: {name}").text
    """

    def decorator(func: F) -> F:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)

        wrapper.__method_type__ = MethodType.PRIMITIVE  # type: ignore[attr-defined]
        wrapper.__primitive_read_only__ = read_only  # type: ignore[attr-defined]
        wrapper.__primitive_deterministic__ = deterministic  # type: ignore[attr-defined]
        return cast(F, wrapper)

    return decorator

split_prompt(full_prompt)

Split a full 3-section prompt into its component parts.

Sections with missing markers return empty strings. If no markers are present at all, the entire prompt is returned as the preamble.

Parameters:

Name	Type	Description	Default
full_prompt	str	A prompt, optionally containing DEFINITIONS, CONTEXT, and INSTRUCTIONS sections delimited by ###<<SECTION:BEGIN>>### / ###<<SECTION:END>>### markers.	required

Returns:

Type	Description
PromptSections	A PromptSections instance with the four extracted parts.

Source code in src/opensymbolicai/prompt_utils.py

def split_prompt(full_prompt: str) -> PromptSections:
    """Split a full 3-section prompt into its component parts.

    Sections with missing markers return empty strings. If no markers are
    present at all, the entire prompt is returned as the preamble.

    Args:
        full_prompt: A prompt, optionally containing DEFINITIONS, CONTEXT,
            and INSTRUCTIONS sections delimited by
            ``###<<SECTION:BEGIN>>###`` / ``###<<SECTION:END>>###`` markers.

    Returns:
        A PromptSections instance with the four extracted parts.
    """
    first_marker = full_prompt.find(PROMPT_DEFINITIONS_BEGIN)
    preamble = full_prompt[:first_marker].strip() if first_marker != -1 else full_prompt.strip()

    return PromptSections(
        preamble=preamble,
        definitions=_extract_section(
            full_prompt, PROMPT_DEFINITIONS_BEGIN, PROMPT_DEFINITIONS_END
        ),
        context=_extract_section(
            full_prompt, PROMPT_CONTEXT_BEGIN, PROMPT_CONTEXT_END
        ),
        instructions=_extract_section(
            full_prompt, PROMPT_INSTRUCTIONS_BEGIN, PROMPT_INSTRUCTIONS_END
        ),
    )