Agent

polymathera.colony.agents.base.Agent

Bases: BaseModel

Base agent class representing an autonomous computational entity.

All agents have: - Unique ID and type - Lifecycle state - Optional page binding - Access to system services (VCM, LLM, blackboard)

Control flow should be driven by a reasoning LLM given sufficient context, not hardcoded. Agents adapt behavior based on current context.

Attributes:

Name	Type	Description
agent_id	str	Unique identifier
agent_type	str	Type of agent ("specialized", "general", "service", "supervisor")
state	AgentState	Current lifecycle state
bound_pages	list[str]	Page IDs this agent is bound to (empty for unbound agents)
created_at	float	Creation timestamp
metadata	AgentMetadata	Arbitrary metadata

state = Field(default=(AgentState.INITIALIZED)) class-attribute instance-attribute

agent_id instance-attribute

metadata = Field(default_factory=AgentMetadata) class-attribute instance-attribute

run_step() async

Execute one step of agent logic.

This is the core method that defines agent behavior. Override in subclasses to implement specific agent logic.

This method should be idempotent and handle its own errors.

This method is @hookable, so capabilities can register BEFORE hooks to prepare for the step, AFTER hooks to post-process the step, AROUND hooks to wrap the step execution entirely, and to handle errors during execution.

NOTE: We use the repeated run_step approach instead of a single long-running run method to facilitate easier suspension and state management. This is necessary for distributed agents that may be suspended and resumed across different replicas.

Source code in src/polymathera/colony/agents/base.py

@hookable
@check_isolation
async def run_step(self) -> None:
    """Execute one step of agent logic.

    This is the core method that defines agent behavior. Override in
    subclasses to implement specific agent logic.

    This method should be idempotent and handle its own errors.

    This method is @hookable, so capabilities can register BEFORE hooks
    to prepare for the step,
    AFTER hooks to post-process the step,
    AROUND hooks to wrap the step execution entirely,
    and to handle errors during execution.

    NOTE: We use the repeated `run_step` approach instead of a single
    long-running `run` method to facilitate easier suspension and state
    management. This is necessary for distributed agents that may be
    suspended and resumed across different replicas.
    """
    if self.state not in (AgentState.RUNNING, AgentState.IDLE):
        logger.warning(
            f"Agent {self.agent_id} in state {self.state}, cannot run step"
        )
        return

    # Set up session_id context for the ENTIRE run_step
    # Session_id is propagated from requests via action_policy_state.custom["current_session_id"]
    # This ensures ALL operations (control messages, hooks, capabilities) have session_id
    from .sessions.context import session_id_context
    current_session_id = None
    if self.action_policy_state is not None:
        current_session_id = self.action_policy_state.custom.get("current_session_id")

    with session_id_context(current_session_id):
        # Process control messages from parent/children (mailbox)
        await self._process_control_messages()

        # Check child health (periodic health monitoring)
        await self._check_child_health()

        if self.action_policy_state is None:
            self.action_policy_state = ActionPolicyExecutionState()

        logger.warning(
            f"\n"
            f"  ┌──────────────────────────────────────────────┐\n"
            f"  │  ▶ RUN_STEP: calling execute_iteration       │\n"
            f"  │  agent={self.agent_id:<40}│\n"
            f"  └──────────────────────────────────────────────┘"
        )
        iteration_result = await self.action_policy.execute_iteration(self.action_policy_state)
        action_str, trunc = self._model_to_str(iteration_result.action_executed)
        logger.warning(
            f"  ◀ RUN_STEP returned: success={iteration_result.success} "
            f"completed={iteration_result.policy_completed} "
            f"idle={iteration_result.idle} "
            f"action={action_str} ({trunc})"
        )

        # Policy-driven state transitions
        if iteration_result.policy_completed:
            # CONTINUOUS-mode agents: A long-lived service agent
            # (SessionAgent, system session, future webhook
            # watchers) MUST NOT terminate on a per-policy
            # ``policy_completed`` signal — the policy completed
            # ITS plan (or hit some stuck-detection branch), but
            # the agent's lifetime is bounded by the SESSION,
            # not by any one plan. Without this branch, every
            # per-policy cap (max_code_iterations, MinimalActionPolicy's
            # cap, an event-handler returning done=True, …) silently
            # kills the chat.
            if self.metadata.lifecycle_mode == LifecycleMode.CONTINUOUS:
                logger.info(
                    "Agent %s is CONTINUOUS; downgrading "
                    "policy_completed → IDLE (the policy "
                    "completed its plan; the agent keeps "
                    "running until _stop_requested).",
                    self.agent_id,
                )
                if self.state != AgentState.IDLE:
                    self._idle_since = time.time()
                self.state = AgentState.IDLE
            else:
                self.state = AgentState.STOPPED
        elif iteration_result.idle:
            if self.state != AgentState.IDLE:
                self._idle_since = time.time()
            self.state = AgentState.IDLE
        elif self.state == AgentState.IDLE:
            # Policy returned work while we were IDLE — wake up
            self.state = AgentState.RUNNING
            self._idle_since = None

        if self.state == AgentState.STOPPED:
            logger.info(f"Agent {self.agent_id} has entered STOPPED state")
            # Stop agent gracefully
            await self.stop(reason="policy_completed")

        if self.state == AgentState.IDLE:
            timeout = self.metadata.idle_timeout
            if timeout is not None and hasattr(self, '_idle_since'):
                if (time.time() - self._idle_since) > timeout:
                    logger.info(
                        f"Agent {self.agent_id} idle timeout ({timeout}s) reached, stopping"
                    )
                    await self.stop(reason="idle_timeout")
                    return
            await self.on_idle()

        if iteration_result.error_context:
            logger.error(
                f"Agent {self.agent_id} encountered error in run_step: "
                f"{iteration_result.error_context.error_details}"
            )
            # Optionally, could set state to FAILED here
            # self.state = AgentState.FAILED

        # Sleep interval: longer when IDLE to avoid busy-looping
        sleep_interval = (
            self.metadata.idle_sleep_interval
            if self.state == AgentState.IDLE
            else 0.1
        )
        await asyncio.sleep(sleep_interval)

