Action Policies

BaseActionPolicy

polymathera.colony.agents.patterns.actions.policies.BaseActionPolicy(agent, action_map=None, action_providers=[], io=None, consciousness_streams=None)

Bases: ActionPolicy

Base class for action policies with dataflow and nested policy support.

Provides: - Automatic action dispatcher creation - Integration with agent capabilities - Nested policy execution with scope inheritance - Dispatch with automatic Ref resolution

Subclasses implement plan_step to produce the next action or child policy. The base execute_iteration handles: - Delegating to active child policies - Executing actions returned by plan_step - Setting up child policies returned by plan_step

TODO: For example, we can orchestrate iterative reasoning to follow the pattern (PLAN → ACT → REFLECT → CRITIQUE → ADAPT) by adding AgentCapabilities that implement each step as an action executor, and then implementing plan_step to select the next action based on the current state. This can be enforced by: - Restricting available actions in the action dispatcher depending on the last completed step, or - Using an ActionPolicy subclass that implements the iterative pattern by overriding execute_iteration to enforce the sequence of steps, and only calling plan_step to get parameters for each step, or - Prompting the LLM planner with this workflow.

Example

class MyPolicy(BaseActionPolicy):
    io = ActionPolicyIO(
        inputs={"query": str},
        outputs={"result": dict}
    )

    async def plan_step(self, state) -> Action | None:
        # Return None when policy is complete
        if state.custom.get("done"):
            return None

        # Return an Action to execute
        return Action(
            action_id="analyze_001",
            agent_id=self.agent.agent_id,
            action_type="analyze",
            parameters={"query": state.scope.get("query")}
        )

        # Or return an ActionPolicy for nested execution
        # return ChildPolicy(self.agent)

Source code in src/polymathera/colony/agents/patterns/actions/policies.py

def __init__(
    self,
    agent: Agent,
    action_map: list[ActionGroup] | None = None,
    action_providers: list[Any] = [],
    io: ActionPolicyIO | None = None, # Declare I/O contract (override in subclasses)
    consciousness_streams: list[ConsciousnessStream] | None = None,
):
    super().__init__(agent)
    self._action_map = action_map
    self._action_providers = action_providers
    self._action_dispatcher: ActionDispatcher | None = None
    self.io: ActionPolicyIO = io or ActionPolicyIO()
    self._consciousness_streams: list[ConsciousnessStream] = list(
        consciousness_streams or []
    )
    # Stream event sources (PR-Sub-2a). Sources are registered via
    # ``attach_source`` (typically by the agent setup that also
    # built the streams) and attached during ``initialize`` so
    # ``self._action_dispatcher`` is already constructed for
    # sources that need it (e.g. ToolResultSource).
    self._stream_sources: list[Any] = []
    self._attached_source_ids: set[int] = set()
    # Tool-result sources subscribe separately so the policy can
    # fan post-dispatch ``ActionResult.data`` into ``"tool_output"``
    # entries without invoking the full source ABC each time.
    self._tool_result_sources: list[Any] = []
    # LLM-cluster failure backoff. Extracted into its own class so
    # the host stays focused on plan-step / dispatch concerns —
    # see ``llm_failure_backoff.py`` for the contract.
    from .llm_failure_backoff import LLMFailureBackoff
    self._llm_failure_backoff = LLMFailureBackoff(agent)

execute_iteration(state) async

Run one iteration, then per-iteration stream maintenance.

