状态与持久化

            ./"$script" || echo "::notice::$(basename "$script") failed (non-fatal)"
          done

      - name: Commit results
        if: steps.work.outputs.mode != ''
        run: |
          git config user.name "aeonframework"
          git config user.email "aeonframework@proton.me"
          rm -f ./notify .notify-sent-hashes  # Remove generated notify script + dedup log before committing

          CURRENT_BRANCH=$(git branch --show-current)
          LABEL="${{ steps.work.outputs.label }}"

        # a dictionary of conversations, default value is list
        if chat_messages is None:
            self._oai_messages = defaultdict(list)
        else:
            self._oai_messages = chat_messages

        self._oai_system_message = [{"content": system_message, "role": "system"}]
        self._description = description if description is not None else system_message
        self._is_termination_msg = (
            is_termination_msg
            if is_termination_msg is not None
            else (lambda x: content_str(x.get("content")) == "TERMINATE")

# --- Persistence Hooks ---
class PersistenceHooks(RunHooks[MasterContext]):  # type: ignore[misc]
    """Custom `RunHooks` implementation for loading and saving `ThreadManager` state.

    This class integrates with the `agents.Runner` lifecycle to automatically
    save message history at the end of a run using user-provided callback
    functions. Loading relies on the `ThreadManager` initialization, which
    invokes the same callbacks to seed the in-memory store.

    Note:
        The signatures for `load_threads_callback` and `save_threads_callback` now

    timestamp = Column(DateTime, server_default=func.now())


class Agent(Base):
    __tablename__ = "agent"
    id = Column(
        UUID(as_uuid=True) if DATABASE_TYPE != "sqlite" else String,
        primary_key=True,
        default=get_new_id if DATABASE_TYPE == "sqlite" else uuid.uuid4,
    )
    name = Column(Text, nullable=False)
    provider_id = Column(
        UUID(as_uuid=True) if DATABASE_TYPE != "sqlite" else String,

 * Session manager that provides isolation between concurrent sessions
 * Generic over the type of state stored for each session
 */
export class SessionManager<T extends SessionState> {
  /** Storage provider for persistence */
  private storage: StorageProvider;

  /** Default state generator function */
  private defaultStateGenerator: (sessionId: SessionId) => T;

  /** Namespace for storage keys */
  private storageNamespace: string;

            raise ValueError("Skillbook serialization must be a JSON object.")
        return cls.from_dict(payload)

    def save_to_file(self, path: str, exclude_embeddings: bool = False) -> None:
        file_path = Path(path)
        sidecar_path = _embedding_sidecar_path(file_path)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        with file_path.open("w", encoding="utf-8") as f:
            f.write(self.dumps(exclude_embeddings=True))

        if exclude_embeddings:
            return

    """The task context."""


class AgentState(BaseModel):
    """The agent state that should be saved and loaded from storage."""

    session_id: str = Field(default_factory=lambda: uuid.uuid4().hex)
    """The session id of the agent. Normally, each session will maintain one
    independent agent state for each agent."""

    summary: str | list[TextBlock | DataBlock] = ""
    """The compressed summary of the context, which will be prepended to the
    context when feed into the LLM."""

      throw e;
    }
  },
  run: async ({ documentId, ingestJob, cleanup: shouldCleanup }) => {
    const db = getDb();

    // Update document status to processing and get document configuration
    const document = await db.document.update({
      where: { id: documentId },
      data: {
        status: DocumentStatus.PROCESSING,
        processingAt: new Date(),
      },

    def reset(self):
        self.environment.reset()

    def save_result(self, plan: str, result: str, spend: float):
        """Save the result to the result file"""
        result_file_path = "./results/" + self.task + ".txt"
        os.makedirs(os.path.dirname(result_file_path), exist_ok=True)
        with open(result_file_path, "w") as f:
            f.write("[Final Plan]\n" + plan + "\n\n")
            f.write("[Result]\n" + result)
            f.write(f"[Spent]\n${spend}")

}

impl LocalLangModel {
    pub async fn try_new(
        model: impl Into<String>,
        config: Option<LocalLangModelConfig>,
    ) -> anyhow::Result<Self> {
        let mut strm = Self::try_new_stream(model, config);
        while let Some(v) = strm.next().await {
            if let Some(result) = v?.result {
                return Ok(result);
            }
        }

from workflow.infra.providers.llm.iflytek_spark.schemas import StreamOutputMsg


class WorkflowEngineCtx(BaseModel):
    """
    Workflow engine execution context.

    Contains all necessary state and configuration for workflow execution,
    including variable pool, node status, dependencies, and execution chains.
    """

    # Variable pool for storing and passing variables during execution
    variable_pool: VariablePool

        """Clear the context."""
        self._messages = []

    async def save_state(self) -> Mapping[str, Any]:
        return ChatCompletionContextState(messages=self._messages).model_dump()

    async def load_state(self, state: Mapping[str, Any]) -> None:
        self._messages = ChatCompletionContextState.model_validate(state).messages


class ChatCompletionContextState(BaseModel):
    messages: List[LLMMessage] = Field(default_factory=list)

 *
 * @see {@link https://github.com/botpress/botpress/blob/master/packages/llmz/examples/14_worker_snapshot/index.ts} Example usage
 */
export class Snapshot implements Serializable<Snapshot.JSON> {
  public readonly id: string
  public readonly reason?: string
  public readonly stack: string
  public readonly toolCall?: ToolCall
  public variables: Variable[]
  #status: SnapshotStatus

  /**
   * Gets the current status of the snapshot.

        if self.logger.console:
            self.logger.console.print_task(prompt)

        # Session restoration: if session passed, restore it (stateless API continuation)
        if session is not None:
            self.current_session = {
                'session_id': session.get('session_id'),
                'messages': list(session.get('messages', [])),
                'trace': list(session.get('trace', [])),
                'turn': session.get('turn', 0)
            }
            # Start YAML session logging with session_id for thread safety
            self.logger.start_session(self.system_prompt, session_id=session.get('session_id'))

- **Workflow Engine**: Emits job steps as `JobRequest` and advances runs.
- **Workers**: Subscribe to job subjects, read context pointers, execute, emit
  `JobResult` with result/artifact pointers, and send heartbeats.
- **Redis**: Stores workflow state, job metadata, and pointer payloads.
- **NATS**: Transport for CAP v2 packets (with optional JetStream durability).

## 3) CAP v2 Wire Contracts (selected fields)

CAP v2 is the canonical protocol; Cordum does not duplicate these definitions.

### 3.1 BusPacket (Envelope)

`BusPacket` carries all bus traffic:

    /// assert_eq!(uri.dimension, cortex_mem_core::Dimension::Session);
    /// assert_eq!(uri.category, "abc123");
    /// ```
    pub fn parse(uri: &str) -> Result<CortexUri> {
        // 1. Validate scheme
        if !uri.starts_with("cortex://") {
            return Err(Error::InvalidScheme);
        }

        // 2. Split path and query
        let uri_without_scheme = &uri[9..]; // Skip "cortex://"
        let (path_part, query_part) = uri_without_scheme
            .split_once('?')

    return value


class FlowState(BaseModel):
    """Base model for all flow states, ensuring each state has a unique ID."""

    id: str = Field(
        default_factory=lambda: str(uuid4()),
        description="Unique identifier for the flow state",
    )


T = TypeVar("T", bound=dict[str, Any] | BaseModel)

   * await sandbox.pause()
   * ```
   */
  async pause(opts?: ConnectionOpts): Promise<boolean> {
    return await SandboxApi.pause(this.sandboxId, this.resolveApiOpts(opts))
  }

  /**
   * @deprecated Use {@link Sandbox.pause} instead.
   */
  async betaPause(opts?: ConnectionOpts): Promise<boolean> {
    return await SandboxApi.betaPause(this.sandboxId, this.resolveApiOpts(opts))
  }

        return res

    def to_dict(self) -> dict[str, Any]:
        """
        Serializes the pipeline to a dictionary.

        This is meant to be an intermediate representation but it can be also used to save a pipeline to file.

        :returns:
            Dictionary with serialized data.
        """
        components = {}

pub use instances::InstanceStatus;

/// Schema version - bump on any schema change.
const SCHEMA_VERSION: i32 = 17;
pub const DEV_ROOT_KV_KEY: &str = "config:dev_root";
const MIGRATIONS: &[(i32, &str)] = &[(
    17,
    "ALTER TABLE instances ADD COLUMN terminal_preset_requested TEXT DEFAULT '';
     ALTER TABLE instances ADD COLUMN terminal_preset_effective TEXT DEFAULT '';
     UPDATE instances
     SET terminal_preset_effective = json_extract(launch_context, '$.terminal_preset')
     WHERE launch_context != '' AND json_valid(launch_context) AND json_extract(launch_context, '$.terminal_preset') IS NOT NULL;",
)];

"""

FTS_SQL = """
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
    content
);

CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages BEGIN
    INSERT INTO messages_fts(rowid, content) VALUES (
        new.id,
        COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
    );
END;

                    )

        # Determine entry point (may differ if resuming)
        # Check if resuming from checkpoint
        if session_state and session_state.get("resume_from_checkpoint") and checkpoint_store:
            checkpoint_id = session_state["resume_from_checkpoint"]
            try:
                checkpoint = await checkpoint_store.load_checkpoint(checkpoint_id)

                if checkpoint:
                    self.logger.info(f"🔄 Resuming from checkpoint: {checkpoint_id} (node: {checkpoint.current_node})")
                    checkpoint_run_id = checkpoint.run_id or LEGACY_RUN_ID
                    self._run_id = checkpoint_run_id

            super().__setattr__('_agents', _agents)
        super().__setattr__(__name, __value)

    def state_dict(self, session_id=None, prefix='', destination=None) -> Dict:
        if destination is None:
            destination = {}
        if self.memory is not None:
            if session_id not in self.memory.memory_map:
                warnings.warn(f'No session id {session_id} in {prefix}memory')
            memory = self.memory.get(session_id)
            saved_memory = memory and memory.save() or []
            destination.update({prefix + 'memory': saved_memory})
        for name, agent in getattr(self, '_agents', {}).items():

        ]
        awrap_model_call_handler = _chain_async_model_call_handlers(async_handlers)

    base_state = state_schema if state_schema is not None else AgentState
    # Build an ordered list: middleware schemas first (in registration order),
    # base_state last so it wins any field conflict.  This lets the caller's
    # explicit state_schema override middleware annotations — e.g. passing
    # a DeltaChannel-annotated schema wins over BinaryOperatorAggregate from
    # AgentState without requiring a post-compilation patch.
    state_schemas: list[type] = [*(m.state_schema for m in middleware), base_state]

    resolved_state_schema, input_schema, output_schema = _resolve_schemas(state_schemas)

                instructions="",
                sampling_params={"strategy": {"type": "greedy"}},
                toolgroups=toolgroups,
                enable_session_persistence=True,
            )
        elif agent_type == AgentChoice.Memory:
            vector_db_ids = agent_params.get("vector_db_ids", [])
            toolgroups = [
                {
                    "name": "builtin::rag",
                    "args": {
                        "vector_db_ids": vector_db_ids,
                        "query_config": QueryConfig(

        return self._ensure_tools_are_async(cast(List[BaseTool], tools))

    async def _init_context(self, ctx: Context, ev: AgentWorkflowStartEvent) -> None:
        """Initialize the context once, if needed."""
        if not await ctx.store.get("memory", default=None):
            default_memory = ev.get("memory", default=None)
            default_memory = default_memory or ChatMemoryBuffer.from_defaults(
                llm=self.llm or Settings.llm
            )
            await ctx.store.set("memory", default_memory)
        if not await ctx.store.get("state", default=None):
            await ctx.store.set("state", copy.deepcopy(self.initial_state))

        if should_save:
            await self._save_checkpoint(iteration_id, current_count)

    async def _save_checkpoint(self, iteration_id: int, completion_count: int) -> None:
        """
        Saves a checkpoint with the specific naming convention.

        Args:
            iteration_id: The iteration ID of the task that just finished.
            completion_count: The total number of completed tasks so far.
        """
        # Requirement 1: Directory format: checkpoint-iter-{iteration_id}-{completion_count}
        dir_name = f"checkpoint-iter-{iteration_id}-{completion_count}"

console.print(f"\n[bold]Refined Final output:[/bold]\n{refined_output}")

with open(filename, 'w') as file:
    file.write(exchange_log)
print(f"\nFull exchange log saved to {filename}")

  SuspendOptions,
} from './step';
import { forwardAgentStreamChunk } from './stream-utils';
import type {
  DefaultEngineType,
  DynamicMapping,
  ExtractSchemaFromStep,
  ExtractSchemaType,
  PathsToStringProps,
  SerializedStep,
  SerializedStepFlowEntry,
  StepFlowEntry,
  StepResult,

        if "env_desc" in data:
            self.env.desc = data["env_desc"]

    def serialize(self, stg_path: Path = None):
        stg_path = SERDESER_PATH.joinpath("team") if stg_path is None else stg_path
        team_info_path = stg_path.joinpath("team.json")
        serialized_data = self.model_dump()
        serialized_data["context"] = self.env.context.serialize()

        write_json_file(team_info_path, serialized_data)

    @classmethod
    def deserialize(cls, stg_path: Path, context: Context = None) -> "Team":

var ErrAgentNotFound = errors.New("agent not found")

type AgentState struct {
	Gid       uint64 `json:"gid,omitempty"`
	Id        string `json:"id" db:"constraint=unique"`
	Name      string `json:"name"`
	Status    string `json:"status"`
	Data      string `json:"data,omitempty"`
	UpdatedAt string `json:"updated"`
}

func WriteAgentState(ctx context.Context, state AgentState) error {

        if not ephemeral and self._extra_hooks:
            hook = CompositeHook([loop_hook] + self._extra_hooks)

        async def _checkpoint(payload: dict[str, Any]) -> None:
            if session is None:
                return
            self._set_runtime_checkpoint(session, payload)

        async def _drain_pending(*, limit: int = _MAX_INJECTIONS_PER_TURN) -> list[dict[str, Any]]:
            """Drain follow-up messages from the pending queue.

            When no messages are immediately available but sub-agents
            spawned in this dispatch are still running, blocks until at

   * version that records task descriptions. Executed run results are rejected
   * because their task records are not a replay contract.
   */
  createPlanArtifact(result: TeamRunResult): PlanArtifact {
    if (result.planOnly !== true || !result.tasks) {
      throw new Error('createPlanArtifact requires a plan-only TeamRunResult.')
    }

    return {
      version: 1,
      ...(result.goal !== undefined ? { goal: result.goal } : {}),
      tasks: result.tasks.map((task): PlanTaskArtifact => {
        if (!task.description) {

## How cron works

- Cron runs **inside the Gateway** process (not inside the model).
- Job definitions, runtime state, and run history persist in OpenClaw's shared SQLite state database so restarts do not lose schedules.
- On upgrade, run `openclaw doctor --fix` to import legacy `~/.openclaw/cron/jobs.json`, `jobs-state.json`, and `runs/*.jsonl` files into SQLite and rename them with a `.migrated` suffix. Malformed job rows are skipped from runtime and copied to `jobs-quarantine.json` for later repair or review.
- `cron.store` still names the logical cron store key and doctor import path. After import, editing that JSON file no longer changes active cron jobs; use `openclaw cron add|edit|remove` or the Gateway cron RPC methods instead.
- All cron executions create [background task](/automation/tasks) records.
- On Gateway startup, overdue isolated agent-turn jobs are rescheduled out of the channel-connect window instead of replaying immediately, so Discord/Telegram startup and native-command setup stay responsive after restarts.
- One-shot jobs (`--at`) auto-delete after success by default.
- Isolated cron runs best-effort close tracked browser tabs/processes for their `cron:<jobId>` session when the run completes, so detached browser automation does not leave orphaned processes behind.
- Isolated cron runs that receive the narrow cron self-cleanup grant can still read scheduler status, a self-filtered list of their current job, and that job's run history, so status/heartbea
…

	registryTLS := flag.Bool("registry-tls", false, "use TLS for registry connection")
	registryFingerprint := flag.String("registry-fingerprint", "", "hex SHA-256 fingerprint of registry TLS certificate (required when -registry-trust=pinned)")
	registryTrust := flag.String("registry-trust", "pinned", "trust store for -registry-tls: 'pinned' (verify cert against -registry-fingerprint) or 'system' (OS x509 root store — used for compat-mode registry on registry.pilotprotocol.network:443 with Let's Encrypt)")
	identityPath := flag.String("identity", "", "path to persist Ed25519 identity (enables stable identity across restarts)")
	email := flag.String("email", "", "email address for account identification and key recovery")
	owner := flag.String("owner", "", "(deprecated: use -email) owner identifier for key rotation recovery")
	keepalive := flag.Duration("keepalive", 0, "keepalive probe interval (default 30s)")
	idleTimeout := flag.Duration("idle-timeout", 0, "idle connection timeout (default 120s)")
	synRate := flag.Int("syn-rate-limit", 0, "max SYN packets per second (default 100)")
	maxConnsPerPort := flag.Int("max-conns-per-port", 0, "max connections per port (default 1024)")
	maxConnsTotal := flag.Int("max-conns-total", 0, "max total connections (default 4096)")
	timeWait := flag.Duration("time-wait", 0, "TIME_WAIT duration (default 10s)")
	public := flag.Bool("public", false, "make this node's endpoint publicly visible (default: private)")

from pipecat.utils.base_object import BaseObject


class FrameSerializer(BaseObject):
    """Abstract base class for frame serialization implementations.

    Defines the interface for converting frames to/from serialized formats
    for transmission or storage. Subclasses must implement the core
    serialize/deserialize methods.
    """

    class InputParams(BaseModel):
        """Base configuration parameters for FrameSerializer.

from .agent import Agent


class Session:
    """
    A simple wrapper around PraisonAI's existing stateful capabilities.
    
    Provides a unified API for:
    - Session management with persistent state
    - Memory operations (short-term, long-term, user-specific)
    - Knowledge base operations
    - Agent state management
    - Remote agent connectivity

/// This class is used to manage the lifecycle of an agent thread.
/// The thread can be not-start, started or ended.
/// </remarks>
public abstract class AgentThread
{
    /// <summary>
    /// Gets the id of the current thread.
    /// </summary>
    public virtual string? Id { get; protected set; }

    /// <summary>
    /// Gets a value indicating whether the thread has been deleted.
    /// </summary>

logger = logging.getLogger(__name__)


class SessionManager(HookProvider, ABC):
    """Abstract interface for managing sessions.

    A session manager is in charge of persisting the conversation and state of an agent across its interaction.
    Changes made to the agents conversation, state, or other attributes should be persisted immediately after
    they are changed. The different methods introduced in this class are called at important lifecycle events
    for an agent, and should be persisted in the session.
    """

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:

// --- SQLite Database ---
const DB_PATH = IS_BUILD_BOOTSTRAP ? ':memory:' : path.join(DATA_DIR, 'swarmclaw.db')
const db = new Database(DB_PATH)
if (!IS_BUILD_BOOTSTRAP) {
  db.pragma('journal_mode = WAL')
  db.pragma('busy_timeout = 15000')
  db.pragma('synchronous = NORMAL')
  db.pragma('cache_size = -64000')
  db.pragma('mmap_size = 268435456')
}
db.pragma('foreign_keys = ON')

        """

        def autosave_loop():
            while self.autosave:
                try:
                    self.save()
                    if self.verbose:
                        logger.debug(
                            f"Autosaved agent state (interval: {interval}s)"
                        )
                except Exception as e:
                    logger.error(f"Autosave failed: {e}")
                time.sleep(interval)

                self._finalize_step_result(step_result, context)


class StorageConnectionStep(Step):
    """Setup storage connection for memory and database operations."""
    
    @property
    def name(self) -> str:
        return "storage_connection"
    
    @property
    def description(self) -> str:
        return "Setup storage connection"

    #
    # Serialization methods
    #
    def dumps(self) -> str:
        """Serialize the Agent instance to a JSON string."""
        return json.dumps(self.to_dict())

    @classmethod
    def loads(
        cls,
        data: str,
        agent_progress_callback: Optional[
            Callable[[AgentStatusType, dict, str], None]