stop(reason='completed') async

Stop agent execution.

This method is @hookable, so capabilities can register BEFORE hooks to flush memories, AFTER hooks for cleanup, etc.

Parameters:

Name	Type	Description	Default
reason	str	Why the agent is stopping. Stored in _stop_reason for tracing and run status updates. Common values: "policy_completed", "idle_timeout", "max_iterations", "error", "cancelled", "stop_requested".	'completed'

Source code in src/polymathera/colony/agents/base.py

@hookable
async def stop(self, reason: str = "completed") -> None:
    """Stop agent execution.

    This method is @hookable, so capabilities can register BEFORE hooks
    to flush memories, AFTER hooks for cleanup, etc.

    Args:
        reason: Why the agent is stopping. Stored in ``_stop_reason``
            for tracing and run status updates. Common values:
            ``"policy_completed"``, ``"idle_timeout"``,
            ``"max_iterations"``, ``"error"``, ``"cancelled"``,
            ``"stop_requested"``.
    """
    self._stop_requested = True
    self._stop_reason = reason
    self.state = AgentState.STOPPED
    self._running = False
    self.action_policy_state = None
    # Parent-agent notification + cross-replica registry sync are
    # handled OUTSIDE Agent.stop() so this method stays Agent-pure
    # (no manager / event-bus reach-throughs):
    # - The manager's ``AgentManagerBase._run_loop`` finally block
    #   propagates this terminal state to the AgentSystem registry
    #   via ``update_agent_state`` so cluster-wide readers
    #   (``fetch_agent_info``, ``get_agent_status``, the dashboard)
    #   see the truth.
    # - ``MemoryLifecycleHooks.flush_before_shutdown`` (a BEFORE
    #   hook on this method) emits
    #   ``LifecycleSignalProtocol.terminated_key`` on the colony
    #   control-plane lifecycle scope; the payload carries
    #   ``parent_agent_id`` + ``stop_reason`` so the spawning
    #   parent's subscription can react without polling.

    # Fire stop_callbacks AFTER state has flipped to STOPPED so
    # callbacks observe the terminal state, not the in-progress
    # one. Errors in one callback don't block the others — they
    # log and the next callback still runs. This is the generic
    # lifecycle-hook surface; Agent stays free of any specific
    # cleanup concerns (mission spawn-gate, billing, audit,
    # …) — those install themselves at spawn time via
    # ``stop_callbacks=[…]`` on the blueprint.
    for callback in list(self.stop_callbacks):
        try:
            await callback(self, reason)
        except Exception:  # noqa: BLE001
            logger.exception(
                "Stop callback raised for agent %s — continuing "
                "with the remaining callbacks.", self.agent_id,
            )

get_blackboard(scope_id=None, backend_type='redis', enable_events=True) async

Get blackboard for reading/writing shared state.

Parameters:

Name	Type	Description	Default
scope_id	str | None	Scope identifier (defaults to agent_id for shared scope)	None
backend_type	str	Backend type for the blackboard (e.g., "redis")	'redis'
enable_events	bool	Whether to enable events on the blackboard	True

Returns:

Type	Description
EnhancedBlackboard	Blackboard instance

Example

# Get shared blackboard with parent
parent_id = self.metadata.parent_id
board = await self.get_blackboard(scope_id=parent_id)

# Write results
await board.write("analysis_complete", my_results)

# Parent reads
result = await board.read("analysis_complete")

Source code in src/polymathera/colony/agents/base.py

async def get_blackboard(
    self,
    scope_id: str | None = None,
    backend_type: str = "redis",
    enable_events: bool = True,
) -> EnhancedBlackboard:
    """Get blackboard for reading/writing shared state.

    Args:
        scope_id: Scope identifier (defaults to `agent_id` for shared scope)
        backend_type: Backend type for the blackboard (e.g., "redis")
        enable_events: Whether to enable events on the blackboard

    Returns:
        Blackboard instance

    Example:
        ```python
        # Get shared blackboard with parent
        parent_id = self.metadata.parent_id
        board = await self.get_blackboard(scope_id=parent_id)

        # Write results
        await board.write("analysis_complete", my_results)

        # Parent reads
        result = await board.read("analysis_complete")
        ```
    """
    if not self._manager:
        raise RuntimeError(f"Agent {self.agent_id} not attached to manager")

    # Default scope_id to agent_id for shared scope
    if scope_id is None:
        from .scopes import ScopeUtils
        scope_id = ScopeUtils.get_agent_level_scope(self)

    return await self._manager.get_blackboard(
        scope_id=scope_id,
        backend_type=backend_type,
        enable_events=enable_events,
    )