The maintenance (flush new entries to each compaction-enabled stream's durable log + run the token-budget safety-net) runs in a finally so it happens however the iteration exits — the last iteration's entries are durably persisted even on an early return or exception. _maintain_streams never raises and is a no-op for legacy (non-compacted) streams.

Source code in src/polymathera/colony/agents/patterns/actions/policies.py

@hookable
@override
async def execute_iteration(
    self,
    state: ActionPolicyExecutionState,
) -> ActionPolicyIterationResult:
    """Run one iteration, then per-iteration stream maintenance.

    The maintenance (flush new entries to each compaction-enabled
    stream's durable log + run the token-budget safety-net) runs in
    a ``finally`` so it happens however the iteration exits — the
    last iteration's entries are durably persisted even on an early
    return or exception. ``_maintain_streams`` never raises and is a
    no-op for legacy (non-compacted) streams.
    """
    try:
        return await self._execute_iteration_inner(state)
    finally:
        await self._maintain_streams()

plan_step(state) async

Produce the next action to execute.

Override this method to implement policy-specific planning logic.

For hierarchical composition, spawn child agents instead of nesting policies. Use self.agent.spawn_child_agents() with appropriate action policies for child agents.

Parameters:

Name	Type	Description	Default
state	ActionPolicyExecutionState	Execution state for this policy	required

Returns:

Type	Description
Action | None	Action: Execute this action
Action | None	None: Skip this iteration. Set state.custom["policy_complete"] = True before returning None to signal that the policy is finished.

Example

async def plan_step(self, state) -> Action | None:
    phase = state.custom.get("phase", "act")

    if phase == "act":
        action = self._get_next_action(state)
        if action is None:
            state.custom["policy_complete"] = True
            return None
        state.custom["phase"] = "process"
        return action

    elif phase == "process":
        # Do some processing without dispatching an action
        self._process_results(state)
        state.custom["phase"] = "act"
        return None  # Skip iteration, continue policy

Source code in src/polymathera/colony/agents/patterns/actions/policies.py

async def plan_step(
    self,
    state: ActionPolicyExecutionState
) -> Action | None:
    """Produce the next action to execute.

    Override this method to implement policy-specific planning logic.

    For hierarchical composition, spawn child agents instead of nesting
    policies. Use `self.agent.spawn_child_agents()` with appropriate
    action policies for child agents.

    Args:
        state: Execution state for this policy

    Returns:
        - Action: Execute this action
        - None: Skip this iteration. Set `state.custom["policy_complete"] = True`
            before returning None to signal that the policy is finished.

    Example:
        ```python
        async def plan_step(self, state) -> Action | None:
            phase = state.custom.get("phase", "act")

            if phase == "act":
                action = self._get_next_action(state)
                if action is None:
                    state.custom["policy_complete"] = True
                    return None
                state.custom["phase"] = "process"
                return action

            elif phase == "process":
                # Do some processing without dispatching an action
                self._process_results(state)
                state.custom["phase"] = "act"
                return None  # Skip iteration, continue policy
        ```
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} must implement plan_step"
    )

action_executor decorator

polymathera.colony.agents.patterns.actions.dispatcher.action_executor(action_key=None, *, input_schema=None, output_schema=None, reads=None, writes=None, exclude_from_planning=False, planning_summary=None, tags=None, interruptible=False, emits_lifecycle=True)

Decorator to turn any method into an action executor.

Automatically infers input/output schemas from type hints if not provided.

Parameters:

Name	Type	Description	Default
action_key	str | ActionType | None	Key identifying the action type. If None, uses method name.	None
input_schema	type[BaseModel] | None	Optional Pydantic model for input validation. If None, inferred from method signature.	None
output_schema	type[BaseModel] | None	Optional Pydantic model for output validation. If None, inferred from return type hint.	None
reads	list[str] | None	List of scope variable names this action reads.	None
writes	list[str] | None	List of scope variable names this action writes.	None
exclude_from_planning	bool	If True, this action is not exposed to the LLM planner. Use this for actions that are only meant to be invoked programmatically in response to events (e.g., game moves in response to spawned agent events). Default is False.	False
tags	frozenset[str] | None	Optional domain/modality tags for this action (e.g., frozenset({"memory", "expensive"})). Used for future per-action tag-based filtering and grouping.	None
interruptible	bool	If True, the dispatcher wraps this action's execution in an asyncio.Task so it can be cancelled mid-flight via ActionDispatcher.cancel_current_action() (e.g., in response to a /abort or /replace slash command). The action body MUST be cancel-safe: any external resources it acquires must be released in finally / async with blocks, and intermediate state must be consistent at every await. When the action is cancelled, the dispatcher returns ActionResult(success=False, completed=True, cancelled=True, ...) instead of raising. Default False — only opt in for genuinely long-running actions (LLM calls, network I/O, sandboxed shell, large analyses) where the user benefits from being able to interrupt them.	False
emits_lifecycle	bool	If True (default), the codegen action policy publishes policy:action_started and policy:action_completed lifecycle events around dispatch — the chat UI spinner and the trace ring buffer both consume these. Set False for actions whose semantics make a "running" badge wrong: idle waits (wait_for_next_event), publish-only narrative emits (emit_mission_status), etc. The protocol stops announcing as "running" something that is by definition not running; every subscriber inherits the fix. The LLM never sees this flag — it is purely framework-facing. Per [[fix-the-class-not-the-instance]] the invariant "idle waits are not work" lives at the single emission source, not at every subscriber.	True

Example

@action_executor()
async def route_query(
    self,
    query: str,
    max_results: int = 10
) -> list[str]:
    '''Route query to find relevant pages.'''
    ...

@action_executor(writes=["analysis_result"])
async def analyze_pages(
    self,
    page_ids: list[str],
    goal: str
) -> AnalysisResult:
    '''Analyze pages for the given goal.'''
    ...

# Event-driven action not visible to planner
@action_executor(exclude_from_planning=True)
async def submit_move(self, game_id: str, move: dict) -> None:
    '''Submit move in response to game event.'''
    ...

Source code in src/polymathera/colony/agents/patterns/actions/dispatcher.py

def action_executor(
    action_key: str | ActionType | None = None,
    *,
    input_schema: type[BaseModel] | None = None,
    output_schema: type[BaseModel] | None = None,
    reads: list[str] | None = None,
    writes: list[str] | None = None,
    exclude_from_planning: bool = False,
    planning_summary: str | None = None,
    tags: frozenset[str] | None = None,
    interruptible: bool = False,
    emits_lifecycle: bool = True,
):
    """Decorator to turn any method into an action executor.

    Automatically infers input/output schemas from type hints if not provided.

    Args:
        action_key: Key identifying the action type. If None, uses method name.
        input_schema: Optional Pydantic model for input validation.
            If None, inferred from method signature.
        output_schema: Optional Pydantic model for output validation.
            If None, inferred from return type hint.
        reads: List of scope variable names this action reads.
        writes: List of scope variable names this action writes.
        exclude_from_planning: If True, this action is not exposed to the LLM
            planner. Use this for actions that are only meant to be invoked
            programmatically in response to events (e.g., game moves in response
            to spawned agent events). Default is False.
        tags: Optional domain/modality tags for this action (e.g., frozenset({"memory", "expensive"})).
            Used for future per-action tag-based filtering and grouping.
        interruptible: If True, the dispatcher wraps this action's execution in
            an asyncio.Task so it can be cancelled mid-flight via
            ``ActionDispatcher.cancel_current_action()`` (e.g., in response to a
            ``/abort`` or ``/replace`` slash command). The action body MUST be
            cancel-safe: any external resources it acquires must be released in
            ``finally`` / ``async with`` blocks, and intermediate state must be
            consistent at every ``await``. When the action is cancelled, the
            dispatcher returns ``ActionResult(success=False, completed=True,
            cancelled=True, ...)`` instead of raising. Default False — only opt
            in for genuinely long-running actions (LLM calls, network I/O,
            sandboxed shell, large analyses) where the user benefits from being
            able to interrupt them.
        emits_lifecycle: If True (default), the codegen action policy
            publishes ``policy:action_started`` and ``policy:action_completed``
            lifecycle events around dispatch — the chat UI spinner and the
            trace ring buffer both consume these. Set False for actions
            whose semantics make a "running" badge wrong: idle waits
            (``wait_for_next_event``), publish-only narrative emits
            (``emit_mission_status``), etc. The protocol stops announcing
            as "running" something that is by definition not running;
            every subscriber inherits the fix. The LLM never sees this
            flag — it is purely framework-facing. Per
            [[fix-the-class-not-the-instance]] the invariant "idle waits
            are not work" lives at the single emission source, not at
            every subscriber.

    Example:
        ```python
        @action_executor()
        async def route_query(
            self,
            query: str,
            max_results: int = 10
        ) -> list[str]:
            '''Route query to find relevant pages.'''
            ...

        @action_executor(writes=["analysis_result"])
        async def analyze_pages(
            self,
            page_ids: list[str],
            goal: str
        ) -> AnalysisResult:
            '''Analyze pages for the given goal.'''
            ...

        # Event-driven action not visible to planner
        @action_executor(exclude_from_planning=True)
        async def submit_move(self, game_id: str, move: dict) -> None:
            '''Submit move in response to game event.'''
            ...
        ```
    """
    def decorator(func):
        # Store action key
        func._action_key = action_key or func.__name__

        # Infer schemas from type hints if not provided
        func._action_input_schema = input_schema or _infer_input_schema(func)
        func._action_output_schema = output_schema or _infer_output_schema(func)

        # Store scope read/write declarations
        func._action_reads = reads or []
        func._action_writes = writes or []

        # Store planning visibility flag
        func._action_exclude_from_planning = exclude_from_planning

        # Store concise planning summary (used instead of full docstring in prompts)
        func._action_planning_summary = planning_summary

        # Store tags for future per-action tag-based filtering
        func._action_tags = tags or frozenset()

        # Store interruptibility flag — dispatched as a cancellable task when True.
        func._action_interruptible = bool(interruptible)

        # Store lifecycle-emission flag — codegen policy reads this to
        # decide whether to publish ``policy:action_started`` /
        # ``policy:action_completed`` around dispatch. See the
        # decorator's ``emits_lifecycle`` argument for the contract.
        func._action_emits_lifecycle = bool(emits_lifecycle)

        return func

    return decorator